fix(claims): verify query values and schemas

Author: Anionix

src/format_bench/claims/tsfile.py

@@ -7,21 +7,41 @@
 import pyarrow as pa
 import pyarrow.parquet as pq
 
+from format_bench.fair import FairOperation, result_evidence
 from format_bench.runner import stats_ms
 
 
 TABLE_NAME = "sensors"
 
 
-def _measure(function: Callable[[], int], warmups: int, iterations: int) -> dict:
+def _normalize_result(name: str, value: Any) -> Any:
+    if value is None:
+        return None
+    if isinstance(value, bytes):
+        value = value.decode("utf-8")
+    if name in {"temperature", "pressure"}:
+        return float(value)
+    if name == "sequence":
+        return int(value)
+    if name == "active":
+        return bool(value)
+    return str(value)
+
+
+def _measure(function: Callable[[], pa.Table], warmups: int, iterations: int) -> dict:
+    result = pa.table({})
     for _ in range(warmups):
-        function()
-    samples, result = [], 0
+        result = function()
+    samples = []
     for _ in range(iterations):
         started = time.perf_counter_ns()
         result = function()
         samples.append((time.perf_counter_ns() - started) / 1_000_000)
-    return {"timing": stats_ms(samples), "result": result}
+    return {
+        "timing": stats_ms(samples),
+        "result": result.num_rows,
+        "evidence": result_evidence(result, FairOperation.READ_ALL),
+    }
 
 
 def _schema() -> tuple[Any, list[str], list[Any]]:
