diff --git a/.github/workflows/develop-post-merge.yml b/.github/workflows/develop-post-merge.yml index b3a358976..af54f93e6 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 @@ -209,15 +219,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 @@ -413,14 +431,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/_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: [] diff --git a/_project/DONE/uat-hardening/planning/uat-execute-path-unification.yaml b/_project/DONE/uat-hardening/planning/uat-execute-path-unification.yaml index 478ebdd13..733de1f23 100644 --- a/_project/DONE/uat-hardening/planning/uat-execute-path-unification.yaml +++ b/_project/DONE/uat-hardening/planning/uat-execute-path-unification.yaml @@ -218,7 +218,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/DONE/uat-hardening/planning/uat-resume-retirement-artifact-durability.yaml b/_project/DONE/uat-hardening/planning/uat-resume-retirement-artifact-durability.yaml index 506f1a5ae..f5432653e 100644 --- a/_project/DONE/uat-hardening/planning/uat-resume-retirement-artifact-durability.yaml +++ b/_project/DONE/uat-hardening/planning/uat-resume-retirement-artifact-durability.yaml @@ -169,10 +169,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" 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/benchbox/core/tpcds/power_test.py b/benchbox/core/tpcds/power_test.py index 3cff0400d..ed7c9b822 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 @@ -330,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: @@ -340,7 +347,7 @@ def _execute_one_query( { "execution_time_seconds": execution_time, "success": True, - "result_count": len(rows), + "result_count": result_count, } ) result.queries_successful += 1 @@ -365,6 +372,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 +609,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 +643,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..d36bda32f 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"): @@ -326,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: @@ -338,7 +346,7 @@ def run(self) -> TPCHPowerTestResult: { "execution_time_seconds": execution_time, "success": True, - "result_count": self._query_result_count(cursor, rows), + "result_count": result_count, } ) @@ -423,14 +431,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/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/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/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() diff --git a/tests/integration/test_tpc_power_tests.py b/tests/integration/test_tpc_power_tests.py index 7ed429320..410d697dd 100644 --- a/tests/integration/test_tpc_power_tests.py +++ b/tests/integration/test_tpc_power_tests.py @@ -266,6 +266,96 @@ 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 == [] + + 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 922eab3ea..16b340bfe 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,111 @@ 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_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 @@ -496,6 +602,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.""" 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) 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"