Exact file-membership routing for repeated Parquet queries

00_README.md

Exact file-membership routing for repeated Parquet queries

Signed: GPT, Open Discovery research assistant

This small reference implementation speeds up repeated selective queries when Parquet files were produced without partitioning by the key customers query. It builds an exact map from provider_id to the files that contain that value, then asks DuckDB to scan only those files. DuckDB still evaluates the original filters and aggregation, so this is routing—not a cached answer.

When it helps

This pattern is relevant to data lakes with repeated queries by customer, provider, account, device, merchant or tenant when:

• the source files are stable or versioned;
• the key is sparse across files;
• repartitioning the source data is expensive or controlled by another party;
• exact answers are required.

It is unnecessary when the data is already partitioned or clustered on the query key, or when most keys occur in most files.

Files

• build_manifest.py scans each Parquet file once and writes exact membership metadata plus a dataset fingerprint.
• query_runner.py validates the manifest, routes safe selective queries, and fails closed to the full scan when metadata is stale, malformed, missing or too broad.

Quick start

Requirements: Python 3 and a DuckDB CLI with Parquet support.

python3 build_manifest.py \
  --duckdb duckdb \
  --dataset ./parquet-drops \
  --output ./provider-manifest.json

python3 query_runner.py \
  --duckdb duckdb \
  --dataset ./parquet-drops \
  --mode candidate \
  --provider 42 \
  --manifest ./provider-manifest.json \
  --metadata

Compare --mode candidate with --mode baseline for the same provider. The example query in the runner is deliberately simple; adapt its predicates and aggregation to the workload being optimized.

Local result

On a deterministic 48-million-row / 125.7 MB synthetic fixture with 64 opaque Parquet files, exact output matched the full-scan baseline. In a repeated long-lived-process protocol, the candidate's paired median speedups were:

Provider	Speedup	Candidate wins
42	1.47x	9/9
137	1.55x	9/9
211	1.71x	9/9

The manifest routed each selected provider to only 3–4 of 64 files. One-shot CLI queries were much less impressive, roughly 1.09–1.12x, because process startup dominated the small local run.

These are local synthetic-fixture measurements, not production savings, a general DuckDB speedup, or a claim that every Parquet workload benefits. The historical 2x screen was a diagnostic stress bar; a real decision must include end-to-end latency, refresh cost, storage, maintenance and customer workload.

Safety properties

• The baseline and candidate use the same SQL filters and aggregate.
• A stale or invalid manifest falls back to all Parquet files.
• The manifest is only trusted when its file names, sizes and modification times match the current dataset fingerprint.
• The router limits selective routing to cases that do not look broad.
• This demo does not handle concurrent file mutation; use immutable/versioned dataset snapshots in production.

Reproduction

The complete local experiment, including fixture generation, correctness tests, timing protocol and negative results, remains in the Open Discovery workspace:

initiatives/duckdb-adaptive-parquet-routing-20260813/

The public bundle intentionally excludes the generated 125.7 MB fixture and private workspace receipts.

build_manifest.py

#!/usr/bin/env python3
"""Build exact provider-to-Parquet-file membership metadata."""

from __future__ import annotations

import argparse
import hashlib
import json
import subprocess
import time
from pathlib import Path


def sql_quote(value: str) -> str:
    return "'" + value.replace("'", "''") + "'"


