diff --git a/CHANGELOG.md b/CHANGELOG.md index 0be6542687..be4eda978d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -339,7 +339,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 runs with an actionable error instead. - Result metadata records genuine engine versions (`engine_version`, `engine_version_source: sql_query`) for DuckDB, SQLite and DataFusion queried directly from the active connection rather than package versions. - +- A successful physical sort rewrite (CTAS or constraint-preserving in-place sort) + is recorded as an executed post-load ledger statement and shown in the run + receipt as attested (`Recorded; not catalog-checkable`). Attested evidence never + satisfies catalog-verifiable intents and never promotes a run to verified on + its own. - ClickHouse tuning introspection bounds table queries in SQL by ledger tables and measures snapshot truncation directly on raw query results. - The Atom feed uses canonical `/blog/` self and entry links while retaining stable entry IDs for existing subscribers. - Deprecated option aliases now show their warning. The Polars `streaming` diff --git a/benchbox/core/tuning/introspection.py b/benchbox/core/tuning/introspection.py index c7e3578e24..00cea577be 100644 --- a/benchbox/core/tuning/introspection.py +++ b/benchbox/core/tuning/introspection.py @@ -27,6 +27,7 @@ TRANSIENT = "transient" MAINTENANCE = "maintenance" UNVERIFIABLE = "unverifiable" +ATTESTED = "attested" _VERIFIABLE_VERDICTS = frozenset({CORROBORATED, ABSENT, MISMATCH, UNVERIFIABLE}) @@ -728,6 +729,30 @@ def _match_object(intent: _Intent, state: IntrospectedState) -> tuple[str, Intro return MISMATCH, named or same_table[0] +def _attested_sort_columns(statement_text: str) -> tuple[str, ...]: + from benchbox.core.tuning.applied_ledger import _split_sql_script + + segments = [segment for segment in _split_sql_script(statement_text) if segment.strip()] + for segment in segments: + order = _ORDER_BY_RE.search(segment) + if order: + return normalize_columns(order.group("cols")) + for segment in segments: + keyword = _ORDER_BY_KEYWORD_RE.search(segment) + if keyword is None: + continue + tail = segment[keyword.end() :] + columns: list[str] = [] + for chunk in tail.split(","): + token = re.sub(r"(?i)\b(asc|desc|nulls\s+first|nulls\s+last)\b", " ", chunk).strip() + token = token.strip().strip(";") + if token: + columns.append(token) + if columns: + return normalize_columns(columns) + return () + + def corroborate(ledger: AppliedTuningLedger, introspected_state: IntrospectedState | None) -> IntrospectionReceipt: platform = introspected_state.platform if introspected_state else "unknown" state_error = None @@ -758,6 +783,20 @@ def corroborate(ledger: AppliedTuningLedger, introspected_state: IntrospectedSta ) continue + if stmt.phase == PHASE_POST_LOAD and stmt.mechanism == "sorted_ingestion": + entries.append( + ReceiptEntry( + statement=stmt.statement, + phase=stmt.phase, + verdict=ATTESTED, + kind=KIND_SORT_KEY, + table=stmt.table, + expected_columns=_attested_sort_columns(str(stmt.statement or "")), + reason="physical sort executed; no catalog footprint", + ) + ) + continue + klass, intents = _classify(stmt) if klass == TRANSIENT: @@ -857,6 +896,7 @@ def corroborate(ledger: AppliedTuningLedger, introspected_state: IntrospectedSta __all__ = [ "ABSENT", + "ATTESTED", "CONSTRAINT_FOREIGN_KEY", "CONSTRAINT_PRIMARY_KEY", "CONSTRAINT_UNIQUE", diff --git a/benchbox/platforms/base/sorted_ingestion.py b/benchbox/platforms/base/sorted_ingestion.py index 37f685c26e..d238dda45f 100644 --- a/benchbox/platforms/base/sorted_ingestion.py +++ b/benchbox/platforms/base/sorted_ingestion.py @@ -206,6 +206,34 @@ def _recording_ctas_connection(self, connection: Any) -> Any: execute_verbs=("sql",), ) + def _attach_sorted_ingestion_evidence( + self, statements: list[str], ledger_statement: str, validated_table: str + ) -> None: + from benchbox.core.tuning.applied_ledger import EXECUTED, PHASE_POST_LOAD + + ledger = getattr(self, "_applied_tuning_ledger", None) + if ledger is None: + return + upgraded = False + for statement in statements: + for recorded in ledger.statements: + if ( + recorded.status == EXECUTED + and recorded.phase == PHASE_POST_LOAD + and recorded.mechanism is None + and recorded.statement == statement + ): + recorded.mechanism = "sorted_ingestion" + recorded.table = validated_table + upgraded = True + if not upgraded: + ledger.record( + ledger_statement, + PHASE_POST_LOAD, + mechanism="sorted_ingestion", + table=validated_table, + ) + def _execute_ctas_sort( self, table_name: str, @@ -285,6 +313,7 @@ def _execute_ctas_sort( raise self._sorted_ingestion_applied_tables.append(validated_table) self._sorted_ingestion_total_apply_seconds += elapsed_seconds(apply_start) + self._attach_sorted_ingestion_evidence(statements, ledger_statement, validated_table) sorted_col_names = ", ".join(column.name for column in sorted_columns) self.log_verbose(f"Applied CTAS sorting to {table_name}: {sorted_col_names}") return True diff --git a/results-explorer/src/components/RunReceipt.tsx b/results-explorer/src/components/RunReceipt.tsx index 479675442c..3b81f5b5b0 100644 --- a/results-explorer/src/components/RunReceipt.tsx +++ b/results-explorer/src/components/RunReceipt.tsx @@ -457,7 +457,9 @@ function AppliedReceiptDrilldown({ raw }: { raw?: string | null }) { } function AppliedReceiptEntryBody({ entry }: { entry: AppliedReceiptEntry }) { - const verdict = receiptText(entry.verdict); + const rawVerdict = receiptText(entry.verdict); + const verdict = + rawVerdict === "attested" ? "attested — Recorded; not catalog-checkable" : rawVerdict; const statement = receiptText(entry.statement); const fields: { label: string; value: string }[] = [ { label: "Kind", value: receiptText(entry.kind) }, diff --git a/results-explorer/src/components/__tests__/RunReceipt.test.tsx b/results-explorer/src/components/__tests__/RunReceipt.test.tsx index aaab077443..5d61a06a65 100644 --- a/results-explorer/src/components/__tests__/RunReceipt.test.tsx +++ b/results-explorer/src/components/__tests__/RunReceipt.test.tsx @@ -535,6 +535,34 @@ describe("RunReceipt applied-tuning receipt drill-down", () => { expect(within(divergent).getByText("catalog reports a different index definition")).toBeTruthy(); }); + it("labels an attested sort entry as recorded but not catalog-checkable", () => { + const receipt = JSON.stringify({ + platform: "duckdb", + corroborated: false, + summary: { attested: 1 }, + entries: [ + { + statement: "CREATE OR REPLACE TABLE lineitem AS SELECT * FROM lineitem ORDER BY l_shipdate", + phase: "post_load", + verdict: "attested", + kind: "sort_key", + table: "lineitem", + expected_columns: ["l_shipdate"], + reason: "physical sort executed; no catalog footprint", + }, + ], + }); + render(); + + const region = screen.getByRole("region", { name: "Run receipt" }); + const drilldown = within(region).getByTestId("applied-receipt-drilldown"); + const entries = within(drilldown).getAllByTestId("applied-receipt-entry"); + expect(entries.length).toBe(1); + const [attested] = entries as [HTMLElement]; + expect(within(attested).getByText("attested — Recorded; not catalog-checkable")).toBeTruthy(); + expect(within(attested).getByText("physical sort executed; no catalog footprint")).toBeTruthy(); + }); + it("omits fields the receipt did not record rather than inventing them", () => { const sparse = JSON.stringify({ entries: [{ verdict: "corroborated" }] }); render(); diff --git a/tests/unit/core/tuning/test_sort_evidence_attested.py b/tests/unit/core/tuning/test_sort_evidence_attested.py new file mode 100644 index 0000000000..f8d0340947 --- /dev/null +++ b/tests/unit/core/tuning/test_sort_evidence_attested.py @@ -0,0 +1,190 @@ +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from benchbox.core.tuning.applied_ledger import PHASE_POST_LOAD, AppliedTuningLedger +from benchbox.core.tuning.introspection import ( + ATTESTED, + KIND_INDEX, + IntrospectedObject, + IntrospectedState, + _attested_sort_columns, + corroborate, +) + +pytestmark = [ + pytest.mark.unit, + pytest.mark.fast, +] + + +SORT_STATEMENT = "CREATE OR REPLACE TABLE lineitem AS SELECT * FROM lineitem ORDER BY l_shipdate, l_orderkey" +INDEX_STATEMENT = 'CREATE INDEX idx_lineitem_sort ON "lineitem" ("l_shipdate")' +ATTESTED_REASON = "physical sort executed; no catalog footprint" + + +def _index_state() -> IntrospectedState: + return IntrospectedState( + platform="duckdb", + objects=[IntrospectedObject(kind=KIND_INDEX, table="lineitem", columns=("l_shipdate",))], + ) + + +def _sort_ledger() -> AppliedTuningLedger: + ledger = AppliedTuningLedger() + ledger.record(INDEX_STATEMENT, PHASE_POST_LOAD) + ledger.record(SORT_STATEMENT, PHASE_POST_LOAD, mechanism="sorted_ingestion", table="lineitem") + return ledger + + +class TestAttestedVerdict: + def test_executed_rewrite_is_attested_not_catalog_checked(self) -> None: + receipt = corroborate(_sort_ledger(), _index_state()) + + attested = [entry for entry in receipt.entries if entry.verdict == ATTESTED] + assert len(attested) == 1 + assert attested[0].reason == ATTESTED_REASON + assert attested[0].kind == "sort_key" + assert attested[0].table == "lineitem" + assert list(attested[0].expected_columns) == ["l_shipdate", "l_orderkey"] + + def test_index_corroboration_plus_rewrite_is_applied_verified(self) -> None: + receipt = corroborate(_sort_ledger(), _index_state()) + + assert receipt.corroborated is True + assert any(entry.verdict == "corroborated" for entry in receipt.entries) + assert any(entry.verdict == ATTESTED for entry in receipt.entries) + + def test_attested_only_never_reaches_verified(self) -> None: + ledger = AppliedTuningLedger() + ledger.record(SORT_STATEMENT, PHASE_POST_LOAD, mechanism="sorted_ingestion", table="lineitem") + + receipt = corroborate(ledger, _index_state()) + + assert receipt.corroborated is False + assert [entry.verdict for entry in receipt.entries] == [ATTESTED] + + def test_requested_sort_without_rewrite_is_dropped_and_blocks(self) -> None: + ledger = AppliedTuningLedger() + ledger.record(INDEX_STATEMENT, PHASE_POST_LOAD) + ledger.record_dropped("sorting:LINEITEM (L_SHIPDATE)", "adapter rendered no statement") + + receipt = corroborate(ledger, _index_state()) + + assert receipt.corroborated is False + assert [entry.verdict for entry in receipt.entries if entry.verdict == ATTESTED] == [] + assert len(ledger.dropped) == 1 + + def test_failed_rewrite_blocks_corroboration(self) -> None: + from benchbox.core.tuning.applied_ledger import STATEMENT_FAILED + + ledger = AppliedTuningLedger() + ledger.record( + SORT_STATEMENT, + PHASE_POST_LOAD, + status=STATEMENT_FAILED, + mechanism="sorted_ingestion", + table="lineitem", + error=RuntimeError("denied"), + ) + + receipt = corroborate(ledger, _index_state()) + + assert receipt.corroborated is False + assert all(entry.verdict != ATTESTED for entry in receipt.entries) + + def test_attested_verdict_is_excluded_from_gate_totals(self) -> None: + receipt = corroborate(_sort_ledger(), _index_state()) + + assert receipt.summary["gate_relevant_total"] == 1 + assert receipt.summary[ATTESTED] == 1 + + +class TestAttestedSortColumns: + def test_parenthesized_order_by(self) -> None: + assert _attested_sort_columns("CREATE TABLE t (a INT) ORDER BY (a, b)") == ("a", "b") + + def test_bare_order_by_with_directions(self) -> None: + assert _attested_sort_columns("CREATE TABLE t AS SELECT * FROM t ORDER BY a DESC, b ASC") == ("a", "b") + + def test_missing_order_by_is_empty(self) -> None: + assert _attested_sort_columns("CREATE TABLE t (a INT)") == () + + +class TestAttestedPromotion: + def _adapter(self, ledger: AppliedTuningLedger) -> MagicMock: + adapter = MagicMock() + adapter._applied_tuning_ledger = ledger + introspector = MagicMock() + introspector.introspect.return_value = _index_state() + adapter.get_tuning_introspector.return_value = introspector + return adapter + + def test_rewrite_plus_index_promotes_to_verified(self) -> None: + from benchbox.platforms.base.tuning_trust import corroborate_applied_ledger + + status, payload = corroborate_applied_ledger(self._adapter(_sort_ledger()), MagicMock(), "applied_unverified") + + assert status == "applied_verified" + assert payload is not None + + def test_attested_only_stays_unverified(self) -> None: + from benchbox.platforms.base.tuning_trust import corroborate_applied_ledger + + ledger = AppliedTuningLedger() + ledger.record(SORT_STATEMENT, PHASE_POST_LOAD, mechanism="sorted_ingestion", table="lineitem") + status, _payload = corroborate_applied_ledger(self._adapter(ledger), MagicMock(), "applied_unverified") + + assert status == "applied_unverified" + + +class TestCtasSortRecordsLedgerStatement: + @pytest.fixture(autouse=True) + def _require_duckdb(self): + pytest.importorskip("duckdb") + + def test_successful_rewrite_records_post_load_statement(self) -> None: + import duckdb + + from benchbox.platforms.duckdb import DuckDBAdapter + from tests.unit.platforms.test_duckdb_sorted_loading import ( + _make_table_tuning_with_sort, + _make_tuning_config, + ) + + conn = duckdb.connect(":memory:") + conn.execute("CREATE TABLE lineitem (l_orderkey INTEGER, l_shipdate DATE)") + conn.execute("INSERT INTO lineitem VALUES (2, '2024-02-01'), (1, '2024-01-01')") + + with patch("benchbox.platforms.duckdb.duckdb"): + adapter = DuckDBAdapter() + adapter._applied_tuning_ledger = AppliedTuningLedger() + + result = adapter.apply_ctas_sort( + "lineitem", _make_tuning_config({"lineitem": _make_table_tuning_with_sort(["l_shipdate"])}), conn + ) + + assert result is True + recorded = [ + statement + for statement in adapter._applied_tuning_ledger.statements + if statement.mechanism == "sorted_ingestion" and statement.phase == PHASE_POST_LOAD + ] + assert len(recorded) == 1 + assert recorded[0].table == "lineitem" + post_load_executed = [ + statement + for statement in adapter._applied_tuning_ledger.statements + if statement.phase == PHASE_POST_LOAD and statement.status == "executed" + ] + assert len(post_load_executed) == 1 + + from benchbox.core.tuning.introspection import ATTESTED, IntrospectedState, corroborate + + receipt = corroborate(adapter._applied_tuning_ledger, IntrospectedState(platform="duckdb", objects=[])) + assert [entry.verdict for entry in receipt.entries] == [ATTESTED] + assert receipt.entries[0].reason == "physical sort executed; no catalog footprint" + assert list(receipt.entries[0].expected_columns) == ["l_shipdate"] + assert receipt.corroborated is False