Add Dask

[alexey-milovidov]

Closes #892.

Adds a Dask entry. Dask is "pandas in parallel", so this mirrors the existing pandas port: a small FastAPI server wraps a Dask DataFrame behind the standard install/start/check/stop/load/query/data-size interface, and queries.sql holds one Python expression per line that the server eval()s — exactly the pattern used by pandas and polars-dataframe.

Design

• Parallelism: a dask.distributed.LocalCluster of worker processes gives real multi-core execution. The frame is persist()ed into cluster memory (and wait()ed on) so load time is comparable to the pandas/polars in-memory model. BENCH_DURABLE=no, so the harness reloads before each cold query and rolls that into the cold try, as with the other in-memory dataframe systems.
• Dataset: reads the partitioned parquet (hits_0..99.parquet) — one partition per file is Dask's natural layout (BENCH_DOWNLOAD_SCRIPT=download-hits-parquet-partitioned).
• Lazy execution: the server materialises each result with dask.compute() and times eval() + compute() together (mirroring pandas, where eval() does all the work). dask.compute() recurses into tuples/lists, so the tuple queries (3, 7) and the 90-sum query (30) compute in one pass.

Query translation

Queries stay close to the pandas port, diverging only where Dask's API does:

• Q10, Q23 (COUNT(DISTINCT) alongside other aggregates): Dask's built-in groupby.agg has no nunique, so these use Dask's documented custom nunique aggregation.
• Q39–Q43 (LIMIT … OFFSET): Dask has no positional iloc, so these take the small top-N with nlargest and slice in pandas after compute() (SQL-faithful, matching how polars-dataframe does it — not the pandas port's nlargest(10).iloc[1000:] quirk that returns empty).
• Q8, Q26, Q27: Dask Series has no sort_values, so these use nlargest / frame sort_values.

Validation

All 43 queries were validated against a synthetic dataset under both the threaded scheduler and a real distributed LocalCluster, and end-to-end through the HTTP server. Verified that cluster teardown on shutdown leaves no orphaned worker processes across the harness's repeated stop/start/reload cycles. Real timings still need a run on a benchmark machine (aarch64 dev box can't produce the x86 result set).

🤖 Generated with Claude Code

[github-actions]

Results for dask are ready for: c6a.metal.
The result files are committed as af0df52.

Logs:

• dask on c6a.metal: benchmark log, cloud-init log

[github-actions]

The run of dask on c6a.4xlarge did not produce results.
The run of dask on c8g.4xlarge did not produce results.

Logs:

• dask on c6a.4xlarge: benchmark log, cloud-init log
• dask on c8g.4xlarge: benchmark log, cloud-init log

[github-actions]

Results for dask are ready for: c6a.metal, c7a.metal-48xl, c8g.metal-48xl.
The result files are committed as d4a9521.

Logs:

• dask on c8g.metal-48xl: benchmark log, cloud-init log
• dask on c6a.metal: benchmark log, cloud-init log
• dask on c7a.metal-48xl: benchmark log, cloud-init log

[github-actions]

The run of dask on c6a.2xlarge did not produce results.
The run of dask on c6a.large did not produce results.
The run of dask on c6a.xlarge did not produce results.
The run of dask on t3a.small did not produce results.

Logs:

• dask on c6a.large: benchmark log, cloud-init log
• dask on c6a.xlarge: benchmark log, cloud-init log
• dask on c6a.2xlarge: benchmark log, cloud-init log
• dask on t3a.small: benchmark log, cloud-init log