From b354e792af7047635cad055322b0b3a1bf54944f Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Tue, 31 Mar 2026 18:26:24 +0800 Subject: [PATCH 1/5] feat(backend): add explain() interface with typed PostgresExplainResult MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement backend.explain(expression, options) for PostgresBackend and AsyncPostgresBackend, following the sync/async parity principle. Changes: - New postgres/explain/ submodule with PostgresExplainPlanLine (wraps the single "QUERY PLAN" column per row) and PostgresExplainResult with analyze_index_usage() / is_full_scan / is_index_used / is_covering_index helpers (text-matching on Seq Scan / Index Scan / Index Only Scan keywords) - PostgresBackend and AsyncPostgresBackend now mixin SyncExplainBackendMixin and AsyncExplainBackendMixin respectively - Override _parse_explain_result() to extract "QUERY PLAN" key from each raw row and construct typed PostgresExplainPlanLine objects - Add format_explain_statement() to PostgresDialect using PostgreSQL bracket-option syntax: EXPLAIN [(ANALYZE, FORMAT JSON)] stmt - Export PostgresExplainResult and PostgresExplainPlanLine from the package __init__ (postgres execute() already included EXPLAIN in its DQL detection tuple — no fix needed) - Add integration tests covering protocol check, result structure, full-scan / index detection, EXPLAIN ANALYZE, and async parity --- .../backend/impl/postgres/__init__.py | 4 + .../impl/postgres/backend/async_backend.py | 9 +- .../backend/impl/postgres/backend/sync.py | 9 +- .../backend/impl/postgres/dialect.py | 36 +++ .../backend/impl/postgres/explain/__init__.py | 10 + .../backend/impl/postgres/explain/types.py | 97 ++++++ .../feature/backend/test_backend_explain.py | 282 ++++++++++++++++++ 7 files changed, 445 insertions(+), 2 deletions(-) create mode 100644 src/rhosocial/activerecord/backend/impl/postgres/explain/__init__.py create mode 100644 src/rhosocial/activerecord/backend/impl/postgres/explain/types.py create mode 100644 tests/rhosocial/activerecord_postgres_test/feature/backend/test_backend_explain.py diff --git a/src/rhosocial/activerecord/backend/impl/postgres/__init__.py b/src/rhosocial/activerecord/backend/impl/postgres/__init__.py index 131e5a70..f53168c0 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/__init__.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/__init__.py @@ -22,6 +22,7 @@ from .dialect import PostgresDialect from .transaction import PostgresTransactionManager, AsyncPostgresTransactionManager from .types import PostgresEnumType +from .explain import PostgresExplainResult, PostgresExplainPlanLine from .statements import ( CreateEnumTypeExpression, DropEnumTypeExpression, @@ -187,6 +188,9 @@ # PostgreSQL-specific Type Helpers "PostgresEnumType", "PostgresEnumAdapter", + # PostgreSQL EXPLAIN Result Types + "PostgresExplainResult", + "PostgresExplainPlanLine", # PostgreSQL DDL Statements "CreateEnumTypeExpression", "DropEnumTypeExpression", diff --git a/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py b/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py index 1521d555..8ccc7ace 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py @@ -30,6 +30,7 @@ from rhosocial.activerecord.backend.base import AsyncStorageBackend from rhosocial.activerecord.backend.introspection.backend_mixin import IntrospectorBackendMixin +from rhosocial.activerecord.backend.explain import AsyncExplainBackendMixin from rhosocial.activerecord.backend.errors import ( ConnectionError, DatabaseError, @@ -60,7 +61,7 @@ from ..transaction import AsyncPostgresTransactionManager -class AsyncPostgresBackend(IntrospectorBackendMixin, PostgresBackendMixin, AsyncStorageBackend): +class AsyncPostgresBackend(AsyncExplainBackendMixin, IntrospectorBackendMixin, PostgresBackendMixin, AsyncStorageBackend): """Asynchronous PostgreSQL-specific backend implementation.""" def __init__(self, **kwargs): @@ -674,5 +675,11 @@ def log(self, level: int, message: str): # Fallback logging print(f"[{logging.getLevelName(level)}] {message}") + def _parse_explain_result(self, raw_rows, sql, duration): + """Return a typed :class:`PostgresExplainResult` for PostgreSQL EXPLAIN output.""" + from ..explain import PostgresExplainResult, PostgresExplainPlanLine + rows = [PostgresExplainPlanLine(line=r.get("QUERY PLAN", "")) for r in raw_rows] + return PostgresExplainResult(raw_rows=raw_rows, sql=sql, duration=duration, rows=rows) + __all__ = ["AsyncPostgresBackend"] diff --git a/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py b/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py index 43cb2216..e7e28650 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py @@ -29,6 +29,7 @@ from rhosocial.activerecord.backend.base import StorageBackend from rhosocial.activerecord.backend.introspection.backend_mixin import IntrospectorBackendMixin +from rhosocial.activerecord.backend.explain import SyncExplainBackendMixin from rhosocial.activerecord.backend.errors import ( ConnectionError, DatabaseError, @@ -60,7 +61,7 @@ from ..introspection import SyncPostgreSQLIntrospector -class PostgresBackend(IntrospectorBackendMixin, PostgresBackendMixin, StorageBackend): +class PostgresBackend(SyncExplainBackendMixin, IntrospectorBackendMixin, PostgresBackendMixin, StorageBackend): """PostgreSQL-specific backend implementation.""" def __init__(self, **kwargs): @@ -675,5 +676,11 @@ def log(self, level: int, message: str): # Fallback logging print(f"[{logging.getLevelName(level)}] {message}") + def _parse_explain_result(self, raw_rows, sql, duration): + """Return a typed :class:`PostgresExplainResult` for PostgreSQL EXPLAIN output.""" + from ..explain import PostgresExplainResult, PostgresExplainPlanLine + rows = [PostgresExplainPlanLine(line=r.get("QUERY PLAN", "")) for r in raw_rows] + return PostgresExplainResult(raw_rows=raw_rows, sql=sql, duration=duration, rows=rows) + __all__ = ["PostgresBackend"] diff --git a/src/rhosocial/activerecord/backend/impl/postgres/dialect.py b/src/rhosocial/activerecord/backend/impl/postgres/dialect.py index d7287b6d..47e417f1 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/dialect.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/dialect.py @@ -147,6 +147,7 @@ CreateMaterializedViewExpression, DropMaterializedViewExpression, RefreshMaterializedViewExpression, + ExplainExpression, ) @@ -396,6 +397,41 @@ def supports_explain_format(self, format_type: str) -> bool: supported_formats = ["TEXT", "XML", "JSON", "YAML"] return format_type_upper in supported_formats + def format_explain_statement(self, explain_expr: "ExplainExpression") -> str: + """Build the PostgreSQL EXPLAIN prefix using bracket-option syntax. + + PostgreSQL syntax: ``EXPLAIN [ ( option [, ...] ) ] statement`` + + Supported options assembled here: + - ``ANALYZE`` + - ``FORMAT { TEXT | XML | JSON | YAML }`` + - ``VERBOSE`` (passed through ``ExplainOptions.verbose`` if present) + + The prefix is returned without a trailing newline; the caller appends + the inner SQL separated by a single space. + """ + from rhosocial.activerecord.backend.expression.statements import ExplainType + + options = explain_expr.options + if options is None: + return "EXPLAIN" + + opts: list = [] + + if options.analyze: + opts.append("ANALYZE") + + if options.format is not None: + fmt_name = options.format.name if hasattr(options.format, "name") else str(options.format) + opts.append(f"FORMAT {fmt_name.upper()}") + elif options.type is not None and options.type == ExplainType.QUERY_PLAN: + # PostgreSQL has no QUERY PLAN keyword; plain EXPLAIN is equivalent + pass + + if opts: + return "EXPLAIN (" + ", ".join(opts) + ")" + return "EXPLAIN" + def supports_graph_match(self) -> bool: """Whether graph query MATCH clause is supported.""" # PostgreSQL doesn't have native MATCH clause like some other systems diff --git a/src/rhosocial/activerecord/backend/impl/postgres/explain/__init__.py b/src/rhosocial/activerecord/backend/impl/postgres/explain/__init__.py new file mode 100644 index 00000000..15ba783c --- /dev/null +++ b/src/rhosocial/activerecord/backend/impl/postgres/explain/__init__.py @@ -0,0 +1,10 @@ +# src/rhosocial/activerecord/backend/impl/postgres/explain/__init__.py +"""PostgreSQL-specific EXPLAIN result types.""" + +from .types import IndexUsage, PostgresExplainPlanLine, PostgresExplainResult + +__all__ = [ + "IndexUsage", + "PostgresExplainPlanLine", + "PostgresExplainResult", +] diff --git a/src/rhosocial/activerecord/backend/impl/postgres/explain/types.py b/src/rhosocial/activerecord/backend/impl/postgres/explain/types.py new file mode 100644 index 00000000..afd2edda --- /dev/null +++ b/src/rhosocial/activerecord/backend/impl/postgres/explain/types.py @@ -0,0 +1,97 @@ +# src/rhosocial/activerecord/backend/impl/postgres/explain/types.py +""" +PostgreSQL-specific EXPLAIN result types. + +PostgreSQL's EXPLAIN output (default TEXT format) produces one row per +plan line, each row having a single column named ``QUERY PLAN``. The +text in that column uses indentation and keywords like ``Seq Scan``, +``Index Scan``, and ``Index Only Scan`` to describe the query plan. +""" +from typing import List + +from pydantic import BaseModel + +from rhosocial.activerecord.backend.explain.types import BaseExplainResult + +try: + from typing import Literal +except ImportError: # Python 3.8 + from typing_extensions import Literal # type: ignore[assignment] + +# --------------------------------------------------------------------------- +# Index-usage label (mirrors SQLite / MySQL for a consistent API) +# --------------------------------------------------------------------------- +IndexUsage = Literal["full_scan", "index_with_lookup", "covering_index", "unknown"] + + +class PostgresExplainPlanLine(BaseModel): + """One line from PostgreSQL's plain-text EXPLAIN output. + + The column returned by psycopg is named ``"QUERY PLAN"`` (case-sensitive). + The field is aliased as ``line`` for convenience. + """ + + line: str + + model_config = {"populate_by_name": True} + + +class PostgresExplainResult(BaseExplainResult): + """PostgreSQL EXPLAIN result (default TEXT format). + + Contains index-usage analysis helpers that mirror the SQLite and MySQL + equivalents for a consistent cross-backend API. + + Analysis is performed via text-matching on the plan lines, looking for + PostgreSQL-specific keywords: + - ``Seq Scan`` → full table scan + - ``Index Scan`` → index lookup with table access + - ``Index Only Scan`` → covering index (no heap access) + - ``Bitmap`` patterns → treated as index_with_lookup + """ + + rows: List[PostgresExplainPlanLine] = [] + + # ------------------------------------------------------------------ + # Index-usage analysis + # ------------------------------------------------------------------ + + def analyze_index_usage(self) -> IndexUsage: + """Analyse index usage from the EXPLAIN output lines. + + Returns one of: ``"full_scan"``, ``"index_with_lookup"``, + ``"covering_index"``, or ``"unknown"``. + """ + if not self.rows: + return "unknown" + + combined = " ".join(r.line for r in self.rows).upper() + + # Covering index is the most specific — check first + if "INDEX ONLY SCAN" in combined: + return "covering_index" + + # Any index scan (with heap access) + if "INDEX SCAN" in combined or "BITMAP INDEX SCAN" in combined or "BITMAP HEAP SCAN" in combined: + return "index_with_lookup" + + # Full sequential scan + if "SEQ SCAN" in combined: + return "full_scan" + + return "unknown" + + @property + def is_full_scan(self) -> bool: + """``True`` when no index is used (sequential scan).""" + return self.analyze_index_usage() == "full_scan" + + @property + def is_index_used(self) -> bool: + """``True`` when any index is used.""" + return self.analyze_index_usage() in ("index_with_lookup", "covering_index") + + @property + def is_covering_index(self) -> bool: + """``True`` when an index-only scan eliminates heap access.""" + return self.analyze_index_usage() == "covering_index" diff --git a/tests/rhosocial/activerecord_postgres_test/feature/backend/test_backend_explain.py b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_backend_explain.py new file mode 100644 index 00000000..02f47f2e --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_backend_explain.py @@ -0,0 +1,282 @@ +# tests/rhosocial/activerecord_postgres_test/feature/backend/test_backend_explain.py +""" +Integration tests for PostgresBackend.explain() and AsyncPostgresBackend.explain(). + +These tests require a real PostgreSQL connection configured via postgres_scenarios.yaml. +The tests create temporary tables, run EXPLAIN, and verify the typed result objects. +""" +import pytest +import pytest_asyncio + +from rhosocial.activerecord.backend.explain import ( + SyncExplainBackendProtocol, + AsyncExplainBackendProtocol, +) +from rhosocial.activerecord.backend.expression import RawSQLExpression +from rhosocial.activerecord.backend.expression.statements import ExplainOptions +from rhosocial.activerecord.backend.impl.postgres import ( + PostgresExplainResult, + PostgresExplainPlanLine, +) + + +# --------------------------------------------------------------------------- +# Schema helpers (PostgreSQL syntax) +# --------------------------------------------------------------------------- + +_SETUP_SQL = """ + DROP TABLE IF EXISTS explain_order_items CASCADE; + DROP TABLE IF EXISTS explain_orders CASCADE; + + CREATE TABLE explain_orders ( + id SERIAL PRIMARY KEY, + status VARCHAR(20) NOT NULL, + amount NUMERIC(10,2) + ); + CREATE INDEX idx_orders_status ON explain_orders(status); + + CREATE TABLE explain_order_items ( + id SERIAL PRIMARY KEY, + order_id INTEGER NOT NULL, + sku VARCHAR(50) NOT NULL, + qty INTEGER NOT NULL DEFAULT 1 + ); + CREATE INDEX idx_items_order_id_sku ON explain_order_items(order_id, sku); + + INSERT INTO explain_orders (status, amount) VALUES + ('pending', 10.00), ('pending', 20.00), + ('shipped', 30.00), ('delivered', 40.00); + + INSERT INTO explain_order_items (order_id, sku, qty) VALUES + (1, 'A001', 1), (1, 'A002', 2), + (2, 'B001', 1), (3, 'A001', 3); +""" + +_CLEANUP_SQL = """ + DROP TABLE IF EXISTS explain_order_items CASCADE; + DROP TABLE IF EXISTS explain_orders CASCADE; +""" + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture(scope="function") +def indexed_backend(postgres_backend): + """Sync backend with test tables and indexes.""" + postgres_backend.executescript(_SETUP_SQL) + yield postgres_backend + try: + postgres_backend.executescript(_CLEANUP_SQL) + except Exception: + pass + + +@pytest_asyncio.fixture(scope="function") +async def async_indexed_backend(async_postgres_backend): + """Async backend with test tables and indexes.""" + await async_postgres_backend.executescript(_SETUP_SQL) + yield async_postgres_backend + try: + await async_postgres_backend.executescript(_CLEANUP_SQL) + except Exception: + pass + + +# --------------------------------------------------------------------------- +# Protocol checking +# --------------------------------------------------------------------------- + +class TestExplainProtocol: + def test_sync_backend_implements_protocol(self, postgres_backend): + assert isinstance(postgres_backend, SyncExplainBackendProtocol) + + @pytest.mark.asyncio + async def test_async_backend_implements_protocol(self, async_postgres_backend): + assert isinstance(async_postgres_backend, AsyncExplainBackendProtocol) + + +# --------------------------------------------------------------------------- +# Sync explain – basic structure +# --------------------------------------------------------------------------- + +class TestSyncExplainBasic: + def test_explain_returns_postgres_explain_result(self, indexed_backend): + dialect = indexed_backend.dialect + expr = RawSQLExpression(dialect, "SELECT * FROM explain_orders") + result = indexed_backend.explain(expr) + assert isinstance(result, PostgresExplainResult) + + def test_result_has_rows(self, indexed_backend): + dialect = indexed_backend.dialect + expr = RawSQLExpression(dialect, "SELECT * FROM explain_orders") + result = indexed_backend.explain(expr) + assert len(result.rows) > 0 + + def test_result_row_type(self, indexed_backend): + dialect = indexed_backend.dialect + expr = RawSQLExpression(dialect, "SELECT * FROM explain_orders") + result = indexed_backend.explain(expr) + for row in result.rows: + assert isinstance(row, PostgresExplainPlanLine) + assert isinstance(row.line, str) + assert len(row.line) > 0 + + def test_result_sql_starts_with_explain(self, indexed_backend): + dialect = indexed_backend.dialect + expr = RawSQLExpression(dialect, "SELECT * FROM explain_orders") + result = indexed_backend.explain(expr) + assert result.sql.upper().startswith("EXPLAIN") + assert "explain_orders" in result.sql.lower() + + def test_result_has_duration(self, indexed_backend): + dialect = indexed_backend.dialect + expr = RawSQLExpression(dialect, "SELECT * FROM explain_orders") + result = indexed_backend.explain(expr) + assert result.duration >= 0.0 + + def test_result_has_raw_rows(self, indexed_backend): + dialect = indexed_backend.dialect + expr = RawSQLExpression(dialect, "SELECT * FROM explain_orders") + result = indexed_backend.explain(expr) + assert isinstance(result.raw_rows, list) + assert len(result.raw_rows) == len(result.rows) + # Each raw row must have the 'QUERY PLAN' key + for raw in result.raw_rows: + assert "QUERY PLAN" in raw + + +# --------------------------------------------------------------------------- +# Sync explain – index usage analysis +# --------------------------------------------------------------------------- + +class TestSyncExplainIndexAnalysis: + def test_full_scan_detection(self, indexed_backend): + """SELECT * FROM table without WHERE → Seq Scan → full_scan.""" + dialect = indexed_backend.dialect + # ANALYZE forces actual execution so the planner won't cheat + result = indexed_backend.explain( + RawSQLExpression(dialect, "SELECT * FROM explain_orders") + ) + assert result.analyze_index_usage() == "full_scan" + assert result.is_full_scan is True + assert result.is_index_used is False + assert result.is_covering_index is False + + def test_index_used_detection(self, indexed_backend): + """SELECT * … WHERE indexed_col = ? → Index Scan (with heap access).""" + dialect = indexed_backend.dialect + result = indexed_backend.explain( + RawSQLExpression(dialect, "SELECT * FROM explain_orders WHERE status = 'pending'") + ) + usage = result.analyze_index_usage() + # Small table may still use Seq Scan; if it uses an index, verify the flag + if usage != "full_scan": + assert result.is_index_used is True + assert result.is_full_scan is False + + def test_covering_index_detection(self, indexed_backend): + """SELECT indexed_cols WHERE indexed_col = ? → Index Only Scan.""" + dialect = indexed_backend.dialect + result = indexed_backend.explain( + RawSQLExpression( + dialect, + "SELECT order_id, sku FROM explain_order_items WHERE order_id = 1" + ) + ) + usage = result.analyze_index_usage() + # PostgreSQL may choose Index Only Scan for covered queries + assert usage in ("covering_index", "index_with_lookup", "full_scan") + if usage == "covering_index": + assert result.is_covering_index is True + + def test_plan_lines_contain_text(self, indexed_backend): + """Plan lines should contain meaningful plan text.""" + dialect = indexed_backend.dialect + result = indexed_backend.explain( + RawSQLExpression(dialect, "SELECT * FROM explain_orders") + ) + combined = " ".join(r.line for r in result.rows) + # Should mention the table name or a scan type + assert any(kw in combined.upper() for kw in ("SCAN", "EXPLAIN_ORDERS", "SEQ")) + + +# --------------------------------------------------------------------------- +# Sync explain – ANALYZE option +# --------------------------------------------------------------------------- + +class TestSyncExplainAnalyze: + def test_explain_analyze_returns_result(self, indexed_backend): + """EXPLAIN (ANALYZE) executes the query and returns runtime stats.""" + dialect = indexed_backend.dialect + if not dialect.supports_explain_analyze(): + pytest.skip("Dialect does not support EXPLAIN ANALYZE") + opts = ExplainOptions(analyze=True) + result = indexed_backend.explain( + RawSQLExpression(dialect, "SELECT * FROM explain_orders"), + opts, + ) + assert isinstance(result, PostgresExplainResult) + assert len(result.rows) > 0 + assert "ANALYZE" in result.sql.upper() + # ANALYZE output typically includes "actual time" + combined = " ".join(r.line for r in result.rows).lower() + assert "actual" in combined or "time" in combined + + def test_explain_analyze_sql_prefix(self, indexed_backend): + """Verify the generated SQL uses bracket syntax: EXPLAIN (ANALYZE).""" + dialect = indexed_backend.dialect + if not dialect.supports_explain_analyze(): + pytest.skip("Dialect does not support EXPLAIN ANALYZE") + opts = ExplainOptions(analyze=True) + result = indexed_backend.explain( + RawSQLExpression(dialect, "SELECT 1"), + opts, + ) + assert result.sql.upper().startswith("EXPLAIN (ANALYZE)") + + +# --------------------------------------------------------------------------- +# Async explain – mirror of sync tests +# --------------------------------------------------------------------------- + +class TestAsyncExplainBasic: + @pytest.mark.asyncio + async def test_explain_returns_postgres_explain_result(self, async_indexed_backend): + dialect = async_indexed_backend.dialect + expr = RawSQLExpression(dialect, "SELECT * FROM explain_orders") + result = await async_indexed_backend.explain(expr) + assert isinstance(result, PostgresExplainResult) + + @pytest.mark.asyncio + async def test_result_has_rows(self, async_indexed_backend): + dialect = async_indexed_backend.dialect + expr = RawSQLExpression(dialect, "SELECT * FROM explain_orders") + result = await async_indexed_backend.explain(expr) + assert len(result.rows) > 0 + + @pytest.mark.asyncio + async def test_result_sql_and_duration(self, async_indexed_backend): + dialect = async_indexed_backend.dialect + expr = RawSQLExpression(dialect, "SELECT * FROM explain_orders") + result = await async_indexed_backend.explain(expr) + assert result.sql.upper().startswith("EXPLAIN") + assert result.duration >= 0.0 + + @pytest.mark.asyncio + async def test_full_scan_detection(self, async_indexed_backend): + dialect = async_indexed_backend.dialect + result = await async_indexed_backend.explain( + RawSQLExpression(dialect, "SELECT * FROM explain_orders") + ) + assert result.is_full_scan is True + + @pytest.mark.asyncio + async def test_raw_rows_have_query_plan_key(self, async_indexed_backend): + dialect = async_indexed_backend.dialect + result = await async_indexed_backend.explain( + RawSQLExpression(dialect, "SELECT * FROM explain_orders") + ) + for raw in result.raw_rows: + assert "QUERY PLAN" in raw From e3ea70c01f845c079d1f9cb49c5bc95ca70c1357 Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Tue, 31 Mar 2026 23:03:44 +0800 Subject: [PATCH 2/5] fix(dialect): return complete (sql, params) tuple from format_explain_statement The overridden format_explain_statement() in PostgresDialect was returning a plain str (the EXPLAIN prefix only), while ExplainExpression.to_sql() passes the return value directly to backend_mixin which unpacks it as (sql, params). This caused a ValueError at runtime. Fix by building the full EXPLAIN SQL string inside the method and returning (sql, params) consistent with the base class contract. Also fix test_backend_explain.py: replace executescript() calls (a MySQL-only method) with per-statement execute() calls, matching the pattern used in all other PostgreSQL backend tests. Add EXPLAIN documentation (en_US and zh_CN) and update the feature index README files. --- .../postgres_specific_features/README.md | 1 + .../postgres_specific_features/explain.md | 294 ++++++++++++++++++ .../postgres_specific_features/README.md | 1 + .../postgres_specific_features/explain.md | 294 ++++++++++++++++++ .../backend/impl/postgres/dialect.py | 14 +- .../feature/backend/test_backend_explain.py | 52 ++-- 6 files changed, 622 insertions(+), 34 deletions(-) create mode 100644 docs/en_US/postgres_specific_features/explain.md create mode 100644 docs/zh_CN/postgres_specific_features/explain.md diff --git a/docs/en_US/postgres_specific_features/README.md b/docs/en_US/postgres_specific_features/README.md index 068300cf..57232011 100644 --- a/docs/en_US/postgres_specific_features/README.md +++ b/docs/en_US/postgres_specific_features/README.md @@ -10,6 +10,7 @@ PostgreSQL offers many advanced features beyond standard SQL. This section cover - **[RETURNING Clause](./returning.md)**: Get data back from DML operations - **[Protocol Support Matrix](./protocol_support.md)**: Comprehensive protocol support and version compatibility - **[Database Introspection](./introspection.md)**: Query metadata using pg_catalog +- **[EXPLAIN Support](./explain.md)**: Query execution plan analysis and performance diagnostics ## Feature Highlights diff --git a/docs/en_US/postgres_specific_features/explain.md b/docs/en_US/postgres_specific_features/explain.md new file mode 100644 index 00000000..2f633ea8 --- /dev/null +++ b/docs/en_US/postgres_specific_features/explain.md @@ -0,0 +1,294 @@ +# EXPLAIN Support + +## Overview + +The PostgreSQL backend provides comprehensive support for the `EXPLAIN` statement, enabling query performance analysis and execution plan inspection. This feature integrates with the ActiveRecord query builder through a unified API while exposing PostgreSQL-specific behaviors. + +Key capabilities: +- Basic `EXPLAIN` for execution plan visualization +- `EXPLAIN ANALYZE` for actual runtime statistics (all modern versions) +- Multiple output formats: `TEXT`, `XML`, `JSON`, `YAML` +- PostgreSQL's bracketed option syntax (`EXPLAIN (ANALYZE, FORMAT JSON)`) +- Structured result parsing with `PostgresExplainResult` +- Index usage analysis helpers (`Seq Scan`, `Index Scan`, `Index Only Scan`) +- Full sync and async support + +## Basic Usage + +### Accessing EXPLAIN via Query Builder + +```python +from rhosocial.activerecord.backend.expression.statements import ExplainFormat + +# Simple EXPLAIN — returns PostgresExplainResult +result = User.query().explain().all() + +# Inspect the generated SQL prefix +print(result.sql) # EXPLAIN SELECT ... + +# Print the execution time (seconds) +print(result.duration) +``` + +### Reading the Result Rows + +```python +result = User.query().explain().all() + +for row in result.rows: + print(row.line) +``` + +Each row is a `PostgresExplainPlanLine` with a single field: + +| Field | Type | Description | +|---|---|---| +| `line` | `str` | One line of the query plan text (e.g. `"Seq Scan on users (cost=0.00..8.27 rows=1 width=...)"`). | + +PostgreSQL returns each plan node as an indented text line in the `"QUERY PLAN"` column. All lines together form the full execution plan tree. + +## Output Formats + +PostgreSQL supports four output formats, all available on any modern server version: + +### TEXT Format (Default) + +```python +# Default — no FORMAT option emitted +result = User.query().explain().all() +# Generates: EXPLAIN SELECT ... +``` + +### JSON Format + +```python +result = User.query().explain(format=ExplainFormat.JSON).all() +# Generates: EXPLAIN (FORMAT JSON) SELECT ... +``` + +### XML Format + +```python +result = User.query().explain(format=ExplainFormat.XML).all() +# Generates: EXPLAIN (FORMAT XML) SELECT ... +``` + +### YAML Format + +```python +result = User.query().explain(format=ExplainFormat.YAML).all() +# Generates: EXPLAIN (FORMAT YAML) SELECT ... +``` + +## EXPLAIN ANALYZE + +`EXPLAIN ANALYZE` executes the query and returns actual runtime statistics alongside cost estimates. Use it to compare estimated vs. actual row counts and timing. + +```python +# Basic ANALYZE +result = User.query().explain(analyze=True).all() +# Generates: EXPLAIN (ANALYZE) SELECT ... + +# ANALYZE with JSON format +result = User.query().explain(analyze=True, format=ExplainFormat.JSON).all() +# Generates: EXPLAIN (ANALYZE, FORMAT JSON) SELECT ... + +# ANALYZE with YAML format +result = User.query().explain(analyze=True, format=ExplainFormat.YAML).all() +# Generates: EXPLAIN (ANALYZE, FORMAT YAML) SELECT ... +``` + +**Important Notes:** +- `EXPLAIN ANALYZE` **executes the query** against the database. Avoid using it on write statements in production without understanding the side effects. +- Supported on all modern PostgreSQL versions (no minimum version constraint within the supported range). + +## SQL Syntax Generated + +PostgreSQL uses a parenthesized option syntax, distinct from MySQL's flat syntax: + +``` +EXPLAIN [ ( option [, ...] ) ] statement +``` + +Options are comma-separated inside parentheses: + +```python +# No options +# → EXPLAIN SELECT ... + +# analyze only +# → EXPLAIN (ANALYZE) SELECT ... + +# format only +# → EXPLAIN (FORMAT JSON) SELECT ... + +# both analyze and format +# → EXPLAIN (ANALYZE, FORMAT JSON) SELECT ... +``` + +The `ANALYZE` option always appears before `FORMAT` in the option list. + +## Index Usage Analysis + +`PostgresExplainResult` provides helpers for a quick index usage assessment based on text matching across all plan lines: + +```python +result = User.query().where(User.c.email == "alice@example.com").explain().all() + +usage = result.analyze_index_usage() +print(usage) # "full_scan" | "index_with_lookup" | "covering_index" | "unknown" + +# Convenience properties +if result.is_full_scan: + print("Warning: sequential scan detected") + +if result.is_covering_index: + print("Optimal: index only scan in use") + +if result.is_index_used: + print("An index is being used") +``` + +### Index Usage Rules + +The analysis concatenates all plan lines and performs case-insensitive text matching: + +| Matched keyword (case-insensitive) | `analyze_index_usage()` return | +|---|---| +| `"INDEX ONLY SCAN"` (checked first) | `"covering_index"` | +| `"INDEX SCAN"`, `"BITMAP INDEX SCAN"`, or `"BITMAP HEAP SCAN"` | `"index_with_lookup"` | +| `"SEQ SCAN"` | `"full_scan"` | +| None of the above | `"unknown"` | + +**Note:** `"INDEX ONLY SCAN"` is checked before `"INDEX SCAN"` to avoid misclassification. + +## Async API + +The async backend provides the same interface. Use `await` for the terminal method: + +```python +async def analyze_queries(): + # Simple async EXPLAIN + result = await User.query().explain().all_async() + + # ANALYZE with JSON format + result = await User.query().explain( + analyze=True, + format=ExplainFormat.JSON + ).all_async() + + for row in result.rows: + print(row.line) +``` + +## Checking Format Support at Runtime + +Use `dialect` methods to guard option availability: + +```python +dialect = backend.dialect + +# Check ANALYZE support (always True for PostgreSQL) +if dialect.supports_explain_analyze(): + result = User.query().explain(analyze=True).all() + +# Check a specific format (all four formats always supported) +if dialect.supports_explain_format("JSON"): + result = User.query().explain(format=ExplainFormat.JSON).all() +``` + +## Combining EXPLAIN with Complex Queries + +EXPLAIN works with the full query builder chain: + +```python +# JOIN query +result = ( + User.query() + .join(Order, User.c.id == Order.c.user_id) + .where(User.c.status == "active") + .explain(format=ExplainFormat.JSON) + .all() +) + +# GROUP BY / aggregate +result = ( + Order.query() + .group_by(Order.c.user_id) + .explain(analyze=True) + .count(Order.c.id) +) + +# Subquery +result = ( + User.query() + .where(User.c.id.in_(Order.query().select(Order.c.user_id))) + .explain() + .all() +) +``` + +## Comparison with MySQL EXPLAIN + +Both backends share the same query builder API but differ in SQL syntax and result structure: + +| Feature | PostgreSQL | MySQL | +|---|---|---| +| SQL syntax | `EXPLAIN (ANALYZE, FORMAT JSON)` | `EXPLAIN ANALYZE FORMAT=JSON` | +| Option delimiter | Comma-separated, parenthesized | Space-separated, no parentheses | +| Supported formats | TEXT, XML, JSON, YAML | TEXT, JSON (5.6.5+), TREE (8.0.16+) | +| ANALYZE support | All modern versions | 8.0.18+ | +| Result row type | `PostgresExplainPlanLine` (1 field) | `MySQLExplainRow` (12 fields) | +| Covering index indicator | `"Index Only Scan"` in plan text | `"Using index"` in `extra` column | + +## Version Compatibility + +| Feature | Minimum PostgreSQL Version | +|---|---| +| Basic `EXPLAIN` | All supported versions | +| `EXPLAIN FORMAT=TEXT/XML/JSON/YAML` | All supported versions | +| `EXPLAIN ANALYZE` | All supported versions | +| `EXPLAIN (ANALYZE, FORMAT JSON)` | All supported versions | + +The PostgreSQL backend targets PostgreSQL 9.6 and later; all EXPLAIN features described here are available on 9.6+. + +## API Reference + +### `ExplainOptions` Fields Used by PostgreSQL + +| Field | Type | Default | Description | +|---|---|---|---| +| `analyze` | `bool` | `False` | Emit `ANALYZE` option | +| `format` | `Optional[ExplainFormat]` | `None` | Output format (`TEXT`/`XML`/`JSON`/`YAML`) | + +Other `ExplainOptions` fields (such as `verbose`, `buffers`, `timing`, `settings`) are accepted but not yet emitted by the current PostgreSQL dialect implementation. + +### `PostgresExplainResult` Methods + +| Method / Property | Description | +|---|---| +| `rows` | `List[PostgresExplainPlanLine]` — parsed plan lines | +| `sql` | The full SQL string that was explained | +| `duration` | Query execution time in seconds | +| `raw_rows` | Raw rows as returned by psycopg (list of dicts with `"QUERY PLAN"` key) | +| `analyze_index_usage()` | Returns `"full_scan"`, `"index_with_lookup"`, `"covering_index"`, or `"unknown"` | +| `is_full_scan` | `True` if `analyze_index_usage() == "full_scan"` | +| `is_index_used` | `True` if any index is used | +| `is_covering_index` | `True` if an index-only scan is used | + +### `PostgresDialect` Methods + +| Method | Return | Description | +|---|---|---| +| `supports_explain_analyze()` | `bool` | Always `True` for PostgreSQL | +| `supports_explain_format(fmt)` | `bool` | `True` for `TEXT`, `XML`, `JSON`, `YAML` | +| `format_explain_statement(expr)` | `str` | Builds `EXPLAIN [(ANALYZE, FORMAT X)]` prefix | + +## Best Practices + +- **Use JSON format for programmatic analysis.** `FORMAT JSON` produces a structured plan tree that is easier to parse than plain text and provides the most detail. +- **Reserve `EXPLAIN ANALYZE` for development.** The statement executes the query, so use it only when you need actual row counts and timing figures. +- **Leverage all four formats for different audiences.** Use `TEXT` for human inspection, `JSON`/`YAML` for tooling, and `XML` for XML-based reporting pipelines. +- **Combine with index analysis.** Use `result.is_full_scan` as a quick smoke test in integration tests to detect inadvertent sequential scans. + +💡 *AI Prompt: Explain how to use EXPLAIN ANALYZE with JSON format on PostgreSQL to diagnose slow queries in rhosocial-activerecord.* diff --git a/docs/zh_CN/postgres_specific_features/README.md b/docs/zh_CN/postgres_specific_features/README.md index 510b0db8..8d7d73ab 100644 --- a/docs/zh_CN/postgres_specific_features/README.md +++ b/docs/zh_CN/postgres_specific_features/README.md @@ -10,6 +10,7 @@ PostgreSQL 提供了许多超越标准 SQL 的高级功能。本节介绍 Postgr - **[RETURNING 子句](./returning.md)**: 从 DML 操作获取返回数据 - **[协议支持矩阵](./protocol_support.md)**: 完整的协议支持和版本兼容性 - **[数据库内省](./introspection.md)**: 使用 pg_catalog 查询元数据 +- **[EXPLAIN 支持](./explain.md)**: 查询执行计划分析与性能诊断 ## 功能亮点 diff --git a/docs/zh_CN/postgres_specific_features/explain.md b/docs/zh_CN/postgres_specific_features/explain.md new file mode 100644 index 00000000..db1bde31 --- /dev/null +++ b/docs/zh_CN/postgres_specific_features/explain.md @@ -0,0 +1,294 @@ +# EXPLAIN 支持 + +## 概述 + +PostgreSQL 后端提供对 `EXPLAIN` 语句的完整支持，可用于分析查询性能并查看执行计划。该功能通过统一的 API 与 ActiveRecord 查询构建器集成，同时暴露 PostgreSQL 特有的行为。 + +主要能力： +- 基础 `EXPLAIN`，用于查看执行计划 +- `EXPLAIN ANALYZE`，返回实际运行时统计数据（所有现代版本均支持） +- 多种输出格式：`TEXT`、`XML`、`JSON`、`YAML` +- PostgreSQL 括号式选项语法（`EXPLAIN (ANALYZE, FORMAT JSON)`） +- 结构化结果解析，通过 `PostgresExplainResult` 访问 +- 索引使用分析辅助方法（`Seq Scan`、`Index Scan`、`Index Only Scan`） +- 同步与异步双模式支持 + +## 基本用法 + +### 通过查询构建器使用 EXPLAIN + +```python +from rhosocial.activerecord.backend.expression.statements import ExplainFormat + +# 简单 EXPLAIN — 返回 PostgresExplainResult +result = User.query().explain().all() + +# 查看生成的 SQL 前缀 +print(result.sql) # EXPLAIN SELECT ... + +# 打印执行耗时（秒） +print(result.duration) +``` + +### 读取结果行 + +```python +result = User.query().explain().all() + +for row in result.rows: + print(row.line) +``` + +每行是一个 `PostgresExplainPlanLine`，只有一个字段： + +| 字段 | 类型 | 说明 | +|---|---|---| +| `line` | `str` | 一行计划文本（如 `"Seq Scan on users (cost=0.00..8.27 rows=1 width=...)"`） | + +PostgreSQL 将每个计划节点以缩进文本行的形式写入 `"QUERY PLAN"` 列。所有行合在一起构成完整的执行计划树。 + +## 输出格式 + +PostgreSQL 支持四种输出格式，所有现代服务器版本均可使用： + +### TEXT 格式（默认） + +```python +# 默认情况下不生成 FORMAT 选项 +result = User.query().explain().all() +# 生成: EXPLAIN SELECT ... +``` + +### JSON 格式 + +```python +result = User.query().explain(format=ExplainFormat.JSON).all() +# 生成: EXPLAIN (FORMAT JSON) SELECT ... +``` + +### XML 格式 + +```python +result = User.query().explain(format=ExplainFormat.XML).all() +# 生成: EXPLAIN (FORMAT XML) SELECT ... +``` + +### YAML 格式 + +```python +result = User.query().explain(format=ExplainFormat.YAML).all() +# 生成: EXPLAIN (FORMAT YAML) SELECT ... +``` + +## EXPLAIN ANALYZE + +`EXPLAIN ANALYZE` 会实际执行查询，并在代价估算的基础上同时返回真实的运行时统计数据，可用于对比预估行数与实际行数及耗时。 + +```python +# 基础 ANALYZE +result = User.query().explain(analyze=True).all() +# 生成: EXPLAIN (ANALYZE) SELECT ... + +# 带 JSON 格式的 ANALYZE +result = User.query().explain(analyze=True, format=ExplainFormat.JSON).all() +# 生成: EXPLAIN (ANALYZE, FORMAT JSON) SELECT ... + +# 带 YAML 格式的 ANALYZE +result = User.query().explain(analyze=True, format=ExplainFormat.YAML).all() +# 生成: EXPLAIN (ANALYZE, FORMAT YAML) SELECT ... +``` + +**重要说明：** +- `EXPLAIN ANALYZE` **会真正执行查询**。在生产环境中对写操作使用时请谨慎评估副作用。 +- 所有现代 PostgreSQL 版本均支持，在支持的版本范围内无最低版本限制。 + +## 生成的 SQL 语法 + +PostgreSQL 使用括号式选项语法，与 MySQL 的平铺式语法不同： + +``` +EXPLAIN [ ( option [, ...] ) ] 语句 +``` + +多个选项在括号内以逗号分隔： + +```python +# 无选项 +# → EXPLAIN SELECT ... + +# 仅 analyze=True +# → EXPLAIN (ANALYZE) SELECT ... + +# 仅指定 format +# → EXPLAIN (FORMAT JSON) SELECT ... + +# 同时指定两者 +# → EXPLAIN (ANALYZE, FORMAT JSON) SELECT ... +``` + +`ANALYZE` 选项始终在选项列表中排在 `FORMAT` 之前。 + +## 索引使用分析 + +`PostgresExplainResult` 提供基于文本匹配的索引使用快速评估，通过对所有计划行进行关键词搜索： + +```python +result = User.query().where(User.c.email == "alice@example.com").explain().all() + +usage = result.analyze_index_usage() +print(usage) # "full_scan" | "index_with_lookup" | "covering_index" | "unknown" + +# 便捷属性 +if result.is_full_scan: + print("警告：检测到顺序扫描") + +if result.is_covering_index: + print("最优：正在使用 Index Only Scan") + +if result.is_index_used: + print("索引已被使用") +``` + +### 索引使用判断规则 + +分析将所有计划行拼接后进行大小写不敏感的文本匹配： + +| 匹配关键词（不区分大小写） | `analyze_index_usage()` 返回值 | +|---|---| +| `"INDEX ONLY SCAN"`（优先检查） | `"covering_index"` | +| `"INDEX SCAN"`、`"BITMAP INDEX SCAN"` 或 `"BITMAP HEAP SCAN"` | `"index_with_lookup"` | +| `"SEQ SCAN"` | `"full_scan"` | +| 以上均不匹配 | `"unknown"` | + +**注意：** `"INDEX ONLY SCAN"` 优先于 `"INDEX SCAN"` 检查，以避免误判。 + +## 异步 API + +异步后端提供相同接口，在终止方法处使用 `await`： + +```python +async def analyze_queries(): + # 简单异步 EXPLAIN + result = await User.query().explain().all_async() + + # 带 JSON 格式的 ANALYZE + result = await User.query().explain( + analyze=True, + format=ExplainFormat.JSON + ).all_async() + + for row in result.rows: + print(row.line) +``` + +## 运行时检查格式支持 + +使用 `dialect` 方法在调用前检查选项可用性： + +```python +dialect = backend.dialect + +# 检查 ANALYZE 支持（PostgreSQL 始终为 True） +if dialect.supports_explain_analyze(): + result = User.query().explain(analyze=True).all() + +# 检查特定格式（四种格式始终支持） +if dialect.supports_explain_format("JSON"): + result = User.query().explain(format=ExplainFormat.JSON).all() +``` + +## 与复杂查询组合使用 + +EXPLAIN 可与完整的查询构建器链式调用配合： + +```python +# JOIN 查询 +result = ( + User.query() + .join(Order, User.c.id == Order.c.user_id) + .where(User.c.status == "active") + .explain(format=ExplainFormat.JSON) + .all() +) + +# GROUP BY / 聚合 +result = ( + Order.query() + .group_by(Order.c.user_id) + .explain(analyze=True) + .count(Order.c.id) +) + +# 子查询 +result = ( + User.query() + .where(User.c.id.in_(Order.query().select(Order.c.user_id))) + .explain() + .all() +) +``` + +## 与 MySQL EXPLAIN 的对比 + +两个后端共享相同的查询构建器 API，但在 SQL 语法和结果结构上存在差异： + +| 特性 | PostgreSQL | MySQL | +|---|---|---| +| SQL 语法 | `EXPLAIN (ANALYZE, FORMAT JSON)` | `EXPLAIN ANALYZE FORMAT=JSON` | +| 选项分隔符 | 括号内逗号分隔 | 空格分隔，无括号 | +| 支持的格式 | TEXT、XML、JSON、YAML | TEXT、JSON（5.6.5+）、TREE（8.0.16+） | +| ANALYZE 支持 | 所有现代版本 | 8.0.18+ | +| 结果行类型 | `PostgresExplainPlanLine`（1 个字段） | `MySQLExplainRow`（12 个字段） | +| 覆盖索引标志 | 计划文本中含 `"Index Only Scan"` | `extra` 列含 `"Using index"` | + +## 版本兼容性 + +| 功能 | 最低 PostgreSQL 版本 | +|---|---| +| 基础 `EXPLAIN` | 所有支持版本 | +| `EXPLAIN FORMAT=TEXT/XML/JSON/YAML` | 所有支持版本 | +| `EXPLAIN ANALYZE` | 所有支持版本 | +| `EXPLAIN (ANALYZE, FORMAT JSON)` | 所有支持版本 | + +PostgreSQL 后端面向 PostgreSQL 9.6 及更高版本；本文档描述的所有 EXPLAIN 功能在 9.6+ 均可使用。 + +## API 参考 + +### PostgreSQL 使用的 `ExplainOptions` 字段 + +| 字段 | 类型 | 默认值 | 说明 | +|---|---|---|---| +| `analyze` | `bool` | `False` | 生成 `ANALYZE` 选项 | +| `format` | `Optional[ExplainFormat]` | `None` | 输出格式（`TEXT`/`XML`/`JSON`/`YAML`） | + +其他 `ExplainOptions` 字段（如 `verbose`、`buffers`、`timing`、`settings`）会被接受，但当前 PostgreSQL 方言实现尚未将其写入生成的 SQL。 + +### `PostgresExplainResult` 方法 + +| 方法 / 属性 | 说明 | +|---|---| +| `rows` | `List[PostgresExplainPlanLine]`，解析后的计划行 | +| `sql` | 被 EXPLAIN 的完整 SQL 字符串 | +| `duration` | 查询执行耗时（秒） | +| `raw_rows` | psycopg 返回的原始行数据（包含 `"QUERY PLAN"` 键的字典列表） | +| `analyze_index_usage()` | 返回 `"full_scan"`、`"index_with_lookup"`、`"covering_index"` 或 `"unknown"` | +| `is_full_scan` | `analyze_index_usage() == "full_scan"` 时为 `True` | +| `is_index_used` | 有任意索引被使用时为 `True` | +| `is_covering_index` | 使用了 Index Only Scan 时为 `True` | + +### `PostgresDialect` 方法 + +| 方法 | 返回值 | 说明 | +|---|---|---| +| `supports_explain_analyze()` | `bool` | PostgreSQL 始终为 `True` | +| `supports_explain_format(fmt)` | `bool` | `TEXT`、`XML`、`JSON`、`YAML` 均为 `True` | +| `format_explain_statement(expr)` | `str` | 构建 `EXPLAIN [(ANALYZE, FORMAT X)]` 前缀 | + +## 最佳实践 + +- **程序化分析推荐使用 JSON 格式。** `FORMAT JSON` 提供最为详细的结构化计划树，比纯文本更易解析。 +- **`EXPLAIN ANALYZE` 仅用于开发阶段。** 该语句会真正执行查询，仅在需要实际行数和耗时数据时使用。 +- **针对不同受众选择不同格式。** 人工查阅使用 `TEXT`，工具集成使用 `JSON`/`YAML`，XML 报表管道使用 `XML`。 +- **结合索引分析。** 在集成测试中使用 `result.is_full_scan` 作为快速冒烟测试，及时发现意外的顺序扫描。 + +💡 *AI 提示词：说明如何在 rhosocial-activerecord 中使用 PostgreSQL 的 EXPLAIN ANALYZE FORMAT JSON 诊断慢查询。* diff --git a/src/rhosocial/activerecord/backend/impl/postgres/dialect.py b/src/rhosocial/activerecord/backend/impl/postgres/dialect.py index 47e417f1..697b3b94 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/dialect.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/dialect.py @@ -397,8 +397,8 @@ def supports_explain_format(self, format_type: str) -> bool: supported_formats = ["TEXT", "XML", "JSON", "YAML"] return format_type_upper in supported_formats - def format_explain_statement(self, explain_expr: "ExplainExpression") -> str: - """Build the PostgreSQL EXPLAIN prefix using bracket-option syntax. + def format_explain_statement(self, explain_expr: "ExplainExpression") -> tuple: + """Build the PostgreSQL EXPLAIN SQL string and return (sql, params). PostgreSQL syntax: ``EXPLAIN [ ( option [, ...] ) ] statement`` @@ -406,15 +406,13 @@ def format_explain_statement(self, explain_expr: "ExplainExpression") -> str: - ``ANALYZE`` - ``FORMAT { TEXT | XML | JSON | YAML }`` - ``VERBOSE`` (passed through ``ExplainOptions.verbose`` if present) - - The prefix is returned without a trailing newline; the caller appends - the inner SQL separated by a single space. """ from rhosocial.activerecord.backend.expression.statements import ExplainType + statement_sql, statement_params = explain_expr.statement.to_sql() options = explain_expr.options if options is None: - return "EXPLAIN" + return f"EXPLAIN {statement_sql}", statement_params opts: list = [] @@ -429,8 +427,8 @@ def format_explain_statement(self, explain_expr: "ExplainExpression") -> str: pass if opts: - return "EXPLAIN (" + ", ".join(opts) + ")" - return "EXPLAIN" + return "EXPLAIN (" + ", ".join(opts) + ") " + statement_sql, statement_params + return f"EXPLAIN {statement_sql}", statement_params def supports_graph_match(self) -> bool: """Whether graph query MATCH clause is supported.""" diff --git a/tests/rhosocial/activerecord_postgres_test/feature/backend/test_backend_explain.py b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_backend_explain.py index 02f47f2e..d35c6621 100644 --- a/tests/rhosocial/activerecord_postgres_test/feature/backend/test_backend_explain.py +++ b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_backend_explain.py @@ -24,38 +24,34 @@ # Schema helpers (PostgreSQL syntax) # --------------------------------------------------------------------------- -_SETUP_SQL = """ - DROP TABLE IF EXISTS explain_order_items CASCADE; - DROP TABLE IF EXISTS explain_orders CASCADE; - - CREATE TABLE explain_orders ( +_SETUP_STMTS = [ + "DROP TABLE IF EXISTS explain_order_items CASCADE", + "DROP TABLE IF EXISTS explain_orders CASCADE", + """CREATE TABLE explain_orders ( id SERIAL PRIMARY KEY, status VARCHAR(20) NOT NULL, amount NUMERIC(10,2) - ); - CREATE INDEX idx_orders_status ON explain_orders(status); - - CREATE TABLE explain_order_items ( + )""", + "CREATE INDEX idx_orders_status ON explain_orders(status)", + """CREATE TABLE explain_order_items ( id SERIAL PRIMARY KEY, order_id INTEGER NOT NULL, sku VARCHAR(50) NOT NULL, qty INTEGER NOT NULL DEFAULT 1 - ); - CREATE INDEX idx_items_order_id_sku ON explain_order_items(order_id, sku); - - INSERT INTO explain_orders (status, amount) VALUES + )""", + "CREATE INDEX idx_items_order_id_sku ON explain_order_items(order_id, sku)", + """INSERT INTO explain_orders (status, amount) VALUES ('pending', 10.00), ('pending', 20.00), - ('shipped', 30.00), ('delivered', 40.00); - - INSERT INTO explain_order_items (order_id, sku, qty) VALUES + ('shipped', 30.00), ('delivered', 40.00)""", + """INSERT INTO explain_order_items (order_id, sku, qty) VALUES (1, 'A001', 1), (1, 'A002', 2), - (2, 'B001', 1), (3, 'A001', 3); -""" + (2, 'B001', 1), (3, 'A001', 3)""", +] -_CLEANUP_SQL = """ - DROP TABLE IF EXISTS explain_order_items CASCADE; - DROP TABLE IF EXISTS explain_orders CASCADE; -""" +_CLEANUP_STMTS = [ + "DROP TABLE IF EXISTS explain_order_items CASCADE", + "DROP TABLE IF EXISTS explain_orders CASCADE", +] # --------------------------------------------------------------------------- @@ -65,10 +61,12 @@ @pytest.fixture(scope="function") def indexed_backend(postgres_backend): """Sync backend with test tables and indexes.""" - postgres_backend.executescript(_SETUP_SQL) + for stmt in _SETUP_STMTS: + postgres_backend.execute(stmt) yield postgres_backend try: - postgres_backend.executescript(_CLEANUP_SQL) + for stmt in _CLEANUP_STMTS: + postgres_backend.execute(stmt) except Exception: pass @@ -76,10 +74,12 @@ def indexed_backend(postgres_backend): @pytest_asyncio.fixture(scope="function") async def async_indexed_backend(async_postgres_backend): """Async backend with test tables and indexes.""" - await async_postgres_backend.executescript(_SETUP_SQL) + for stmt in _SETUP_STMTS: + await async_postgres_backend.execute(stmt) yield async_postgres_backend try: - await async_postgres_backend.executescript(_CLEANUP_SQL) + for stmt in _CLEANUP_STMTS: + await async_postgres_backend.execute(stmt) except Exception: pass From a70f5d600800c7bb0cbe9dd48558ea185f6417a4 Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Wed, 1 Apr 2026 21:12:17 +0800 Subject: [PATCH 3/5] feat(backend): add dual-layer connection recovery for PostgreSQL Implement automatic reconnection mechanism with two layers: - Plan A: Pre-query connection check in _get_cursor() using psycopg's closed/broken attributes - Plan B: Error retry mechanism in execute() detecting connection errors via SQLSTATE codes and error message patterns Supported connection error SQLSTATE codes: - 08000-08007: Connection exception class - 57P01-57P04: Server shutdown class Added fixtures for connection resilience testing and comprehensive documentation covering connection loss scenarios and recovery behavior. --- docs/en_US/troubleshooting/connection.md | 208 ++++++++++- docs/zh_CN/troubleshooting/connection.md | 208 ++++++++++- .../impl/postgres/backend/async_backend.py | 231 ++++++++++-- .../backend/impl/postgres/backend/sync.py | 231 ++++++++++-- .../feature/backend/conftest.py | 76 ++++ .../backend/test_connection_resilience.py | 340 ++++++++++++++++++ 6 files changed, 1224 insertions(+), 70 deletions(-) create mode 100644 tests/rhosocial/activerecord_postgres_test/feature/backend/test_connection_resilience.py diff --git a/docs/en_US/troubleshooting/connection.md b/docs/en_US/troubleshooting/connection.md index a3228306..7de866e0 100644 --- a/docs/en_US/troubleshooting/connection.md +++ b/docs/en_US/troubleshooting/connection.md @@ -105,4 +105,210 @@ psql -U postgres -c "CREATE DATABASE test;" createdb -U postgres test ``` -💡 *AI Prompt:* "How to troubleshoot PostgreSQL connection errors?" +## Connection Loss and Automatic Recovery + +### Overview + +In long-running applications, database connections may be dropped for various reasons. The PostgreSQL backend implements a dual-layer protection mechanism to ensure automatic recovery when connections are lost. + +### Common Connection Loss Scenarios + +| Scenario | Cause | SQLSTATE Codes | +|----------|-------|----------------| +| `idle_in_transaction_session_timeout` expiry | Transaction idle time exceeds PostgreSQL's setting | 08006 | +| Connection terminated | DBA executes `pg_terminate_backend()` | 57P01 | +| Network instability | Network issues cause TCP connection to drop | 08006 | +| Server restart | PostgreSQL server restart or crash | 08006, 57P01 | +| Firewall timeout | Firewall closes long-idle TCP connections | 08006 | +| Admin shutdown | Server is being shut down by administrator | 57P01, 57P02 | + +### Automatic Recovery Mechanism + +The PostgreSQL backend implements two layers of automatic recovery: + +#### Plan A: Pre-Query Connection Check + +Before each query execution, the backend automatically checks the connection status: + +```python +def _get_cursor(self): + """Get a database cursor, ensuring connection is active.""" + if not self._connection: + # No connection, establish new one + self.connect() + elif self._connection.closed or self._connection.broken: + # Connection lost, reconnect + self._connection = None + self.connect() + return self._connection.cursor() +``` + +**Features**: +- Proactive checking, detects issues before query execution +- Uses psycopg v3's `closed` and `broken` attributes to check connection state +- Completely transparent to the application layer + +**Note**: Unlike MySQL's `is_connected()` method which actively polls the server, psycopg v3's `closed` and `broken` attributes are updated lazily. They may not immediately reflect server-side disconnection until the next operation is attempted. This is why Plan B (error retry) is also necessary. + +#### Plan B: Error Retry Mechanism + +When a connection error occurs during query execution, the backend automatically retries: + +```python +# PostgreSQL connection error SQLSTATE codes +CONNECTION_ERROR_SQLSTATES = { + "08000", # CONNECTION_EXCEPTION + "08001", # SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION + "08003", # CONNECTION_DOES_NOT_EXIST + "08004", # SQLSERVER_REJECTED_ESTABLISHMENT_OF_SQLCONNECTION + "08006", # CONNECTION_FAILURE + "08007", # TRANSACTION_RESOLUTION_UNKNOWN + "08P01", # PROTOCOL_VIOLATION + "57P01", # ADMIN_SHUTDOWN + "57P02", # CRASH_SHUTDOWN + "57P03", # CANNOT_CONNECT_NOW + "57P04", # DATABASE_DROPPED +} + +async def execute(self, sql, params=None, *, options=None, max_retries=2): + for attempt in range(max_retries + 1): + try: + return await super().execute(sql, params, options=options) + except OperationalError as e: + if self._is_connection_error_message(str(e)) and attempt < max_retries: + await self._reconnect() + continue + raise +``` + +**Features**: +- Reactive recovery, triggered on query failure +- Maximum 2 retries +- Only retries on connection errors, other errors are raised directly +- Detects connection errors via SQLSTATE codes and error message patterns + +### Manual Keep-Alive Mechanism + +For scenarios requiring proactive connection maintenance, use the `ping()` method: + +```python +# Check connection status without auto-reconnect +is_alive = backend.ping(reconnect=False) + +# Check connection status with auto-reconnect if disconnected +is_alive = backend.ping(reconnect=True) +``` + +#### Keep-Alive Best Practices + +In multi-process worker scenarios, implement periodic keep-alive: + +```python +import threading +import time + +def keepalive_worker(backend, interval=60): + """Background keep-alive thread""" + while True: + time.sleep(interval) + if backend.ping(reconnect=True): + logger.debug("Connection keepalive successful") + else: + logger.warning("Connection keepalive failed") + +# Start keep-alive thread +keepalive_thread = threading.Thread( + target=keepalive_worker, + args=(backend, 60), + daemon=True +) +keepalive_thread.start() +``` + +### Async Backend Support + +The async backend (`AsyncPostgresBackend`) provides the same connection recovery mechanism: + +```python +# Async ping +is_alive = await async_backend.ping(reconnect=True) + +# Async queries also auto-reconnect +result = await async_backend.execute("SELECT 1") +``` + +### Best Practices + +#### 1. Configure PostgreSQL Timeout Parameters Appropriately + +```sql +-- View current settings +SHOW idle_in_transaction_session_timeout; +SHOW statement_timeout; + +-- Recommended settings (adjust based on your requirements) +SET GLOBAL idle_in_transaction_session_timeout = '10min'; +-- statement_timeout should be set per-session based on query complexity +``` + +#### 2. Use Connection Pool + +For high-concurrency scenarios, configure connection pooling: + +```python +config = PostgreSQLConnectionConfig( + host='localhost', + database='mydb', + # Connection pool settings + # Note: psycopg v3 has built-in connection pool support +) +``` + +#### 3. Multi-Process Worker Scenarios + +Each worker process should have its own backend instance: + +```python +def worker_process(worker_id, config): + """Worker process entry point""" + # Create independent backend instance within the process + backend = PostgresBackend(connection_config=config) + backend.connect() + + try: + # Execute tasks + do_work(backend) + finally: + backend.disconnect() +``` + +#### 4. Monitor Connection Status + +```python +import logging + +# Enable backend logging +logging.getLogger('rhosocial.activerecord.backend').setLevel(logging.DEBUG) + +# Backend will automatically log connection recovery events +# DEBUG: Connection lost, reconnecting... +# DEBUG: Reconnected successfully +``` + +### Connection Error SQLSTATE Codes Reference + +| SQLSTATE | Name | Description | +|----------|------|-------------| +| 08000 | CONNECTION_EXCEPTION | General connection exception | +| 08001 | SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION | Client unable to establish connection | +| 08003 | CONNECTION_DOES_NOT_EXIST | Connection does not exist | +| 08004 | SQLSERVER_REJECTED_ESTABLISHMENT_OF_SQLCONNECTION | Server rejected connection | +| 08006 | CONNECTION_FAILURE | Connection failure | +| 08007 | TRANSACTION_RESOLUTION_UNKNOWN | Transaction resolution unknown | +| 08P01 | PROTOCOL_VIOLATION | Protocol violation | +| 57P01 | ADMIN_SHUTDOWN | Administrator commanded shutdown | +| 57P02 | CRASH_SHUTDOWN | Server crash | +| 57P03 | CANNOT_CONNECT_NOW | Cannot connect now | +| 57P04 | DATABASE_DROPPED | Database dropped | + +💡 *AI Prompt:* "How does PostgreSQL backend automatically recover from connection losses? What is the dual-layer protection mechanism?" diff --git a/docs/zh_CN/troubleshooting/connection.md b/docs/zh_CN/troubleshooting/connection.md index a483f3c1..64dc25c3 100644 --- a/docs/zh_CN/troubleshooting/connection.md +++ b/docs/zh_CN/troubleshooting/connection.md @@ -105,4 +105,210 @@ psql -U postgres -c "CREATE DATABASE test;" createdb -U postgres test ``` -💡 *AI 提示词：* "如何排查 PostgreSQL 连接错误？" +## 连接丢失与自动恢复 + +### 概述 + +在长时间运行的应用程序中，数据库连接可能因各种原因被断开。PostgreSQL 后端实现了双层保护机制，确保连接丢失时能够自动恢复。 + +### 常见连接丢失场景 + +| 场景 | 原因 | SQLSTATE 代码 | +|------|------|---------------| +| `idle_in_transaction_session_timeout` 超时 | 事务空闲时间超过 PostgreSQL 设置 | 08006 | +| 连接被终止 | DBA 执行 `pg_terminate_backend()` | 57P01 | +| 网络不稳定 | 网络问题导致 TCP 连接断开 | 08006 | +| 服务器重启 | PostgreSQL 服务器重启或崩溃 | 08006, 57P01 | +| 防火墙超时 | 防火墙关闭长时间空闲的 TCP 连接 | 08006 | +| 管理员关闭 | 服务器正在被管理员关闭 | 57P01, 57P02 | + +### 自动恢复机制 + +PostgreSQL 后端实现了两层自动恢复： + +#### 方案 A：查询前连接检查 + +在每次查询执行前，后端自动检查连接状态： + +```python +def _get_cursor(self): + """获取数据库游标，确保连接处于活动状态""" + if not self._connection: + # 无连接，建立新连接 + self.connect() + elif self._connection.closed or self._connection.broken: + # 连接已断开，重新连接 + self._connection = None + self.connect() + return self._connection.cursor() +``` + +**特点**： +- 主动检查，在查询执行前检测问题 +- 使用 psycopg v3 的 `closed` 和 `broken` 属性检查连接状态 +- 对应用层完全透明 + +**注意**：与 MySQL 的 `is_connected()` 方法主动轮询服务器不同，psycopg v3 的 `closed` 和 `broken` 属性是惰性更新的。它们可能不会立即反映服务器端的断开，直到尝试下一次操作。这就是为什么方案 B（错误重试）也是必要的。 + +#### 方案 B：错误重试机制 + +当查询执行过程中发生连接错误时，后端自动重试： + +```python +# PostgreSQL 连接错误 SQLSTATE 代码 +CONNECTION_ERROR_SQLSTATES = { + "08000", # CONNECTION_EXCEPTION + "08001", # SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION + "08003", # CONNECTION_DOES_NOT_EXIST + "08004", # SQLSERVER_REJECTED_ESTABLISHMENT_OF_SQLCONNECTION + "08006", # CONNECTION_FAILURE + "08007", # TRANSACTION_RESOLUTION_UNKNOWN + "08P01", # PROTOCOL_VIOLATION + "57P01", # ADMIN_SHUTDOWN + "57P02", # CRASH_SHUTDOWN + "57P03", # CANNOT_CONNECT_NOW + "57P04", # DATABASE_DROPPED +} + +async def execute(self, sql, params=None, *, options=None, max_retries=2): + for attempt in range(max_retries + 1): + try: + return await super().execute(sql, params, options=options) + except OperationalError as e: + if self._is_connection_error_message(str(e)) and attempt < max_retries: + await self._reconnect() + continue + raise +``` + +**特点**： +- 被动恢复，在查询失败时触发 +- 最多重试 2 次 +- 仅对连接错误重试，其他错误直接抛出 +- 通过 SQLSTATE 代码和错误消息模式检测连接错误 + +### 手动保活机制 + +对于需要主动维护连接的场景，使用 `ping()` 方法： + +```python +# 检查连接状态，不自动重连 +is_alive = backend.ping(reconnect=False) + +# 检查连接状态，如果断开则自动重连 +is_alive = backend.ping(reconnect=True) +``` + +#### 保活最佳实践 + +在多进程 Worker 场景中，实现定期保活： + +```python +import threading +import time + +def keepalive_worker(backend, interval=60): + """后台保活线程""" + while True: + time.sleep(interval) + if backend.ping(reconnect=True): + logger.debug("连接保活成功") + else: + logger.warning("连接保活失败") + +# 启动保活线程 +keepalive_thread = threading.Thread( + target=keepalive_worker, + args=(backend, 60), + daemon=True +) +keepalive_thread.start() +``` + +### 异步后端支持 + +异步后端（`AsyncPostgresBackend`）提供相同的连接恢复机制： + +```python +# 异步 ping +is_alive = await async_backend.ping(reconnect=True) + +# 异步查询也会自动重连 +result = await async_backend.execute("SELECT 1") +``` + +### 最佳实践 + +#### 1. 合理配置 PostgreSQL 超时参数 + +```sql +-- 查看当前设置 +SHOW idle_in_transaction_session_timeout; +SHOW statement_timeout; + +-- 推荐设置（根据您的需求调整） +SET GLOBAL idle_in_transaction_session_timeout = '10min'; +-- statement_timeout 应根据查询复杂度在每个会话中设置 +``` + +#### 2. 使用连接池 + +对于高并发场景，配置连接池： + +```python +config = PostgreSQLConnectionConfig( + host='localhost', + database='mydb', + # 连接池设置 + # 注意：psycopg v3 内置连接池支持 +) +``` + +#### 3. 多进程 Worker 场景 + +每个 Worker 进程应有自己的后端实例： + +```python +def worker_process(worker_id, config): + """Worker 进程入口""" + # 在进程内创建独立的后端实例 + backend = PostgresBackend(connection_config=config) + backend.connect() + + try: + # 执行任务 + do_work(backend) + finally: + backend.disconnect() +``` + +#### 4. 监控连接状态 + +```python +import logging + +# 启用后端日志 +logging.getLogger('rhosocial.activerecord.backend').setLevel(logging.DEBUG) + +# 后端会自动记录连接恢复事件 +# DEBUG: Connection lost, reconnecting... +# DEBUG: Reconnected successfully +``` + +### 连接错误 SQLSTATE 代码参考 + +| SQLSTATE | 名称 | 描述 | +|----------|------|------| +| 08000 | CONNECTION_EXCEPTION | 一般连接异常 | +| 08001 | SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION | 客户端无法建立连接 | +| 08003 | CONNECTION_DOES_NOT_EXIST | 连接不存在 | +| 08004 | SQLSERVER_REJECTED_ESTABLISHMENT_OF_SQLCONNECTION | 服务器拒绝连接 | +| 08006 | CONNECTION_FAILURE | 连接失败 | +| 08007 | TRANSACTION_RESOLUTION_UNKNOWN | 事务解决状态未知 | +| 08P01 | PROTOCOL_VIOLATION | 协议违规 | +| 57P01 | ADMIN_SHUTDOWN | 管理员命令关闭 | +| 57P02 | CRASH_SHUTDOWN | 服务器崩溃 | +| 57P03 | CANNOT_CONNECT_NOW | 当前无法连接 | +| 57P04 | DATABASE_DROPPED | 数据库已删除 | + +💡 *AI 提示词：* "PostgreSQL 后端如何自动从连接丢失中恢复？双层保护机制是什么？" diff --git a/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py b/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py index 8ccc7ace..8aa14b70 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py @@ -345,8 +345,19 @@ async def disconnect(self): raise OperationalError(f"Error during PostgreSQL disconnection: {str(e)}") from None async def _get_cursor(self): - """Get a database cursor, ensuring connection is active.""" + """Get a database cursor, ensuring connection is active. + + This method implements automatic connection health checking (Plan A): + - Checks if connection object exists + - Checks if connection is still valid using closed/broken attributes + - Automatically reconnects if connection was lost + """ if not self._connection: + self.log(logging.DEBUG, "No connection, connecting...") + await self.connect() + elif self._connection.closed or self._connection.broken: + self.log(logging.DEBUG, "Connection lost, reconnecting...") + self._connection = None # Clear the closed/broken connection await self.connect() return self._connection.cursor() @@ -507,8 +518,173 @@ async def _detect_extensions(self) -> Dict[str, PostgresExtensionInfo]: return extensions - async def execute(self, sql: str, params: Optional[Tuple] = None, *, options=None, **kwargs) -> QueryResult: - """Execute a SQL statement with optional parameters asynchronously.""" + async def ping(self, reconnect: bool = True) -> bool: + """ + Ping the PostgreSQL server to check if the async connection is alive. + + Args: + reconnect: If True, attempt to reconnect if the connection is dead. + If False, just return the current connection status. + + Returns: + True if the connection is alive (or was successfully reconnected), + False if the connection is dead and reconnect is False or reconnection failed. + """ + try: + if not self._connection: + if reconnect: + await self.connect() + return True + else: + return False + + # Check connection status without triggering auto-reconnect + if self._connection.closed or self._connection.broken: + if reconnect: + self._connection = None + await self.connect() + return True + else: + return False + + # Execute a simple query to verify connection is actually working + cursor = await self._get_cursor() + await cursor.execute("SELECT 1") + await cursor.fetchone() + await cursor.close() + + return True + except psycopg.Error as e: + self.log(logging.WARNING, f"PostgreSQL async connection ping failed: {str(e)}") + if reconnect: + try: + self._connection = None + await self.connect() + return True + except Exception as connect_error: + self.log(logging.ERROR, f"Failed to reconnect after ping failure: {str(connect_error)}") + return False + return False + + # PostgreSQL connection error SQLSTATE codes + # Reference: https://www.postgresql.org/docs/current/errcodes-appendix.html + CONNECTION_ERROR_SQLSTATES = { + "08000", # CONNECTION_EXCEPTION + "08001", # SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION + "08003", # CONNECTION_DOES_NOT_EXIST + "08004", # SQLSERVER_REJECTED_ESTABLISHMENT_OF_SQLCONNECTION + "08006", # CONNECTION_FAILURE + "08007", # TRANSACTION_RESOLUTION_UNKNOWN + "08P01", # PROTOCOL_VIOLATION + "57P01", # ADMIN_SHUTDOWN + "57P02", # CRASH_SHUTDOWN + "57P03", # CANNOT_CONNECT_NOW + "57P04", # DATABASE_DROPPED + } + + def _is_connection_error(self, error: Exception) -> bool: + """ + Check if an error indicates a connection loss. + + This method identifies errors that result from lost or invalid connections, + which should trigger automatic reconnection attempts. + + Args: + error: The exception to check + + Returns: + True if the error indicates a connection problem, False otherwise + """ + if isinstance(error, PsycopgOperationalError): + # Check SQLSTATE code if available + sqlstate = getattr(error, 'sqlstate', None) + if sqlstate and sqlstate in self.CONNECTION_ERROR_SQLSTATES: + return True + + # Check for common connection error patterns in message + error_msg = str(error).lower() + connection_error_patterns = [ + 'connection', + 'connect', + 'closed', + 'terminated', + 'unexpectedly', + 'broken pipe', + 'reset by peer', + 'server closed', + 'server has gone away', + ] + for pattern in connection_error_patterns: + if pattern in error_msg: + return True + + return False + + def _is_connection_error_message(self, error_msg: str) -> bool: + """ + Check if an error message indicates a connection loss. + + This method identifies error messages that result from lost or invalid connections, + which should trigger automatic reconnection attempts. + + Args: + error_msg: The error message to check + + Returns: + True if the message indicates a connection problem, False otherwise + """ + error_msg_lower = error_msg.lower() + connection_error_patterns = [ + 'connection', + 'connect', + 'closed', + 'terminated', + 'terminating connection due to administrator command', + 'unexpectedly', + 'broken pipe', + 'reset by peer', + 'server closed', + 'server has gone away', + 'ssl connection has been closed', + 'consuming input failed', + ] + for pattern in connection_error_patterns: + if pattern in error_msg_lower: + return True + return False + + async def _reconnect(self) -> bool: + """ + Attempt to reconnect to the PostgreSQL server. + + Returns: + True if reconnection was successful, False otherwise + """ + try: + self._connection = None + await self.connect() + return True + except Exception as e: + self.log(logging.ERROR, f"Failed to reconnect: {str(e)}") + return False + + async def execute(self, sql: str, params: Optional[Tuple] = None, *, options=None, max_retries: int = 2, **kwargs) -> QueryResult: + """ + Execute a SQL statement with optional parameters asynchronously. + + This method wraps the base execute method with automatic reconnection + logic for connection errors (Plan B: Error Retry Mechanism). + + Args: + sql: SQL statement to execute + params: Optional parameters for the SQL statement + options: Execution options + max_retries: Maximum number of retry attempts for connection errors + **kwargs: Additional keyword arguments + + Returns: + QueryResult containing the query results + """ from rhosocial.activerecord.backend.options import ExecutionOptions, StatementType # If no options provided, create default options from kwargs @@ -528,7 +704,7 @@ async def execute(self, sql: str, params: Optional[Tuple] = None, *, options=Non options = ExecutionOptions( stmt_type=stmt_type, - process_result_set=None, # Let the base logic determine this based on stmt_type + process_result_set=None, column_adapters=column_adapters, column_mapping=column_mapping, ) @@ -540,36 +716,23 @@ async def execute(self, sql: str, params: Optional[Tuple] = None, *, options=Non if "column_adapters" in kwargs: options.column_adapters = kwargs["column_adapters"] - return await super().execute(sql, params, options=options) - - async def ping(self, reconnect: bool = True) -> bool: - """Ping the PostgreSQL server to check if the async connection is alive.""" - try: - if not self._connection: - if reconnect: - await self.connect() - return True - else: - return False - - # Execute a simple query to check connection - cursor = await self._get_cursor() - await cursor.execute("SELECT 1") - await cursor.fetchone() - await cursor.close() - - return True - except psycopg.Error as e: - self.log(logging.WARNING, f"PostgreSQL async connection ping failed: {str(e)}") - if reconnect: - try: - await self.disconnect() - await self.connect() - return True - except Exception as connect_error: - self.log(logging.ERROR, f"Failed to reconnect after ping failure: {str(connect_error)}") - return False - return False + # Retry loop for connection errors (Plan B) + # We catch the framework's OperationalError because the original PsycopgOperationalError + # has already been converted by _handle_error() in the base class. + for attempt in range(max_retries + 1): + try: + return await super().execute(sql, params, options=options) + except OperationalError as e: + error_msg = str(e) + if self._is_connection_error_message(error_msg) and attempt < max_retries: + self.log(logging.WARNING, + f"Connection error on attempt {attempt + 1}, retrying... Error: {error_msg}") + if await self._reconnect(): + continue + # If reconnection fails, break and raise the error + self.log(logging.ERROR, f"Failed to reconnect after connection error") + break + raise async def _handle_error(self, error: Exception) -> None: """Handle PostgreSQL-specific errors asynchronously.""" diff --git a/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py b/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py index e7e28650..9f69e8a5 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py @@ -346,8 +346,19 @@ def disconnect(self): raise OperationalError(f"Error during PostgreSQL disconnection: {str(e)}") from None def _get_cursor(self): - """Get a database cursor, ensuring connection is active.""" + """Get a database cursor, ensuring connection is active. + + This method implements automatic connection health checking (Plan A): + - Checks if connection object exists + - Checks if connection is still valid using closed/broken attributes + - Automatically reconnects if connection was lost + """ if not self._connection: + self.log(logging.DEBUG, "No connection, connecting...") + self.connect() + elif self._connection.closed or self._connection.broken: + self.log(logging.DEBUG, "Connection lost, reconnecting...") + self._connection = None # Clear the closed/broken connection self.connect() return self._connection.cursor() @@ -508,8 +519,173 @@ def _detect_extensions(self) -> Dict[str, PostgresExtensionInfo]: return extensions - def execute(self, sql: str, params: Optional[Tuple] = None, *, options=None, **kwargs) -> QueryResult: - """Execute a SQL statement with optional parameters.""" + def ping(self, reconnect: bool = True) -> bool: + """ + Ping the PostgreSQL server to check if the connection is alive. + + Args: + reconnect: If True, attempt to reconnect if the connection is dead. + If False, just return the current connection status. + + Returns: + True if the connection is alive (or was successfully reconnected), + False if the connection is dead and reconnect is False or reconnection failed. + """ + try: + if not self._connection: + if reconnect: + self.connect() + return True + else: + return False + + # Check connection status without triggering auto-reconnect + if self._connection.closed or self._connection.broken: + if reconnect: + self._connection = None + self.connect() + return True + else: + return False + + # Execute a simple query to verify connection is actually working + cursor = self._get_cursor() + cursor.execute("SELECT 1") + cursor.fetchone() + cursor.close() + + return True + except psycopg.Error as e: + self.log(logging.WARNING, f"PostgreSQL connection ping failed: {str(e)}") + if reconnect: + try: + self._connection = None + self.connect() + return True + except Exception as connect_error: + self.log(logging.ERROR, f"Failed to reconnect after ping failure: {str(connect_error)}") + return False + return False + + # PostgreSQL connection error SQLSTATE codes + # Reference: https://www.postgresql.org/docs/current/errcodes-appendix.html + CONNECTION_ERROR_SQLSTATES = { + "08000", # CONNECTION_EXCEPTION + "08001", # SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION + "08003", # CONNECTION_DOES_NOT_EXIST + "08004", # SQLSERVER_REJECTED_ESTABLISHMENT_OF_SQLCONNECTION + "08006", # CONNECTION_FAILURE + "08007", # TRANSACTION_RESOLUTION_UNKNOWN + "08P01", # PROTOCOL_VIOLATION + "57P01", # ADMIN_SHUTDOWN + "57P02", # CRASH_SHUTDOWN + "57P03", # CANNOT_CONNECT_NOW + "57P04", # DATABASE_DROPPED + } + + def _is_connection_error(self, error: Exception) -> bool: + """ + Check if an error indicates a connection loss. + + This method identifies errors that result from lost or invalid connections, + which should trigger automatic reconnection attempts. + + Args: + error: The exception to check + + Returns: + True if the error indicates a connection problem, False otherwise + """ + if isinstance(error, PsycopgOperationalError): + # Check SQLSTATE code if available + sqlstate = getattr(error, 'sqlstate', None) + if sqlstate and sqlstate in self.CONNECTION_ERROR_SQLSTATES: + return True + + # Check for common connection error patterns in message + error_msg = str(error).lower() + connection_error_patterns = [ + 'connection', + 'connect', + 'closed', + 'terminated', + 'unexpectedly', + 'broken pipe', + 'reset by peer', + 'server closed', + 'server has gone away', + ] + for pattern in connection_error_patterns: + if pattern in error_msg: + return True + + return False + + def _reconnect(self) -> bool: + """ + Attempt to reconnect to the PostgreSQL server. + + Returns: + True if reconnection was successful, False otherwise + """ + try: + self._connection = None + self.connect() + return True + except Exception as e: + self.log(logging.ERROR, f"Failed to reconnect: {str(e)}") + return False + + def _is_connection_error_message(self, error_msg: str) -> bool: + """ + Check if an error message indicates a connection loss. + + This method identifies error messages that result from lost or invalid connections, + which should trigger automatic reconnection attempts. + + Args: + error_msg: The error message to check + + Returns: + True if the message indicates a connection problem, False otherwise + """ + error_msg_lower = error_msg.lower() + connection_error_patterns = [ + 'connection', + 'connect', + 'closed', + 'terminated', + 'terminating connection due to administrator command', + 'unexpectedly', + 'broken pipe', + 'reset by peer', + 'server closed', + 'server has gone away', + 'ssl connection has been closed', + 'consuming input failed', + ] + for pattern in connection_error_patterns: + if pattern in error_msg_lower: + return True + return False + + def execute(self, sql: str, params: Optional[Tuple] = None, *, options=None, max_retries: int = 2, **kwargs) -> QueryResult: + """ + Execute a SQL statement with optional parameters. + + This method wraps the base execute method with automatic reconnection + logic for connection errors (Plan B: Error Retry Mechanism). + + Args: + sql: SQL statement to execute + params: Optional parameters for the SQL statement + options: Execution options + max_retries: Maximum number of retry attempts for connection errors + **kwargs: Additional keyword arguments + + Returns: + QueryResult containing the query results + """ from rhosocial.activerecord.backend.options import ExecutionOptions, StatementType # If no options provided, create default options from kwargs @@ -529,7 +705,7 @@ def execute(self, sql: str, params: Optional[Tuple] = None, *, options=None, **k options = ExecutionOptions( stmt_type=stmt_type, - process_result_set=None, # Let the base logic determine this based on stmt_type + process_result_set=None, column_adapters=column_adapters, column_mapping=column_mapping, ) @@ -541,36 +717,23 @@ def execute(self, sql: str, params: Optional[Tuple] = None, *, options=None, **k if "column_adapters" in kwargs: options.column_adapters = kwargs["column_adapters"] - return super().execute(sql, params, options=options) - - def ping(self, reconnect: bool = True) -> bool: - """Ping the PostgreSQL server to check if the connection is alive.""" - try: - if not self._connection: - if reconnect: - self.connect() - return True - else: - return False - - # Execute a simple query to check connection - cursor = self._get_cursor() - cursor.execute("SELECT 1") - cursor.fetchone() - cursor.close() - - return True - except psycopg.Error as e: - self.log(logging.WARNING, f"PostgreSQL connection ping failed: {str(e)}") - if reconnect: - try: - self.disconnect() - self.connect() - return True - except Exception as connect_error: - self.log(logging.ERROR, f"Failed to reconnect after ping failure: {str(connect_error)}") - return False - return False + # Retry loop for connection errors (Plan B) + # We catch the framework's OperationalError because the original PsycopgOperationalError + # has already been converted by _handle_error() in the base class. + for attempt in range(max_retries + 1): + try: + return super().execute(sql, params, options=options) + except OperationalError as e: + error_msg = str(e) + if self._is_connection_error_message(error_msg) and attempt < max_retries: + self.log(logging.WARNING, + f"Connection error on attempt {attempt + 1}, retrying... Error: {error_msg}") + if self._reconnect(): + continue + # If reconnection fails, break and raise the error + self.log(logging.ERROR, f"Failed to reconnect after connection error") + break + raise def _handle_error(self, error: Exception) -> None: """Handle PostgreSQL-specific errors.""" diff --git a/tests/rhosocial/activerecord_postgres_test/feature/backend/conftest.py b/tests/rhosocial/activerecord_postgres_test/feature/backend/conftest.py index 0fde141a..ad2852bb 100644 --- a/tests/rhosocial/activerecord_postgres_test/feature/backend/conftest.py +++ b/tests/rhosocial/activerecord_postgres_test/feature/backend/conftest.py @@ -142,3 +142,79 @@ async def async_postgres_backend(request): backend.scenario_name = scenario_name # Attach for test reference yield backend await provider.async_cleanup() + + +# --- Non-parameterized Single Backend Fixtures --- + + +@pytest.fixture(scope="function") +def postgres_backend_single(): + """Non-parameterized fixture using the first available scenario.""" + scenario_names = get_scenario_names() + if not scenario_names: + pytest.skip("No PostgreSQL scenarios configured") + scenario_name = scenario_names[0] + provider = BackendFeatureProvider() + backend = provider.setup_backend(scenario_name) + backend.scenario_name = scenario_name + yield backend + provider.cleanup() + + +@pytest_asyncio.fixture(scope="function") +async def async_postgres_backend_single(): + """Non-parameterized async fixture using the first available scenario.""" + scenario_names = get_scenario_names() + if not scenario_names: + pytest.skip("No PostgreSQL scenarios configured") + scenario_name = scenario_names[0] + provider = BackendFeatureProvider() + backend = await provider.setup_async_backend(scenario_name) + backend.scenario_name = scenario_name + yield backend + await provider.async_cleanup() + + +# --- Control Backend for Session Modification Tests --- + + +@pytest.fixture(scope="function") +def postgres_control_backend(): + """ + Dedicated control backend for tests that modify PostgreSQL session settings. + + This fixture provides an independent backend instance for operations that + need to control or interfere with the main test backend, such as: + - Terminating other connections (pg_terminate_backend) + - Setting global variables + - Monitoring other connections + + The fixture is NOT parameterized to ensure consistent behavior across + all PostgreSQL scenarios. + """ + scenario_names = get_scenario_names() + if not scenario_names: + pytest.skip("No PostgreSQL scenarios configured") + scenario_name = scenario_names[0] + provider = BackendFeatureProvider() + backend = provider.setup_backend(scenario_name) + yield backend + provider.cleanup() + + +@pytest_asyncio.fixture(scope="function") +async def async_postgres_control_backend(): + """ + Dedicated async control backend for tests that modify PostgreSQL session settings. + + This fixture provides an independent async backend instance for operations that + need to control or interfere with the main test backend. + """ + scenario_names = get_scenario_names() + if not scenario_names: + pytest.skip("No PostgreSQL scenarios configured") + scenario_name = scenario_names[0] + provider = BackendFeatureProvider() + backend = await provider.setup_async_backend(scenario_name) + yield backend + await provider.async_cleanup() diff --git a/tests/rhosocial/activerecord_postgres_test/feature/backend/test_connection_resilience.py b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_connection_resilience.py new file mode 100644 index 00000000..93be3c45 --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_connection_resilience.py @@ -0,0 +1,340 @@ +# tests/rhosocial/activerecord_postgres_test/feature/backend/test_connection_resilience.py +""" +Connection Resilience Tests + +Tests for PostgreSQL backend's ability to handle connection loss scenarios: +1. Connection terminated by pg_terminate_backend() +2. Automatic reconnection via Plan A (pre-query check) and Plan B (error retry) +3. Manual ping reconnection + +These tests verify the implementation of: +- Plan A: Pre-query connection check in _get_cursor() +- Plan B: Error retry mechanism in execute() + +Both synchronous (PostgresBackend) and asynchronous (AsyncPostgresBackend) are tested. + +Note: Unlike MySQL's is_connected() which actively checks connection state, +psycopg v3's closed/broken attributes only update after an operation fails. +Therefore, tests focus on actual query execution and reconnection behavior. +""" +import pytest +import pytest_asyncio +import asyncio +import time +import logging + +from rhosocial.activerecord.backend.impl.postgres import PostgresBackend, AsyncPostgresBackend + + +logger = logging.getLogger(__name__) + + +def print_separator(title: str): + """Print a visual separator for test sections.""" + print("\n" + "=" * 70) + print(f" {title}") + print("=" * 70) + + +# ============================================================================ +# Synchronous Backend Tests +# ============================================================================ + + +class TestConnectionStateAttributes: + """Tests for the closed/broken attributes in various states.""" + + def test_connection_not_closed_when_connected(self, postgres_backend_single: PostgresBackend): + """Verify closed and broken are False for active connection.""" + assert postgres_backend_single._connection is not None + assert postgres_backend_single._connection.closed is False + assert postgres_backend_single._connection.broken is False + + def test_connection_closed_after_disconnect(self, postgres_backend_single: PostgresBackend): + """Verify closed is True after explicit disconnect.""" + postgres_backend_single.disconnect() + # After disconnect, _connection should be None + assert postgres_backend_single._connection is None + + +class TestTerminateConnectionRecovery: + """Tests for pg_terminate_backend() recovery via Plan A + Plan B.""" + + def test_terminate_triggers_reconnection( + self, + postgres_backend_single: PostgresBackend, + postgres_control_backend: PostgresBackend + ): + """Test that pg_terminate_backend() triggers automatic reconnection.""" + print_separator("Test: pg_terminate_backend Recovery") + + # 1. Get current connection PID + result = postgres_backend_single.execute("SELECT pg_backend_pid() AS pid") + original_pid = result.data[0]['pid'] + print(f"Original connection PID: {original_pid}") + + # 2. Terminate the connection + print(f"Terminating connection {original_pid}...") + postgres_control_backend.execute(f"SELECT pg_terminate_backend({original_pid})") + time.sleep(1) + + # 3. Execute query - should trigger reconnection (Plan A or Plan B) + print("Executing query after connection terminated...") + result = postgres_backend_single.execute("SELECT 1 AS test") + print(f"Query succeeded: {result}") + assert result.data[0]['test'] == 1 + + # 4. Verify new connection PID + new_pid_result = postgres_backend_single.execute("SELECT pg_backend_pid() AS pid") + new_pid = new_pid_result.data[0]['pid'] + print(f"New connection PID: {new_pid}") + assert new_pid != original_pid + + def test_multiple_queries_after_reconnection( + self, + postgres_backend_single: PostgresBackend, + postgres_control_backend: PostgresBackend + ): + """Test that multiple queries work correctly after reconnection.""" + print_separator("Test: Multiple Queries After Reconnection") + + # Terminate connection + result = postgres_backend_single.execute("SELECT pg_backend_pid() AS pid") + original_pid = result.data[0]['pid'] + postgres_control_backend.execute(f"SELECT pg_terminate_backend({original_pid})") + time.sleep(1) + + # Execute multiple queries + for i in range(5): + result = postgres_backend_single.execute(f"SELECT {i} AS value") + assert result.data[0]['value'] == i + print(f"Query {i + 1}/5 succeeded") + + +class TestIdleTimeoutRecovery: + """Tests for idle_in_transaction_session_timeout recovery.""" + + def test_idle_timeout_triggers_reconnection( + self, + postgres_backend_single: PostgresBackend, + postgres_control_backend: PostgresBackend + ): + """ + Test that idle_in_transaction_session_timeout triggers automatic reconnection. + + Note: We test idle_in_transaction_session_timeout instead of statement_timeout + because it's more relevant to connection pooling scenarios. + """ + print_separator("Test: idle_in_transaction_session_timeout Recovery") + + # 1. Get original timeout + result = postgres_backend_single.execute( + "SHOW idle_in_transaction_session_timeout" + ) + original_timeout = result.data[0]['idle_in_transaction_session_timeout'] + print(f"Original idle_in_transaction_session_timeout: {original_timeout}") + + # 2. Set short timeout (5 seconds) in a transaction + test_timeout = "5s" + print(f"Setting idle_in_transaction_session_timeout to {test_timeout}...") + postgres_backend_single.execute( + f"SET SESSION idle_in_transaction_session_timeout = '{test_timeout}'" + ) + + # Verify setting + result = postgres_backend_single.execute( + "SHOW idle_in_transaction_session_timeout" + ) + current_timeout = result.data[0]['idle_in_transaction_session_timeout'] + print(f"Current timeout: {current_timeout}") + + # 3. Wait for timeout + print("Waiting 7 seconds for timeout...") + time.sleep(7) + + # 4. Execute query - should trigger reconnection + print("Executing query after timeout...") + try: + result = postgres_backend_single.execute("SELECT 1 AS test") + print(f"Query succeeded: {result}") + assert result.data[0]['test'] == 1 + finally: + # Restore original timeout via control backend + try: + postgres_control_backend.execute( + f"SET GLOBAL idle_in_transaction_session_timeout = '{original_timeout}'" + ) + except Exception: + pass # Ignore cleanup errors + + +class TestPingReconnect: + """Tests for manual ping reconnection.""" + + def test_ping_reconnect_after_terminate( + self, + postgres_backend_single: PostgresBackend, + postgres_control_backend: PostgresBackend + ): + """Test that ping(reconnect=True) can restore a terminated connection.""" + print_separator("Test: Ping Reconnect") + + # 1. Get current connection PID + result = postgres_backend_single.execute("SELECT pg_backend_pid() AS pid") + original_pid = result.data[0]['pid'] + print(f"Original connection PID: {original_pid}") + + # 2. Terminate the connection + print(f"Terminating connection {original_pid}...") + postgres_control_backend.execute(f"SELECT pg_terminate_backend({original_pid})") + time.sleep(1) + + # 3. Ping with reconnect - should succeed + print("Calling ping(reconnect=True)...") + ping_result = postgres_backend_single.ping(reconnect=True) + print(f"Ping result: {ping_result}") + assert ping_result is True, "Ping should succeed with reconnect=True" + + # 4. Verify connection is restored + assert postgres_backend_single._connection is not None + assert postgres_backend_single._connection.closed is False + + # 5. Verify new connection PID + new_pid_result = postgres_backend_single.execute("SELECT pg_backend_pid() AS pid") + new_pid = new_pid_result.data[0]['pid'] + print(f"New connection PID: {new_pid}") + assert new_pid != original_pid + + def test_ping_alive_connection_returns_true(self, postgres_backend_single: PostgresBackend): + """Test that ping returns True for alive connections.""" + print_separator("Test: Ping Alive Connection") + + ping_result = postgres_backend_single.ping(reconnect=False) + print(f"Ping result for alive connection: {ping_result}") + assert ping_result is True + + +# ============================================================================ +# Asynchronous Backend Tests +# ============================================================================ + + +class TestAsyncConnectionStateAttributes: + """Async tests for the closed/broken attributes in various states.""" + + @pytest.mark.asyncio + async def test_connection_not_closed_when_connected(self, async_postgres_backend_single: AsyncPostgresBackend): + """Verify closed and broken are False for active connection.""" + assert async_postgres_backend_single._connection is not None + assert async_postgres_backend_single._connection.closed is False + assert async_postgres_backend_single._connection.broken is False + + @pytest.mark.asyncio + async def test_connection_closed_after_disconnect(self, async_postgres_backend_single: AsyncPostgresBackend): + """Verify closed is True after explicit disconnect.""" + await async_postgres_backend_single.disconnect() + # After disconnect, _connection should be None + assert async_postgres_backend_single._connection is None + + +class TestAsyncTerminateConnectionRecovery: + """Async tests for pg_terminate_backend() recovery.""" + + @pytest.mark.asyncio + async def test_terminate_triggers_reconnection( + self, + async_postgres_backend_single: AsyncPostgresBackend, + async_postgres_control_backend: AsyncPostgresBackend + ): + """Test that pg_terminate_backend() triggers automatic reconnection.""" + print_separator("Test: Async pg_terminate_backend Recovery") + + # 1. Get current connection PID + result = await async_postgres_backend_single.execute("SELECT pg_backend_pid() AS pid") + original_pid = result.data[0]['pid'] + print(f"Original connection PID: {original_pid}") + + # 2. Terminate the connection + print(f"Terminating connection {original_pid}...") + await async_postgres_control_backend.execute(f"SELECT pg_terminate_backend({original_pid})") + await asyncio.sleep(1) + + # 3. Execute query - should trigger reconnection + print("Executing query after connection terminated...") + result = await async_postgres_backend_single.execute("SELECT 1 AS test") + print(f"Query succeeded: {result}") + assert result.data[0]['test'] == 1 + + # 4. Verify new connection PID + new_pid_result = await async_postgres_backend_single.execute("SELECT pg_backend_pid() AS pid") + new_pid = new_pid_result.data[0]['pid'] + print(f"New connection PID: {new_pid}") + assert new_pid != original_pid + + @pytest.mark.asyncio + async def test_multiple_queries_after_reconnection( + self, + async_postgres_backend_single: AsyncPostgresBackend, + async_postgres_control_backend: AsyncPostgresBackend + ): + """Test that multiple queries work correctly after reconnection.""" + print_separator("Test: Async Multiple Queries After Reconnection") + + # Terminate connection + result = await async_postgres_backend_single.execute("SELECT pg_backend_pid() AS pid") + original_pid = result.data[0]['pid'] + await async_postgres_control_backend.execute(f"SELECT pg_terminate_backend({original_pid})") + await asyncio.sleep(1) + + # Execute multiple queries + for i in range(5): + result = await async_postgres_backend_single.execute(f"SELECT {i} AS value") + assert result.data[0]['value'] == i + print(f"Query {i + 1}/5 succeeded") + + +class TestAsyncPingReconnect: + """Async tests for manual ping reconnection.""" + + @pytest.mark.asyncio + async def test_ping_reconnect_after_terminate( + self, + async_postgres_backend_single: AsyncPostgresBackend, + async_postgres_control_backend: AsyncPostgresBackend + ): + """Test that ping(reconnect=True) can restore a terminated connection.""" + print_separator("Test: Async Ping Reconnect") + + # 1. Get current connection PID + result = await async_postgres_backend_single.execute("SELECT pg_backend_pid() AS pid") + original_pid = result.data[0]['pid'] + print(f"Original connection PID: {original_pid}") + + # 2. Terminate the connection + print(f"Terminating connection {original_pid}...") + await async_postgres_control_backend.execute(f"SELECT pg_terminate_backend({original_pid})") + await asyncio.sleep(1) + + # 3. Ping with reconnect - should succeed + print("Calling ping(reconnect=True)...") + ping_result = await async_postgres_backend_single.ping(reconnect=True) + print(f"Ping result: {ping_result}") + assert ping_result is True, "Ping should succeed with reconnect=True" + + # 4. Verify connection is restored + assert async_postgres_backend_single._connection.closed is False + + # 5. Verify new connection PID + new_pid_result = await async_postgres_backend_single.execute("SELECT pg_backend_pid() AS pid") + new_pid = new_pid_result.data[0]['pid'] + print(f"New connection PID: {new_pid}") + assert new_pid != original_pid + + @pytest.mark.asyncio + async def test_ping_alive_connection_returns_true(self, async_postgres_backend_single: AsyncPostgresBackend): + """Test that ping returns True for alive connections.""" + print_separator("Test: Async Ping Alive Connection") + + ping_result = await async_postgres_backend_single.ping(reconnect=False) + print(f"Ping result for alive connection: {ping_result}") + assert ping_result is True From e2bd448e4064b9497fde876927ce394555be0275 Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Thu, 2 Apr 2026 01:29:34 +0800 Subject: [PATCH 4/5] ci: update rhosocial-activerecord installation source to git+https Install from release/v1.0.0.dev21 branch instead of PyPI dev20. --- .github/workflows/test.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index b3b23dde..a06c5b22 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -67,7 +67,7 @@ jobs: - name: Install activerecord and testsuite from PyPI run: | - pip install rhosocial-activerecord==1.0.0.dev20 + pip install git+https://github.com/rhosocial/python-activerecord@release/v1.0.0.dev21 pip install rhosocial-activerecord-testsuite==1.0.0.dev7 - name: Create postgres scenario config From 7e493d88e23297356c8409834704c9002dfdb7be Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Thu, 2 Apr 2026 14:46:45 +0800 Subject: [PATCH 5/5] chore: add changelog fragment for PR #16 --- changelog.d/16.added.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/16.added.md diff --git a/changelog.d/16.added.md b/changelog.d/16.added.md new file mode 100644 index 00000000..8e871dad --- /dev/null +++ b/changelog.d/16.added.md @@ -0,0 +1 @@ +Added EXPLAIN clause support with typed PostgresExplainResult and dual-layer connection recovery for PostgreSQL backend