@@ -181,26 +201,50 @@ def run_tsfile_claim(
     end = min(points_per_device, start + max(1, points_per_device // 10))
     device = f"device-{device_index:03d}"
 
-    def ts_query() -> int:
-        rows = 0
+    def ts_query() -> pa.Table:
+        rows = []
         with TsFileReader(str(tsfile_path)) as reader:
             with reader.query_table(
                 TABLE_NAME, columns, start, end - 1, tag_eq("device", device)
             ) as result:
                 while result.next():
-                    rows += 1
-        return rows
+                    rows.append(
+                        {
+                            name: _normalize_result(
+                                name, result.get_value_by_name(name)
+                            )
+                            for name in columns
+                        }
+                    )
+        return pa.Table.from_pylist(
+            rows,
+            schema=pa.schema(
+                [
+                    pa.field("device", pa.string()),
+                    pa.field("site", pa.string()),
+                    pa.field("temperature", pa.float64()),
+                    pa.field("pressure", pa.float64()),
+                    pa.field("active", pa.bool_()),
+                    pa.field("sequence", pa.int64()),
+                ]
+            ),
+        )
 
-    def parquet_query() -> int:
+    def parquet_query() -> pa.Table:
         return pq.read_table(
             parquet_path,
             columns=columns,
             filters=[("device", "=", device), ("timestamp", ">=", start), ("timestamp", "<", end)],
-        ).num_rows
+        )
 
     ts_result = _measure(ts_query, warmups, iterations)
     parquet_result = _measure(parquet_query, warmups, iterations)
-    status = "MEASURED" if ts_result["result"] == parquet_result["result"] else "FAILED"
+    status = (
+        "MEASURED"
+        if ts_result["result"] == parquet_result["result"]
+        and ts_result["evidence"] == parquet_result["evidence"]
+        else "FAILED"
+    )
     return {
         "status": status,
         "rows": devices * points_per_device,
@@ -213,4 +257,8 @@ def parquet_query() -> int:
         },
         "bytes": {"tsfile": tsfile_path.stat().st_size, "parquet": parquet_path.stat().st_size},
         "timing": {"tsfile": ts_result, "parquet": parquet_result},
+        "evidence": {
+            "tsfile": ts_result["evidence"],
+            "parquet": parquet_result["evidence"],
+        },
     }

src/format_bench/claims/vortex.py

@@ -10,25 +10,31 @@
 import vortex
 import vortex.expr as ve
 
+from format_bench.fair import FairOperation, result_evidence
 from format_bench.runner import stats_ms
 
 
 ROW_GROUP_SIZE = 4096
 PROJECTION = ["full_name", "repo_stars"]
 
 
-def _measure(function: Callable[[], int], warmups: int, iterations: int) -> dict:
+def _measure(function: Callable[[], pa.Table], warmups: int, iterations: int) -> dict:
+    result = pa.table({})
     for _ in range(warmups):
-        function()
-    samples, result = [], 0
+        result = function()
+    samples = []
     for _ in range(iterations):
         started = time.perf_counter_ns()
         result = function()
         samples.append((time.perf_counter_ns() - started) / 1_000_000)
-    return {"timing": stats_ms(samples), "result": result}
+    return {
+        "timing": stats_ms(samples),
+        "result": result.num_rows,
+        "evidence": result_evidence(result, FairOperation.READ_ALL),
+    }
 
 
-def _parquet_random_take(path: Path, indices: list[int]) -> int:
+def _parquet_random_take(path: Path, indices: list[int]) -> pa.Table:
     parquet = pq.ParquetFile(path)
     groups = sorted({index // ROW_GROUP_SIZE for index in indices})
     table = parquet.read_row_groups(groups, columns=PROJECTION)
@@ -37,7 +43,7 @@ def _parquet_random_take(path: Path, indices: list[int]) -> int:
         offsets[group] = offset
         offset += parquet.metadata.row_group(group).num_rows
     local = [offsets[index // ROW_GROUP_SIZE] + index % ROW_GROUP_SIZE for index in indices]
-    return table.take(pa.array(local, type=pa.int64())).num_rows
+    return table.take(pa.array(local, type=pa.int64()))
 
 
 def _write_compact(table: pa.Table, path: Path) -> None:
@@ -51,13 +57,12 @@ def _vortex_scan(
     *,
     expr=None,
     indices=None,
-) -> int:
+) -> pa.Table:
     source = vortex.open(str(path))
     return (
         source.scan(PROJECTION, expr=expr, indices=indices)
         .read_all()
         .to_arrow_table()
-        .num_rows
     )
 
 
@@ -78,21 +83,21 @@ def _variant(
     vortex_indices = vortex.array(pa.array(indices, type=pa.uint64()))
     operations = {
         "full_projection": (
-            lambda: pq.read_table(parquet_path, columns=PROJECTION).num_rows,
+            lambda: pq.read_table(parquet_path, columns=PROJECTION),
             lambda: _vortex_scan(vortex_path),
         ),
         "filter_popular": (
             lambda: pq.read_table(
                 parquet_path, columns=PROJECTION, filters=[("repo_stars", ">", 100000)]
-            ).num_rows,
+            ),
             lambda: _vortex_scan(
                 vortex_path, expr=ve.column("repo_stars") > 100000
             ),
         ),
         "filter_none": (
             lambda: pq.read_table(
                 parquet_path, columns=PROJECTION, filters=[("repo_stars", ">", 99_999_999)]
-            ).num_rows,
+            ),
             lambda: _vortex_scan(
                 vortex_path, expr=ve.column("repo_stars") > 99_999_999
             ),
@@ -106,7 +111,10 @@ def _variant(
     for operation, (parquet_fn, vortex_fn) in operations.items():
         parquet_result = _measure(parquet_fn, warmups, iterations)
         vortex_result = _measure(vortex_fn, warmups, iterations)
-        if parquet_result["result"] != vortex_result["result"]:
+        if (
+            parquet_result["result"] != vortex_result["result"]
+            or parquet_result["evidence"] != vortex_result["evidence"]
+        ):
             raise ValueError(f"stress result mismatch for {name}/{operation}")
         measured[operation] = {"parquet": parquet_result, "vortex": vortex_result}
     return {

src/format_bench/fair.py

@@ -39,6 +39,11 @@ class ResultEvidence(TypedDict):
     normalized_hash: str
 
 
+def _evidence_type(field: pa.Field) -> str:
+    # Arrow string_view is a storage-level view of the same UTF-8 logical type.
+    return "string" if pa.types.is_string_view(field.type) else str(field.type)
+
+
 def operations_for(manifest: Mapping[str, object] | None = None) -> tuple[str, ...]:
     if manifest is not None and "workloads" in manifest:
         workloads = load_workloads(manifest)
@@ -131,7 +136,11 @@ def result_evidence(
         "rows": table.num_rows,
         "columns": table.column_names,
         "schema": [
-            {"name": field.name, "type": str(field.type), "nullable": field.nullable}
+            {
+                "name": field.name,
+                "type": _evidence_type(field),
+                "nullable": field.nullable,
+            }
             for field in table.schema
         ],
         "row_order": "ORDER_SENSITIVE" if order_sensitive else "ORDER_INSENSITIVE",

tests/test_fair.py

@@ -34,6 +34,26 @@ def test_result_evidence_includes_type_and_nullability() -> None:
     )
 
 
+def test_result_evidence_distinguishes_same_rows_with_different_values() -> None:
+    original = pa.table({"value": [1, 2]})
+    changed = pa.table({"value": [1, 3]})
+
+    assert original.num_rows == changed.num_rows
+    assert result_evidence(original, FairOperation.READ_ALL) != result_evidence(
+        changed, FairOperation.READ_ALL
+    )
+
+
+def test_result_evidence_distinguishes_same_rows_with_different_schema() -> None:
+    original = pa.table({"value": pa.array([1, 2], type=pa.int64())})
+    changed = pa.table({"value": pa.array([1, 2], type=pa.float64())})
+
+    assert original.num_rows == changed.num_rows
+    assert result_evidence(original, FairOperation.READ_ALL) != result_evidence(
+        changed, FairOperation.READ_ALL
+    )
+
+
 def test_result_evidence_remains_order_insensitive() -> None:
     table = pa.table({"value": [1, 2, 3]})
     reversed_table = table.take(pa.array([2, 1, 0]))

tests/test_tsfile_claim.py

@@ -1,8 +1,13 @@
+import sys
 from pathlib import Path
+from types import ModuleType
 
+import pyarrow as pa
 import pytest
 
+import format_bench.claims.tsfile as tsfile_claim
 from format_bench.claims import run_tsfile_claim
+from format_bench.fair import FairOperation, result_evidence
 
 
 def test_tsfile_claim_matches_parquet_results(tmp_path: Path) -> None:
@@ -15,10 +20,44 @@ def test_tsfile_claim_matches_parquet_results(tmp_path: Path) -> None:
     assert result["rows"] == 20
     assert result["timing"]["tsfile"]["result"] == 1
     assert result["timing"]["parquet"]["result"] == 1
+    assert result["evidence"]["tsfile"] == result["evidence"]["parquet"]
+    assert result["evidence"]["tsfile"]["rows"] == 1
+    assert result["evidence"]["tsfile"]["normalized_hash"]
     assert result["bytes"]["tsfile"] > 0
     assert result["writer_settings"]["tsfile"]["default_compression_type_"] == "LZ4"
     assert result["writer_settings"]["tsfile"]["time_encoding_type_"] == "TS_2DIFF"
     assert result["writer_settings"]["parquet"] == {
         "compression": "zstd",
         "row_group_size": 10,
     }
+
+
+def test_tsfile_claim_rejects_equal_counts_with_different_values(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    fake_tsfile = ModuleType("tsfile")
+    fake_tsfile.TsFileReader = object
+    fake_tsfile.tag_eq = lambda *_: None
+    monkeypatch.setitem(sys.modules, "tsfile", fake_tsfile)
+
+    def fake_write(tsfile_path: Path, parquet_path: Path, *_: int):
+        tsfile_path.write_bytes(b"tsfile")
+        parquet_path.write_bytes(b"parquet")
+        return {"tsfile": 0.0, "parquet": 0.0}, {}
+
+    original = result_evidence(pa.table({"value": [1]}), FairOperation.READ_ALL)
+    changed = result_evidence(pa.table({"value": [2]}), FairOperation.READ_ALL)
+    measurements = iter(
+        [
+            {"timing": {}, "result": 1, "evidence": original},
+            {"timing": {}, "result": 1, "evidence": changed},
+        ]
+    )
+    monkeypatch.setattr(tsfile_claim, "_write_datasets", fake_write)
+    monkeypatch.setattr(tsfile_claim, "_measure", lambda *_: next(measurements))
+
+    result = tsfile_claim.run_tsfile_claim(
+        tmp_path, devices=1, points_per_device=1, warmups=0, iterations=1
+    )
+
+    assert result["status"] == "FAILED"

tests/test_vortex_adapter.py

@@ -1,11 +1,13 @@
 import json
 from pathlib import Path
 
+import pyarrow as pa
 import pytest
 
 import format_bench.claims.vortex as vortex_claim
 from format_bench.canonical import canonical_hash, query_counts, read_csv
 from format_bench.claims import run_vortex_stress
+from format_bench.fair import FairOperation, result_evidence
 from format_bench.formats import VortexAdapter
 
 
@@ -48,4 +50,23 @@ def tracked_open(path: str):
     for variant in ("sorted", "unsorted"):
         for operation in result[variant]["operations"].values():
             assert operation["parquet"]["result"] == operation["vortex"]["result"]
+            assert operation["parquet"]["evidence"] == operation["vortex"]["evidence"]
     assert open_count == 8
+
+
+def test_vortex_stress_rejects_equal_counts_with_different_values(
+    tmp_path: Path, fixture_contract, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _, table = fixture_contract
+    original = result_evidence(pa.table({"value": [1]}), FairOperation.READ_ALL)
+    changed = result_evidence(pa.table({"value": [2]}), FairOperation.READ_ALL)
+    measurements = iter(
+        [
+            {"result": 1, "evidence": original},
+            {"result": 1, "evidence": changed},
+        ]
+    )
+    monkeypatch.setattr(vortex_claim, "_measure", lambda *_: next(measurements))
+
+    with pytest.raises(ValueError, match="stress result mismatch"):
+        run_vortex_stress(table, tmp_path, rows=8, warmups=0, iterations=1)