Fix transaction_primitives on Snowflake and BigQuery

benchbox/cli/orchestrator.py

@@ -151,10 +151,14 @@ def _get_benchmark_instance(self, config: BenchmarkConfig, system_profile):
         # the get_data_source_benchmark() instance method (no
         # DATA_SOURCE_BENCHMARK class attribute) were not resolved at
         # construction, so redirect them to the shared root now. Class-attr
-        # sharers were already constructed with it.
+        # sharers were already constructed with it. Benchmarks that generate
+        # their own output (e.g. tpcds_obt, which transforms source data into
+        # a separate OBT table) must keep their own output_dir: redirecting
+        # would hide their generated files from the loader.
         if getattr(benchmark_class, "DATA_SOURCE_BENCHMARK", None) is None:
             data_source = getattr(benchmark_instance, "get_data_source_benchmark", lambda: None)()
-            if data_source and self.custom_output_dir is None:
+            generates_own_output = getattr(benchmark_instance, "GENERATES_OWN_OUTPUT", False)
+            if data_source and self.custom_output_dir is None and not generates_own_output:
                 shared_path = self.directory_manager.get_datagen_path(data_source.lower(), config.scale_factor)
                 benchmark_instance.output_dir = shared_path
 

benchbox/core/tpcds_obt/benchmark.py

