From 93d263fd2e2435bca27a98fa23ce161b5b9dd6e6 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 11 Jul 2026 19:37:34 +0000 Subject: [PATCH 1/7] fix(test): narrow archive-skip exception + correct non-ASCII match (#1128 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit archive_dir's broad `except Exception` converted a checksum/manifest integrity failure into a silent pytest.skip, defeating the point of a suite that exists to catch corrupted canonical data. Narrowed to DownloadError/ImportError/OSError (true availability failures); a ChecksumMismatchError now propagates and fails the test. DuckDB's `~` operator is regexp_full_match, not a contains-match, so the UTF-8 fidelity check only selected rows consisting entirely of one non-ASCII character - a manual DuckDB check confirmed it matched zero rows even on data containing "Amélie". Switched to regexp_matches() and recomputed the pinned UTF8_FIDELITY counts/hashes against the real archive with the corrected query (title.title: 5->155075 rows, etc.). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01SKw2UhgHuzYPo3y5MfarD6 --- .../test_joinorder_full_archive.py | 54 +++++++++++++++++-- 1 file changed, 49 insertions(+), 5 deletions(-) diff --git a/tests/integration/test_joinorder_full_archive.py b/tests/integration/test_joinorder_full_archive.py index 0a1ddceaa..2c6f1cdd9 100644 --- a/tests/integration/test_joinorder_full_archive.py +++ b/tests/integration/test_joinorder_full_archive.py @@ -70,10 +70,16 @@ # (count of rows containing a non-ASCII byte, sha256 over the ordered # id\x1fvalue\x1e stream of those rows). A latin1 misread, truncation, or # re-encode would change the count or the hash. +# +# #1128 review: these were originally pinned against a buggy WHERE clause +# (`column ~ '[^\x00-\x7F]'` - DuckDB's `~` is regexp_full_match, so it only +# matched rows consisting ENTIRELY of one non-ASCII character), which +# selected a near-empty set. Recomputed against the corrected +# regexp_matches() contains-match query below. UTF8_FIDELITY: dict[tuple[str, str], tuple[int, str]] = { - ("title", "title"): (5, "99cc1eb7d57881fbf7966886718c359570ce85045ab759912e9fa278de582bc9"), - ("char_name", "name"): (4, "73f997632839dcd98adff8b5fe89785c255cb6a450e63589eb8d5b2e9f0a6389"), - ("aka_title", "title"): (2, "7f140d945dc9b7e0a759bf9131e628f23467decdcf82750060a616d1b2afe28f"), + ("title", "title"): (155075, "b615f169051a3c855d0f3209134f9c70eff2e789e4c58efd7da3aa86d554e8e9"), + ("char_name", "name"): (191219, "49c688de8cf3a7069a8d29d17af1ee55c814bd4f7c3f726d4795b42cfd674d94"), + ("aka_title", "title"): (41318, "19a3c9634ba4c99ae6480122feb5256965bdc19d95a64fcb38bae5b7f92a0d6a"), } @@ -114,15 +120,52 @@ def archive_dir() -> Path: Reuses a locally present/cached archive (no re-download); skips when the archive cannot be materialized (offline, missing optional deps, etc.). """ + from benchbox.core.data_fetch.errors import DownloadError from benchbox.core.joinorder.benchmark import JoinOrderBenchmark try: paths = JoinOrderBenchmark().generate_data() - except Exception as exc: # noqa: BLE001 - any fetch/verify failure => skip + except (DownloadError, ImportError, OSError) as exc: + # True availability failures only (offline, missing the optional + # zstandard extraction dependency, local filesystem issues) - skip. + # A ChecksumMismatchError/ManifestValidationError means the archive + # WAS materialized but is corrupt or stale, which is exactly the + # class of defect this suite exists to catch; letting it propagate + # as a real failure (not caught here) is intentional. pytest.skip(f"canonical joinorder archive unavailable: {exc}") return paths[0].parent +def test_archive_dir_fixture_skips_only_on_availability_failures(monkeypatch: pytest.MonkeyPatch) -> None: + """#1128 review: a corrupt/stale archive (ChecksumMismatchError) must FAIL + this suite, not silently pytest.skip it away - that is exactly the class + of defect the full-archive assurance tests exist to catch. Only true + availability failures (offline, missing the zstandard extraction + dependency, local filesystem issues) should skip.""" + from benchbox.core.data_fetch.errors import ChecksumMismatchError, DownloadError + from benchbox.core.joinorder.benchmark import JoinOrderBenchmark + + fixture_fn = archive_dir.__wrapped__ + + monkeypatch.setattr( + JoinOrderBenchmark, + "generate_data", + lambda self: (_ for _ in ()).throw(DownloadError("offline")), + ) + with pytest.raises(pytest.skip.Exception): + fixture_fn() + + monkeypatch.setattr( + JoinOrderBenchmark, + "generate_data", + lambda self: (_ for _ in ()).throw( + ChecksumMismatchError(path="title.parquet", expected_sha256="a" * 64, actual_sha256="b" * 64) + ), + ) + with pytest.raises(ChecksumMismatchError): + fixture_fn() + + @pytest.fixture(scope="module") def archive_conn(archive_dir: Path) -> Any: conn = duckdb.connect(database=":memory:") @@ -210,7 +253,8 @@ def test_full_archive_utf8_multibyte_fidelity(archive_conn: Any) -> None: for (table, column), (expected_count, expected_sha256) in UTF8_FIDELITY.items(): rows = archive_conn.execute( f"SELECT id, {_quote_ident(column)} FROM {_quote_ident(table)} " - f"WHERE {_quote_ident(column)} ~ '[^\\x00-\\x7F]' ORDER BY id, {_quote_ident(column)}" + f"WHERE regexp_matches({_quote_ident(column)}, '[^\\x00-\\x7F]') " + f"ORDER BY id, {_quote_ident(column)}" ).fetchall() assert len(rows) == expected_count digest = hashlib.sha256() From 1d247005c1ea923641e8ac38c225c4b40cc25686 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 11 Jul 2026 19:38:29 +0000 Subject: [PATCH 2/7] docs(todo): correct scope_limit on completed fix-session-isolation-placeholder-rows (#1130 review) The primary path (w0/w1, actually taken) landed #1117 as-is rather than the fallback test-only rewrite, so this completed item's scope_limit only naming the fallback's test file understated what was actually authorized/touched. Added the two paths #1117 changed (connection_wrappers.py + its unit test file) for an accurate audit trail; no scope reopened, both were already merged. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01SKw2UhgHuzYPo3y5MfarD6 --- .../active/fix-session-isolation-placeholder-rows.yaml | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/_project/DONE/auto-revert-incident/active/fix-session-isolation-placeholder-rows.yaml b/_project/DONE/auto-revert-incident/active/fix-session-isolation-placeholder-rows.yaml index 0ba776aca..ceb45bdf9 100644 --- a/_project/DONE/auto-revert-incident/active/fix-session-isolation-placeholder-rows.yaml +++ b/_project/DONE/auto-revert-incident/active/fix-session-isolation-placeholder-rows.yaml @@ -124,5 +124,12 @@ verification: scope_limit: only_modify: - "tests/integration/test_throughput_session_isolation.py" + # #1130 review: the primary path (w0/w1, actually taken) lands #1117 as-is + # rather than the fallback test-only rewrite; #1117 changed these two + # paths. Recorded post-hoc for an accurate scope_limit on this completed + # item's audit trail - both were already merged via #1117 before this + # correction, not a re-opening of scope. + - "benchbox/platforms/base/connection_wrappers.py" + - "tests/unit/platforms/base/test_connection_wrappers.py" deps: needs: [] From 1c1e2eabe049baf27a9a11ae571e5edeba237edd Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 11 Jul 2026 19:43:19 +0000 Subject: [PATCH 3/7] fix(ci): require all gate signatures + detect coverage-only failures in develop-post-merge (#1136 review) Two gaps in the signature-aware auto-revert decision: 1. The current-run artifact check only required at least one signature.json to exist (compgen -G glob match), not all four expected gate artifacts. If one gate crashed/timed out before its always()-run signature step could upload, the diff still ran on the remaining three - a missing gate's real failure was silently excluded instead of forcing fail-open. Now checks each of the four expected signature- directories explicitly. 2. fast-test/medium-test's signature emission trusted the junit XML unconditionally: a job can fail for a reason junit never records (e.g. pytest-cov's --cov-fail-under gate - every testcase passes, the step still fails), producing an empty-but-"clean" signature that made the diff see no new failures and silently suppress a real revert. Added a --job-failed flag to scripts/post_merge_signature.py's build command: when the job failed and the junit-derived signature is empty, it falls back to a job-failure descriptor instead of trusting the false-clean junit. A job that failed WITH real testcase failures keeps the more precise junit signature. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01SKw2UhgHuzYPo3y5MfarD6 --- .github/workflows/develop-post-merge.yml | 61 ++++++++++++++------ scripts/post_merge_signature.py | 22 ++++++- tests/unit/test_post_merge_signature.py | 73 ++++++++++++++++++++++++ 3 files changed, 139 insertions(+), 17 deletions(-) diff --git a/.github/workflows/develop-post-merge.yml b/.github/workflows/develop-post-merge.yml index 03b8012a8..2726cff41 100644 --- a/.github/workflows/develop-post-merge.yml +++ b/.github/workflows/develop-post-merge.yml @@ -85,16 +85,26 @@ jobs: - name: Emit fast-test signature # Bare python is intentional: the signature builder is stdlib-only. # Falls back to a job-failure descriptor if pytest crashed before - # writing junit XML at all (e.g. a collection-time crash), or left a - # truncated/unparseable file behind (killed mid-write) - a signature - # artifact must ALWAYS upload, or the auto-revert diff would compare - # against a partial current signature and could wrongly suppress. + # writing junit XML at all (e.g. a collection-time crash), left a + # truncated/unparseable file behind (killed mid-write), or failed for + # a reason junit never records (e.g. pytest-cov's --cov-fail-under + # gate, where every testcase passes but the step still fails) - a + # signature artifact must ALWAYS upload and must never look clean + # when the job actually failed, or the auto-revert diff would + # compare against a partial/false-clean current signature and could + # wrongly suppress. if: always() shell: bash run: | set -euo pipefail + JOB_FAILED_FLAG="" + if [ "${{ job.status }}" = "failure" ]; then + JOB_FAILED_FLAG="--job-failed" + fi if [ -f fast-test-junit.xml ]; then - python scripts/post_merge_signature.py build --job fast-test --junit fast-test-junit.xml --out signature.json \ + python scripts/post_merge_signature.py build --job fast-test --junit fast-test-junit.xml \ + --failed-step "Run CI fast-test mirror (job failed with no testcase-level failures, e.g. coverage threshold)" \ + $JOB_FAILED_FLAG --out signature.json \ || python scripts/post_merge_signature.py build --job fast-test --failed-step "Run CI fast-test mirror (junit xml unparseable)" --out signature.json else python scripts/post_merge_signature.py build --job fast-test --failed-step "Run CI fast-test mirror (no junit xml produced)" --out signature.json @@ -198,15 +208,23 @@ jobs: - name: Emit medium-test signature # Bare python is intentional: the signature builder is stdlib-only. - # Same missing/unparseable-junit fallback as fast-test: a signature - # artifact must always upload so the auto-revert diff never runs on - # a partial current signature. + # Same missing/unparseable-junit AND job-failed-but-junit-is-clean + # fallbacks as fast-test: a signature artifact must always upload and + # must never look clean when the job actually failed, or the + # auto-revert diff could run on a partial/false-clean current + # signature. if: always() shell: bash run: | set -euo pipefail + JOB_FAILED_FLAG="" + if [ "${{ job.status }}" = "failure" ]; then + JOB_FAILED_FLAG="--job-failed" + fi if [ -f medium-test-junit.xml ]; then - python scripts/post_merge_signature.py build --job medium-test --junit medium-test-junit.xml --out signature.json \ + python scripts/post_merge_signature.py build --job medium-test --junit medium-test-junit.xml \ + --failed-step "Run medium speed tier (job failed with no testcase-level failures)" \ + $JOB_FAILED_FLAG --out signature.json \ || python scripts/post_merge_signature.py build --job medium-test --failed-step "Run medium speed tier (junit xml unparseable)" --out signature.json else python scripts/post_merge_signature.py build --job medium-test --failed-step "Run medium speed tier (no junit xml produced)" --out signature.json @@ -402,14 +420,25 @@ jobs: # Merge the four gate jobs' per-job signature artifacts (one # subdirectory per artifact, downloaded above) into a single - # combined signature for this run. - if compgen -G "current-signatures/*/signature.json" > /dev/null; then - if ! jq -s '{failure_ids: ([.[].failure_ids] | add | unique | sort)}' \ - current-signatures/*/signature.json > current-combined.json; then - fail_open_reason="Could not combine this run's signature artifacts (malformed signature JSON)." + # combined signature for this run. Require ALL FOUR expected gate + # artifacts, not just "at least one": if one gate crashes/times out + # before its always()-run signature step can upload (or the upload + # itself fails), the other three still leave signature.json files + # behind, so a bare compgen glob check would silently diff a + # partial current signature and could suppress a revert the missing + # gate's real failure warranted. + EXPECTED_SIGNATURE_JOBS=(lint fast-test explorer-tokens medium-test) + missing_signatures=() + for job in "${EXPECTED_SIGNATURE_JOBS[@]}"; do + if [ ! -f "current-signatures/signature-${job}/signature.json" ]; then + missing_signatures+=("${job}") fi - else - fail_open_reason="No current-run signature artifacts were found." + done + if [ "${#missing_signatures[@]}" -gt 0 ]; then + fail_open_reason="Missing current-run signature artifact(s) for: ${missing_signatures[*]} (gate job crashed, timed out, or was cancelled before its signature could upload)." + elif ! jq -s '{failure_ids: ([.[].failure_ids] | add | unique | sort)}' \ + current-signatures/*/signature.json > current-combined.json; then + fail_open_reason="Could not combine this run's signature artifacts (malformed signature JSON)." fi if [ -z "${fail_open_reason}" ]; then diff --git a/scripts/post_merge_signature.py b/scripts/post_merge_signature.py index c56786fb2..9c522950d 100644 --- a/scripts/post_merge_signature.py +++ b/scripts/post_merge_signature.py @@ -129,6 +129,16 @@ def _build_command(args: argparse.Namespace) -> int: try: if args.junit: signature = build_signature_from_junit(args.job, args.junit) + # A gate can fail for a reason junit never records - e.g. pytest-cov's + # --cov-fail-under gate, which fails the step (and job) while every + # individual testcase passes, leaving zero / elements. + # Trusting an empty junit-derived signature in that case would make + # the diff see no new failures and silently suppress a real revert. + # Only override when the job actually failed AND junit found nothing - + # a job that failed WITH real testcase failures keeps its junit + # signature (more precise than the coarse job-level descriptor). + if args.job_failed and not signature["failure_ids"]: + signature = build_signature_from_job_failure(args.job, args.failed_step) else: signature = build_signature_from_job_failure(args.job, args.failed_step) except SignatureError as exc: @@ -171,7 +181,17 @@ def main(argv: list[str] | None = None) -> int: build_parser.add_argument("--junit", type=Path, help="Path to a junit XML report") build_parser.add_argument( "--failed-step", - help="Name of the failed step, for non-junit jobs (omit if the job succeeded)", + help="Name of the failed step, for non-junit jobs (omit if the job succeeded). " + "Combined with --junit and --job-failed, this is also the fallback descriptor " + "used when the job failed but junit recorded no testcase-level failures.", + ) + build_parser.add_argument( + "--job-failed", + action="store_true", + help="The job failed overall (e.g. job.status == 'failure'). With --junit, an " + "empty junit-derived signature is then treated as untrustworthy (a job can fail " + "for a reason junit never records, e.g. a coverage-threshold gate) and replaced " + "with a --failed-step job-failure descriptor instead of a false-clean signature.", ) build_parser.add_argument("--out", type=Path, required=True, help="Path to write the signature JSON") build_parser.set_defaults(func=_build_command) diff --git a/tests/unit/test_post_merge_signature.py b/tests/unit/test_post_merge_signature.py index cd9a527ed..2e59fbe86 100644 --- a/tests/unit/test_post_merge_signature.py +++ b/tests/unit/test_post_merge_signature.py @@ -243,6 +243,79 @@ def test_cli_build_from_junit_writes_signature_file(tmp_path: Path) -> None: assert len(written["failure_ids"]) == 2 +def test_cli_build_job_failed_with_clean_junit_falls_back_to_job_failure(tmp_path: Path) -> None: + # #1136 review: a gate can fail for a reason junit never records (e.g. + # pytest-cov's --cov-fail-under gate - every testcase passes, the step + # still fails). Trusting the empty junit-derived signature here would + # make the diff see no new failures and silently suppress a real revert. + junit_path = tmp_path / "junit.xml" + junit_path.write_text(JUNIT_ALL_PASS, encoding="utf-8") + out_path = tmp_path / "signature.json" + + exit_code = main( + [ + "build", + "--job", + "fast-test", + "--junit", + str(junit_path), + "--failed-step", + "coverage threshold not met", + "--job-failed", + "--out", + str(out_path), + ] + ) + + assert exit_code == 0 + written = json.loads(out_path.read_text(encoding="utf-8")) + assert written["kind"] == "job-failure" + assert written["failure_ids"] == ["fast-test:coverage threshold not met"] + + +def test_cli_build_job_failed_with_real_junit_failures_keeps_junit_signature(tmp_path: Path) -> None: + # A job that failed WITH real testcase failures keeps the more precise + # junit-derived signature - --job-failed only overrides an EMPTY one. + junit_path = tmp_path / "junit.xml" + junit_path.write_text(JUNIT_WITH_FAILURES, encoding="utf-8") + out_path = tmp_path / "signature.json" + + exit_code = main( + [ + "build", + "--job", + "fast-test", + "--junit", + str(junit_path), + "--failed-step", + "should not be used", + "--job-failed", + "--out", + str(out_path), + ] + ) + + assert exit_code == 0 + written = json.loads(out_path.read_text(encoding="utf-8")) + assert written["kind"] == "junit" + assert len(written["failure_ids"]) == 2 + + +def test_cli_build_without_job_failed_keeps_clean_junit_signature(tmp_path: Path) -> None: + # The default (no --job-failed, i.e. the job actually succeeded) must + # keep behaving exactly as before: a clean junit stays a clean signature. + junit_path = tmp_path / "junit.xml" + junit_path.write_text(JUNIT_ALL_PASS, encoding="utf-8") + out_path = tmp_path / "signature.json" + + exit_code = main(["build", "--job", "fast-test", "--junit", str(junit_path), "--out", str(out_path)]) + + assert exit_code == 0 + written = json.loads(out_path.read_text(encoding="utf-8")) + assert written["kind"] == "junit" + assert written["failure_ids"] == [] + + def test_cli_build_from_job_failure_writes_signature_file(tmp_path: Path) -> None: out_path = tmp_path / "signature.json" From 673b09025ef400f616f3faeb34f62678a8564d7e Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 11 Jul 2026 19:51:47 +0000 Subject: [PATCH 4/7] fix(platforms): route count-only power fetches through row_count (#1137 review) TPC-H/TPC-DS power test harnesses called cursor.fetchall() solely to count/discard rows (result_count is the only consumer), which now trips PlatformAdapterCursor's placeholder-materialization warning on every query for a normal validation-mode run - noise that defeats the warning's purpose of flagging genuine VALUE-dependent placeholder consumption. - tpch/power_test.py, tpcds/power_test.py: `_query_result_count` checks platform_result["rows_returned"] first (never materializes) and only falls back to fetchall() for cursors with no reported count (raw DB-API cursors, test doubles). - tpcds/power_test.py's warm-up loop drains via row_count() when available, falling back to fetchall() only for cursors without it. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01SKw2UhgHuzYPo3y5MfarD6 --- benchbox/core/tpcds/power_test.py | 38 +++++++-- benchbox/core/tpch/power_test.py | 20 +++-- tests/integration/test_tpc_power_tests.py | 48 +++++++++++ tests/integration/test_tpcds_power_test.py | 99 ++++++++++++++++++++++ 4 files changed, 195 insertions(+), 10 deletions(-) diff --git a/benchbox/core/tpcds/power_test.py b/benchbox/core/tpcds/power_test.py index 3cff0400d..32717941b 100644 --- a/benchbox/core/tpcds/power_test.py +++ b/benchbox/core/tpcds/power_test.py @@ -315,7 +315,6 @@ def _execute_one_query( if hasattr(connection, "set_query_context"): connection.set_query_context(query_display_id, stream_id=self.config.stream_id) cursor = connection.execute(query_text) - rows = cursor.fetchall() if hasattr(cursor, "fetchall") else [] if hasattr(cursor, "platform_result"): result_dict = cursor.platform_result @@ -340,7 +339,7 @@ def _execute_one_query( { "execution_time_seconds": execution_time, "success": True, - "result_count": len(rows), + "result_count": self._query_result_count(cursor), } ) result.queries_successful += 1 @@ -365,6 +364,27 @@ def _execute_one_query( result.query_results.append(query_result) result.queries_executed += 1 + @staticmethod + def _query_result_count(cursor: Any) -> int: + """Return the true result cardinality for adapter cursors when available. + + Checks platform_result["rows_returned"] first - this never + materializes the cursor's row list, so a count-only power run never + trips PlatformAdapterCursor's placeholder-materialization warning + (#1137: that warning exists to catch VALUE-dependent consumers of + fabricated placeholder rows, not this count-only path). Only calls + cursor.fetchall() as a fallback when no reported count is available + (raw DB-API cursors, test doubles). + """ + platform_result = getattr(cursor, "platform_result", None) + if isinstance(platform_result, dict): + reported = platform_result.get("rows_returned") + if isinstance(reported, int) and reported >= 0: + return reported + if hasattr(cursor, "fetchall"): + return len(cursor.fetchall()) + return 0 + def _finalize_power_metrics(self, result: TPCDSPowerTestResult, start_time: float) -> None: """Compute Power@Size, total time, success flag - TPC-DS requires >=70% query success.""" total_execution_time = elapsed_seconds(start_time) @@ -581,7 +601,16 @@ def _warm_up_database(self) -> None: for query in warm_up_queries: try: cursor = self.connection.execute(query) - cursor.fetchall() + # row_count() reads the already-executed platform_result + # directly and never materializes/warns (#1137); fall back to + # fetchall() to drain a raw DB-API cursor that has no + # row_count(). The result is discarded either way - this is + # purely a drain/no-op, not a data dependency. + counter = getattr(cursor, "row_count", None) + if callable(counter): + counter() + elif hasattr(cursor, "fetchall"): + cursor.fetchall() except Exception: pass # Ignore warm-up failures @@ -606,8 +635,7 @@ def _execute_query(self, query_id: Any, query_text: str) -> dict[str, Any]: try: cursor = self.connection.execute(query_text) - rows = cursor.fetchall() - result["result_count"] = len(rows) + result["result_count"] = self._query_result_count(cursor) self.connection.commit() except Exception as e: result["status"] = "error" diff --git a/benchbox/core/tpch/power_test.py b/benchbox/core/tpch/power_test.py index 72112f95f..615e0dba9 100644 --- a/benchbox/core/tpch/power_test.py +++ b/benchbox/core/tpch/power_test.py @@ -310,7 +310,6 @@ def run(self) -> TPCHPowerTestResult: self.connection.set_query_context(query_id, stream_id=self.config.stream_id) cursor = self.connection.execute(query_text) - rows = cursor.fetchall() if hasattr(cursor, "fetchall") else [] # Check for validation failures from platform adapter if hasattr(cursor, "platform_result"): @@ -338,7 +337,7 @@ def run(self) -> TPCHPowerTestResult: { "execution_time_seconds": execution_time, "success": True, - "result_count": self._query_result_count(cursor, rows), + "result_count": self._query_result_count(cursor), } ) @@ -423,14 +422,25 @@ def _result_digest_from_cursor(cursor: Any) -> str | None: return None @staticmethod - def _query_result_count(cursor: Any, rows: list[Any]) -> int: - """Return the true result cardinality for adapter cursors when available.""" + def _query_result_count(cursor: Any) -> int: + """Return the true result cardinality for adapter cursors when available. + + Checks platform_result["rows_returned"] first - this never + materializes the cursor's row list, so a count-only power run never + trips PlatformAdapterCursor's placeholder-materialization warning + (#1137: that warning exists to catch VALUE-dependent consumers of + fabricated placeholder rows, not this count-only path). Only calls + cursor.fetchall() as a fallback when no reported count is available + (raw DB-API cursors, test doubles). + """ platform_result = getattr(cursor, "platform_result", None) if isinstance(platform_result, dict): reported = platform_result.get("rows_returned") if isinstance(reported, int) and reported >= 0: return reported - return len(rows) + if hasattr(cursor, "fetchall"): + return len(cursor.fetchall()) + return 0 def get_all_queries(self) -> dict[str, str]: """Get all queries for the power test.""" diff --git a/tests/integration/test_tpc_power_tests.py b/tests/integration/test_tpc_power_tests.py index 7ed429320..061e1e9aa 100644 --- a/tests/integration/test_tpc_power_tests.py +++ b/tests/integration/test_tpc_power_tests.py @@ -266,6 +266,54 @@ def test_tpch_power_test_uses_adapter_reported_row_count(self, tpch_mock_benchma assert result.success assert result.query_results[0]["result_count"] == 42 + def test_tpch_power_test_count_only_path_never_warns(self, tpch_mock_benchmark, caplog): + """#1137 review: the power harness must count via row_count() (never + materializing/warning), not eagerly call fetchall() on every query. + + Uses a FRESH, not-yet-materialized cursor - unlike + test_tpch_power_test_uses_adapter_reported_row_count above, which + already calls fetchall() on its cursor before the run and would mask + a regression here.""" + import logging + + cursor = PlatformAdapterCursor( + { + "status": "SUCCESS", + "rows_returned": 42, + "first_row": ("sentinel",), + "query_id": "6", + } + ) + assert cursor._rows is None # not yet materialized + + mock_connection = Mock() + mock_connection.execute.return_value = cursor + + power_test = TPCHPowerTest( + benchmark=tpch_mock_benchmark, + connection=mock_connection, + scale_factor=1.0, + seed=1, + validation=False, + warm_up=False, + query_subset=["6"], + ) + + with caplog.at_level(logging.WARNING, logger="benchbox.platforms.base.connection_wrappers"): + result = power_test.run() + + assert result.success + assert result.query_results[0]["result_count"] == 42 + assert cursor._rows is None # still never materialized + # Filtered by logger name: query_subset itself logs an unrelated, + # expected non-compliance warning from benchbox.core.tpch.power_test. + connection_wrapper_warnings = [ + r + for r in caplog.records + if r.name == "benchbox.platforms.base.connection_wrappers" and r.levelno == logging.WARNING + ] + assert connection_wrapper_warnings == [] + @patch("rich.console.Console") def test_tpch_power_test_execution_flow(self, mock_console, tpch_mock_benchmark): """Test TPC-H power test execution flow.""" diff --git a/tests/integration/test_tpcds_power_test.py b/tests/integration/test_tpcds_power_test.py index 922eab3ea..d4e5cdb7f 100644 --- a/tests/integration/test_tpcds_power_test.py +++ b/tests/integration/test_tpcds_power_test.py @@ -27,6 +27,7 @@ TPCDSPowerTest, TPCDSPowerTestResult as PowerTestResult, # Alias for backward compatibility ) +from benchbox.platforms.base.connection_wrappers import PlatformAdapterCursor # Mark all tests in this file as integration tests pytestmark = [ @@ -414,6 +415,60 @@ def test_query_execution(self, mock_db_connection): mock_cursor.fetchall.assert_called() mock_connection.commit.assert_called() + def test_execute_query_count_only_path_never_warns(self, caplog): + """#1137 review: _execute_query must count via a PlatformAdapterCursor's + reported rows_returned (never materializing/warning), not eagerly call + fetchall() when a real count is already available.""" + import logging + + mock_benchmark = self.create_mock_benchmark() + cursor = PlatformAdapterCursor( + {"status": "SUCCESS", "rows_returned": 7, "first_row": ("sentinel",), "query_id": "1"} + ) + mock_connection = Mock() + mock_connection.execute.return_value = cursor + + power_test = TPCDSPowerTest(benchmark=mock_benchmark, connection_string="sqlite::memory:") + power_test.connection = mock_connection + + with caplog.at_level(logging.WARNING, logger="benchbox.platforms.base.connection_wrappers"): + result = power_test._execute_query(1, "SELECT 1 as test") + + assert result["status"] == "success" + assert result["result_count"] == 7 + assert cursor._rows is None # never materialized + connection_wrapper_warnings = [ + r + for r in caplog.records + if r.name == "benchbox.platforms.base.connection_wrappers" and r.levelno == logging.WARNING + ] + assert connection_wrapper_warnings == [] + + def test_warm_up_database_never_warns(self, caplog): + """#1137 review: warm-up queries drain via row_count(), not fetchall(), + so a real PlatformAdapterCursor never trips the placeholder warning.""" + import logging + + mock_benchmark = self.create_mock_benchmark() + cursor = PlatformAdapterCursor({"status": "SUCCESS", "rows_returned": 3, "query_id": "warmup"}) + mock_connection = Mock() + mock_connection.execute.return_value = cursor + + power_test = TPCDSPowerTest(benchmark=mock_benchmark, connection_string="sqlite::memory:", warm_up=True) + power_test.connection = mock_connection + + with caplog.at_level(logging.WARNING, logger="benchbox.platforms.base.connection_wrappers"): + power_test._warm_up_database() + + assert mock_connection.execute.call_count >= 5 + assert cursor._rows is None # never materialized + connection_wrapper_warnings = [ + r + for r in caplog.records + if r.name == "benchbox.platforms.base.connection_wrappers" and r.levelno == logging.WARNING + ] + assert connection_wrapper_warnings == [] + def test_execute_one_query_propagates_captured_plan_fields(self): """--capture-plans / --normalize-plan-literals on TPC-DS power: the adapter's captured plan/fingerprints live on cursor.platform_result, but _execute_one_query @@ -496,6 +551,50 @@ def test_execute_one_query_omits_plan_fields_when_capture_disabled(self): assert "query_plan" not in query_result assert "plan_fingerprint" not in query_result + def test_execute_one_query_count_only_path_never_warns(self, caplog): + """#1137 review: _execute_one_query must count via a + PlatformAdapterCursor's reported rows_returned (never + materializing/warning) rather than eagerly calling fetchall().""" + import logging + + from benchbox.core.tpcds.power_test import TPCDSPowerTestConfig, TPCDSPowerTestResult + + mock_benchmark = self.create_mock_benchmark() + power_test = TPCDSPowerTest(benchmark=mock_benchmark, connection_string="sqlite::memory:") + power_test.config = TPCDSPowerTestConfig(scale_factor=1.0) + + cursor = PlatformAdapterCursor( + {"status": "SUCCESS", "rows_returned": 5, "first_row": ("sentinel",), "query_id": "1"} + ) + mock_connection = Mock(spec=["execute", "commit"]) + mock_connection.execute.return_value = cursor + + result = TPCDSPowerTestResult( + config=power_test.config, + start_time="", + end_time="", + total_time=0.0, + power_at_size=0.0, + queries_executed=0, + queries_successful=0, + query_results=[], + success=True, + errors=[], + ) + + with caplog.at_level(logging.WARNING, logger="benchbox.platforms.base.connection_wrappers"): + power_test._execute_one_query(0, 1, mock_connection, stream_param_seed=1, result=result) + + query_result = result.query_results[0] + assert query_result["result_count"] == 5 + assert cursor._rows is None # never materialized + connection_wrapper_warnings = [ + r + for r in caplog.records + if r.name == "benchbox.platforms.base.connection_wrappers" and r.levelno == logging.WARNING + ] + assert connection_wrapper_warnings == [] + @patch("benchbox.core.tpcds.power_test.DatabaseConnection") def test_query_execution_error_handling(self, mock_db_connection): """Test error handling during query execution.""" From dd7872e830eb443d3478a871e58b839a61c448e9 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 11 Jul 2026 19:55:07 +0000 Subject: [PATCH 5/7] docs(todo): fix inverted grep exit-code gates in 2 uat-hardening TODOs (#1138 review) Both verification commands defined success as "no grep hits" but grep exits 1 on no match and 0 on a match - the gates had the pass/fail logic backwards, so a future implementer's scripted verification would report success/failure exactly opposite of the real outcome. - uat-execute-path-unification.yaml: `grep ... && echo FOUND || echo CLEAN` always exits 0 regardless of which branch prints - the FOUND branch now exits 1. - uat-resume-retirement-artifact-durability.yaml: both "no hits = success" pipelines now invert their exit status with `!`, matching the `! grep -q ...` idiom already used elsewhere in this TODO tree. Verified empirically against the real repo (dead code still present - correctly reports FOUND, exit 1) and simulated clean/dirty/filtered cases for the resume-reference gates. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01SKw2UhgHuzYPo3y5MfarD6 --- .../uat-hardening/planning/uat-execute-path-unification.yaml | 2 +- .../planning/uat-resume-retirement-artifact-durability.yaml | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/_project/TODO/uat-hardening/planning/uat-execute-path-unification.yaml b/_project/TODO/uat-hardening/planning/uat-execute-path-unification.yaml index 8372a6d39..cd4f50364 100644 --- a/_project/TODO/uat-hardening/planning/uat-execute-path-unification.yaml +++ b/_project/TODO/uat-hardening/planning/uat-execute-path-unification.yaml @@ -114,7 +114,7 @@ verification: command: "uv run -- python -m pytest tests/uat/ -m fast -q" expected_output: "all pass including the new parity test" - description: "Dead-code removal is complete" - command: "grep -rn 'env_without_pythonpath\\|iter_argv_for_log' tests/uat/ && echo FOUND || echo CLEAN" + command: "grep -rn 'env_without_pythonpath\\|iter_argv_for_log' tests/uat/ && { echo FOUND; exit 1; } || echo CLEAN" expected_output: "CLEAN" scope_limit: diff --git a/_project/TODO/uat-hardening/planning/uat-resume-retirement-artifact-durability.yaml b/_project/TODO/uat-hardening/planning/uat-resume-retirement-artifact-durability.yaml index 5ac6d9155..2bcd7a783 100644 --- a/_project/TODO/uat-hardening/planning/uat-resume-retirement-artifact-durability.yaml +++ b/_project/TODO/uat-hardening/planning/uat-resume-retirement-artifact-durability.yaml @@ -98,10 +98,10 @@ anti_patterns: verification: - description: "No resume references remain in Python code" - command: "grep -rni 'resume' tests/uat/ --include='*.py' | grep -vi 'resumable\\|datagen'" + command: "! (grep -rni 'resume' tests/uat/ --include='*.py' | grep -vi 'resumable\\|datagen')" expected_output: "no hits" - description: "No resume references remain in Makefile, spec, or operator docs" - command: "grep -ni 'resume' Makefile _project/specs/uat-framework.md docs/operations/uat-framework.md tests/uat/README.md | grep -vi 'resumable\\|datagen'" + command: "! (grep -ni 'resume' Makefile _project/specs/uat-framework.md docs/operations/uat-framework.md tests/uat/README.md | grep -vi 'resumable\\|datagen')" expected_output: "no hits (or only prose explaining the removal)" - description: "UAT unit suite green, including new atomicity/ordering/collision tests" command: "uv run -- python -m pytest tests/uat/ -m fast -q" From 7ecb2848eae31ca6f7fa4ea71e6cfa9d5bb23579 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 11 Jul 2026 19:58:55 +0000 Subject: [PATCH 6/7] fix(throughput): cancel queued (not-yet-running) streams before nonblocking shutdown (#1141 review) executor.shutdown(wait=False) left cancel_futures at its default False, so when max_workers < num_streams, a future still QUEUED (never dispatched to a worker) at the timeout deadline would start executing once a worker eventually freed up - well after execute() had already returned and counted that stream as timed out. That created extra, unexpected DB work overlapping with whatever phase ran next. cancel_futures=True closes the gap: it cancels only futures that have not started running (Python's documented ThreadPoolExecutor.shutdown semantics), leaving already-running (leaked/abandoned) futures untouched - the can't-forcibly-cancel-a-running-thread invariant is unaffected. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01SKw2UhgHuzYPo3y5MfarD6 --- benchbox/core/throughput/runner.py | 53 +++++++++++++------ .../throughput-result-alignment.md | 36 +++++++++---- tests/unit/core/throughput/test_runner.py | 31 +++++++++++ 3 files changed, 93 insertions(+), 27 deletions(-) diff --git a/benchbox/core/throughput/runner.py b/benchbox/core/throughput/runner.py index bd0d3ed2c..4e5a4be65 100644 --- a/benchbox/core/throughput/runner.py +++ b/benchbox/core/throughput/runner.py @@ -46,23 +46,31 @@ Non-blocking shutdown (return without joining a leaked thread) ---------------------------------------------------------------- ``execute()`` manages the ``ThreadPoolExecutor`` manually (no ``with`` -block) so it can call ``executor.shutdown(wait=False)`` in a ``finally`` -instead of relying on the context manager's implicit +block) so it can call ``executor.shutdown(wait=False, cancel_futures=True)`` +in a ``finally`` instead of relying on the context manager's implicit ``shutdown(wait=True)``. This means ``execute()`` returns as soon as its own timeout/accounting window closes -- bounded by ``config.stream_timeout`` plus classification overhead -- rather than blocking until every submitted thread finishes. On the healthy path (no timeout fires), every future is already ``done()`` -by the time ``as_completed()`` returns, so ``shutdown(wait=False)`` is a -no-op: there is nothing left to join, and default-timeout behavior for -healthy runs is unchanged. On the hung path, any future still not -``done()`` when the deadline elapses is *abandoned*, not killed -- Python -cannot forcibly stop a running thread. ``shutdown(wait=False)`` only stops -the executor from accepting new submissions; it does not touch threads -already running, so the leaked thread keeps executing in the background -until ``stream_fn`` itself returns (naturally, or via cooperative -cancellation if enabled) and exits. +by the time ``as_completed()`` returns, so this call is a no-op: there is +nothing left to join or cancel, and default-timeout behavior for healthy +runs is unchanged. On the hung path, two distinct futures can remain +outstanding at the deadline: + +* **Running** (already dispatched to a worker thread): *abandoned*, not + killed -- Python cannot forcibly stop a running thread. ``cancel_futures`` + never touches these (Python's documented semantics only cancel futures + that have not started running); the leaked thread keeps executing in the + background until ``stream_fn`` itself returns (naturally, or via + cooperative cancellation if enabled) and exits. +* **Queued** (submitted, but ``max_workers < num_streams`` left it still + waiting for a worker): ``cancel_futures=True`` cancels these outright, so + they never start at all. Without this, a queued stream would begin + executing only *after* ``execute()`` has already returned and counted it + as timed-out -- extra DB work that can overlap with whatever phase runs + next. **Zombie connection/accounting semantics** (see also ``docs/development/throughput-result-alignment.md``): a leaked stream's @@ -306,11 +314,24 @@ def _record_completed_future( # (abandoned, not killed) thread. On the healthy path every # future is already done() by this point, so this is a no-op -- # nothing to join, default timeout behavior for healthy runs is - # unchanged. On the hung path, any leaked thread keeps running - # to completion in the background; its connection is closed by - # its own stream_fn's finally whenever that eventually happens. - # See the module docstring "Non-blocking shutdown" section. - executor.shutdown(wait=False) + # unchanged. On the hung path, any already-RUNNING leaked thread + # keeps running to completion in the background; its connection + # is closed by its own stream_fn's finally whenever that + # eventually happens. + # + # cancel_futures=True: when max_workers < num_streams, a future + # can still be QUEUED (never dispatched to a worker) at the + # deadline -- distinct from a leaked RUNNING future. Without + # this, such a future would start executing only AFTER + # execute() has already returned and counted it as timed-out, + # creating extra DB work that can overlap with whatever phase + # runs next. cancel_futures only cancels futures that have not + # started running (Python's documented semantics); an + # already-running future is never touched by it, so the + # can't-forcibly-cancel-a-running-thread invariant above is + # unaffected. See the module docstring "Non-blocking shutdown" + # section. + executor.shutdown(wait=False, cancel_futures=True) @staticmethod def compute_metrics( diff --git a/docs/development/throughput-result-alignment.md b/docs/development/throughput-result-alignment.md index 8fba02494..fb762856b 100644 --- a/docs/development/throughput-result-alignment.md +++ b/docs/development/throughput-result-alignment.md @@ -243,21 +243,35 @@ itself for as long as the stream kept running, and with it the whole throughput phase and any subsequent phase (e.g. data maintenance). **Fix:** `execute()` now manages the `ThreadPoolExecutor` manually (no -`with` block) and calls `executor.shutdown(wait=False)` in a `finally`. -`shutdown(wait=False)` only stops the executor from accepting new -submissions — it does not touch, join, or wait on threads that are already -running. This is deliberately **not** a thread-kill: Python cannot forcibly -stop a running thread, and this change does not attempt to. A leaked -thread is *abandoned*, not killed — it keeps running `stream_fn` to -completion in the background, entirely decoupled from the caller of -`execute()`. +`with` block) and calls `executor.shutdown(wait=False, cancel_futures=True)` +in a `finally`. `shutdown(wait=False)` only stops the executor from +accepting new submissions — it does not touch, join, or wait on threads +that are already running. This is deliberately **not** a thread-kill: +Python cannot forcibly stop a running thread, and this change does not +attempt to. A leaked (already-**running**) thread is *abandoned*, not +killed — it keeps running `stream_fn` to completion in the background, +entirely decoupled from the caller of `execute()`. + +`cancel_futures=True` handles a distinct case: when `max_workers < +num_streams`, a future can still be **queued** (submitted, but never +dispatched to a worker thread) at the timeout deadline — see "Known +limitation" above. Without `cancel_futures=True`, such a future would sit +in the executor's queue and start executing only once a worker frees up +(e.g. when an earlier hung stream eventually returns), which could be well +after `execute()` has already returned and counted that stream as +timed-out — extra, unexpected DB work overlapping with whatever runs next. +`cancel_futures` only cancels futures that have not started running +(Python's documented `ThreadPoolExecutor.shutdown` semantics); an +already-running future is never touched by it, so the +can't-forcibly-cancel-a-running-thread invariant above is unaffected. **Healthy-path behavior is unchanged.** When every stream completes before the timeout (or no timeout is set), every future is already `done()` by the time the `as_completed()`/fallback loop finishes, so -`shutdown(wait=False)` has nothing left to wait for — it is a no-op -compared to the old `shutdown(wait=True)`. Only the hung-stream shutdown -path changes; default timeout behavior for healthy runs is unaffected. +`shutdown(wait=False, cancel_futures=True)` has nothing left to wait for or +cancel — it is a no-op compared to the old `shutdown(wait=True)`. Only the +hung-stream shutdown path changes; default timeout behavior for healthy +runs is unaffected. ### Zombie connection ownership diff --git a/tests/unit/core/throughput/test_runner.py b/tests/unit/core/throughput/test_runner.py index 945a9be1f..aaa4d1875 100644 --- a/tests/unit/core/throughput/test_runner.py +++ b/tests/unit/core/throughput/test_runner.py @@ -600,3 +600,34 @@ def test_mock_truthy_config_healthy_path_unaffected(self) -> None: cancel_events = getattr(config, "_stream_cancel_events", None) assert isinstance(cancel_events, dict) and len(cancel_events) == 1 assert all(not event.is_set() for event in cancel_events.values()) + + def test_queued_stream_never_starts_after_max_workers_throttled_timeout(self) -> None: + """#1141 review: with max_workers < num_streams, a future can still + be QUEUED (never dispatched to a worker) at the timeout deadline - + distinct from a leaked RUNNING future. Without cancel_futures=True, + that queued stream would start executing once the sole worker frees + up (when the hung stream 0 finally returns), well after execute() + has already returned and counted stream 1 as timed out too - extra + DB work overlapping with whatever runs next.""" + started = threading.Event() + + def stream_fn(stream_id: int, seed: int, cfg: _FakeConfig) -> ThroughputStreamResult: + if stream_id == 0: + time.sleep(self.HANG_SLEEP) # occupies the sole worker past the timeout + else: + started.set() # only reachable if stream 1 was ever dispatched + return _make_stream_result(stream_id) + + config = _FakeConfig(num_streams=2, max_workers=1, stream_timeout=self.TINY_TIMEOUT, cancel_on_timeout=False) + result = _make_result() + logger = logging.getLogger("test-throughput-runner") + + StreamRunner.execute(stream_fn, config, result, logger) + + assert result.streams_executed == 2 + assert len(result.errors) == 2 + + # Wait well past HANG_SLEEP (when the sole worker frees up) - a + # cancelled queued future must never be dispatched, even once a + # worker becomes free. + assert not started.wait(timeout=self.HANG_SLEEP + 1.0) From d452d05fd09358271a6bfa79c73fe62440fffa65 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 12 Jul 2026 01:25:00 +0000 Subject: [PATCH 7/7] fix(power-test): drain raw DB-API cursor before commit (#1144 review) #1137's fix routed count-only power-test paths through row_count()/rows_returned to avoid a placeholder-materialization warning, but for raw DB-API cursors without platform_result, _query_result_count still falls back to fetchall() - and that fallback call had moved to after connection.commit(). Some drivers with unbuffered SELECT results reject or invalidate commit() while rows are still unread, which would misrecord successful power queries as failures. Restores the pre-#1137 drain-before-commit ordering in both tpch/power_test.py and tpcds/power_test.py::_execute_one_query while keeping #1137's benefit (no eager fetchall() when rows_returned is already available). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01SKw2UhgHuzYPo3y5MfarD6 --- benchbox/core/tpcds/power_test.py | 10 ++++- benchbox/core/tpch/power_test.py | 11 ++++- tests/integration/test_tpc_power_tests.py | 42 ++++++++++++++++++ tests/integration/test_tpcds_power_test.py | 51 ++++++++++++++++++++++ 4 files changed, 112 insertions(+), 2 deletions(-) diff --git a/benchbox/core/tpcds/power_test.py b/benchbox/core/tpcds/power_test.py index 32717941b..ed7c9b822 100644 --- a/benchbox/core/tpcds/power_test.py +++ b/benchbox/core/tpcds/power_test.py @@ -329,6 +329,14 @@ def _execute_one_query( # fallback in _attach_captured_plans. propagate_plan_capture_fields(result_dict, query_result) + # Count BEFORE commit (#1144 review): when rows_returned + # isn't available, _query_result_count falls back to + # cursor.fetchall(). Some raw DB-API drivers with unbuffered + # SELECT results can reject/invalidate commit() while rows + # are still unread, so draining must happen first - matches + # the pre-#1137 ordering. + result_count = self._query_result_count(cursor) + if hasattr(connection, "commit"): connection.commit() finally: @@ -339,7 +347,7 @@ def _execute_one_query( { "execution_time_seconds": execution_time, "success": True, - "result_count": self._query_result_count(cursor), + "result_count": result_count, } ) result.queries_successful += 1 diff --git a/benchbox/core/tpch/power_test.py b/benchbox/core/tpch/power_test.py index 615e0dba9..d36bda32f 100644 --- a/benchbox/core/tpch/power_test.py +++ b/benchbox/core/tpch/power_test.py @@ -325,6 +325,15 @@ def run(self) -> TPCHPowerTestResult: # exact key rather than the ambiguous public-id fallback. propagate_plan_capture_fields(result_dict, query_result) + # Count BEFORE commit (#1144 review): when + # rows_returned isn't available, _query_result_count + # falls back to cursor.fetchall(). Some raw DB-API + # drivers with unbuffered SELECT results can + # reject/invalidate commit() while rows are still + # unread, so draining must happen first - matches + # the pre-#1137 ordering. + result_count = self._query_result_count(cursor) + if hasattr(self.connection, "commit"): self.connection.commit() finally: @@ -337,7 +346,7 @@ def run(self) -> TPCHPowerTestResult: { "execution_time_seconds": execution_time, "success": True, - "result_count": self._query_result_count(cursor), + "result_count": result_count, } ) diff --git a/tests/integration/test_tpc_power_tests.py b/tests/integration/test_tpc_power_tests.py index 061e1e9aa..410d697dd 100644 --- a/tests/integration/test_tpc_power_tests.py +++ b/tests/integration/test_tpc_power_tests.py @@ -314,6 +314,48 @@ def test_tpch_power_test_count_only_path_never_warns(self, tpch_mock_benchmark, ] assert connection_wrapper_warnings == [] + def test_tpch_power_test_drains_raw_cursor_before_commit(self, tpch_mock_benchmark): + """#1144 review: for a raw DB-API cursor (no platform_result, so + _query_result_count falls back to fetchall()), draining must happen + BEFORE commit() - some drivers with unbuffered SELECT results + reject/invalidate commit() while rows are still unread. #1137's fix + moved the fetchall()-based count to after commit(); this pins the + pre-#1137 ordering back in place.""" + + class _CommitBeforeDrainSensitiveCursor: + def __init__(self, rows: list) -> None: + self._rows = rows + self.drained = False + + def fetchall(self) -> list: + self.drained = True + return self._rows + + cursor = _CommitBeforeDrainSensitiveCursor([("a",), ("b",), ("c",)]) + mock_connection = Mock(spec=["execute", "commit"]) + mock_connection.execute.return_value = cursor + + def _commit_requires_drained_cursor() -> None: + assert cursor.drained, "commit() was called before the cursor was drained" + + mock_connection.commit.side_effect = _commit_requires_drained_cursor + + power_test = TPCHPowerTest( + benchmark=tpch_mock_benchmark, + connection=mock_connection, + scale_factor=1.0, + seed=1, + validation=False, + warm_up=False, + query_subset=["6"], + ) + + result = power_test.run() + + assert result.success + assert result.query_results[0]["result_count"] == 3 + mock_connection.commit.assert_called_once() + @patch("rich.console.Console") def test_tpch_power_test_execution_flow(self, mock_console, tpch_mock_benchmark): """Test TPC-H power test execution flow.""" diff --git a/tests/integration/test_tpcds_power_test.py b/tests/integration/test_tpcds_power_test.py index d4e5cdb7f..16b340bfe 100644 --- a/tests/integration/test_tpcds_power_test.py +++ b/tests/integration/test_tpcds_power_test.py @@ -469,6 +469,57 @@ def test_warm_up_database_never_warns(self, caplog): ] assert connection_wrapper_warnings == [] + def test_execute_one_query_drains_raw_cursor_before_commit(self): + """#1144 review: for a raw DB-API cursor (no platform_result, so + _query_result_count falls back to fetchall()), draining must happen + BEFORE commit() - some drivers with unbuffered SELECT results + reject/invalidate commit() while rows are still unread. #1137's fix + moved the fetchall()-based count to after commit(); this pins the + pre-#1137 ordering back in place.""" + from benchbox.core.tpcds.power_test import TPCDSPowerTestConfig, TPCDSPowerTestResult + + class _CommitBeforeDrainSensitiveCursor: + def __init__(self, rows: list) -> None: + self._rows = rows + self.drained = False + + def fetchall(self) -> list: + self.drained = True + return self._rows + + mock_benchmark = self.create_mock_benchmark() + power_test = TPCDSPowerTest(benchmark=mock_benchmark, connection_string="sqlite::memory:") + power_test.config = TPCDSPowerTestConfig(scale_factor=1.0) + + cursor = _CommitBeforeDrainSensitiveCursor([("a",), ("b",)]) + mock_connection = Mock(spec=["execute", "commit"]) + mock_connection.execute.return_value = cursor + + def _commit_requires_drained_cursor() -> None: + assert cursor.drained, "commit() was called before the cursor was drained" + + mock_connection.commit.side_effect = _commit_requires_drained_cursor + + result = TPCDSPowerTestResult( + config=power_test.config, + start_time="", + end_time="", + total_time=0.0, + power_at_size=0.0, + queries_executed=0, + queries_successful=0, + query_results=[], + success=True, + errors=[], + ) + + power_test._execute_one_query(0, 1, mock_connection, stream_param_seed=1, result=result) + + query_result = result.query_results[0] + assert query_result["success"] is True + assert query_result["result_count"] == 2 + mock_connection.commit.assert_called_once() + def test_execute_one_query_propagates_captured_plan_fields(self): """--capture-plans / --normalize-plan-literals on TPC-DS power: the adapter's captured plan/fingerprints live on cursor.platform_result, but _execute_one_query