diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 35ea6a8..692e496 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -44,7 +44,7 @@ env:
   # The previous pin, the fix/schema-name-propagation-gaps branch, predates all
   # of that layer, so the imports this backend now makes do not resolve against
   # it.
-  CORE_PKG: "rhosocial-activerecord @ git+https://github.com/rhosocial/python-activerecord.git@release/v1.0.0.dev30"
+  CORE_PKG: "rhosocial-activerecord @ git+https://github.com/rhosocial/python-activerecord.git@feature/logging-execution"
   TESTSUITE_PKG: "rhosocial-activerecord-testsuite @ git+https://github.com/rhosocial/python-activerecord-testsuite.git@release/v1.0.0.dev17"
 
 jobs:
diff --git a/changelog.d/9.changed.md b/changelog.d/9.changed.md
new file mode 100644
index 0000000..fb085a3
--- /dev/null
+++ b/changelog.d/9.changed.md
@@ -0,0 +1 @@
+Adapted this backend to the core execution logging contract: structured execution summaries, shared sync/async logger ownership, redacted SQL logging, and bridges for the shared relation/query/basic contracts.
diff --git a/src/rhosocial/activerecord/backend/impl/bigquery/async_transaction.py b/src/rhosocial/activerecord/backend/impl/bigquery/async_transaction.py
index 31a1815..7c7ebe9 100644
--- a/src/rhosocial/activerecord/backend/impl/bigquery/async_transaction.py
+++ b/src/rhosocial/activerecord/backend/impl/bigquery/async_transaction.py
@@ -2,6 +2,8 @@
 from contextlib import asynccontextmanager
 from typing import Optional
 
+from rhosocial.activerecord.logging.defaults import get_logger
+
 
 class AsyncBigQueryTransactionManager:
     def __init__(self, backend):
@@ -20,8 +22,7 @@ async def transaction(self, isolation_level: Optional[str] = None):
             except Exception:
                 # The original exception must propagate; a backend that
                 # rejects ROLLBACK (e.g. the emulator) should not mask it.