@@ -127,6 +127,11 @@ class TPCDSOBTBenchmark(BaseBenchmark):
     while OBT-specific transformations are stored separately.
     """
 
+    # OBT generates its own output table from TPC-DS source data, so the CLI
+    # must not redirect its output_dir to the shared TPC-DS root: that would
+    # hide the generated OBT parquet from the cloud loader.
+    GENERATES_OWN_OUTPUT = True
+
     def __init__(
         self,
         scale_factor: float = 1.0,

benchbox/core/transaction_primitives/benchmark.py

@@ -29,6 +29,10 @@
 from benchbox.core.transaction_primitives.operations import TransactionOperationsManager
 from benchbox.core.transaction_primitives.schema import STAGING_TABLES, get_all_staging_tables_sql, get_create_table_sql
 from benchbox.core.transactional.benchmark_base import TransactionalBenchmarkBase
+from benchbox.sql_compat.rules.execution_filter.cloud_transaction_primitives import (
+    BIGQUERY_TRANSACTION_PRIMITIVES_OPERATION_SKIPS,
+    SNOWFLAKE_TRANSACTION_PRIMITIVES_OPERATION_SKIPS,
+)
 from benchbox.sql_compat.rules.execution_filter.databricks_transaction_primitives import (
     DATABRICKS_TRANSACTION_PRIMITIVES_OPERATION_SKIPS,
 )
@@ -708,8 +712,10 @@ def execute_operation(
 
         try:
             platform_operation_skips = {
+                "bigquery": BIGQUERY_TRANSACTION_PRIMITIVES_OPERATION_SKIPS,
                 "databricks": DATABRICKS_TRANSACTION_PRIMITIVES_OPERATION_SKIPS,
                 "duckdb": DUCKDB_TRANSACTION_PRIMITIVES_OPERATION_SKIPS,
+                "snowflake": SNOWFLAKE_TRANSACTION_PRIMITIVES_OPERATION_SKIPS,
                 "pg_duckdb": PG_DUCKDB_TRANSACTION_PRIMITIVES_OPERATION_SKIPS,
                 "timescaledb": TIMESCALEDB_TRANSACTION_PRIMITIVES_OPERATION_SKIPS,
             }

benchbox/core/transaction_primitives/catalog/operations.yaml

@@ -235,6 +235,9 @@ operations:
     expected_rows: 1
   cleanup_sql: "DELETE FROM txn_orders WHERE o_orderkey BETWEEN 9990001 AND 9990010\n"
   expected_rows_affected: 10
+  platform_overrides:
+    snowflake: "BEGIN TRANSACTION;\nINSERT INTO txn_orders\nWITH order_data AS (\n  SELECT 9990000 + n AS o_orderkey,\n         1 AS o_custkey,\n         'O' AS o_orderstatus,\n         1000.0 * n AS o_totalprice,\n         DATE '1998-01-01' AS o_orderdate,\n         '5-LOW' AS o_orderpriority,\n         'Clerk#000000001' AS o_clerk,\n         0 AS o_shippriority,\n         'cte_test' AS o_comment\n  FROM (SELECT unnest(generate_series(1, 10)) AS n) t\n)\nSELECT * FROM order_data;\nUPDATE txn_orders SET o_totalprice = o_totalprice * 2 WHERE o_orderkey BETWEEN 9990001 AND 9990010;\nCOMMIT;\n"
+    bigquery: "BEGIN TRANSACTION;\nINSERT INTO txn_orders\nWITH order_data AS (\n  SELECT 9990000 + n AS o_orderkey,\n         1 AS o_custkey,\n         'O' AS o_orderstatus,\n         1000.0 * n AS o_totalprice,\n         DATE '1998-01-01' AS o_orderdate,\n         '5-LOW' AS o_orderpriority,\n         'Clerk#000000001' AS o_clerk,\n         0 AS o_shippriority,\n         'cte_test' AS o_comment\n  FROM (SELECT unnest(generate_series(1, 10)) AS n) t\n)\nSELECT * FROM order_data;\nUPDATE txn_orders SET o_totalprice = o_totalprice * 2 WHERE o_orderkey BETWEEN 9990001 AND 9990010;\nCOMMIT;\n"
 - id: transaction_nested_subquery_updates
   category: advanced
   description: Tests UPDATE with nested subqueries within transaction

benchbox/core/transactional/benchmark_base.py

@@ -32,6 +32,18 @@
 # SQL ROLLBACK is needed to close a transaction opened by catalog BEGIN SQL.
 _SQL_ROLLBACK_AFTER_ERROR_DIALECTS = frozenset({"databricks"})
 _UNNEST_GENERATE_SERIES_RE = re.compile(r"unnest\(\s*generate_series\((?P<args>[^()]*)\)\s*\)", re.IGNORECASE)
+# The catalog writes integer series as ``(SELECT unnest(generate_series(a, b)) AS n) t``.
+# Snowflake and BigQuery have neither function (verified live: "Unknown functions
+# GENERATE_SERIES, UNNEST" and "Unexpected keyword UNNEST"); rewrite the whole
+# derived-table body into each engine's series generator.
+_UNNEST_SERIES_SELECT_RE = re.compile(
+    r"SELECT\s+unnest\(\s*generate_series\(\s*(?P<lo>[^,()]+?)\s*,\s*(?P<hi>[^,()]+?)\s*\)\s*\)\s+AS\s+(?P<alias>\w+)",
+    re.IGNORECASE,
+)
+_SERIES_SELECT_TEMPLATES = {
+    "snowflake": "SELECT value::INT AS {alias} FROM TABLE(FLATTEN(ARRAY_GENERATE_RANGE({lo}, {hi} + 1)))",
+    "bigquery": "SELECT {alias} FROM UNNEST(GENERATE_ARRAY({lo}, {hi})) AS {alias}",
+}
 _SET_THEN_BEGIN_ISOLATION_RE = re.compile(
     r"SET\s+TRANSACTION\s+ISOLATION\s+LEVEL\s+(?P<level>REPEATABLE\s+READ|SERIALIZABLE|READ\s+COMMITTED)\s*;\s*"
     r"BEGIN\s+TRANSACTION\s*;",
@@ -337,7 +349,13 @@ def _lookup_platform_override(
 
     def _rewrite_transactional_sql_for_platform(self, sql: str, platform_key: str | None) -> str:
         """Apply narrow transactional catalog rewrites for the active SQL dialect."""
-        if (platform_key or "").lower() not in _POSTGRES_SERIES_DIALECTS:
+        dialect = (platform_key or "").lower()
+        template = _SERIES_SELECT_TEMPLATES.get(dialect)
+        if template is not None:
+            return _UNNEST_SERIES_SELECT_RE.sub(
+                lambda m: template.format(lo=m.group("lo"), hi=m.group("hi"), alias=m.group("alias")), sql
+            )
+        if dialect not in _POSTGRES_SERIES_DIALECTS:
             return sql
         sql = _UNNEST_GENERATE_SERIES_RE.sub(r"generate_series(\g<args>)", sql)
         return _SET_THEN_BEGIN_ISOLATION_RE.sub(r"BEGIN TRANSACTION ISOLATION LEVEL \g<level>;", sql)

benchbox/platforms/bigquery.py

@@ -695,6 +695,7 @@ def preprocess_operation_sql(self, query_id: str, operation: Any) -> str | None:
             flags=re.IGNORECASE,
         )
         rewritten = re.sub(r"\bINTERVAL\s+'(\d+)'\s+([A-Za-z]+)", r"INTERVAL \1 \2", rewritten)
+        rewritten = self._numeric_decimal_literals(rewritten)
         statements = split_sql_statements(rewritten)
         parts = []
         changed = False
@@ -715,6 +716,25 @@ def preprocess_operation_sql(self, query_id: str, operation: Any) -> str | None:
             rewritten = rewritten[: bare_insert.start()] + bare_insert.group(1) + "INSERT ROW" + bare_insert.group(2)
         return rewritten
 
+    @staticmethod
+    def _numeric_decimal_literals(sql: str) -> str:
+        """Type bare decimal literals in operation SQL as NUMERIC.
+
+        BigQuery types ``1000.0`` as FLOAT64 and refuses to insert FLOAT64 into
+        the NUMERIC money and discount columns of the TPC-H staging tables
+        (verified live: "Query column 4 has type FLOAT64 which cannot be
+        inserted into column o_totalprice, which has type NUMERIC"). Every
+        decimal literal in the write catalogs targets those columns, so
+        ``NUMERIC '1000.0'`` keeps the arithmetic exact and the insert legal.
+        String literals and identifiers are left untouched.
+        """
+        import re
+
+        parts = re.split(r"('(?:[^'\\]|\\.|'')*')", sql)
+        for index in range(0, len(parts), 2):
+            parts[index] = re.sub(r"(?<![\w.'])(\d+\.\d+)(?![\w.])", r"NUMERIC '\1'", parts[index])
+        return "".join(parts)
+
     def _get_connection_params(self, **connection_config) -> dict[str, Any]:
         """Get standardized connection parameters."""
         return {
@@ -2307,6 +2327,8 @@ def _qualify_single_statement(self, statement: str, allow_fallback: bool = True)
             if not allow_fallback:
                 return statement
             table_names = list(self._FALLBACK_QUALIFY_TABLES)
+        if self._batch_temp_tables:
+            table_names = [name for name in table_names if name not in self._batch_temp_tables]
 
         # Blank string literals and comments length-preservingly so matches
         # found in the masked copy align with the original statement.
@@ -2403,7 +2425,27 @@ def _qualify_table_names(self, query: str, allow_fallback: bool = True) -> str:
         if not statements:
             return query
 
-        qualified_parts = [self._qualify_single_statement(stmt, allow_fallback=allow_fallback) for stmt in statements]
+        # Script-scoped temporary tables live outside the dataset, so their
+        # names must stay bare everywhere in the batch (verified live: a
+        # qualified temp-table name fails "Not found: Table ...TEMP_...").
+        import re
+
+        temp_tables = {
+            m.group(1).upper()
+            for stmt in statements
+            for m in re.finditer(
+                r"(?i)\bCREATE\s+(?:OR\s+REPLACE\s+)?TEMP(?:ORARY)?\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?`?(\w+)`?",
+                stmt,
+            )
+        }
+        previous_temp_tables = self._batch_temp_tables
+        self._batch_temp_tables = temp_tables
+        try:
+            qualified_parts = [
+                self._qualify_single_statement(stmt, allow_fallback=allow_fallback) for stmt in statements
+            ]
+        finally:
+            self._batch_temp_tables = previous_temp_tables
         result = ";\n".join(qualified_parts)
         if query.rstrip().endswith(";"):
             result += ";"
@@ -2412,6 +2454,8 @@ def _qualify_table_names(self, query: str, allow_fallback: bool = True) -> str:
         return result
 
     _FROM_SCAN_TOKEN = None
+    # Temp-table names created in the batch currently being qualified.
+    _batch_temp_tables: frozenset[str] | set[str] = frozenset()
 
     @staticmethod
     def _from_scan_token():

benchbox/platforms/databricks/adapter.py

@@ -1972,7 +1972,7 @@ def load_data(
                         existing_tables,
                         data_source,
                     )
-                    table_stats[table_name.upper()] = row_count
+                    table_stats[table_name.lower()] = row_count
                     load_time = elapsed_seconds(load_start)
 
                     per_table_timings[table_name.upper()] = {
@@ -1986,7 +1986,7 @@ def load_data(
 
                 except Exception as e:
                     self.logger.error(f"Failed to load {table_name}: {str(e)[:200]}")
-                    table_stats[table_name.upper()] = 0
+                    table_stats[table_name.lower()] = 0
                     per_table_timings[table_name.upper()] = {
                         "copy_into_ms": 0,
                         "optimize_ms": 0,
@@ -2529,7 +2529,7 @@ def create_external_tables(
                 cursor.execute(f"CREATE TABLE {table_name_upper} USING PARQUET LOCATION '{location}'")
                 cursor.execute(f"SELECT COUNT(*) FROM {table_name_upper}")
                 result = cursor.fetchone()
-                table_stats[table_name_upper] = int(result[0]) if result else 0
+                table_stats[table_name_lower] = int(result[0]) if result else 0
 
         finally:
             cursor.close()

benchbox/sql_compat/rules/execution_filter/cloud_transaction_primitives.py

@@ -0,0 +1,57 @@
+"""Snowflake and BigQuery execution-filter rules for transaction_primitives operation gaps."""
+
+from __future__ import annotations
+
+from benchbox.sql_compat.actions import CompatAction
+from benchbox.sql_compat.context import Phase
+from benchbox.sql_compat.decision import CompatibilityDecision, FailureMode, SkipQueryPayload, SupportLevel
+from benchbox.sql_compat.registry import REGISTRY
+
+_SAVEPOINT = "{engine} has no SAVEPOINT statement."
+_ISOLATION = "{engine} rejects SET TRANSACTION ISOLATION LEVEL."
+_TEMP_TABLE = (
+    "BigQuery runs each multi-statement script as chained single-statement jobs, "
+    "so a TEMP TABLE created in one statement is invisible to the next."
+)
+
+_OPERATIONS = {
+    "transaction_savepoint_nested": _SAVEPOINT,
+    "transaction_savepoint_deep_nesting": _SAVEPOINT,
+    "transaction_isolation_read_committed": _ISOLATION,
+    "transaction_isolation_repeatable_read": _ISOLATION,
+    "transaction_isolation_serializable": _ISOLATION,
+}
+
+_BIGQUERY_ONLY_OPERATIONS = {
+    "transaction_create_temp_table": _TEMP_TABLE,
+}
+
+SNOWFLAKE_TRANSACTION_PRIMITIVES_OPERATION_SKIPS = {
+    query_id: reason.format(engine="Snowflake") for query_id, reason in _OPERATIONS.items()
+}
+BIGQUERY_TRANSACTION_PRIMITIVES_OPERATION_SKIPS = {
+    query_id: reason.format(engine="BigQuery") for query_id, reason in _OPERATIONS.items()
+} | dict(_BIGQUERY_ONLY_OPERATIONS)
+
+for _platform, _skips in (
+    ("snowflake", SNOWFLAKE_TRANSACTION_PRIMITIVES_OPERATION_SKIPS),
+    ("bigquery", BIGQUERY_TRANSACTION_PRIMITIVES_OPERATION_SKIPS),
+):
+    for _query_id, _reason in _skips.items():
+        REGISTRY.register(
+            CompatibilityDecision(
+                rule_id=f"execution_filter.{_platform}.transaction_primitives.{_query_id}",
+                action=CompatAction.SKIP_QUERY,
+                support_level=SupportLevel.SKIPPED_QUERY,
+                failure_mode=FailureMode.UNSUPPORTED_FEATURE,
+                payload=SkipQueryPayload(
+                    reason=f"{_reason} Evidence: live {_platform} probe on 2026-09-26.",
+                    query_id=_query_id,
+                ),
+                reason=_reason,
+            ),
+            Phase.EXECUTION_FILTER,
+            _platform,
+            benchmark="transaction_primitives",
+            query_id=_query_id,
+        )

docs/compat/capability-matrix.md

@@ -4,7 +4,7 @@
 
 Every rule registered in `benchbox.sql_compat` is listed below. The registry is the authoritative source of compatibility policy; this document is regenerated from it. See [adr-sql-compat-phase-aware-pipeline.md](../development/adr/adr-sql-compat-phase-aware-pipeline.md) for the design.
 
-**Total registered rules:** 545
+**Total registered rules:** 556
 
 **Platforms covered:** 35
 
@@ -13,7 +13,7 @@ Every rule registered in `benchbox.sql_compat` is listed below. The registry is
 | platform | benchmark_gate | query_source | query_adapter | schema_emit | ddl_optimize | execution_filter | total |
 |---|---|---|---|---|---|---|---|
 | athena | - | - | - | - | 1 | - | 1 |
-| bigquery | - | 2 | 25 | 2 | 1 | 2 | 32 |
+| bigquery | - | 2 | 25 | 2 | 1 | 8 | 38 |
 | clickhouse | - | 19 | 3 | 4 | 1 | - | 27 |
 | clickhouse-cloud | - | - | 2 | - | - | 21 | 23 |
 | clickhouse-local | 1 | - | 2 | - | - | 21 | 24 |
@@ -37,7 +37,7 @@ Every rule registered in `benchbox.sql_compat` is listed below. The registry is
 | questdb | 1 | - | - | - | 1 | - | 2 |
 | redshift | - | - | - | 2 | 1 | - | 3 |
 | singlestore | - | - | - | - | 4 | - | 4 |
-| snowflake | - | 7 | 2 | 2 | 1 | 2 | 14 |
+| snowflake | - | 7 | 2 | 2 | 1 | 7 | 19 |
 | spark | - | 6 | 11 | 2 | 1 | - | 20 |
 | sqlite | - | 1 | - | - | - | - | 1 |
 | starrocks | - | 15 | 1 | 2 | 1 | - | 19 |
@@ -92,6 +92,12 @@ Every rule registered in `benchbox.sql_compat` is listed below. The registry is
 | ddl_optimize | platform-wide | rewrite_ddl | REWRITTEN | SYNTAX_ERROR | `ddl_optimize.bigquery.all.convert_to_bigquery_table` |
 | execution_filter | benchmark=tpchavoc, query=1_v7 | skip_query | SKIPPED_QUERY | UNSUPPORTED_FEATURE | `execution_filter.bigquery.tpchavoc.1_v7` |
 | execution_filter | benchmark=tpchavoc, query=2_v2 | skip_query | SKIPPED_QUERY | UNSUPPORTED_FEATURE | `execution_filter.bigquery.tpchavoc.2_v2` |
+| execution_filter | benchmark=transaction_primitives, query=transaction_create_temp_table | skip_query | SKIPPED_QUERY | UNSUPPORTED_FEATURE | `execution_filter.bigquery.transaction_primitives.transaction_create_temp_table` |
+| execution_filter | benchmark=transaction_primitives, query=transaction_isolation_read_committed | skip_query | SKIPPED_QUERY | UNSUPPORTED_FEATURE | `execution_filter.bigquery.transaction_primitives.transaction_isolation_read_committed` |
+| execution_filter | benchmark=transaction_primitives, query=transaction_isolation_repeatable_read | skip_query | SKIPPED_QUERY | UNSUPPORTED_FEATURE | `execution_filter.bigquery.transaction_primitives.transaction_isolation_repeatable_read` |
+| execution_filter | benchmark=transaction_primitives, query=transaction_isolation_serializable | skip_query | SKIPPED_QUERY | UNSUPPORTED_FEATURE | `execution_filter.bigquery.transaction_primitives.transaction_isolation_serializable` |
+| execution_filter | benchmark=transaction_primitives, query=transaction_savepoint_deep_nesting | skip_query | SKIPPED_QUERY | UNSUPPORTED_FEATURE | `execution_filter.bigquery.transaction_primitives.transaction_savepoint_deep_nesting` |
+| execution_filter | benchmark=transaction_primitives, query=transaction_savepoint_nested | skip_query | SKIPPED_QUERY | UNSUPPORTED_FEATURE | `execution_filter.bigquery.transaction_primitives.transaction_savepoint_nested` |
 
 ### clickhouse
 
@@ -641,6 +647,11 @@ Every rule registered in `benchbox.sql_compat` is listed below. The registry is
 | ddl_optimize | platform-wide | rewrite_ddl | REWRITTEN | SYNTAX_ERROR | `ddl_optimize.snowflake.all.optimize_table_definition` |
 | execution_filter | benchmark=tpchavoc, query=1_v7 | skip_query | SKIPPED_QUERY | UNSUPPORTED_FEATURE | `execution_filter.snowflake.tpchavoc.1_v7` |
 | execution_filter | benchmark=tpchavoc, query=2_v2 | skip_query | SKIPPED_QUERY | UNSUPPORTED_FEATURE | `execution_filter.snowflake.tpchavoc.2_v2` |
+| execution_filter | benchmark=transaction_primitives, query=transaction_isolation_read_committed | skip_query | SKIPPED_QUERY | UNSUPPORTED_FEATURE | `execution_filter.snowflake.transaction_primitives.transaction_isolation_read_committed` |
+| execution_filter | benchmark=transaction_primitives, query=transaction_isolation_repeatable_read | skip_query | SKIPPED_QUERY | UNSUPPORTED_FEATURE | `execution_filter.snowflake.transaction_primitives.transaction_isolation_repeatable_read` |
+| execution_filter | benchmark=transaction_primitives, query=transaction_isolation_serializable | skip_query | SKIPPED_QUERY | UNSUPPORTED_FEATURE | `execution_filter.snowflake.transaction_primitives.transaction_isolation_serializable` |
+| execution_filter | benchmark=transaction_primitives, query=transaction_savepoint_deep_nesting | skip_query | SKIPPED_QUERY | UNSUPPORTED_FEATURE | `execution_filter.snowflake.transaction_primitives.transaction_savepoint_deep_nesting` |
+| execution_filter | benchmark=transaction_primitives, query=transaction_savepoint_nested | skip_query | SKIPPED_QUERY | UNSUPPORTED_FEATURE | `execution_filter.snowflake.transaction_primitives.transaction_savepoint_nested` |
 
 ### spark