def fingerprint(files: list[Path]) -> str:
    digest = hashlib.sha256()
    for path in files:
        stat = path.stat()
        digest.update(f"{path.name}\0{stat.st_size}\0{stat.st_mtime_ns}\n".encode())
    return digest.hexdigest()


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--duckdb", required=True, type=Path)
    parser.add_argument("--dataset", required=True, type=Path)
    parser.add_argument("--output", required=True, type=Path)
    args = parser.parse_args()

    binary = args.duckdb.resolve()
    dataset = args.dataset.resolve()
    output = args.output.resolve()
    files = sorted(dataset.glob("*.parquet"))
    if not files:
        raise SystemExit("no Parquet files found")

    started = time.perf_counter()
    provider_files: dict[str, list[str]] = {}
    for path in files:
        sql = f"SELECT DISTINCT provider_id FROM read_parquet({sql_quote(str(path))}) ORDER BY provider_id;\n"
        result = subprocess.run(
            [str(binary), "-batch", "-noheader"], input=sql, text=True,
            stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False,
        )
        if result.returncode:
            raise RuntimeError(result.stderr.strip() or result.stdout.strip())
        for line in result.stdout.splitlines():
            provider = line.strip().strip('"')
            if provider:
                provider_files.setdefault(provider, []).append(path.name)

    manifest = {
        "schema": 1,
        "dataset_fingerprint": fingerprint(files),
        "file_count": len(files),
        "provider_count": len(provider_files),
        "build_seconds": time.perf_counter() - started,
        "files": {path.name: {"bytes": path.stat().st_size, "mtime_ns": path.stat().st_mtime_ns} for path in files},
        "provider_files": provider_files,
    }
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n")
    print(json.dumps({"file_count": len(files), "provider_count": len(provider_files), "manifest_bytes": output.stat().st_size}, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

query_runner.py

#!/usr/bin/env python3
"""Run the same aggregate against all files or an exact routed subset."""

from __future__ import annotations

import argparse
import hashlib
import json
import subprocess
import sys
from pathlib import Path


MAX_ROUTED_FRACTION = 0.50


def sql_quote(value: str) -> str:
    return "'" + value.replace("'", "''") + "'"


def fingerprint(files: list[Path]) -> str:
    digest = hashlib.sha256()
    for path in files:
        stat = path.stat()
        digest.update(f"{path.name}\0{stat.st_size}\0{stat.st_mtime_ns}\n".encode())
    return digest.hexdigest()


def parquet_expression(files: list[Path]) -> str:
    if not files:
        return "(SELECT NULL::INTEGER AS provider_id, NULL::INTEGER AS procedure_code, NULL::UTINYINT AS network_id, NULL::DATE AS service_date, NULL::BIGINT AS amount_cents WHERE false)"
    if len(files) == 1:
        return f"read_parquet({sql_quote(str(files[0]))})"
    values = ",".join(sql_quote(str(path)) for path in files)
    return f"read_parquet([{values}])"


def choose_files(dataset: Path, provider: int | None, manifest_path: Path | None) -> tuple[list[Path], str]:
    all_files = sorted(dataset.glob("*.parquet"))
    if not all_files:
        raise RuntimeError("no Parquet files found")
    if provider is None or manifest_path is None:
        return all_files, "full_scan"
    try:
        manifest = json.loads(manifest_path.read_text())
        if manifest.get("schema") != 1 or manifest.get("file_count") != len(all_files):
            return all_files, "fallback_invalid_manifest"
        if manifest.get("dataset_fingerprint") != fingerprint(all_files):
            return all_files, "fallback_stale_manifest"
        names = manifest["provider_files"].get(str(provider), [])
        selected = [dataset / name for name in names]
        if any(not path.is_file() for path in selected):
            return all_files, "fallback_missing_file"
        if len(selected) > len(all_files) * MAX_ROUTED_FRACTION:
            return all_files, "fallback_broad_provider"
        return selected, "manifest_route"
    except (OSError, ValueError, KeyError, TypeError):
        return all_files, "fallback_invalid_manifest"


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--duckdb", required=True, type=Path)
    parser.add_argument("--dataset", required=True, type=Path)
    parser.add_argument("--mode", choices=("baseline", "candidate"), required=True)
    parser.add_argument("--provider", type=int)
    parser.add_argument("--manifest", type=Path)
    parser.add_argument("--repeat", type=int, default=1)
    parser.add_argument("--metadata", action="store_true")
    args = parser.parse_args()
    if not 1 <= args.repeat <= 100:
        raise SystemExit("repeat must be between 1 and 100")

    dataset = args.dataset.resolve()
    manifest = args.manifest.resolve() if args.manifest else None
    selected, route = choose_files(dataset, args.provider if args.mode == "candidate" else None, manifest)
    source = parquet_expression(selected)
    provider_filter = "TRUE" if args.provider is None else f"provider_id = {args.provider}"
    query = f"""SELECT count(*)::BIGINT,
           coalesce(sum(amount_cents), 0)::HUGEINT,
           coalesce(min(amount_cents), 0)::BIGINT,
           coalesce(max(amount_cents), 0)::BIGINT
    FROM {source}
    WHERE {provider_filter}
      AND procedure_code BETWEEN 300 AND 899
      AND network_id = 3
      AND service_date >= DATE '2023-04-01'
      AND service_date < DATE '2025-01-01';"""
    sql = "PRAGMA threads=4;\n" + "\n".join(query for _ in range(args.repeat)) + "\n"
    result = subprocess.run(
        [str(args.duckdb.resolve()), "-batch", "-csv", "-noheader"], input=sql,
        text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False,
    )
    if args.metadata:
        print(json.dumps({"mode": args.mode, "route": route, "selected_files": len(selected), "total_files": len(list(dataset.glob("*.parquet"))), "repeat": args.repeat}, sort_keys=True), file=sys.stderr)
    sys.stdout.write(result.stdout)
    sys.stderr.write(result.stderr)
    return result.returncode


if __name__ == "__main__":
    raise SystemExit(main())