-                import logging
-                logging.getLogger(__name__).warning(
+                get_logger("rhosocial.activerecord.transaction").warning(
                     "Failed to roll back BigQuery transaction", exc_info=True
                 )
             raise
diff --git a/src/rhosocial/activerecord/backend/impl/bigquery/backend/async_backend.py b/src/rhosocial/activerecord/backend/impl/bigquery/backend/async_backend.py
index 46e1e06..46f3139 100644
--- a/src/rhosocial/activerecord/backend/impl/bigquery/backend/async_backend.py
+++ b/src/rhosocial/activerecord/backend/impl/bigquery/backend/async_backend.py
@@ -1,9 +1,15 @@
 """BigQuery asynchronous backend implementation."""
+import logging
 from typing import Any, Dict, List, Optional, Tuple
 
 from rhosocial.activerecord.backend.base import AsyncStorageBackend
+from rhosocial.activerecord.backend.base.execution_logging import (
+    log_execution_error, log_execution_start,
+)
 from rhosocial.activerecord.backend.errors import ConnectionError, DatabaseError
+from rhosocial.activerecord.backend.options import ExecutionOptions, LogContext
 from rhosocial.activerecord.backend.result import QueryResult
+from rhosocial.activerecord.backend.schema import StatementType
 from rhosocial.activerecord.backend.dialect.exceptions import UnsupportedFeatureError
 from rhosocial.activerecord.backend.introspection.backend_mixin import (
     IntrospectorBackendMixin,
@@ -12,6 +18,7 @@
 from ..config import BigQueryConnectionConfig
 from ..dialect import BigQueryDialect
 from ..async_transaction import AsyncBigQueryTransactionManager
+from .backend import log_result_summary
 from .qualified_names import BigQueryQualifiedNameMixin
 
 
@@ -76,6 +83,7 @@ async def connect(self) -> None:
                 project = getattr(config, 'project', None)
                 self._client = bigquery.Client(project=project, client_options=client_options)
             self._connected = True
+            self.log(logging.DEBUG, "Connected to BigQuery")
         except Exception as e:
             raise ConnectionError(f"Failed to connect to BigQuery: {e}") from e
 
@@ -125,8 +133,23 @@ def _create_introspector(self) -> Any:
             self, AsyncBigQueryIntrospectorExecutor(self)
         )
 
-    async def execute(self, sql: str, params: Optional[Tuple] = None, returning: Optional[Any] = None, column_adapters: Optional[Dict] = None, *, options: Optional[Any] = None) -> QueryResult:
-        from rhosocial.activerecord.backend.schema import StatementType
+    async def execute(
+        self,
+        sql: str,
+        params: Optional[Tuple] = None,
+        returning: Optional[Any] = None,
+        column_adapters: Optional[Dict] = None,
+        *,
+        options: Optional[Any] = None,
+    ) -> QueryResult:
+        """
+        Execute a SQL statement against BigQuery asynchronously.
+
+        Mirrors :meth:`BigQueryBackend.execute`: the core execution pipeline
+        is replaced (there is no DB-API cursor), so this method emits the
+        shared ``DEBUG`` SQL record and ``INFO`` summary through
+        :mod:`rhosocial.activerecord.backend.base.execution_logging`.
+        """
         from .backend import BigQueryBackend
 
         if options is not None:
@@ -139,12 +162,18 @@ async def execute(self, sql: str, params: Optional[Tuple] = None, returning: Opt
         else:
             column_mapping = None
             is_select = sql.strip().upper().startswith(("SELECT", "WITH"))
+        stmt_type = getattr(options, 'stmt_type', None) if options is not None else None
+        if stmt_type is None:
+            stmt_type = StatementType.DQL if is_select else StatementType.OTHER
+        log_execution_start(self, sql, params, options)
         try:
             if getattr(self.config, 'api_endpoint', None):
                 # Emulator REST fast path: see BigQueryBackend.execute.
-                return BigQueryBackend._execute_via_rest(
+                result = BigQueryBackend._execute_via_rest(
                     self, sql, params, is_select, column_adapters, column_mapping
                 )
+                log_result_summary(self, stmt_type, sql, result, options)
+                return result
             if not getattr(self, '_connected', False) or getattr(self, '_client', None) is None:
                 await self.connect()
             job_config = BigQueryBackend._build_job_config(params)
@@ -154,7 +183,6 @@ async def execute(self, sql: str, params: Optional[Tuple] = None, returning: Opt
                 exc = job.exception()
                 if exc is not None:
                     raise exc
-                from rhosocial.activerecord.backend.errors import DatabaseError
                 raise DatabaseError(str(job.errors))
             if is_select:
                 rows_iter = job.result()
@@ -173,18 +201,26 @@ async def execute(self, sql: str, params: Optional[Tuple] = None, returning: Opt
                 affected_rows = job.num_dml_affected_rows or 0
                 result_data = None
             duration = (job.ended - job.started).total_seconds() if job.started and job.ended else None
-            return QueryResult(data=result_data, affected_rows=affected_rows, duration=duration)
+            result = QueryResult(data=result_data, affected_rows=affected_rows, duration=duration)
+            log_result_summary(self, stmt_type, sql, result, options)
+            return result
         except Exception as e:
+            log_execution_error(self, e, options)
             self._handle_error(e)
 
     # -- Client-side PK generation (see BigQueryBackend for rationale) ----------
 
     async def _next_pk_value(self, table: str, schema_name, pk_column: str) -> int:
-        from rhosocial.activerecord.backend.options import ExecutionOptions
-        from rhosocial.activerecord.backend.schema import StatementType
-
         sql, params = self._next_pk_expression(table, schema_name, pk_column).to_sql()
-        result = await self.execute(sql, params, options=ExecutionOptions(stmt_type=StatementType.DQL))
+        options = ExecutionOptions(
+            stmt_type=StatementType.DQL,
+            log_context=LogContext(
+                operation="select",
+                object_name=table,
+                schema=schema_name,
+            ),
+        )
+        result = await self.execute(sql, params, options=options)
         row = result.data[0]
         value = row.get("next_id") if isinstance(row, dict) else row[0]
         key = (schema_name, table, pk_column)
@@ -247,13 +283,12 @@ async def bulk_insert(self, options) -> QueryResult:
 
     async def bulk_update(self, options) -> QueryResult:
         # See BigQueryBackend.bulk_update: explicit ELSE branch per CASE.
+        from rhosocial.activerecord.backend.base.operations import execution_log_context
         from rhosocial.activerecord.backend.expression import (
             Column, Literal, UpdateStatement, ComparisonPredicate, InPredicate,
         )
         from rhosocial.activerecord.backend.expression.advanced_functions import CaseExpression
         from rhosocial.activerecord.backend.expression.objects import Table
-        from rhosocial.activerecord.backend.options import ExecutionOptions
-        from rhosocial.activerecord.backend.schema import StatementType
 
         pk_col = Column(self.dialect, options.pk_column)
         assignments = {}
@@ -283,10 +318,11 @@ async def bulk_update(self, options) -> QueryResult:
             stmt_type=StatementType.DML,
             column_adapters=options.column_adapters,
             column_mapping=options.column_mapping,
+            log_context=execution_log_context(options, "bulk_update"),
         )
         result = await self.execute(sql, params, options=exec_options)
         if options.auto_commit:
-            self._handle_auto_commit_if_needed()
+            await self._handle_auto_commit_if_needed()
         if result.affected_rows == 0 and self._is_emulator_mode():
             result.affected_rows = len(getattr(options, "pk_values", None) or [])
         return result
@@ -299,11 +335,16 @@ def _is_emulator_mode(self) -> bool:
     async def _count_matching(self, table: str, schema_name, where):
         if where is None:
             return None
-        from rhosocial.activerecord.backend.options import ExecutionOptions
-        from rhosocial.activerecord.backend.schema import StatementType
-
         sql, params = self._count_expression(table, schema_name, where).to_sql()
-        result = await self.execute(sql, params, options=ExecutionOptions(stmt_type=StatementType.DQL))
+        options = ExecutionOptions(
+            stmt_type=StatementType.DQL,
+            log_context=LogContext(
+                operation="count",
+                object_name=table,
+                schema=schema_name,
+            ),
+        )
+        result = await self.execute(sql, params, options=options)
         if not result.data:
             return None
         row = result.data[0]
@@ -410,12 +451,12 @@ async def introspect_and_adapt(self) -> None:
         """See :meth:`BigQueryBackend.introspect_and_adapt`."""
         if not getattr(self, '_client', None):
             return
-        import logging
         try:
             dataset_name = getattr(self.config, 'dataset', None)
             if dataset_name:
                 self._client.get_dataset(dataset_name)
             else:
                 list(self._client.list_datasets(max_results=1))
+            self.log(logging.DEBUG, "BigQuery introspection completed")
         except Exception as e:
             self.log(logging.WARNING, f"BigQuery introspection failed: {e}")
diff --git a/src/rhosocial/activerecord/backend/impl/bigquery/backend/backend.py b/src/rhosocial/activerecord/backend/impl/bigquery/backend/backend.py
index 983cbd8..4381d2f 100644
--- a/src/rhosocial/activerecord/backend/impl/bigquery/backend/backend.py
+++ b/src/rhosocial/activerecord/backend/impl/bigquery/backend/backend.py
@@ -3,10 +3,15 @@
 from typing import Any, Dict, List, Optional, Tuple
 
 from rhosocial.activerecord.backend.base import StorageBackend
+from rhosocial.activerecord.backend.base.execution_logging import (
+    log_execution_error, log_execution_start, log_execution_summary,
+)
 from rhosocial.activerecord.backend.errors import (
     ConnectionError, DatabaseError, IntegrityError, QueryError,
 )
+from rhosocial.activerecord.backend.options import ExecutionOptions, LogContext
 from rhosocial.activerecord.backend.result import QueryResult
+from rhosocial.activerecord.backend.schema import StatementType
 from rhosocial.activerecord.backend.dialect.exceptions import UnsupportedFeatureError
 from rhosocial.activerecord.backend.introspection.backend_mixin import (
     IntrospectorBackendMixin,
@@ -18,6 +23,45 @@
 from .qualified_names import BigQueryQualifiedNameMixin
 
 
+class ExecutionSummaryCursor:
+    """Cursor stand-in for the shared execution summary.
+
+    BigQuery has no DB-API cursor: the affected-row count already lives on
+    the :class:`~rhosocial.activerecord.backend.result.QueryResult`.  The
+    shared summary helper only reads ``rowcount``, so the count is exposed
+    through that one attribute instead of duplicating the helper.
+    """
+
+    __slots__ = ("rowcount",)
+
+    def __init__(self, rowcount: int) -> None:
+        self.rowcount = rowcount
+
+
+def log_result_summary(
+    backend: Any,
+    stmt_type: StatementType,
+    sql: str,
+    result: QueryResult,
+    options: Optional[ExecutionOptions],
+) -> None:
+    """Emit the shared ``INFO`` summary for one completed BigQuery execution.
+
+    The canonical DEBUG SQL line and INFO DML/DQL/DDL/TCL summary stay in
+    :mod:`rhosocial.activerecord.backend.base.execution_logging`; this
+    adapter only supplies the row count and rows BigQuery already returned.
+    """
+    log_execution_summary(
+        backend,
+        stmt_type,
+        sql,
+        ExecutionSummaryCursor(result.affected_rows or 0),
+        result.data,
+        result.duration or 0.0,
+        options,
+    )
+
+
 class BigQueryBackend(
     IntrospectorBackendMixin, BigQueryQualifiedNameMixin, StorageBackend
 ):
@@ -81,7 +125,7 @@ def connect(self) -> None:
                     client_options=client_options,
                 )
             self._connected = True
-            self.log(logging.INFO, "Connected to BigQuery")
+            self.log(logging.DEBUG, "Connected to BigQuery")
         except Exception as e:
             raise ConnectionError(f"Failed to connect to BigQuery: {e}") from e
 
@@ -299,17 +343,30 @@ def _infer_param_type(value: Any) -> str:
             return 'BYTES'
         return 'STRING'
 
-    def execute(self, sql: str, params: Optional[Tuple] = None, returning: Optional[Any] = None, column_adapters: Optional[Dict] = None, *, options: Optional[Any] = None) -> QueryResult:
+    def execute(
+        self,
+        sql: str,
+        params: Optional[Tuple] = None,
+        returning: Optional[Any] = None,
+        column_adapters: Optional[Dict] = None,
+        *,
+        options: Optional[Any] = None,
+    ) -> QueryResult:
         """
         Execute a SQL statement against BigQuery.
 
         Accepts the base-class style ``options`` (ExecutionOptions) keyword.
         Statement type resolution order: options.stmt_type >
         process_result_set heuristic > SQL prefix detection.
-        """
-        from rhosocial.activerecord.backend.result import QueryResult
-        from rhosocial.activerecord.backend.schema import StatementType
 
+        This backend replaces the core execution pipeline (there is no DB-API
+        cursor), so it owns the two execution log records the core layer
+        defines: one ``DEBUG`` record with the SQL text and redacted
+        parameters, and one ``INFO`` summary.  Both come from
+        :mod:`rhosocial.activerecord.backend.base.execution_logging`; the
+        backend-specific part is only the row count, which BigQuery reports
+        on the job rather than on a cursor.
+        """
         if options is not None:
             column_adapters = column_adapters or getattr(options, 'column_adapters', None)
             column_mapping = getattr(options, 'column_mapping', None)
@@ -320,13 +377,21 @@ def execute(self, sql: str, params: Optional[Tuple] = None, returning: Optional[
         else:
             column_mapping = None
             is_select = sql.strip().upper().startswith(("SELECT", "WITH"))
+        stmt_type = getattr(options, 'stmt_type', None) if options is not None else None
+        if stmt_type is None:
+            stmt_type = StatementType.DQL if is_select else StatementType.OTHER
+        log_execution_start(self, sql, params, options)
         try:
             if self._is_emulator_mode():
                 # Emulator fast path: one ``jobs.query`` REST call per SQL
                 # statement. The full client job lifecycle (submit + poll +
                 # getQueryResults) costs ~6s per statement against the
                 # emulator while a single jobs.query call completes in ~2s.
-                return self._execute_via_rest(sql, params, is_select, column_adapters, column_mapping)
+                result = self._execute_via_rest(
+                    sql, params, is_select, column_adapters, column_mapping
+                )
+                log_result_summary(self, stmt_type, sql, result, options)
+                return result
             if not getattr(self, '_connected', False) or getattr(self, '_client', None) is None:
                 self.connect()
             job_config = self._build_job_config(params)
@@ -359,8 +424,11 @@ def execute(self, sql: str, params: Optional[Tuple] = None, returning: Optional[
                 affected_rows = job.num_dml_affected_rows or 0
                 result_data = None
             duration = (job.ended - job.started).total_seconds() if job.started and job.ended else None
-            return QueryResult(data=result_data, affected_rows=affected_rows, duration=duration)
+            result = QueryResult(data=result_data, affected_rows=affected_rows, duration=duration)
+            log_result_summary(self, stmt_type, sql, result, options)
+            return result
         except Exception as e:
+            log_execution_error(self, e, options)
             self._handle_error(e)
 
     @property
@@ -558,12 +626,17 @@ def _raise_http_error(status: int, message: str):
     # only intended for single-writer workloads.
 
     def _next_pk_value(self, table: str, schema_name: Optional[str], pk_column: str) -> int:
-        from rhosocial.activerecord.backend.options import ExecutionOptions
-        from rhosocial.activerecord.backend.schema import StatementType
-
         key = (schema_name, table, pk_column)
         sql, params = self._next_pk_expression(table, schema_name, pk_column).to_sql()
-        result = self.execute(sql, params, options=ExecutionOptions(stmt_type=StatementType.DQL))
+        options = ExecutionOptions(
+            stmt_type=StatementType.DQL,
+            log_context=LogContext(
+                operation="select",
+                object_name=table,
+                schema=schema_name,
+            ),
+        )
+        result = self.execute(sql, params, options=options)
         row = result.data[0]
         value = row.get("next_id") if isinstance(row, dict) else row[0]
         # Stay monotonic across deletes: MAX(pk)+1 reuses the id of a just
@@ -640,14 +713,12 @@ def bulk_update(self, options) -> QueryResult:
         # rows, and the BigQuery analyzer rejects an untyped NULL CASE result
         # ("Value of type INT64 cannot be assigned to ..."). Reusing the
         # column itself is also the safe semantics.
-        from rhosocial.activerecord.backend.base.operations import is_sql_expression
+        from rhosocial.activerecord.backend.base.operations import execution_log_context
         from rhosocial.activerecord.backend.expression import (
             Column, Literal, UpdateStatement, ComparisonPredicate, InPredicate,
         )
         from rhosocial.activerecord.backend.expression.advanced_functions import CaseExpression
         from rhosocial.activerecord.backend.expression.objects import Table
-        from rhosocial.activerecord.backend.options import ExecutionOptions
-        from rhosocial.activerecord.backend.schema import StatementType
 
         pk_col = Column(self.dialect, options.pk_column)
         assignments = {}
@@ -680,6 +751,7 @@ def bulk_update(self, options) -> QueryResult:
             stmt_type=StatementType.DML,
             column_adapters=options.column_adapters,
             column_mapping=options.column_mapping,
+            log_context=execution_log_context(options, "bulk_update"),
         )
         result = self.execute(sql, params, options=exec_options)
         if options.auto_commit:
@@ -702,11 +774,16 @@ def _is_emulator_mode(self) -> bool:
     def _count_matching(self, table: str, schema_name: Optional[str], where) -> Optional[int]:
         if where is None:
             return None
-        from rhosocial.activerecord.backend.options import ExecutionOptions
-        from rhosocial.activerecord.backend.schema import StatementType
-
         sql, params = self._count_expression(table, schema_name, where).to_sql()
-        result = self.execute(sql, params, options=ExecutionOptions(stmt_type=StatementType.DQL))
+        options = ExecutionOptions(
+            stmt_type=StatementType.DQL,
+            log_context=LogContext(
+                operation="count",
+                object_name=table,
+                schema=schema_name,
+            ),
+        )
+        result = self.execute(sql, params, options=options)
         if not result.data:
             return None
         row = result.data[0]
diff --git a/src/rhosocial/activerecord/backend/impl/bigquery/transaction.py b/src/rhosocial/activerecord/backend/impl/bigquery/transaction.py
index 8743e7b..6e9e01c 100644
--- a/src/rhosocial/activerecord/backend/impl/bigquery/transaction.py
+++ b/src/rhosocial/activerecord/backend/impl/bigquery/transaction.py
@@ -2,6 +2,8 @@
 from contextlib import contextmanager
 from typing import Optional
 
+from rhosocial.activerecord.logging.defaults import get_logger
+
 
 class BigQueryTransactionManager:
     def __init__(self, backend):
@@ -21,8 +23,7 @@ def transaction(self, isolation_level: Optional[str] = None):
             except Exception:
                 # The original exception must propagate; a backend that
                 # rejects ROLLBACK (e.g. the emulator) should not mask it.
-                import logging
-                logging.getLogger(__name__).warning(
+                get_logger("rhosocial.activerecord.transaction").warning(
                     "Failed to roll back BigQuery transaction", exc_info=True
                 )
             raise
diff --git a/tests/rhosocial/activerecord_bigquery_test/feature/basic/conftest.py b/tests/rhosocial/activerecord_bigquery_test/feature/basic/conftest.py
new file mode 100644
index 0000000..d17d075
--- /dev/null
+++ b/tests/rhosocial/activerecord_bigquery_test/feature/basic/conftest.py
@@ -0,0 +1,8 @@
+# tests/rhosocial/activerecord_bigquery_test/feature/basic/conftest.py
+"""
+Pytest configuration for basic feature tests.
+
+This file imports fixtures from the corresponding testsuite, making them
+available to the tests in this directory.
+"""
+from rhosocial.activerecord.testsuite.feature.basic.conftest import *  # noqa: F403
diff --git a/tests/rhosocial/activerecord_bigquery_test/feature/basic/nopk/test_keyless_models.py b/tests/rhosocial/activerecord_bigquery_test/feature/basic/nopk/test_keyless_models.py
new file mode 100644
index 0000000..473a20a
--- /dev/null
+++ b/tests/rhosocial/activerecord_bigquery_test/feature/basic/nopk/test_keyless_models.py
@@ -0,0 +1,4 @@
+# tests/rhosocial/activerecord_bigquery_test/feature/basic/nopk/test_keyless_models.py
+"""Shared contract wrapper for keyless models on BigQuery."""
+from rhosocial.activerecord.testsuite.feature.basic.nopk.test_keyless_models import *  # noqa: F403
+from rhosocial.activerecord.testsuite.feature.basic.nopk.test_keyless_models_async import *  # noqa: F403
diff --git a/tests/rhosocial/activerecord_bigquery_test/feature/basic/readonly/test_readonly_models.py b/tests/rhosocial/activerecord_bigquery_test/feature/basic/readonly/test_readonly_models.py
new file mode 100644
index 0000000..c23b6ee
--- /dev/null
+++ b/tests/rhosocial/activerecord_bigquery_test/feature/basic/readonly/test_readonly_models.py
@@ -0,0 +1,4 @@
+# tests/rhosocial/activerecord_bigquery_test/feature/basic/readonly/test_readonly_models.py
+"""Shared contract wrapper for read-only models on BigQuery."""
+from rhosocial.activerecord.testsuite.feature.basic.readonly.test_readonly_models import *  # noqa: F403
+from rhosocial.activerecord.testsuite.feature.basic.readonly.test_readonly_models_async import *  # noqa: F403
diff --git a/tests/rhosocial/activerecord_bigquery_test/feature/logging/__init__.py b/tests/rhosocial/activerecord_bigquery_test/feature/logging/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/rhosocial/activerecord_bigquery_test/feature/logging/test_execution_logging_unit.py b/tests/rhosocial/activerecord_bigquery_test/feature/logging/test_execution_logging_unit.py
new file mode 100644
index 0000000..dc0339f
--- /dev/null
+++ b/tests/rhosocial/activerecord_bigquery_test/feature/logging/test_execution_logging_unit.py
@@ -0,0 +1,413 @@
+# tests/rhosocial/activerecord_bigquery_test/feature/logging/test_execution_logging_unit.py
+"""DB-free unit tests for BigQuery execution logging.
+
+The tests in this module never construct ``google.cloud.bigquery.Client``, never
+reach the network and never need the emulator: the emulator REST fast path and
+the client job path are replaced with in-process fakes. Both the synchronous and
+the asynchronous backend must report the canonical
+``rhosocial.activerecord.backend.bigquery`` logger, and both execution paths must
+emit the shared DEBUG SQL record and the INFO result summary.
+"""
+
+import logging
+import sys
+import types
+
+import pytest
+
+from rhosocial.activerecord.backend.errors import DatabaseError
+from rhosocial.activerecord.backend.expression import (
+    Column, ComparisonPredicate, Literal,
+)
+from rhosocial.activerecord.backend.impl.bigquery.async_transaction import (
+    AsyncBigQueryTransactionManager,
+)
+from rhosocial.activerecord.backend.impl.bigquery.backend.async_backend import (
+    AsyncBigQueryBackend,
+)
+from rhosocial.activerecord.backend.impl.bigquery.backend.backend import (
+    BigQueryBackend,
+    ExecutionSummaryCursor,
+    log_result_summary,
+)
+from rhosocial.activerecord.backend.impl.bigquery.transaction import (
+    BigQueryTransactionManager,
+)
+from rhosocial.activerecord.backend.options import ExecutionOptions, LogContext
+from rhosocial.activerecord.backend.result import QueryResult
+from rhosocial.activerecord.backend.schema import StatementType
+
+CANONICAL_LOGGER = "rhosocial.activerecord.backend.bigquery"
+TRANSACTION_LOGGER = "rhosocial.activerecord.transaction"
+
+
+class _FakeClient:
+    """Minimal stand-in for the google client used by the non-emulator path."""
+
+    def __init__(self, error=None):
+        self._error = error
+
+    def query(self, sql, job_config=None):
+        if self._error is not None:
+            raise self._error
+        raise AssertionError("these tests do not exercise a successful job path")
+
+
+class _FailingConnection:
+    """Connection whose statements always fail (BEGIN and ROLLBACK)."""
+
+    def __init__(self):
+        self.calls = []
+
+    def execute(self, sql):
+        self.calls.append(sql)
+        raise RuntimeError(f"{sql} failed")
+
+
+class _FailingAsyncConnection:
+    """Async twin of :class:`_FailingConnection`."""
+
+    def __init__(self):
+        self.calls = []
+
+    async def execute(self, sql):
+        self.calls.append(sql)
+        raise RuntimeError(f"{sql} failed")
+
+
+def _fake_rest(result):
+    """Build a replacement for ``BigQueryBackend._execute_via_rest``."""
+
+    def _rest(self, sql, params, is_select, column_adapters, column_mapping):
+        return result
+
+    return _rest
+
+
+def _install_fake_google(monkeypatch):
+    """Install just enough of the google namespace for ``connect()``.
+
+    The real packages are not needed (and in some environments fail to import),
+    so every module imported by ``connect()`` is replaced with a fake.
+    """
+
+    google = types.ModuleType("google")
+    cloud = types.ModuleType("google.cloud")
+    bigquery = types.ModuleType("google.cloud.bigquery")
+    auth = types.ModuleType("google.auth")
+    credentials = types.ModuleType("google.auth.credentials")
+    api_core = types.ModuleType("google.api_core")
+    client_options = types.ModuleType("google.api_core.client_options")
+
+    class _FakeBigQueryClient:
+        def __init__(self, *args, **kwargs):
+            self.init_kwargs = kwargs
+
+    bigquery.Client = _FakeBigQueryClient
+    cloud.bigquery = bigquery
+    google.cloud = cloud
+    credentials.AnonymousCredentials = object
+    auth.credentials = credentials
+    google.auth = auth
+    client_options.ClientOptions = lambda **kwargs: kwargs
+    api_core.client_options = client_options
+    google.api_core = api_core
+
+    for name, module in {
+        "google": google,
+        "google.cloud": cloud,
+        "google.cloud.bigquery": bigquery,
+        "google.auth": auth,
+        "google.auth.credentials": credentials,
+        "google.api_core": api_core,
+        "google.api_core.client_options": client_options,
+    }.items():
+        monkeypatch.setitem(sys.modules, name, module)
+
+
+def _records_with_event(caplog, event):
+    return [
+        record for record in caplog.records
+        if getattr(record, "event", None) == event
+    ]
+
+
+def test_backends_use_the_canonical_logger():
+    backend = BigQueryBackend(project="project", dataset="dataset")
+    assert backend._get_logger_name() == CANONICAL_LOGGER
+    assert backend.logger.name == CANONICAL_LOGGER
+    assert isinstance(backend.transaction_manager, BigQueryTransactionManager)
+
+    async_backend = AsyncBigQueryBackend(project="project", dataset="dataset")
+    assert async_backend._get_logger_name() == CANONICAL_LOGGER
+    assert async_backend.logger.name == CANONICAL_LOGGER
+    assert isinstance(
+        async_backend.transaction_manager, AsyncBigQueryTransactionManager
+    )
+
+    cursor = ExecutionSummaryCursor(3)
+    assert cursor.rowcount == 3
+
+
+def test_log_result_summary_covers_dml_and_dql(caplog):
+    backend = BigQueryBackend(project="project", dataset="dataset")
+
+    dml_options = ExecutionOptions(
+        stmt_type=StatementType.DML,
+        log_context=LogContext(operation="update", object_name="users"),
+    )
+    with caplog.at_level(logging.INFO, logger=CANONICAL_LOGGER):
+        log_result_summary(
+            backend,
+            StatementType.DML,
+            "UPDATE users SET name = ?",
+            QueryResult(data=[{"id": 1}], affected_rows=2, duration=0.01),
+            dml_options,
+        )
+    dml_records = _records_with_event(caplog, "dml.update")
+    assert dml_records
+    assert dml_records[0].activerecord["rows_affected"] == 2
+
+    dql_options = ExecutionOptions(
+        stmt_type=StatementType.DQL,
+        log_context=LogContext(operation="select", object_name="users"),
+    )
+    with caplog.at_level(logging.INFO, logger=CANONICAL_LOGGER):
+        log_result_summary(
+            backend,
+            StatementType.DQL,
+            "SELECT * FROM users",
+            QueryResult(data=[{"id": 1}, {"id": 2}], affected_rows=2, duration=0.02),
+            dql_options,
+        )
+    dql_records = _records_with_event(caplog, "dql.select")
+    assert dql_records
+    assert dql_records[0].activerecord["rows_returned"] == 2
+
+
+def test_sync_execute_emulator_path_logs_start_and_summary(caplog, monkeypatch):
+    backend = BigQueryBackend(
+        project="project", dataset="dataset", api_endpoint="http://emulator"
+    )
+    result = QueryResult(data=[{"id": 1}], affected_rows=1, duration=0.01)
+    monkeypatch.setattr(BigQueryBackend, "_execute_via_rest", _fake_rest(result))
+
+    with caplog.at_level(logging.DEBUG, logger=CANONICAL_LOGGER):
+        returned = backend.execute("SELECT 1")
+
+    assert returned is result
+    assert any(
+        record.levelno == logging.DEBUG
+        and "Executing SQL: SELECT 1" in record.getMessage()
+        for record in caplog.records
+    )
+    summary = _records_with_event(caplog, "dql.select")
+    assert summary
+    assert summary[0].activerecord["rows_returned"] == 1
+
+
+def test_sync_execute_error_logs_before_reraise(caplog, monkeypatch):
+    backend = BigQueryBackend(project="project", dataset="dataset")
+    backend._connected = True
+    backend._client = _FakeClient(ValueError("boom"))
+    monkeypatch.setattr(
+        BigQueryBackend, "_classify_error", lambda self, error: "generic"
+    )
+    options = ExecutionOptions(
+        stmt_type=StatementType.DML,
+        log_context=LogContext(operation="update", object_name="users"),
+    )
+
+    with caplog.at_level(logging.DEBUG, logger=CANONICAL_LOGGER):
+        with pytest.raises(DatabaseError, match="boom"):
+            backend.execute("UPDATE users SET name = ?", None, options=options)
+
+    failures = [
+        record for record in caplog.records
+        if record.getMessage().startswith("Query failed")
+    ]
+    assert failures
+    assert failures[0].activerecord["error_type"] == "ValueError"
+
+
+def test_sync_next_pk_value_uses_log_context(monkeypatch):
+    backend = BigQueryBackend(project="project", dataset="dataset")
+    captured = {}
+
+    def fake_execute(sql, params=None, options=None, **kwargs):
+        captured["sql"] = sql
+        captured["options"] = options
+        return QueryResult(data=[{"next_id": 3}], affected_rows=1)
+
+    monkeypatch.setattr(backend, "execute", fake_execute)
+
+    assert backend._next_pk_value("users", "dataset", "id") == 3
+    context = captured["options"].log_context
+    assert context.operation == "select"
+    assert context.object_name == "users"
+    assert context.schema == "dataset"
+    assert "MAX" in captured["sql"].upper()
+
+
+def test_sync_count_matching_uses_log_context(monkeypatch):
+    backend = BigQueryBackend(project="project", dataset="dataset")
+    captured = {}
+
+    def fake_execute(sql, params=None, options=None, **kwargs):
+        captured["options"] = options
+        return QueryResult(data=[{"n": 3}], affected_rows=1)
+
+    monkeypatch.setattr(backend, "execute", fake_execute)
+
+    dialect = backend.dialect
+    where = ComparisonPredicate(
+        dialect, "=", Column(dialect, "id"), Literal(dialect, 7)
+    )
+    assert backend._count_matching("users", "dataset", where) == 3
+    assert captured["options"].log_context.operation == "count"
+    assert backend._count_matching("users", "dataset", None) is None
+
+
+def test_sync_connect_logs_without_google_client(caplog, monkeypatch):
+    _install_fake_google(monkeypatch)
+    backend = BigQueryBackend(project="project", dataset="dataset")
+
+    with caplog.at_level(logging.DEBUG, logger=CANONICAL_LOGGER):
+        backend.connect()
+
+    assert backend._connected is True
+    assert backend._client is not None
+    assert any(
+        record.getMessage() == "Connected to BigQuery"
+        for record in caplog.records
+    )
+
+
+@pytest.mark.asyncio
+async def test_async_execute_emulator_path_logs_start_and_summary(caplog, monkeypatch):
+    backend = AsyncBigQueryBackend(
+        project="project", dataset="dataset", api_endpoint="http://emulator"
+    )
+    result = QueryResult(data=[{"id": 1}], affected_rows=1, duration=0.01)
+    monkeypatch.setattr(BigQueryBackend, "_execute_via_rest", _fake_rest(result))
+
+    with caplog.at_level(logging.DEBUG, logger=CANONICAL_LOGGER):
+        returned = await backend.execute("SELECT 1")
+
+    assert returned is result
+    assert any(
+        record.levelno == logging.DEBUG
+        and "Executing SQL: SELECT 1" in record.getMessage()
+        for record in caplog.records
+    )
+    summary = _records_with_event(caplog, "dql.select")
+    assert summary
+    assert summary[0].activerecord["rows_returned"] == 1
+
+
+@pytest.mark.asyncio
+async def test_async_execute_error_logs_before_reraise(caplog, monkeypatch):
+    backend = AsyncBigQueryBackend(project="project", dataset="dataset")
+    backend._connected = True
+    backend._client = _FakeClient(ValueError("boom"))
+    monkeypatch.setattr(
+        AsyncBigQueryBackend, "_classify_error", lambda self, error: "generic"
+    )
+    options = ExecutionOptions(
+        stmt_type=StatementType.DML,
+        log_context=LogContext(operation="update", object_name="users"),
+    )
+
+    with caplog.at_level(logging.DEBUG, logger=CANONICAL_LOGGER):
+        with pytest.raises(DatabaseError, match="boom"):
+            await backend.execute("UPDATE users SET name = ?", None, options=options)
+
+    failures = [
+        record for record in caplog.records
+        if record.getMessage().startswith("Query failed")
+    ]
+    assert failures
+    assert failures[0].activerecord["error_type"] == "ValueError"
+
+
+@pytest.mark.asyncio
+async def test_async_next_pk_value_uses_log_context(monkeypatch):
+    backend = AsyncBigQueryBackend(project="project", dataset="dataset")
+    captured = {}
+
+    async def fake_execute(sql, params=None, options=None, **kwargs):
+        captured["options"] = options
+        return QueryResult(data=[{"next_id": 4}], affected_rows=1)
+
+    monkeypatch.setattr(backend, "execute", fake_execute)
+
+    assert await backend._next_pk_value("users", "dataset", "id") == 4
+    context = captured["options"].log_context
+    assert context.operation == "select"
+    assert context.object_name == "users"
+    assert context.schema == "dataset"
+
+
+@pytest.mark.asyncio
+async def test_async_count_matching_uses_log_context(monkeypatch):
+    backend = AsyncBigQueryBackend(project="project", dataset="dataset")
+    captured = {}
+
+    async def fake_execute(sql, params=None, options=None, **kwargs):
+        captured["options"] = options
+        return QueryResult(data=[{"n": 5}], affected_rows=1)
+
+    monkeypatch.setattr(backend, "execute", fake_execute)
+
+    dialect = backend.dialect
+    where = ComparisonPredicate(
+        dialect, "=", Column(dialect, "id"), Literal(dialect, 9)
+    )
+    assert await backend._count_matching("users", "dataset", where) == 5
+    assert captured["options"].log_context.operation == "count"
+    assert await backend._count_matching("users", "dataset", None) is None
+
+
+@pytest.mark.asyncio
+async def test_async_connect_logs_without_google_client(caplog, monkeypatch):
+    _install_fake_google(monkeypatch)
+    backend = AsyncBigQueryBackend(project="project", dataset="dataset")
+
+    with caplog.at_level(logging.DEBUG, logger=CANONICAL_LOGGER):
+        await backend.connect()
+
+    assert backend._connected is True
+    assert backend._client is not None
+    assert any(
+        record.getMessage() == "Connected to BigQuery"
+        for record in caplog.records
+    )
+
+
+def test_sync_transaction_logs_rollback_failure(caplog):
+    manager = BigQueryTransactionManager(_FailingConnection())
+
+    with caplog.at_level(logging.WARNING, logger=TRANSACTION_LOGGER):
+        with pytest.raises(RuntimeError, match="BEGIN TRANSACTION failed"):
+            with manager.transaction():
+                pass
+
+    assert any(
+        "Failed to roll back BigQuery transaction" in record.getMessage()
+        for record in caplog.records
+    )
+
+
+@pytest.mark.asyncio
+async def test_async_transaction_logs_rollback_failure(caplog):
+    manager = AsyncBigQueryTransactionManager(_FailingAsyncConnection())
+
+    with caplog.at_level(logging.WARNING, logger=TRANSACTION_LOGGER):
+        with pytest.raises(RuntimeError, match="BEGIN TRANSACTION failed"):
+            async with manager.transaction():
+                pass
+
+    assert any(
+        "Failed to roll back BigQuery transaction" in record.getMessage()
+        for record in caplog.records
+    )
diff --git a/tests/rhosocial/activerecord_bigquery_test/feature/query/basic/test_find_one_or_create.py b/tests/rhosocial/activerecord_bigquery_test/feature/query/basic/test_find_one_or_create.py
new file mode 100644
index 0000000..0e0d5c9
--- /dev/null
+++ b/tests/rhosocial/activerecord_bigquery_test/feature/query/basic/test_find_one_or_create.py
@@ -0,0 +1,4 @@
+# tests/rhosocial/activerecord_bigquery_test/feature/query/basic/test_find_one_or_create.py
+"""Shared contract wrapper for find_one_or_create on BigQuery."""
+from rhosocial.activerecord.testsuite.feature.query.basic.test_find_one_or_create import *  # noqa: F403
+from rhosocial.activerecord.testsuite.feature.query.basic.test_find_one_or_create_async import *  # noqa: F403
diff --git a/tests/rhosocial/activerecord_bigquery_test/feature/relation/conftest.py b/tests/rhosocial/activerecord_bigquery_test/feature/relation/conftest.py
new file mode 100644
index 0000000..a6583d2
--- /dev/null
+++ b/tests/rhosocial/activerecord_bigquery_test/feature/relation/conftest.py
@@ -0,0 +1,8 @@
+# tests/rhosocial/activerecord_bigquery_test/feature/relation/conftest.py
+"""
+Pytest configuration for relation feature tests.
+
+This file imports fixtures from the corresponding testsuite, making them
+available to the tests in this directory.
+"""
+from rhosocial.activerecord.testsuite.feature.relation.conftest import *  # noqa: F403
diff --git a/tests/rhosocial/activerecord_bigquery_test/feature/relation/ordering/test_has_many_order_by.py b/tests/rhosocial/activerecord_bigquery_test/feature/relation/ordering/test_has_many_order_by.py
new file mode 100644
index 0000000..2d7c973
--- /dev/null
+++ b/tests/rhosocial/activerecord_bigquery_test/feature/relation/ordering/test_has_many_order_by.py
@@ -0,0 +1,4 @@
+# tests/rhosocial/activerecord_bigquery_test/feature/relation/ordering/test_has_many_order_by.py
+"""Shared contract wrapper for HasMany(order_by=...) on BigQuery."""
+from rhosocial.activerecord.testsuite.feature.relation.ordering.test_has_many_order_by import *  # noqa: F403
+from rhosocial.activerecord.testsuite.feature.relation.ordering.test_has_many_order_by_async import *  # noqa: F403