You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sedona.apache.org by ji...@apache.org on 2026/07/24 07:48:02 UTC
(sedona-spatialbench) branch main updated: Verify engine correctness in a downstream job that reuses benchmark results (#128)
This is an automated email from the ASF dual-hosted git repository.
jiayuasu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/sedona-spatialbench.git
The following commit(s) were added to refs/heads/main by this push:
new acf21b5 Verify engine correctness in a downstream job that reuses benchmark results (#128)
acf21b5 is described below
commit acf21b55dd67bdee0250aea1fa815e28a394a2d2
Author: Jia Yu <ji...@apache.org>
AuthorDate: Fri Jul 24 00:47:57 2026 -0700
Verify engine correctness in a downstream job that reuses benchmark results (#128)
* Verify engine correctness in a downstream job that reuses benchmark results
Adds a correctness check that reuses the benchmark run instead of re-executing
queries in a second workflow: the benchmark jobs already run every
(engine, query, scale factor), so they now also dump each result, and a
downstream verify job compares those dumps to the committed ground-truth answers
(benchmark/answers/sf<sf>). A mismatch — or a result that reported success but is
missing — fails CI.
run_benchmark.py:
- New --result-dir: reuse the timed run's result (no extra execution) to write
<engine>_<query>_result.csv. DuckDB captures column names during the timed run
so its fetchall result becomes a DataFrame without re-running; other engines
reuse the DataFrame they already return. Normalizes durations -> seconds,
decimals/object-decimals -> float, timestamps -> datetime. Writes csv only, so
no pyarrow filesystem is touched inside a SedonaDB worker; pandas/numpy stay out
of the module top to preserve the import-order isolation. Gated on committed
answers existing for the scale factor, so SF10 (no answers yet) does no extra
work.
benchmark/verify_results.py (new, engine-free):
- Derives comparison semantics from the canonical answer schema (the type-faithful
q<n>.parquet): integer keys/counts, timestamps and strings match exactly, and
only floating-point metrics use rtol/atol.
- The LIMIT-boundary-tie exception is restricted to eligible capped LIMIT queries
and only tolerates a final-row identity swap where every float ordering metric
still ties within tolerance — never a single-row/non-LIMIT query or a numeric
difference.
- Result-presence semantics: reported-success-but-no-dump fails (a silent dump
failure); explicit timeout/error/OOM is tolerated; a framework that produced no
record at all (its whole benchmark job failed, e.g. an install error) is shown
as "no result" but does not fail the gate — it left no answer that could be
wrong, and its own red job surfaces it.
- Robust by construction: a result that cannot be read/compared becomes a warning
cell, never a crash, so one broken framework can't take down the summary.
- Renders a per-engine correctness table to the job summary and exits non-zero on
any real failure.
benchmark.yml:
- Engine jobs pass --result-dir and upload the result csv alongside the timing
json; pandas added to the Spatial Polars / PyCanopy jobs.
- New verify-correctness job: needs the benchmark jobs (runs even if some failed),
installs pandas + pyarrow, runs verify_results.py, always publishes the table to
the job summary (with a fallback note if none was produced), and fails CI on a
mismatch.
Part of #126.
* Fix engine query implementations flagged by the correctness check
The correctness verification (verify_results.py) surfaced five (engine, query)
results that diverged from the SedonaDB/DuckDB ground-truth answers. All are real
implementation bugs, not tolerance issues:
Q5 was missing a whole column. GeoPandas and PyCanopy computed the repeat-customer
trip count for the `> 5` filter but dropped it from the output; the reference (and
the answer) return it as `dropoff_count`. Add it to the projection.
Spatial Polars Q7 used the wrong metre/degree constant. It multiplied the
straight-line degree distance by 111111, while the reference divides by 0.000009
(= x111111.111...); the ~1e-6 gap tripped the tolerance. Use /0.000009 to match the
reference and the other engines.
PyCanopy Q6/Q10 lost precision on avg_distance. t_distance is decimal(15,5), and a
decimal mean stays at scale 5 and rounds the result (e.g. 0.00086 vs 0.000865);
SedonaDB and the other engines average in float. Cast t_distance to float before
averaging.
Verified at SF1 against the committed answers: GeoPandas Q5, Spatial Polars Q7 and
PyCanopy Q10 now match. PyCanopy Q5/Q6 use the same fixes (column projection and the
identical float cast) but need PyCanopy 0.3.3 (Linux-only wheel) to run, so they are
verified by CI rather than locally.
Part of #126.
* Harden the correctness check against review feedback
Four fixes so the correctness gate cannot pass without genuinely verifying results:
- Fail when nothing was verified. A failed artifact download now fails the verify
job (the download step no longer swallows errors), and verify_results.py fails if
zero (engine, query) pairs were actually compared — so a missing/empty results set
can't pass silently. A single framework that produced no result is still reported
(❔) without failing, so the summary and the gate both survive one broken engine.
- Treat an unreadable result as a failure. A result file that exists but cannot be
read/compared is now a failing verdict (was reported but non-failing); the summary
still renders (crash-safe per cell). run_benchmark.py writes each dump to a temp
file and atomically replaces the destination, so a process killed mid-write never
leaves a partial csv.
- Remove Q8 from the LIMIT-boundary-tie tolerance. Q8 orders by an integer count
with an integer-key tiebreaker, so its ordering is deterministic and must match
exactly; the boundary exception only guards float ordering ties and would otherwise
accept an arbitrary final-row count.
- Keep result serialization outside the query timeout. The worker queues its timing
result before serializing; if the process is still alive at the timeout, the runner
checks the queue and, if the query already finished, gives serialization a bounded
grace instead of charging it to the query timeout.
Part of #126.
---
.github/workflows/benchmark.yml | 111 ++++++++-
benchmark/answers/README.md | 29 ++-
benchmark/run_benchmark.py | 187 +++++++++++++--
benchmark/verify_results.py | 378 ++++++++++++++++++++++++++++++
spatialbench-queries/geopandas_queries.py | 4 +-
spatialbench-queries/pycanopy_queries.py | 14 +-
spatialbench-queries/spatial_polars.py | 4 +-
7 files changed, 685 insertions(+), 42 deletions(-)
diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml
index d8ebf4d..07b8f5e 100644
--- a/.github/workflows/benchmark.yml
+++ b/.github/workflows/benchmark.yml
@@ -263,6 +263,7 @@ jobs:
--runs ${{ env.BENCHMARK_RUNS }} \
--timeout ${{ env.QUERY_TIMEOUT }} \
--scale-factor ${{ matrix.scale_factor }} \
+ --result-dir . \
--output duckdb_${{ matrix.query }}_results.json
- name: Upload results
@@ -270,7 +271,10 @@ jobs:
uses: actions/upload-artifact@v7
with:
name: duckdb-${{ matrix.query }}-results-sf${{ matrix.scale_factor }}
- path: duckdb_${{ matrix.query }}_results.json
+ path: |
+ duckdb_${{ matrix.query }}_results.json
+ duckdb_${{ matrix.query }}_result.csv
+ if-no-files-found: warn
retention-days: 30
benchmark-geopandas:
@@ -319,6 +323,7 @@ jobs:
--runs ${{ env.BENCHMARK_RUNS }} \
--timeout ${{ env.QUERY_TIMEOUT }} \
--scale-factor ${{ matrix.scale_factor }} \
+ --result-dir . \
--output geopandas_${{ matrix.query }}_results.json
- name: Upload results
@@ -326,7 +331,10 @@ jobs:
uses: actions/upload-artifact@v7
with:
name: geopandas-${{ matrix.query }}-results-sf${{ matrix.scale_factor }}
- path: geopandas_${{ matrix.query }}_results.json
+ path: |
+ geopandas_${{ matrix.query }}_results.json
+ geopandas_${{ matrix.query }}_result.csv
+ if-no-files-found: warn
retention-days: 30
benchmark-sedonadb:
@@ -380,6 +388,7 @@ jobs:
--runs ${{ env.BENCHMARK_RUNS }} \
--timeout ${{ env.QUERY_TIMEOUT }} \
--scale-factor ${{ matrix.scale_factor }} \
+ --result-dir . \
--output sedonadb_${{ matrix.query }}_results.json
- name: Upload results
@@ -387,7 +396,10 @@ jobs:
uses: actions/upload-artifact@v7
with:
name: sedonadb-${{ matrix.query }}-results-sf${{ matrix.scale_factor }}
- path: sedonadb_${{ matrix.query }}_results.json
+ path: |
+ sedonadb_${{ matrix.query }}_results.json
+ sedonadb_${{ matrix.query }}_result.csv
+ if-no-files-found: warn
retention-days: 30
benchmark-spatial-polars:
@@ -421,9 +433,9 @@ jobs:
- name: Install dependencies
run: |
if [ -n "${{ env.SPATIAL_POLARS_VERSION }}" ]; then
- pip install "spatial-polars[knn]==${{ env.SPATIAL_POLARS_VERSION }}" pyarrow
+ pip install "spatial-polars[knn]==${{ env.SPATIAL_POLARS_VERSION }}" pyarrow pandas
else
- pip install "spatial-polars[knn]" pyarrow
+ pip install "spatial-polars[knn]" pyarrow pandas
fi
echo "Installed Spatial Polars version: $(python -c 'from importlib.metadata import version; print(version("spatial-polars"))')"
@@ -436,6 +448,7 @@ jobs:
--runs ${{ env.BENCHMARK_RUNS }} \
--timeout ${{ env.QUERY_TIMEOUT }} \
--scale-factor ${{ matrix.scale_factor }} \
+ --result-dir . \
--output spatial_polars_${{ matrix.query }}_results.json
- name: Upload results
@@ -443,7 +456,10 @@ jobs:
uses: actions/upload-artifact@v7
with:
name: spatial_polars-${{ matrix.query }}-results-sf${{ matrix.scale_factor }}
- path: spatial_polars_${{ matrix.query }}_results.json
+ path: |
+ spatial_polars_${{ matrix.query }}_results.json
+ spatial_polars_${{ matrix.query }}_result.csv
+ if-no-files-found: warn
retention-days: 30
benchmark-pycanopy:
@@ -477,9 +493,9 @@ jobs:
- name: Install dependencies
run: |
if [ -n "${{ env.PYCANOPY_VERSION }}" ]; then
- pip install "pycanopy==${{ env.PYCANOPY_VERSION }}" pyarrow
+ pip install "pycanopy==${{ env.PYCANOPY_VERSION }}" pyarrow pandas
else
- pip install pycanopy pyarrow
+ pip install pycanopy pyarrow pandas
fi
echo "Installed PyCanopy version: $(python -c 'from importlib.metadata import version; print(version("pycanopy"))')"
@@ -492,6 +508,7 @@ jobs:
--runs ${{ env.BENCHMARK_RUNS }} \
--timeout ${{ env.QUERY_TIMEOUT }} \
--scale-factor ${{ matrix.scale_factor }} \
+ --result-dir . \
--output pycanopy_${{ matrix.query }}_results.json
- name: Upload results
@@ -499,7 +516,10 @@ jobs:
uses: actions/upload-artifact@v7
with:
name: pycanopy-${{ matrix.query }}-results-sf${{ matrix.scale_factor }}
- path: pycanopy_${{ matrix.query }}_results.json
+ path: |
+ pycanopy_${{ matrix.query }}_results.json
+ pycanopy_${{ matrix.query }}_result.csv
+ if-no-files-found: warn
retention-days: 30
summarize-results:
@@ -549,3 +569,76 @@ jobs:
results/
benchmark_summary.md
retention-days: 90
+
+ # ── Correctness verification ──
+ # Depends on (needs) the benchmark jobs and reuses their result dumps
+ # (<engine>_<query>_result.csv) instead of re-running any query. Compares each
+ # against the committed ground-truth answers (benchmark/answers/sf<sf>) in a
+ # clean, engine-free environment. Fails the job — and thus CI — if any engine's
+ # result does not match its committed answer.
+ verify-correctness:
+ name: Verify Correctness (SF${{ matrix.scale_factor }})
+ needs: [parse-scale-factors, benchmark-duckdb, benchmark-geopandas, benchmark-sedonadb, benchmark-spatial-polars, benchmark-pycanopy]
+ if: always() && (needs.benchmark-duckdb.result != 'cancelled' || needs.benchmark-geopandas.result != 'cancelled' || needs.benchmark-sedonadb.result != 'cancelled' || needs.benchmark-spatial-polars.result != 'cancelled' || needs.benchmark-pycanopy.result != 'cancelled')
+ runs-on: ubuntu-latest
+ strategy:
+ matrix:
+ scale_factor: ${{ fromJson(needs.parse-scale-factors.outputs.matrix) }}
+ steps:
+ - uses: actions/checkout@v7
+
+ # No continue-on-error: a failed artifact download must fail the correctness
+ # job rather than let it verify nothing. (A run that produced no artifacts is
+ # not a download error; verify_results.py fails that case via its
+ # "no results verified" guard.)
+ - name: Download all results for this scale factor
+ uses: actions/download-artifact@v8
+ with:
+ pattern: '*-results-sf${{ matrix.scale_factor }}'
+ path: results
+ merge-multiple: true
+
+ - name: Setup Python
+ uses: actions/setup-python@v6
+ with:
+ python-version: '3.11'
+
+ - name: Install pandas
+ run: pip install pandas pyarrow
+
+ - name: Verify results against committed answers
+ id: verify
+ continue-on-error: true
+ run: |
+ python benchmark/verify_results.py \
+ --results-dir results \
+ --scale-factor ${{ matrix.scale_factor }} \
+ --engines ${{ env.BENCHMARK_ENGINES }} \
+ --output correctness_summary.md
+
+ - name: Display and publish correctness summary
+ if: always()
+ run: |
+ if [ -f correctness_summary.md ]; then
+ cat correctness_summary.md
+ cat correctness_summary.md >> "$GITHUB_STEP_SUMMARY"
+ else
+ msg="### ⚠️ Correctness summary not generated (verify_results.py produced no output)."
+ echo "$msg"
+ echo "$msg" >> "$GITHUB_STEP_SUMMARY"
+ fi
+
+ - name: Upload correctness summary
+ if: always()
+ uses: actions/upload-artifact@v7
+ with:
+ name: correctness-summary-sf${{ matrix.scale_factor }}
+ path: correctness_summary.md
+ if-no-files-found: warn
+ retention-days: 90
+
+ - name: Fail if any result did not match
+ if: always() && steps.verify.outcome == 'failure'
+ run: |
+ echo "::error::One or more engine results did not match the committed answers (SF${{ matrix.scale_factor }})."
+ exit 1
diff --git a/benchmark/answers/README.md b/benchmark/answers/README.md
index 04054ec..92d367d 100644
--- a/benchmark/answers/README.md
+++ b/benchmark/answers/README.md
@@ -19,8 +19,10 @@
# SpatialBench ground-truth answers
-Reference results for the SpatialBench queries, used by the correctness harness to
-verify that every participating engine returns the same answer for the same query.
+Reference results for the SpatialBench queries. A downstream CI job
+(`verify_results.py`) reuses the benchmark run's result dumps — instead of
+re-executing any query — and compares each engine's result against these answers.
+A mismatch **fails CI**.
## Layout
@@ -37,10 +39,16 @@ Each query's expected result is committed in two formats, written from the same
normalized frame:
- **`q<n>.parquet`** — the type-faithful canonical answer (timestamps stay
- timestamps, ints stay ints). The correctness harness compares against this.
+ timestamps, ints stay ints). The verify job reads this to know each column's type,
+ so comparison semantics come from the schema: integer keys/counts, timestamps and
+ strings are matched exactly, and only floating-point metrics use tolerance.
- **`q<n>.csv`** — a review companion: GitHub renders it as a table and diffs are
readable when an answer changes.
+The benchmark run dumps each engine's result to a csv (never parquet), so no pyarrow
+filesystem is touched inside a SedonaDB worker; the engine-free verify job then reads
+the answer parquet (types) and that csv (values) and compares them.
+
Most queries are bounded to at most 100 rows (see #124), so the fixtures are tiny. The one
exception is **Q4**, which groups the top-1000 tipped trips by zone and returns one row per
zone (~260 at SF1); it is inherently small and bounded (never more than the number of zones
@@ -67,18 +75,25 @@ Row order is significant and preserved: every query has a deterministic `ORDER B
(with key tiebreakers) followed by `LIMIT`, so the expected rows and their order are
well defined.
-## Comparison semantics (for the harness)
+## Comparison semantics
+
+Columns are compared by position (engines name columns differently, e.g.
+`avg_duration` vs `avg_duration_seconds`), with:
- Integer keys, strings, and timestamps: exact match.
- Floats (distances, areas, IoU, seconds): relative + absolute tolerance
(`rtol=1e-6`, `atol=1e-9`) to absorb cross-engine floating-point differences.
- The final row may legitimately differ across engines when a float metric ties near
- the `LIMIT` boundary; the harness treats a within-tolerance boundary difference as a
- pass.
+ the `LIMIT` boundary; a within-tolerance boundary difference is treated as a pass.
+
+An engine that cannot compute a query (timeout / error / OOM, e.g. DuckDB's Q12
+below) is reported but does **not** fail the job — that is a runtime issue surfaced
+by the benchmark summary, not a wrong answer. Only an actual mismatch fails CI.
## Caveat: Q12 at SF1
DuckDB has no KNN operator, so its Q12 uses a lateral cross-join that is infeasible at
SF1 (it does not finish in reasonable time). Q12 is therefore **not** cross-checked by
DuckDB here — it is validated by the KNN-capable engines (SedonaDB, Spatial Polars,
-PyCanopy) in the correctness harness.
+PyCanopy). In the verify job, DuckDB's Q12 shows as "could not compute" rather than a
+failure.
diff --git a/benchmark/run_benchmark.py b/benchmark/run_benchmark.py
index e2dd5a0..6b04f09 100755
--- a/benchmark/run_benchmark.py
+++ b/benchmark/run_benchmark.py
@@ -26,6 +26,7 @@ on the SpatialBench queries at a specified scale factor.
import argparse
import json
import multiprocessing
+import os
import signal
import sys
import time
@@ -43,6 +44,14 @@ sys.path.append(str(Path(__file__).parent.parent / "spatialbench-queries"))
# Constants
QUERY_COUNT = 12
TABLES = ["building", "customer", "driver", "trip", "vehicle", "zone"]
+DURATION_SUFFIX = "_seconds"
+# When --result-dir is set, the worker serializes the (already computed) result
+# after queueing its timing. If the process is still alive at the query timeout, we
+# briefly probe the queue: a result already there means the query finished and only
+# serialization is running, which then gets its own grace window rather than being
+# charged to the query timeout.
+SERIALIZE_PROBE_SECONDS = 5
+SERIALIZE_GRACE_SECONDS = 120
@dataclass
@@ -97,11 +106,17 @@ def _run_query_in_process(
data_paths: dict[str, str],
query_name: str,
query_sql: str | None,
+ dump_csv: str | None = None,
):
"""Worker function to run a query in a separate process.
This allows us to forcefully terminate queries that hang or consume
too much memory, which SIGALRM cannot do for native code.
+
+ If dump_csv is set, the *timed* result is reused (no re-execution) to write the
+ normalized result there for the correctness verify job. The timing result is put
+ on the queue before dumping, so a dump failure never affects the timing; the
+ verify job reports a missing csv as a failed correctness check.
"""
try:
# For Spatial Polars, ensure the package is imported first to register namespace
@@ -112,7 +127,7 @@ def _run_query_in_process(
benchmark.setup()
try:
start_time = time.perf_counter()
- row_count, _ = benchmark.execute_query(query_name, query_sql)
+ row_count, result = benchmark.execute_query(query_name, query_sql)
elapsed = time.perf_counter() - start_time
result_queue.put({
"status": "success",
@@ -120,6 +135,15 @@ def _run_query_in_process(
"row_count": row_count,
"error_message": None,
})
+ if dump_csv:
+ try:
+ # Write to a temp file and atomically replace, so a process killed
+ # mid-serialization never leaves a partial csv at the final path.
+ tmp_csv = f"{dump_csv}.tmp"
+ normalize(benchmark.dump_frame(result, query_sql)).to_csv(tmp_csv, index=False)
+ os.replace(tmp_csv, dump_csv)
+ except Exception as e:
+ print(f" [dump] failed to write {dump_csv}: {e}", file=sys.stderr, flush=True)
finally:
benchmark.teardown()
except Exception as e:
@@ -131,6 +155,61 @@ def _run_query_in_process(
})
+# ── Result dumping (for correctness verification) ──
+# The benchmark run optionally writes each query's normalized result to a csv so a
+# downstream, engine-free CI job (verify_results.py) can compare it to the committed
+# ground-truth answer. The dump reuses the *timed* run's result (no extra query
+# execution) and writes csv only (never parquet), so no pyarrow filesystem is
+# touched inside a SedonaDB worker. pandas/numpy are imported lazily so the parent
+# never pulls them in before forking a SedonaDB worker (which bundles its own Arrow
+# and segfaults if pyarrow/pandas load first).
+
+
+def normalize(df):
+ """Coerce an engine result to the canonical, engine-neutral form.
+
+ Matches the normalization used to generate the committed answers: durations
+ become float seconds (with a ``_seconds`` suffix), decimals become floats, and
+ timestamps become microsecond datetimes. Object columns of Decimal / timedelta
+ (which DuckDB's fetchall yields) are handled too.
+ """
+ import datetime as _dt
+ import decimal
+
+ import pandas as pd
+
+ df = df.reset_index(drop=True).copy()
+ rename = {}
+ for col in df.columns:
+ s = df[col]
+ if pd.api.types.is_timedelta64_dtype(s):
+ df[col] = s.dt.total_seconds().astype(float)
+ if not str(col).endswith(DURATION_SUFFIX):
+ rename[col] = f"{col}{DURATION_SUFFIX}"
+ elif s.dtype == object:
+ non_null = s.dropna()
+ if len(non_null) and all(isinstance(v, decimal.Decimal) for v in non_null):
+ df[col] = s.astype(float)
+ elif len(non_null) and all(isinstance(v, _dt.timedelta) for v in non_null):
+ df[col] = s.map(lambda v: v.total_seconds() if isinstance(v, _dt.timedelta) else v).astype(float)
+ if not str(col).endswith(DURATION_SUFFIX):
+ rename[col] = f"{col}{DURATION_SUFFIX}"
+ elif pd.api.types.is_datetime64_any_dtype(s):
+ df[col] = s.astype("datetime64[us]")
+ return df.rename(columns=rename)
+
+
+def _to_pandas_frame(result):
+ """Coerce whatever an engine's query returns into a plain pandas DataFrame."""
+ import pandas as pd
+
+ if isinstance(result, pd.DataFrame):
+ return pd.DataFrame(result) # strip GeoDataFrame subclass; keep values
+ if hasattr(result, "to_pandas"): # polars (spatial_polars, pycanopy)
+ return result.to_pandas()
+ return pd.DataFrame(result)
+
+
def get_data_paths(data_dir: str) -> dict[str, str]:
"""Get paths to all data tables.
@@ -189,6 +268,14 @@ class BaseBenchmark(ABC):
"""Execute a query and return (row_count, result)."""
pass
+ def dump_frame(self, result, query: str | None):
+ """Convert a *timed* query result into a pandas DataFrame for dumping.
+
+ Reuses the result from execute_query (no re-execution). DuckDB overrides
+ this because its execute_query returns raw rows without column names.
+ """
+ return _to_pandas_frame(result)
+
def run_query(self, query_name: str, query: str | None = None, timeout: int = 1200) -> BenchmarkResult:
"""Run a single query with timeout handling."""
start_time = time.perf_counter()
@@ -242,6 +329,7 @@ class DuckDBBenchmark(BaseBenchmark):
def __init__(self, data_paths: dict[str, str]):
super().__init__(data_paths, "duckdb")
self._conn = None
+ self._last_columns = None
def setup(self) -> None:
import duckdb
@@ -261,9 +349,17 @@ class DuckDBBenchmark(BaseBenchmark):
self._conn = None
def execute_query(self, query_name: str, query: str | None) -> tuple[int, Any]:
- result = self._conn.execute(query).fetchall()
+ rel = self._conn.execute(query)
+ # Capture column names (cheap, from the cursor) so the timed fetchall result
+ # can be turned into a DataFrame later without re-running the query.
+ self._last_columns = [d[0] for d in rel.description]
+ result = rel.fetchall()
return len(result), result
+ def dump_frame(self, result, query: str | None):
+ import pandas as pd
+ return pd.DataFrame(result, columns=self._last_columns)
+
class GeoPandasBenchmark(BaseBenchmark):
"""GeoPandas benchmark runner."""
@@ -387,6 +483,7 @@ def run_query_isolated(
query_name: str,
query_sql: str | None,
timeout: int,
+ dump_csv: Path | None = None,
) -> BenchmarkResult:
"""Run a single query in an isolated subprocess with hard timeout.
@@ -394,26 +491,55 @@ def run_query_isolated(
1. Native code (C++/Rust) can be forcefully terminated
2. Memory-hungry queries don't affect the main process
3. Crashed queries don't invalidate the benchmark runner
+
+ If dump_csv is set, the timed result is also written there (reused, not re-run)
+ for the correctness verify job.
"""
result_queue = multiprocessing.Queue()
process = multiprocessing.Process(
target=_run_query_in_process,
- args=(result_queue, engine_class, data_paths, query_name, query_sql),
+ args=(result_queue, engine_class, data_paths, query_name, query_sql,
+ str(dump_csv) if dump_csv else None),
)
+ def _kill():
+ process.terminate()
+ process.join(timeout=5)
+ if process.is_alive():
+ process.kill()
+ process.join(timeout=2)
+
+ def _from_queue(result_data):
+ return BenchmarkResult(
+ query=query_name,
+ engine=engine_name,
+ time_seconds=result_data["time_seconds"],
+ row_count=result_data["row_count"],
+ status=result_data["status"],
+ error_message=result_data["error_message"],
+ )
+
process.start()
process.join(timeout=timeout)
if process.is_alive():
- # Query exceeded timeout - forcefully terminate
- process.terminate()
- process.join(timeout=5) # Give it 5 seconds to terminate gracefully
+ # The query didn't signal completion within the timeout. It may, however, have
+ # finished and left only the (reused) result serialization running — that must
+ # not count against the query timeout. The worker queues its timing result
+ # *before* serializing, so a result already on the queue means the query
+ # finished; give serialization a bounded grace. An empty queue means the query
+ # itself is still running: a real timeout.
+ try:
+ result_data = result_queue.get(timeout=SERIALIZE_PROBE_SECONDS)
+ except Exception:
+ result_data = None
- if process.is_alive():
- # Still alive - kill it
- process.kill()
- process.join(timeout=2)
+ if result_data is not None:
+ process.join(timeout=SERIALIZE_GRACE_SECONDS)
+ _kill()
+ return _from_queue(result_data)
+ _kill()
return BenchmarkResult(
query=query_name,
engine=engine_name,
@@ -425,15 +551,7 @@ def run_query_isolated(
# Process completed - get result from queue
try:
- result_data = result_queue.get_nowait()
- return BenchmarkResult(
- query=query_name,
- engine=engine_name,
- time_seconds=result_data["time_seconds"],
- row_count=result_data["row_count"],
- status=result_data["status"],
- error_message=result_data["error_message"],
- )
+ return _from_queue(result_queue.get_nowait())
except Exception:
# Process died without putting result in queue
return BenchmarkResult(
@@ -454,6 +572,7 @@ def run_benchmark(
scale_factor: float,
runs: int = 3,
output_file: str | None = None,
+ result_dir: Path | None = None,
) -> BenchmarkSuite:
"""Generic benchmark runner for any engine.
@@ -550,7 +669,8 @@ def run_benchmark(
for idx, (query_name, query_sql) in enumerate(query_items):
print(f" Running {query_name}...", end=" ", flush=True)
- # First run
+ # First run — also dump the (reused) result for the verify job.
+ dump_csv = result_dir / f"{engine}_{query_name}_result.csv" if result_dir else None
result = run_query_isolated(
engine_class=engine_class,
engine_name=engine,
@@ -558,6 +678,7 @@ def run_benchmark(
query_name=query_name,
query_sql=query_sql,
timeout=timeout,
+ dump_csv=dump_csv,
)
# If first run succeeded and we want multiple runs, do additional runs
@@ -674,6 +795,15 @@ def main():
help="Output file for results")
parser.add_argument("--scale-factor", type=float, default=1,
help="Scale factor of the data (for reporting only)")
+ parser.add_argument("--result-dir", type=str, default=None,
+ help="If set, write each query's normalized result to "
+ "<result-dir>/<engine>_<query>_result.csv for the downstream "
+ "correctness verify job. Only done for scale factors that have "
+ "committed answers (benchmark/answers/sf<sf>).")
+ parser.add_argument("--answers-dir", type=str, default=None,
+ help="Directory of committed answers for this scale factor "
+ "(default: benchmark/answers/sf<sf>). Gates --result-dir: "
+ "results are dumped only when this directory exists.")
args = parser.parse_args()
@@ -696,8 +826,23 @@ def main():
for table, path in data_paths.items():
print(f" {table}: {path}")
+ # Resolve where to dump normalized results (for the downstream verify job).
+ # Only dump for scale factors that have committed answers to compare against.
+ result_dir = None
+ if args.result_dir:
+ sf = args.scale_factor
+ sf_tag = f"sf{int(sf) if float(sf).is_integer() else sf}"
+ answers_dir = Path(args.answers_dir) if args.answers_dir else Path(__file__).parent / "answers" / sf_tag
+ if answers_dir.is_dir():
+ result_dir = Path(args.result_dir)
+ result_dir.mkdir(parents=True, exist_ok=True)
+ print(f"Dumping normalized results to {result_dir} (answers exist at {answers_dir})")
+ else:
+ print(f"No committed answers at {answers_dir} — results not dumped, correctness not checked for this scale factor")
+
results = [
- run_benchmark(engine, data_paths, queries, args.timeout, args.scale_factor, args.runs, args.output)
+ run_benchmark(engine, data_paths, queries, args.timeout, args.scale_factor,
+ args.runs, args.output, result_dir)
for engine in engines
]
diff --git a/benchmark/verify_results.py b/benchmark/verify_results.py
new file mode 100755
index 0000000..ba95b0e
--- /dev/null
+++ b/benchmark/verify_results.py
@@ -0,0 +1,378 @@
+#!/usr/bin/env python3
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Verify SpatialBench correctness by comparing the benchmark run's result dumps
+against the committed ground-truth answers, and render a summary table.
+
+This runs as a downstream CI job that *needs* the benchmark jobs and reuses their
+output: the benchmark run writes each query's normalized result to
+``<engine>_<query>_result.csv`` (see run_benchmark.py --result-dir); this script
+reads those dumps plus the committed answers in ``benchmark/answers/sf<sf>`` and
+compares them. It is engine-free (only pandas/numpy/pyarrow), so none of the
+SedonaDB / pyarrow import-order pitfalls apply here.
+
+Comparison semantics come from the canonical answer schema — the committed
+``q<n>.parquet``, which is type-faithful:
+ * columns compared by position (engines name e.g. avg_duration differently)
+ * integer keys / counts: exact
+ * timestamps: exact (parsed, format-independent)
+ * strings: exact
+ * floating-point metrics (distances, areas, IoU, seconds): rtol=1e-6, atol=1e-9
+ * a within-tolerance tie confined to the final row of a *capped* LIMIT query is
+ tolerated (float ordering metrics can tie across the LIMIT boundary and swap
+ the last row's identity); this exception never applies to non-LIMIT queries,
+ to results below the cap, or when a numeric metric actually differs.
+
+Exit status: non-zero if any engine's result MISMATCHES its committed answer, or a
+result that should exist is missing — so a regression fails CI. An engine that
+explicitly could not compute a query (timeout / error / OOM, e.g. DuckDB's
+lateral-join Q12 at scale) is reported but does not fail the job.
+"""
+
+import argparse
+import json
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+
+QUERY_COUNT = 12
+ENGINES = ["duckdb", "geopandas", "sedonadb", "spatial_polars", "pycanopy"]
+ENGINE_ICONS = {
+ "sedonadb": "🌵 SedonaDB",
+ "duckdb": "🦆 DuckDB",
+ "geopandas": "🐼 GeoPandas",
+ "spatial_polars": "🐻❄️ Spatial Polars",
+ "pycanopy": "🌴 PyCanopy",
+}
+
+# Queries eligible for the LIMIT-boundary-tie tolerance (see #124): bounded by
+# ORDER BY <float metric> ... LIMIT, so two rows can tie on the float metric within
+# tolerance and swap the final row's identity. Q8 is deliberately excluded — it
+# orders by an integer count (nearby_pickup_count) with an integer-key tiebreaker,
+# so its ordering is fully deterministic and must match exactly.
+LIMIT_QUERIES = {"q1", "q5", "q7", "q9", "q10", "q12"}
+LIMIT_CAP = 100
+
+# Verdicts that must fail CI: a wrong value, a result that reported success but is
+# missing, or a result file that exists but cannot be read/compared.
+FAILING = {"fail", "missing", "verify_error"}
+
+VERDICT_SYMBOL = {
+ "pass": "✅",
+ "fail": "❌",
+ "missing": "🚫",
+ "timeout": "⏱️",
+ "oom": "💀",
+ "run_error": "⚠️",
+ "verify_error": "❌",
+ "no_data": "❔",
+ "no_answer": "—",
+}
+
+
+def column_kind(dtype) -> str:
+ """Classify a canonical (answer parquet) column dtype for comparison."""
+ if pd.api.types.is_float_dtype(dtype):
+ return "float"
+ if pd.api.types.is_integer_dtype(dtype):
+ return "int"
+ if pd.api.types.is_datetime64_any_dtype(dtype):
+ return "datetime"
+ return "exact"
+
+
+def compare(answer: pd.DataFrame, result: pd.DataFrame, rtol: float, atol: float) -> list[dict]:
+ """Positional comparison. ``answer`` is the type-faithful parquet frame, whose
+ per-column dtype decides the semantics; ``result`` is the engine's csv dump.
+
+ Returns a list of structured issues (kind / first_row / count / message). Only
+ floating-point columns use tolerance — integer keys, timestamps and strings are
+ matched exactly, so e.g. key 1451371 vs 1451372.0 is a mismatch, not a tie.
+ """
+ if answer.shape[1] != result.shape[1]:
+ return [{
+ "kind": "shape", "first_row": -1, "count": 0,
+ "message": (f"column count differs: answer {answer.shape[1]} "
+ f"({list(answer.columns)}) vs engine {result.shape[1]} ({list(result.columns)})"),
+ }]
+ issues = []
+ if len(answer) != len(result):
+ issues.append({
+ "kind": "shape", "first_row": -1, "count": abs(len(answer) - len(result)),
+ "message": f"row count differs: answer {len(answer)} vs engine {len(result)}",
+ })
+ n = min(len(answer), len(result))
+ for i in range(answer.shape[1]):
+ name = answer.columns[i]
+ kind = column_kind(answer.dtypes.iloc[i])
+ a = answer.iloc[:n, i].reset_index(drop=True)
+ b = result.iloc[:n, i].reset_index(drop=True)
+ if kind == "float":
+ va = pd.to_numeric(a, errors="coerce").to_numpy(dtype=float)
+ vb = pd.to_numeric(b, errors="coerce").to_numpy(dtype=float)
+ bad = ~np.isclose(va, vb, rtol=rtol, atol=atol, equal_nan=True)
+ elif kind == "int":
+ va = pd.to_numeric(a, errors="coerce")
+ vb = pd.to_numeric(b, errors="coerce")
+ bad = ((va != vb) & ~(va.isna() & vb.isna())).to_numpy()
+ elif kind == "datetime":
+ va = pd.to_datetime(a, errors="coerce")
+ vb = pd.to_datetime(b, errors="coerce")
+ bad = ((va != vb) & ~(va.isna() & vb.isna())).to_numpy()
+ else: # exact string match
+ sa = a.astype("string").fillna("<NA>")
+ sb = b.astype("string").fillna("<NA>")
+ bad = (sa != sb).to_numpy()
+ count = int(bad.sum())
+ if count:
+ j = int(np.where(bad)[0][0])
+ issues.append({
+ "kind": kind, "first_row": j, "count": count, "col": i, "name": str(name),
+ "message": (f"col {i} ('{name}', {kind}): {count}/{n} mismatch "
+ f"(first at row {j}: {a.iloc[j]!r} vs {b.iloc[j]!r})"),
+ })
+ return issues
+
+
+def is_boundary_tie(query: str, issues: list[dict], answer: pd.DataFrame) -> bool:
+ """True only for a legitimate LIMIT-boundary tie: an eligible LIMIT query, at the
+ cap, whose only discrepancy is the final row's identity (int key / string), with
+ every floating-point ordering metric still tying within tolerance.
+ """
+ if not issues:
+ return False
+ if query not in LIMIT_QUERIES:
+ return False
+ if len(answer) != LIMIT_CAP:
+ return False
+ last = len(answer) - 1
+ for iss in issues:
+ if iss["kind"] == "shape":
+ return False
+ # every discrepancy must be exactly one differing value, on the final row
+ if iss["first_row"] != last or iss["count"] != 1:
+ return False
+ # a numeric ordering metric that differs is not a tie — it is a wrong answer
+ if iss["kind"] == "float":
+ return False
+ return True
+
+
+def load_statuses(results_dir: Path) -> dict:
+ """Map (engine, query) -> benchmark run status, from the timing JSONs, so a
+ timeout/error/OOM (couldn't run) is distinguished from a missing dump."""
+ statuses = {}
+ for jf in results_dir.glob("*_results.json"):
+ try:
+ data = json.loads(jf.read_text())
+ except Exception:
+ continue
+ for suite in data.get("results", []):
+ engine = suite.get("engine")
+ for r in suite.get("results", []):
+ statuses[(engine, r.get("query"))] = r.get("status")
+ return statuses
+
+
+def verify_one(results_dir: Path, answers_dir: Path, engine: str, query: str,
+ status: str | None, rtol: float, atol: float) -> tuple[str, str | None]:
+ """Return (verdict, detail). Verdicts in FAILING ('fail', 'missing') fail CI."""
+ answer_pq = answers_dir / f"{query}.parquet"
+ result_csv = results_dir / f"{engine}_{query}_result.csv"
+ if not answer_pq.exists():
+ return "no_answer", None
+ if not result_csv.exists():
+ # No result to check.
+ # - explicit runtime failure (timeout/error/OOM): tolerated, the engine
+ # could not compute the query;
+ # - the engine reported SUCCESS but produced no dump: the dump silently
+ # failed, which must fail CI (otherwise the gate passes without checking);
+ # - no record at all (the framework's whole benchmark job failed, e.g. an
+ # install error): reported but not a correctness failure — the framework's
+ # own red job surfaces it, and it left no answer that could be "wrong".
+ if status == "timeout":
+ return "timeout", None
+ if status == "not_started":
+ return "oom", None
+ if status == "error":
+ return "run_error", None
+ if status == "success":
+ return "missing", "benchmark reported success but produced no result (dump failed)"
+ return "no_data", "framework produced no result (its benchmark job failed)"
+ # Reading/comparing one engine's dump must never take down the whole summary:
+ # a bad or unreadable result becomes a ⚠️ cell (not a crash, not a hard failure —
+ # the engine's own benchmark job surfaces the underlying problem).
+ try:
+ answer = pd.read_parquet(answer_pq)
+ result = pd.read_csv(result_csv)
+ issues = compare(answer, result, rtol, atol)
+ except Exception as e:
+ return "verify_error", f"could not read/compare result: {e}"
+ if not issues:
+ return "pass", None
+ if is_boundary_tie(query, issues, answer):
+ return "pass", "boundary tie tolerated"
+ return "fail", "; ".join(i["message"] for i in issues[:3])
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description="Verify engine correctness against committed answers")
+ parser.add_argument("--results-dir", required=True,
+ help="Directory of result dumps (<engine>_<query>_result.csv) and timing JSONs")
+ parser.add_argument("--answers-dir", default=None,
+ help="Committed answers directory (default: benchmark/answers/sf<sf>)")
+ parser.add_argument("--scale-factor", required=True, help="Scale factor being verified")
+ parser.add_argument("--engines", default=",".join(ENGINES))
+ parser.add_argument("--queries", default=None, help="Comma list, e.g. q1,q6 (default all 12)")
+ parser.add_argument("--rtol", type=float, default=1e-6)
+ parser.add_argument("--atol", type=float, default=1e-9)
+ parser.add_argument("--output", default="correctness_summary.md")
+ args = parser.parse_args()
+
+ sf = args.scale_factor
+ sf_tag = f"sf{int(float(sf)) if float(sf).is_integer() else sf}"
+ sf_display = int(float(sf)) if float(sf).is_integer() else sf
+ results_dir = Path(args.results_dir)
+ answers_dir = Path(args.answers_dir) if args.answers_dir else Path(__file__).parent / "answers" / sf_tag
+
+ engines = [e.strip() for e in args.engines.split(",") if e.strip()]
+ queries = ([q.strip() for q in args.queries.split(",")] if args.queries
+ else [f"q{i}" for i in range(1, QUERY_COUNT + 1)])
+
+ # No committed answers for this scale factor -> nothing to verify, nothing to gate.
+ if not answers_dir.is_dir():
+ msg = (f"# ✅ SpatialBench Correctness (SF{sf_display})\n\n"
+ f"No committed answers at `{answers_dir}`, so correctness was not checked "
+ f"for scale factor {sf_display}.\n")
+ Path(args.output).write_text(msg)
+ print(msg)
+ return 0
+
+ statuses = load_statuses(results_dir)
+
+ # verdicts[engine][query] = (verdict, detail)
+ verdicts = {}
+ for engine in engines:
+ verdicts[engine] = {}
+ for query in queries:
+ verdicts[engine][query] = verify_one(
+ results_dir, answers_dir, engine, query,
+ statuses.get((engine, query)), args.rtol, args.atol,
+ )
+
+ failures = [
+ (engine, query, verdict, detail)
+ for engine in engines
+ for query, (verdict, detail) in verdicts[engine].items()
+ if verdict in FAILING
+ ]
+
+ # Number of (engine, query) pairs actually compared against an answer. If this is
+ # zero the gate verified nothing — a total artifact-download failure or a fully
+ # broken benchmark — and must not pass silently.
+ compared = sum(
+ 1
+ for engine in engines
+ for _q, (verdict, _d) in verdicts[engine].items()
+ if verdict in ("pass", "fail")
+ )
+ verified_nothing = compared == 0
+
+ # ── Build the markdown summary ──
+ lines = [
+ f"# ✅ SpatialBench Correctness (SF{sf_display})",
+ "",
+ f"Each engine's result compared against the committed ground-truth answers in "
+ f"[`benchmark/answers/{sf_tag}`](../tree/main/benchmark/answers/{sf_tag}). "
+ "A mismatch (❌), a missing result (🚫), or an unreadable result (⚠️) fails this job.",
+ "",
+ ]
+ if verified_nothing:
+ lines += [
+ "> ❌ **No engine results were verified** — the benchmark artifacts are "
+ "missing or unreadable (likely a download failure). Failing the job rather "
+ "than passing without checking anything.",
+ "",
+ ]
+ lines += [
+ "| Query | " + " | ".join(ENGINE_ICONS.get(e, e.title()) for e in engines) + " |",
+ "|:------|" + "|".join(":---:" for _ in engines) + "|",
+ ]
+ for query in queries:
+ row = f"| **{query.upper()}** |"
+ for engine in engines:
+ verdict, _ = verdicts[engine][query]
+ row += f" {VERDICT_SYMBOL.get(verdict, '❓')} |"
+ lines.append(row)
+
+ # Per-engine tally
+ lines.extend([
+ "",
+ "| Engine | ✅ Correct | ❌ Failed | Not verified |",
+ "|--------|:---------:|:---------:|:------------:|",
+ ])
+ for engine in engines:
+ correct = sum(1 for v, _ in verdicts[engine].values() if v == "pass")
+ failed = sum(1 for v, _ in verdicts[engine].values() if v in FAILING)
+ not_verified = len(queries) - correct - failed
+ lines.append(f"| {ENGINE_ICONS.get(engine, engine.title())} | {correct} | {failed} | {not_verified} |")
+
+ if failures:
+ lines.extend(["", "## ❌ Failures", ""])
+ for engine, query, verdict, detail in failures:
+ sym = VERDICT_SYMBOL.get(verdict, "❌")
+ lines.append(f"- {sym} **{ENGINE_ICONS.get(engine, engine.title())} {query.upper()}**: {detail}")
+
+ lines.extend([
+ "",
+ "| Legend | Meaning |",
+ "|--------|---------|",
+ "| ✅ | Matches the committed answer |",
+ "| ❌ | Does not match, or the result exists but could not be read/compared — fails CI |",
+ "| 🚫 | Reported success but produced no result (dump failed) — fails CI |",
+ "| ⏱️ | Engine could not compute the query in time (not a failure) |",
+ "| 💀 | Runner killed before this query ran, likely OOM (not a failure) |",
+ "| ⚠️ | Engine errored running the query (not a failure) |",
+ "| ❔ | Framework produced no result — its benchmark job failed; see the benchmark summary (not a failure) |",
+ "| — | No committed answer for this query |",
+ "",
+ f"*Generated on {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')}*",
+ ])
+
+ markdown = "\n".join(lines)
+ Path(args.output).write_text(markdown)
+ print(markdown)
+
+ if verified_nothing:
+ print("\nFAILED: no engine results were verified (artifacts missing/unreadable — "
+ "possible download failure).", file=sys.stderr)
+ return 1
+ if failures:
+ print(f"\nFAILED: {len(failures)} (engine, query) result(s) did not match, were "
+ f"missing, or could not be read.", file=sys.stderr)
+ return 1
+ print("\nAll engine results match the committed answers (where a result was produced).")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/spatialbench-queries/geopandas_queries.py b/spatialbench-queries/geopandas_queries.py
index 8090354..ee36525 100644
--- a/spatialbench-queries/geopandas_queries.py
+++ b/spatialbench-queries/geopandas_queries.py
@@ -205,8 +205,8 @@ def q5(data_paths: dict[str, str]) -> DataFrame: # type: ignore[override]
grouped.sort_values(
["monthly_travel_hull_area", "c_custkey", "pickup_month"],
ascending=[False, True, True],
- )[["c_custkey", "c_name", "pickup_month", "monthly_travel_hull_area"]]
- .rename(columns={"c_name": "customer_name"})
+ )[["c_custkey", "c_name", "pickup_month", "monthly_travel_hull_area", "trip_count"]]
+ .rename(columns={"c_name": "customer_name", "trip_count": "dropoff_count"})
.head(100) # Return only the top 100 repeat customer-months (bounded result set)
.reset_index(drop=True)
)
diff --git a/spatialbench-queries/pycanopy_queries.py b/spatialbench-queries/pycanopy_queries.py
index f966fc1..53e5472 100644
--- a/spatialbench-queries/pycanopy_queries.py
+++ b/spatialbench-queries/pycanopy_queries.py
@@ -149,8 +149,12 @@ def q5(data_paths: dict[str, str]) -> pl.DataFrame:
).sort(["monthly_travel_hull_area", "t_custkey", "pickup_month"], descending=[True, False, False])
return (
- grouped.select(["t_custkey", "c_name", "pickup_month", "monthly_travel_hull_area"])
- .rename({"t_custkey": "c_custkey", "c_name": "customer_name"})
+ grouped.select(
+ ["t_custkey", "c_name", "pickup_month", "monthly_travel_hull_area", "trip_count"]
+ )
+ .rename(
+ {"t_custkey": "c_custkey", "c_name": "customer_name", "trip_count": "dropoff_count"}
+ )
.head(100) # Return only the top 100 repeat customer-months (bounded result set)
)
@@ -179,6 +183,9 @@ def q6(data_paths: dict[str, str]) -> pl.DataFrame:
qdf = trip.select(["t_distance", "t_pickuptime", "t_dropofftime"]).with_columns(
pl.Series("qx", qx),
pl.Series("qy", qy),
+ # t_distance is decimal(15,5); average it as float so the result keeps full
+ # precision (a decimal mean stays at scale 5 and rounds off the answer).
+ t_distance=pl.col("t_distance").cast(pl.Float64),
duration_seconds=(pl.col("t_dropofftime") - pl.col("t_pickuptime")).dt.total_seconds(),
)
@@ -287,6 +294,9 @@ def q10(data_paths: dict[str, str]) -> pl.DataFrame:
qdf = trip.with_columns(
pl.Series("qx", qx),
pl.Series("qy", qy),
+ # t_distance is decimal(15,5); average it as float so the result keeps full
+ # precision (a decimal mean stays at scale 5 and rounds off the answer).
+ t_distance=pl.col("t_distance").cast(pl.Float64),
duration_seconds=(pl.col("t_dropofftime") - pl.col("t_pickuptime")).dt.total_seconds(),
).select(["qx", "qy", "t_distance", "duration_seconds"])
diff --git a/spatialbench-queries/spatial_polars.py b/spatialbench-queries/spatial_polars.py
index 2c75e5d..6c22e27 100644
--- a/spatialbench-queries/spatial_polars.py
+++ b/spatialbench-queries/spatial_polars.py
@@ -328,7 +328,9 @@ def q7(data_paths: dict[str, str]) -> DataFrame:
)
.with_columns(
(
- pl.struct(("t_pickuploc", "t_dropoffloc")).spatial.distance() * 111111
+ # meters = degrees / 0.000009 (1 meter ~= 0.000009 degree), matching
+ # the reference SQL and the other engines' conversion constant.
+ pl.struct(("t_pickuploc", "t_dropoffloc")).spatial.distance() / 0.000009
).alias("line_distance_m"),
pl.col("t_distance").alias("reported_distance_m"),
)