From 4ef146e1c1f75ee42e1d0f0d327f9e7d79e00c36 Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Wed, 11 Nov 2026 11:57:03 +0800 Subject: [PATCH 1/5] feat(errors): classify ClickHouse driver errors by taxonomy --- .../impl/clickhouse/backend/backend.py | 14 +- .../impl/clickhouse/mixins/backend_mixin.py | 250 ++++++++++++++++-- .../errors/test_error_taxonomy_contract.py | 4 + .../backend/test_driver_error_mapping.py | 222 ++++++++++++++++ 4 files changed, 467 insertions(+), 23 deletions(-) create mode 100644 tests/rhosocial/activerecord_clickhouse_test/feature/backend/errors/test_error_taxonomy_contract.py create mode 100644 tests/rhosocial/activerecord_clickhouse_test/feature/backend/test_driver_error_mapping.py diff --git a/src/rhosocial/activerecord/backend/impl/clickhouse/backend/backend.py b/src/rhosocial/activerecord/backend/impl/clickhouse/backend/backend.py index 5ecdbcd..47001a2 100644 --- a/src/rhosocial/activerecord/backend/impl/clickhouse/backend/backend.py +++ b/src/rhosocial/activerecord/backend/impl/clickhouse/backend/backend.py @@ -31,7 +31,11 @@ log_execution_error, ) from rhosocial.activerecord.backend.base.operations import execution_log_context -from rhosocial.activerecord.backend.errors import ConnectionError, DatabaseError +from rhosocial.activerecord.backend.errors import ( + ConnectionError, + DatabaseError, + attach_diagnostics, +) from rhosocial.activerecord.backend.options import ( BulkInsertOptions, BulkUpdateOptions, @@ -427,7 +431,13 @@ def connect(self) -> None: logging.DEBUG, f"Failed to connect to ClickHouse database, exception will be re-raised: {str(e)}", ) - raise ConnectionError(f"Failed to connect to ClickHouse: {str(e)}") from e + raise attach_diagnostics( + ConnectionError(f"Failed to connect to ClickHouse: {str(e)}"), + operation="connect", + driver_code=getattr(e, "code", None), + backend="clickhouse", + transient=False, + ) from e def disconnect(self) -> None: """Close connection to ClickHouse database.""" diff --git a/src/rhosocial/activerecord/backend/impl/clickhouse/mixins/backend_mixin.py b/src/rhosocial/activerecord/backend/impl/clickhouse/mixins/backend_mixin.py index 9c6596d..3c93747 100644 --- a/src/rhosocial/activerecord/backend/impl/clickhouse/mixins/backend_mixin.py +++ b/src/rhosocial/activerecord/backend/impl/clickhouse/mixins/backend_mixin.py @@ -1,6 +1,6 @@ # src/rhosocial/activerecord/backend/impl/clickhouse/mixins/backend_mixin.py import logging -from typing import Dict, Tuple, Type +from typing import Dict, Optional, Tuple, Type from rhosocial.activerecord.backend.type_adapter import SQLTypeAdapter @@ -124,20 +124,55 @@ def get_default_adapter_suggestions(self) -> Dict[Type, Tuple[SQLTypeAdapter, Ty return suggestions + # clickhouse-connect 1.7.x exposes the ClickHouse server error code as + # ``Error.code`` and its symbolic name as ``Error.name``. Codes come from + # ClickHouse's ErrorCodes.cpp and are stable across server versions, so they + # are checked before message text. + INTEGRITY_ERROR_CODES = { + 57, # TABLE_ALREADY_EXISTS: a duplicate object name + 70, # CANNOT_CONVERT_TYPE: includes NULL into a non-nullable column + } + QUERY_ERROR_CODES = { + 47, # UNKNOWN_IDENTIFIER + 60, # UNKNOWN_TABLE + 62, # SYNTAX_ERROR + 81, # UNKNOWN_DATABASE + } + TRANSIENT_ERROR_CODES = { + 159, # TIMEOUT_EXCEEDED + 202, # TOO_MANY_SIMULTANEOUS_QUERIES + 241, # MEMORY_LIMIT_EXCEEDED + } + AUTHENTICATION_ERROR_CODES = { + 497, # ACCESS_DENIED + 516, # AUTHENTICATION_FAILED + } + READONLY_ERROR_CODES = { + 164, # READONLY: the server or session refuses writes + } CONNECTION_ERROR_CODES = { + 209, # SOCKET_TIMEOUT + 210, # NETWORK_ERROR + 279, # ALL_CONNECTION_TRIES_FAILED + # HTTP-level codes some driver versions surface instead of a + # ClickHouse server code. 503, 504, 1000, 1002, - 210, - 279, } + def _error_code(self, error: Exception) -> Optional[int]: + """Return the ClickHouse server error code when the driver exposes one.""" + code = getattr(error, "code", None) + if code is None: + code = getattr(error, "errno", None) + return code + def _is_connection_error(self, error: Exception) -> bool: """Check if an error indicates a connection loss.""" - if hasattr(error, "errno"): - if error.errno in self.CONNECTION_ERROR_CODES: - return True + if self._error_code(error) in self.CONNECTION_ERROR_CODES: + return True error_str = str(error).lower() connection_error_patterns = [ @@ -153,38 +188,211 @@ def _is_connection_error(self, error: Exception) -> bool: ] return any(pattern in error_str for pattern in connection_error_patterns) + def _is_authentication_error(self, error: Exception) -> bool: + """Check if an error means the credentials or permissions were refused.""" + if self._error_code(error) in self.AUTHENTICATION_ERROR_CODES: + return True + + error_str = str(error).lower() + authentication_error_patterns = [ + "authentication failed", + "access denied", + "not enough privileges", + "invalid password", + "unknown user", + "there is no user", + ] + return any(pattern in error_str for pattern in authentication_error_patterns) + def _handle_error(self, error: Exception) -> None: - """Handle ClickHouse-specific errors.""" + """Handle ClickHouse-specific errors. + + Recognition order: driver exception class, ClickHouse server error + code, then message text. + + ClickHouse has no general-purpose transactions and no deadlock + detector: the server cannot report a deadlock, so no ``DeadlockError`` + mapping exists here. Raising one would tell callers to retry a failure + that never arrives. Concurrent-query pressure (code 202) and memory + exhaustion (code 241) are infrastructure failures and are marked + transient instead. + """ from clickhouse_connect.driver.exceptions import ( DatabaseError as ClickHouseDatabaseError, Error as ClickHouseError, IntegrityError as ClickHouseIntegrityError, OperationalError as ClickHouseOperationalError, + ProgrammingError as ClickHouseProgrammingError, ) from rhosocial.activerecord.backend.errors import ( + ConnectionError, DatabaseError, IntegrityError, OperationalError, + QueryError, + attach_diagnostics, ) error_msg = str(error) + error_lower = error_msg.lower() + code = self._error_code(error) + backend = "clickhouse" if isinstance(error, ClickHouseIntegrityError): - if "UNIQUE constraint" in error_msg or "Violates unique constraint" in error_msg: - raise IntegrityError(f"Unique constraint violation: {error_msg}") - raise IntegrityError(error_msg) - elif isinstance(error, ClickHouseDatabaseError): - # ClickHouse enforces NOT NULL through its type system, so a - # assignment of NULL to a non-nullable column arrives as a generic - # database error (CANNOT_CONVERT_TYPE) rather than as an integrity - # error. Preserve the cross-backend contract and classify it as a - # not-null violation. - if "Cannot convert NULL to a non-nullable type" in error_msg: - raise IntegrityError(f"cannot be null: {error_msg}") - raise DatabaseError(error_msg) + if ( + code in self.INTEGRITY_ERROR_CODES + or "unique constraint" in error_lower + or "violates unique constraint" in error_lower + ): + self.log(logging.DEBUG, f"Unique constraint violation: {error_msg}") + raise attach_diagnostics( + IntegrityError(f"Unique constraint violation: {error_msg}"), + operation="execute", + driver_code=code, + backend=backend, + ) from error + self.log(logging.DEBUG, f"Integrity error: {error_msg}") + raise attach_diagnostics( + IntegrityError(error_msg), + operation="execute", + driver_code=code, + backend=backend, + ) from error + elif ( + code in self.INTEGRITY_ERROR_CODES + or "cannot convert null" in error_lower + or "non-nullable" in error_lower + ): + # ClickHouse enforces NOT NULL through its type system, so assigning + # NULL to a non-nullable column arrives as CANNOT_CONVERT_TYPE + # (code 70) rather than as a driver integrity error. Preserve the + # cross-backend contract and classify it as a not-null violation. + if "cannot convert null" in error_lower or "non-nullable" in error_lower: + self.log(logging.DEBUG, f"NULL in non-nullable column: {error_msg}") + raise attach_diagnostics( + IntegrityError(f"cannot be null: {error_msg}"), + operation="execute", + driver_code=code, + backend=backend, + ) from error + self.log(logging.DEBUG, f"Integrity violation: {error_msg}") + raise attach_diagnostics( + IntegrityError(f"Integrity violation: {error_msg}"), + operation="execute", + driver_code=code, + backend=backend, + ) from error + elif self._is_authentication_error(error): + # Credentials and permissions do not improve on retry, so this is + # a non-transient ConnectionError rather than OperationalError. + self.log(logging.DEBUG, f"Authentication or permission failure: {error_msg}") + raise attach_diagnostics( + ConnectionError(error_msg), + operation="execute", + driver_code=code, + backend=backend, + transient=False, + ) from error elif isinstance(error, ClickHouseOperationalError): - raise OperationalError(error_msg) + # The HTTP layer raises OperationalError after retrying, so it is + # already a transient infrastructure failure. + transient = ( + code in self.TRANSIENT_ERROR_CODES + or code in self.CONNECTION_ERROR_CODES + or self._is_connection_error(error) + or "timeout exceeded" in error_lower + or "timeout expired" in error_lower + ) + self.log(logging.DEBUG, f"Operational error: {error_msg}") + raise attach_diagnostics( + OperationalError(error_msg), + operation="execute", + driver_code=code, + backend=backend, + transient=transient, + ) from error + elif code in self.TRANSIENT_ERROR_CODES or any( + hint in error_lower + for hint in ( + "timeout exceeded", + "timeout expired", + "memory limit", + "too many simultaneous", + ) + ): + self.log(logging.DEBUG, f"Transient server pressure: {error_msg}") + raise attach_diagnostics( + OperationalError(error_msg), + operation="execute", + driver_code=code, + backend=backend, + transient=True, + ) from error + elif ( + code in self.READONLY_ERROR_CODES + or "readonly" in error_lower + or "read-only" in error_lower + ): + # A read-only server/session is not a model policy failure: report + # it as infrastructure so it is never confused with ReadOnlyError. + self.log(logging.DEBUG, f"Read-only server: {error_msg}") + raise attach_diagnostics( + OperationalError(error_msg), + operation="execute", + driver_code=code, + backend=backend, + transient=False, + ) from error + elif self._is_connection_error(error): + self.log(logging.DEBUG, f"Connection lost: {error_msg}") + raise attach_diagnostics( + OperationalError(error_msg), + operation="execute", + driver_code=code, + backend=backend, + transient=True, + ) from error + elif ( + code in self.QUERY_ERROR_CODES + or isinstance(error, ClickHouseProgrammingError) + or any( + hint in error_lower + for hint in ( + "syntax error", + "unknown identifier", + "unknown table", + "unknown column", + "unknown database", + "missing columns", + "no such column", + "doesn't exist", + "cannot parse", + ) + ) + ): + self.log(logging.DEBUG, f"Query error: {error_msg}") + raise attach_diagnostics( + QueryError(error_msg), + operation="execute", + driver_code=code, + backend=backend, + ) from error + elif isinstance(error, ClickHouseDatabaseError): + self.log(logging.DEBUG, f"Database error: {error_msg}") + raise attach_diagnostics( + DatabaseError(error_msg), + operation="execute", + driver_code=code, + backend=backend, + ) from error elif isinstance(error, ClickHouseError): - raise DatabaseError(error_msg) + self.log(logging.DEBUG, f"ClickHouse error: {error_msg}") + raise attach_diagnostics( + DatabaseError(error_msg), + operation="execute", + driver_code=code, + backend=backend, + ) from error else: + self.log(logging.DEBUG, f"Unexpected error: {error_msg}") raise error diff --git a/tests/rhosocial/activerecord_clickhouse_test/feature/backend/errors/test_error_taxonomy_contract.py b/tests/rhosocial/activerecord_clickhouse_test/feature/backend/errors/test_error_taxonomy_contract.py new file mode 100644 index 0000000..0377e24 --- /dev/null +++ b/tests/rhosocial/activerecord_clickhouse_test/feature/backend/errors/test_error_taxonomy_contract.py @@ -0,0 +1,4 @@ +# tests/rhosocial/activerecord_clickhouse_test/feature/backend/errors/test_error_taxonomy_contract.py +"""Run the testsuite's semantic error taxonomy contract for ClickHouse.""" + +from rhosocial.activerecord.testsuite.feature.backend.errors.test_error_taxonomy import * # noqa: F401,F403 diff --git a/tests/rhosocial/activerecord_clickhouse_test/feature/backend/test_driver_error_mapping.py b/tests/rhosocial/activerecord_clickhouse_test/feature/backend/test_driver_error_mapping.py new file mode 100644 index 0000000..9771448 --- /dev/null +++ b/tests/rhosocial/activerecord_clickhouse_test/feature/backend/test_driver_error_mapping.py @@ -0,0 +1,222 @@ +# tests/rhosocial/activerecord_clickhouse_test/feature/backend/test_driver_error_mapping.py +"""ClickHouse driver exceptions map to semantic error categories. + +Only synthetic driver exceptions are constructed; no ClickHouse server is +touched. The codes come from the ``code`` keyword accepted by +clickhouse-connect 1.7.x ``Error`` classes, mirroring ClickHouse's +ErrorCodes.cpp. The whole module skips when the driver is not installed. +""" + +import pytest + +clickhouse_errors = pytest.importorskip("clickhouse_connect.driver.exceptions") + +from rhosocial.activerecord.backend.errors import ( # noqa: E402 + ConcurrencyError, + ConnectionError, + DataError, + DatabaseError, + InfrastructureError, + IntegrityError, + OperationalError, + QueryError, +) +from rhosocial.activerecord.backend.impl.clickhouse.backend import ( # noqa: E402 + ClickHouseBackend, +) +from rhosocial.activerecord.backend.impl.clickhouse.config import ( # noqa: E402 + ClickHouseConnectionConfig, +) + + +@pytest.fixture +def backend(): + """A backend that never connects: mapping is pure classification.""" + return ClickHouseBackend( + connection_config=ClickHouseConnectionConfig(host="localhost", database="db") + ) + + +def _translated(backend, error): + with pytest.raises(DatabaseError) as info: + backend._handle_error(error) + return info.value + + +def test_unique_constraint_maps_to_integrity(backend): + error = clickhouse_errors.IntegrityError("Violates unique constraint users_email") + + translated = _translated(backend, error) + + assert isinstance(translated, IntegrityError) + assert isinstance(translated, DataError) + assert translated.backend == "clickhouse" + assert translated.operation == "execute" + assert translated.retryable is False + + +def test_duplicate_object_code_maps_to_integrity(backend): + error = clickhouse_errors.DatabaseError( + "Table db.users already exists", + code=57, + name="TABLE_ALREADY_EXISTS", + ) + + translated = _translated(backend, error) + + assert isinstance(translated, IntegrityError) + assert translated.driver_code == 57 + + +def test_null_to_non_nullable_maps_to_integrity(backend): + error = clickhouse_errors.DatabaseError( + "Cannot convert NULL to a non-nullable type", + code=70, + name="CANNOT_CONVERT_TYPE", + ) + + translated = _translated(backend, error) + + assert isinstance(translated, IntegrityError) + assert "cannot be null" in str(translated) + assert translated.driver_code == 70 + + +def test_timeout_maps_to_transient_operational_error(backend): + error = clickhouse_errors.OperationalError( + "Timeout exceeded: max_execution_time", + code=159, + name="TIMEOUT_EXCEEDED", + ) + + translated = _translated(backend, error) + + assert isinstance(translated, OperationalError) + assert isinstance(translated, InfrastructureError) + assert translated.transient is True + assert translated.retryable is True + assert translated.driver_code == 159 + + +def test_concurrency_limit_is_transient_infrastructure(backend): + error = clickhouse_errors.DatabaseError( + "Too many simultaneous queries", + code=202, + name="TOO_MANY_SIMULTANEOUS_QUERIES", + ) + + translated = _translated(backend, error) + + assert isinstance(translated, OperationalError) + assert not isinstance(translated, ConcurrencyError) + assert translated.transient is True + assert translated.retryable is True + + +def test_memory_limit_is_transient_infrastructure(backend): + error = clickhouse_errors.DatabaseError( + "Memory limit exceeded", + code=241, + name="MEMORY_LIMIT_EXCEEDED", + ) + + translated = _translated(backend, error) + + assert isinstance(translated, OperationalError) + assert translated.transient is True + + +def test_unknown_table_maps_to_query_error(backend): + error = clickhouse_errors.DatabaseError( + "Table db.missing doesn't exist", + code=60, + name="UNKNOWN_TABLE", + ) + + translated = _translated(backend, error) + + assert isinstance(translated, QueryError) + assert translated.driver_code == 60 + assert translated.retryable is False + + +def test_unknown_column_maps_to_query_error(backend): + error = clickhouse_errors.DatabaseError( + "Unknown identifier users.missing", + code=47, + name="UNKNOWN_IDENTIFIER", + ) + + translated = _translated(backend, error) + + assert isinstance(translated, QueryError) + assert translated.driver_code == 47 + + +def test_syntax_error_maps_to_query_error(backend): + error = clickhouse_errors.DatabaseError( + "Syntax error near FROM", + code=62, + name="SYNTAX_ERROR", + ) + + translated = _translated(backend, error) + + assert isinstance(translated, QueryError) + assert translated.driver_code == 62 + + +def test_authentication_failure_is_not_retryable(backend): + error = clickhouse_errors.OperationalError( + "default: Authentication failed", + code=516, + name="AUTHENTICATION_FAILED", + ) + + translated = _translated(backend, error) + + assert isinstance(translated, ConnectionError) + assert isinstance(translated, InfrastructureError) + assert translated.transient is False + assert translated.retryable is False + assert translated.driver_code == 516 + + +def test_connection_loss_is_transient(backend): + error = clickhouse_errors.OperationalError("Connection refused by server") + + translated = _translated(backend, error) + + assert isinstance(translated, OperationalError) + assert translated.transient is True + assert translated.retryable is True + + +def test_generic_database_error_falls_back(backend): + error = clickhouse_errors.DatabaseError("Something odd happened") + + translated = _translated(backend, error) + + assert isinstance(translated, DatabaseError) + assert not isinstance( + translated, + (IntegrityError, QueryError, OperationalError, ConcurrencyError), + ) + assert translated.operation == "execute" + assert translated.backend == "clickhouse" + + +def test_generic_driver_error_falls_back(backend): + error = clickhouse_errors.Error("Something odd happened") + + translated = _translated(backend, error) + + assert isinstance(translated, DatabaseError) + assert not isinstance(translated, DataError) + + +def test_non_driver_exception_is_reraised(backend): + error = ValueError("local validation only") + + with pytest.raises(ValueError, match="local validation only"): + backend._handle_error(error) From d55dcf82f21cac2d73121d44c574d147997312ba Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Wed, 11 Nov 2026 13:14:10 +0800 Subject: [PATCH 2/5] fix(execute): align ClickHouse execute error phases --- .../impl/clickhouse/backend/backend.py | 133 +++++--- .../impl/clickhouse/mixins/backend_mixin.py | 12 +- .../backend/test_execute_error_contract.py | 309 ++++++++++++++++++ 3 files changed, 400 insertions(+), 54 deletions(-) create mode 100644 tests/rhosocial/activerecord_clickhouse_test/feature/backend/test_execute_error_contract.py diff --git a/src/rhosocial/activerecord/backend/impl/clickhouse/backend/backend.py b/src/rhosocial/activerecord/backend/impl/clickhouse/backend/backend.py index 47001a2..a5f7ccb 100644 --- a/src/rhosocial/activerecord/backend/impl/clickhouse/backend/backend.py +++ b/src/rhosocial/activerecord/backend/impl/clickhouse/backend/backend.py @@ -16,13 +16,11 @@ from clickhouse_connect.dbapi import connect as clickhouse_connect_dbapi_connect from clickhouse_connect.driver.exceptions import ( Error as ClickHouseError, - OperationalError as ClickHouseOperationalError, ) except ImportError: # pragma: no cover clickhouse_connect = None # type: ignore clickhouse_connect_dbapi_connect = None # type: ignore ClickHouseError = Exception - ClickHouseOperationalError = Exception from rhosocial.activerecord.backend.base import StorageBackend from rhosocial.activerecord.backend.base.execution_logging import ( @@ -34,6 +32,8 @@ from rhosocial.activerecord.backend.errors import ( ConnectionError, DatabaseError, + OperationalError, + TransactionError, attach_diagnostics, ) from rhosocial.activerecord.backend.options import ( @@ -506,6 +506,12 @@ def execute_many( driver's ``executemany`` path does not report per-row counts. The batch DEBUG record and its INFO summary come from the shared execution-logging helpers, so the canonical SQL line is not duplicated here. + + Every failure goes through ``_handle_error`` like ``execute`` does, so a + driver timeout or syntax error carries the same semantic type on both + paths. The batch is auto-committed before the result is returned, and a + cursor that refuses to close cannot replace the error already + propagating. """ if options is None: options = ExecutionOptions(stmt_type=StatementType.DML) @@ -528,6 +534,11 @@ def execute_many( cursor.execute(escaped_sql, params) affected_rows += cursor.rowcount + # Auto-commit before reporting success: a commit failure must abort + # the batch instead of returning a result for rows that were never + # persisted. + self._handle_auto_commit_if_needed() + duration = time.perf_counter() - start_time result = QueryResult(affected_rows=affected_rows, data=None, duration=duration) @@ -537,11 +548,21 @@ def execute_many( except Exception as e: log_execution_error(self, e, options) + if isinstance(e, DatabaseError): + raise self._handle_error(e) - return None + # _handle_error() is required to raise. Keep a defensive fallback so + # a handler that returns cannot turn a failure into a None result. + raise DatabaseError(str(e)) from e finally: if cursor: - cursor.close() + try: + cursor.close() + except Exception as close_error: + self.log( + logging.WARNING, + f"Failed to close cursor after execute_many: {str(close_error)}", + ) def get_server_version(self) -> Tuple[int, int, int]: """Get ClickHouse server version.""" @@ -653,40 +674,50 @@ def _reconnect(self) -> bool: return False def _handle_auto_commit(self) -> None: - """Handle auto commit based on ClickHouse connection and transaction state. + """Commit the connection when autocommit is off and no transaction is active. This method will commit the current connection if: 1. The connection exists and is open 2. There is no active transaction managed by transaction_manager + 3. The connection config has ``autocommit`` disabled It's used by insert/update/delete operations to ensure changes are persisted immediately when auto_commit=True is specified. + + A failed commit is not swallowed: the statement already ran, so + returning a successful result for rows that were never persisted would + report data loss as success. The driver failure is raised as + :class:`TransactionError` with the original exception as its cause. """ + if not self._connection: + return + + if self.in_transaction: + return + + # For ClickHouse, if autocommit is disabled, we need to commit explicitly + if getattr(self.config, "autocommit", True): + return + try: - # Check if connection exists - if not self._connection: - return - - # Check if we're not in an active transaction - if not self.in_transaction: - # For ClickHouse, if autocommit is disabled, we need to commit explicitly - if not getattr(self.config, "autocommit", True): - self._connection.commit() - self.log(logging.DEBUG, "Auto-committed operation (not in active transaction)") - except Exception as e: - # Just log the error but don't raise - this is a convenience feature - self.log(logging.WARNING, f"Failed to auto-commit: {str(e)}") + self._connection.commit() + except Exception as exc: + raise attach_diagnostics( + TransactionError(f"Auto-commit failed: {str(exc)}"), + operation="commit", + backend="clickhouse", + ) from exc + self.log(logging.DEBUG, "Auto-committed operation (not in active transaction)") def _handle_auto_commit_if_needed(self) -> None: """ Handle auto-commit for ClickHouse. - ClickHouse respects the autocommit setting, but we also need to handle explicit commits. + ClickHouse respects the autocommit setting, but we also need to handle + explicit commits. Delegates to :meth:`_handle_auto_commit` so that the + commit-failure contract is identical on both call paths. """ - if not self.in_transaction and self._connection: - if not getattr(self.config, "autocommit", True): - self._connection.commit() - self.log(logging.DEBUG, "Auto-committed operation (not in active transaction)") + self._handle_auto_commit() def execute( self, @@ -701,10 +732,11 @@ def execute( This method combines: 1. Automatic ExecutionOptions construction when none is provided - 2. Retry logic for connection errors (Plan B error recovery) + 2. Retry logic for transient connection failures (Plan B error recovery) - Plan B: If a connection error occurs during execution, it will automatically - attempt to reconnect and retry the query up to max_retries times. + Plan B: If a transient connection failure occurs during execution, it + will automatically attempt to reconnect and retry the query up to + ``max_retries`` times. Args: sql: The SQL statement to execute @@ -717,7 +749,8 @@ def execute( QueryResult object containing execution results Raises: - DatabaseError: If execution fails after all retries + DatabaseError: the translated framework error, raised unchanged once + retries are exhausted or the failure is not transient. """ # If no options provided, create default options from kwargs if options is None: @@ -748,34 +781,30 @@ def execute( if "column_adapters" in kwargs: options.column_adapters = kwargs["column_adapters"] - # Execute with retry logic for connection errors - last_error = None - + # Execute with retry logic for transient connection failures. + # + # ``super().execute()`` already translates driver exceptions through + # ``_handle_error``, so the errors reaching this loop are framework + # ``DatabaseError`` subclasses. Catching driver exception types here + # never matched, and translating the framework error a second time + # rewrote a type that was already correct. Retry only when the + # framework error declares itself transient, and stop after the last + # attempt without touching the type. for attempt in range(max_retries + 1): try: return super().execute(sql, params, options=options) - except (ClickHouseOperationalError, ClickHouseError) as e: - last_error = e - - # Check if this is a connection error that warrants retry - if self._is_connection_error(e) and attempt < max_retries: - self.log(logging.WARNING, f"Connection error on attempt {attempt + 1}/{max_retries + 1}: {str(e)}") - - # Attempt to reconnect - if self._reconnect(): - continue - else: - self.log(logging.DEBUG, "Reconnection failed, aborting retry") - break - else: - # Not a connection error or max retries reached - break - - # All retries exhausted or non-connection error - if last_error: - self._handle_error(last_error) - - # This should not be reached, but for type safety + except (ConnectionError, OperationalError) as exc: + if exc.transient is True and attempt < max_retries and self._reconnect(): + self.log( + logging.WARNING, + f"Transient error on attempt {attempt + 1}/{max_retries + 1}, " + f"reconnected and retrying: {str(exc)}", + ) + continue + raise + + # Unreachable for max_retries >= 0 (the loop always returns or raises); + # kept so a negative max_retries cannot turn a failure into ``None``. raise DatabaseError(f"Execution failed after {max_retries + 1} attempts") def executescript(self, sql_script: str) -> None: diff --git a/src/rhosocial/activerecord/backend/impl/clickhouse/mixins/backend_mixin.py b/src/rhosocial/activerecord/backend/impl/clickhouse/mixins/backend_mixin.py index 3c93747..6282587 100644 --- a/src/rhosocial/activerecord/backend/impl/clickhouse/mixins/backend_mixin.py +++ b/src/rhosocial/activerecord/backend/impl/clickhouse/mixins/backend_mixin.py @@ -207,8 +207,13 @@ def _is_authentication_error(self, error: Exception) -> bool: def _handle_error(self, error: Exception) -> None: """Handle ClickHouse-specific errors. - Recognition order: driver exception class, ClickHouse server error - code, then message text. + A framework :class:`DatabaseError` is re-raised unchanged: it was + already translated once, and re-classifying it here would rewrite a + ``QueryError`` or ``TransactionError`` produced by a later stage of the + execution pipeline into a generic driver failure. + + Recognition order for driver exceptions: driver exception class, + ClickHouse server error code, then message text. ClickHouse has no general-purpose transactions and no deadlock detector: the server cannot report a deadlock, so no ``DeadlockError`` @@ -233,6 +238,9 @@ def _handle_error(self, error: Exception) -> None: attach_diagnostics, ) + if isinstance(error, DatabaseError): + raise error + error_msg = str(error) error_lower = error_msg.lower() code = self._error_code(error) diff --git a/tests/rhosocial/activerecord_clickhouse_test/feature/backend/test_execute_error_contract.py b/tests/rhosocial/activerecord_clickhouse_test/feature/backend/test_execute_error_contract.py new file mode 100644 index 0000000..478fd16 --- /dev/null +++ b/tests/rhosocial/activerecord_clickhouse_test/feature/backend/test_execute_error_contract.py @@ -0,0 +1,309 @@ +# tests/rhosocial/activerecord_clickhouse_test/feature/backend/test_execute_error_contract.py +"""ClickHouse ``execute`` phases follow the core error contract. + +``ExecutionMixin.execute`` is patched so the retry loop sees the framework +errors the real pipeline produces, and a fake connection covers the commit and +close phases. No ClickHouse server is contacted; the module skips when the +driver is not installed. + +Note: ClickHouse has no deadlock detector, so the "lock wait" case of the +contract becomes a server timeout (``OperationalError(transient=True)``). +""" + +import pytest + +pytest.importorskip("clickhouse_connect") + +from clickhouse_connect.driver.exceptions import ( # noqa: E402 + DatabaseError as ClickHouseDatabaseError, + OperationalError as ClickHouseOperationalError, +) + +from rhosocial.activerecord.backend.base.execution import ExecutionMixin # noqa: E402 +from rhosocial.activerecord.backend.errors import ( # noqa: E402 + ConnectionError, + OperationalError, + QueryError, + TransactionError, + attach_diagnostics, +) +from rhosocial.activerecord.backend.impl.clickhouse.backend import ( # noqa: E402 + ClickHouseBackend, +) +from rhosocial.activerecord.backend.impl.clickhouse.config import ( # noqa: E402 + ClickHouseConnectionConfig, +) +from rhosocial.activerecord.backend.result import QueryResult # noqa: E402 + + +class _FakeCursor: + """DB-API cursor that records calls and can fail on purpose.""" + + def __init__(self, *, execute_error=None, close_error=None): + self.execute_error = execute_error + self.close_error = close_error + self.rowcount = 1 + self.lastrowid = None + self.description = None + self.executed = [] + self.closed = False + + def execute(self, sql, params=None, settings=None): # noqa: ARG002 + if self.execute_error is not None: + raise self.execute_error + self.executed.append((sql, params)) + + def fetchone(self): + return (1,) + + def close(self): + self.closed = True + if self.close_error is not None: + raise self.close_error + + +class _FakeConnection: + """Connection whose first cursor serves the liveness probe. + + ``_get_cursor`` spends one cursor on the probe query and then opens the + cursor the caller uses, so the configured failures apply from the second + cursor onward. + """ + + def __init__(self, *, execute_error=None, close_error=None, commit_error=None): + self.execute_error = execute_error + self.close_error = close_error + self.commit_error = commit_error + self.cursors = [] + self.commits = 0 + + def cursor(self): + first = not self.cursors + cursor = _FakeCursor( + execute_error=None if first else self.execute_error, + close_error=None if first else self.close_error, + ) + self.cursors.append(cursor) + return cursor + + def commit(self): + self.commits += 1 + if self.commit_error is not None: + raise self.commit_error + + def close(self): + pass + + +def _make_backend(*, autocommit: bool = True) -> ClickHouseBackend: + """Build a backend without opening a connection.""" + return ClickHouseBackend( + connection_config=ClickHouseConnectionConfig( + host="localhost", + database="test_db", + autocommit=autocommit, + ) + ) + + +@pytest.fixture +def backend(): + return _make_backend() + + +@pytest.fixture +def transient_error(): + """A framework error the retry loop is allowed to retry.""" + return attach_diagnostics( + OperationalError("Timeout exceeded while waiting for server response"), + operation="execute", + backend="clickhouse", + transient=True, + ) + + +@pytest.fixture +def patched_execution(monkeypatch): + """Replace the core pipeline so only the retry loop is under test. + + Returns a helper that installs a script of outcomes (error or result). + """ + + def install(outcomes): + calls = [] + + def fake_execute(self, sql, params=None, *, options=None): # noqa: ARG001 + calls.append(sql) + outcome = outcomes[min(len(calls) - 1, len(outcomes) - 1)] + if isinstance(outcome, BaseException): + raise outcome + return outcome + + monkeypatch.setattr(ExecutionMixin, "execute", fake_execute) + return calls + + return install + + +@pytest.fixture +def reconnects(monkeypatch, backend): + """Record ``_reconnect`` calls; returns the recording list.""" + recorded = [] + + def fake_reconnect(): + recorded.append(True) + return True + + monkeypatch.setattr(backend, "_reconnect", fake_reconnect) + return recorded + + +class TestExecuteRetry: + """Retries depend on ``error.transient``, not on driver exception types.""" + + def test_transient_error_reconnects_then_retries(self, backend, patched_execution, reconnects, + transient_error): + calls = patched_execution([transient_error, QueryResult(affected_rows=1, duration=0.0)]) + + result = backend.execute("INSERT INTO t VALUES (1)") + + assert result.affected_rows == 1 + assert len(calls) == 2 + assert reconnects == [True] + + def test_non_transient_error_is_not_retried(self, backend, patched_execution, reconnects): + error = ConnectionError("Authentication failed for user default") + calls = patched_execution([error]) + + with pytest.raises(ConnectionError) as info: + backend.execute("SELECT 1") + + assert info.value is error + assert len(calls) == 1 + assert reconnects == [] + + def test_transient_connection_error_is_retried(self, backend, patched_execution, reconnects): + error = attach_diagnostics( + ConnectionError("Network unreachable"), + operation="connect", + backend="clickhouse", + transient=True, + ) + calls = patched_execution([error, QueryResult(affected_rows=0, duration=0.0)]) + + result = backend.execute("SELECT 1") + + assert result.affected_rows == 0 + assert len(calls) == 2 + assert reconnects == [True] + + def test_retries_exhausted_raise_original_error_unchanged(self, backend, patched_execution, + reconnects, transient_error): + calls = patched_execution([transient_error]) + + with pytest.raises(OperationalError) as info: + backend.execute("SELECT 1", max_retries=2) + + assert info.value is transient_error + assert len(calls) == 3 + assert len(reconnects) == 2 + + def test_failed_reconnect_stops_retrying(self, backend, patched_execution, reconnects, + transient_error, monkeypatch): + calls = patched_execution([transient_error]) + monkeypatch.setattr(backend, "_reconnect", lambda: False) + + with pytest.raises(OperationalError) as info: + backend.execute("SELECT 1") + + assert info.value is transient_error + assert len(calls) == 1 + + +class TestAutoCommit: + """A failed commit is reported, never swallowed.""" + + def test_commit_failure_raises_transaction_error(self, monkeypatch): + commit_error = RuntimeError("commit rejected by server") + backend = _make_backend(autocommit=False) + backend._connection = _FakeConnection(commit_error=commit_error) + + with pytest.raises(TransactionError) as info: + backend.execute("INSERT INTO t VALUES (1)") + + assert "Auto-commit failed" in str(info.value) + assert info.value.__cause__ is commit_error + assert backend._connection.commits == 1 + + def test_autocommit_enabled_skips_commit(self, backend): + backend._connection = _FakeConnection() + + result = backend.execute("INSERT INTO t VALUES (1)") + + assert result.affected_rows == 1 + assert backend._connection.commits == 0 + + +class TestExecuteMany: + """``execute_many`` uses the same taxonomy and commit contract as ``execute``.""" + + def test_timeout_maps_to_transient_operational_error(self, backend): + driver_error = ClickHouseOperationalError( + "Timeout exceeded while reading from socket", + code=159, + name="TIMEOUT_EXCEEDED", + ) + backend._connection = _FakeConnection(execute_error=driver_error) + + with pytest.raises(OperationalError) as info: + backend.execute_many("INSERT INTO t VALUES (%s)", [(1,), (2,)]) + + assert info.value.transient is True + assert info.value.driver_code == 159 + assert info.value.backend == "clickhouse" + + def test_commit_failure_raises_transaction_error(self, monkeypatch): + commit_error = RuntimeError("commit rejected by server") + backend = _make_backend(autocommit=False) + backend._connection = _FakeConnection(commit_error=commit_error) + + with pytest.raises(TransactionError) as info: + backend.execute_many("INSERT INTO t VALUES (%s)", [(1,), (2,)]) + + assert "Auto-commit failed" in str(info.value) + assert info.value.__cause__ is commit_error + + def test_close_failure_does_not_replace_original_error(self, backend): + syntax_error = ClickHouseDatabaseError( + "Syntax error near FROM", + code=62, + name="SYNTAX_ERROR", + ) + backend._connection = _FakeConnection( + execute_error=syntax_error, + close_error=RuntimeError("cursor close failed"), + ) + + with pytest.raises(QueryError, match="Syntax error near FROM"): + backend.execute_many("INSERT INTO t VALUES (%s)", [(1,)]) + + # The close was attempted, and its failure stayed out of the way. + assert backend._connection.cursors[-1].closed is True + + +class TestHandleError: + """A framework error is translated once, never a second time.""" + + def test_framework_error_is_raised_unchanged(self, backend): + error = attach_diagnostics( + QueryError("already translated"), + operation="execute", + backend="clickhouse", + ) + + with pytest.raises(QueryError) as info: + backend._handle_error(error) + + assert info.value is error + assert type(info.value) is QueryError + assert info.value.operation == "execute" From 6472434b593363d62a8c9381f7e91e7430f59b4d Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Wed, 11 Nov 2026 14:28:07 +0800 Subject: [PATCH 3/5] ci: pin core and testsuite feature/error-taxonomy branches --- .github/workflows/test.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index afc64e1..6cb2cfe 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -66,8 +66,8 @@ jobs: - name: Install activerecord and testsuite run: | - pip install "rhosocial-activerecord @ git+https://github.com/rhosocial/python-activerecord.git@release/v1.0.0.dev30" # NOSONAR - pip install "rhosocial-activerecord-testsuite @ git+https://github.com/rhosocial/python-activerecord-testsuite.git@release/v1.0.0.dev17" + pip install "rhosocial-activerecord @ git+https://github.com/rhosocial/python-activerecord.git@feature/error-taxonomy" # NOSONAR + pip install "rhosocial-activerecord-testsuite @ git+https://github.com/rhosocial/python-activerecord-testsuite.git@feature/error-taxonomy" - name: Install dependencies run: | From 3e1831119dc78a49c55ef75a9f9bfa752ed0841a Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Wed, 11 Nov 2026 15:47:10 +0800 Subject: [PATCH 4/5] docs(changelog): add fragment for the error taxonomy PR (#12) --- changelog.d/12.added.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/12.added.md diff --git a/changelog.d/12.added.md b/changelog.d/12.added.md new file mode 100644 index 0000000..ba2e7eb --- /dev/null +++ b/changelog.d/12.added.md @@ -0,0 +1 @@ +Classified ClickHouse driver errors into the semantic error taxonomy (data, record, policy, concurrency, infrastructure) and aligned `execute` / `execute_many` with the execute-phase error contract: framework errors pass through, auto-commit failures raise `TransactionError`, and only transient connection failures are retried. Added contract tests covering the mapping and phases. From c8c4ccdccea73861a665284f506e7ecd390cd1ba Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Wed, 11 Nov 2026 15:52:57 +0800 Subject: [PATCH 5/5] ci: mark feature-branch pip installs as reviewed (#12) --- .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 6cb2cfe..2e505b7 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -67,7 +67,7 @@ jobs: - name: Install activerecord and testsuite run: | pip install "rhosocial-activerecord @ git+https://github.com/rhosocial/python-activerecord.git@feature/error-taxonomy" # NOSONAR - pip install "rhosocial-activerecord-testsuite @ git+https://github.com/rhosocial/python-activerecord-testsuite.git@feature/error-taxonomy" + pip install "rhosocial-activerecord-testsuite @ git+https://github.com/rhosocial/python-activerecord-testsuite.git@feature/error-taxonomy" # NOSONAR - name: Install dependencies run: |