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