feat: simpatico zonemaps (chunk pruning)

CMakeLists.txt

@@ -246,6 +246,7 @@ endif()
 # cmake-format: off
 set(EXTENSION_SOURCES
     src/compression/compressed_representation.cpp
+    src/compression/simpatico_file_ingest.cpp
     src/compression/compressed_scan.cpp
     src/compression/compression_converters.cpp
     src/compression/plan_register.cpp
@@ -346,6 +347,7 @@ set(EXTENSION_SOURCES
     src/op/scan/gpu_ingestible.cpp
     src/op/scan/parquet_gpu_ingestible.cpp
     src/op/scan/duckdb_native_gpu_ingestible.cpp
+    src/op/scan/simpatico_gpu_ingestible.cpp
     src/op/dynamic_filter/dynamic_filter_publish_plan.cpp
     src/op/dynamic_filter/dynamic_filter_publisher.cpp
     src/op/dynamic_filter/sirius_dynamic_filter.cpp
@@ -677,6 +679,8 @@ endif()
 
 # cmake-format: off
 set(TEST_SOURCES
+    test/cpp/compression/test_hpln_ingest.cpp
+    test/cpp/compression/test_simpatico_ingestible.cpp
     test/cpp/compression/test_compression.cpp
     test/cpp/config/test_config.cpp
     test/cpp/config/test_context.cpp
@@ -735,6 +739,7 @@ set(TEST_SOURCES
     test/cpp/integration/test_gpu_execution_semantic_cast_fallback.cpp
     test/cpp/integration/test_parquet_null_predicate_pushdown.cpp
     test/cpp/integration/test_query_lifecycle_slot.cpp
+    test/cpp/integration/test_read_simpatico_sql.cpp
     test/cpp/integration/test_gpu_execution_tpcds_nulls.cpp
     test/cpp/integration/test_gpu_execution_tpch.cpp
     test/cpp/integration/test_gpu_execution_tpch_mgpu_audit.cpp
@@ -779,6 +784,7 @@ set(TEST_SOURCES
     test/cpp/scan_manager/test_insert_delta_job.cpp
     test/cpp/scan_manager/test_memory_prefetcher_accounting.cpp
     test/cpp/scan_manager/test_mvcc_mask_job.cpp
+    test/cpp/scan_manager/test_pin_clustering.cpp
     test/cpp/scan_manager/test_pinned_chunk_stats.cpp
     test/cpp/scan_manager/test_pinned_entry_column_lookup.cpp
     test/cpp/operator/test_physical_grouped_aggregate_merge_mgpu.cpp

bench/chunk-skipping/.gitignore

@@ -0,0 +1,4 @@
+# Benchmark output: per-run logs, per-query result dumps and generated configs.
+# Regenerate with run-sweep.sh / run-sweep-sf1000.sh; only the runtime CSVs are worth
+# keeping, and those live in runtimes/.
+results*/

bench/chunk-skipping/analyze.py

@@ -0,0 +1,42 @@
+import os, re, csv, glob, collections, json
+RES=os.environ.get('RES', os.path.dirname(os.path.abspath(__file__))+'/results')
+runs={}
+for d in sorted(glob.glob(RES+'/tpch_*')):
+    m=re.search(r'_([0-9]+(?:GB|MB))_prune-(on|off)$', d)
+    if not m: continue
+    batch, arm = m.group(1), m.group(2)
+    best=collections.defaultdict(lambda: 1e9)
+    f=d+'/csv/runtimes.csv'
+    if not os.path.exists(f): continue
+    for r in csv.DictReader(open(f)):
+        best[r['query']]=min(best[r['query']], float(r['runtime_s']))
+    # pruning evidence
+    pruned=0; total=0
+    for lf in glob.glob(d+'/log_dir/*.log'):
+        for line in open(lf, errors='ignore'):
+            mm=re.search(r'zone-map pruning for pinned entry \'(\w+)\'.*?: (\d+)/(\d+) chunks', line)
+            if mm: pruned+=int(mm.group(2)); total+=int(mm.group(3))
+    runs[(batch,arm)]={'best':dict(best),'pruned':pruned,'total':total}
+
+def order(b):
+    n=float(b[:-2]); return n*1024 if b.endswith('GB') else n
+batches=sorted({k[0] for k in runs}, key=order, reverse=True)
+qs=sorted({q for v in runs.values() for q in v['best']}, key=lambda x:int(x[1:]))
+
+print(f"{'batch':>8}{'suite ON':>11}{'suite OFF':>11}{'delta':>10}{'delta %':>9}   chunks pruned (ON)")
+for b in batches:
+    on,off=runs.get((b,'on')),runs.get((b,'off'))
+    if not on or not off: continue
+    so=sum(on['best'].get(q,0) for q in qs); sf=sum(off['best'].get(q,0) for q in qs)
+    frac = f"{on['pruned']}/{on['total']} = {100*on['pruned']/max(on['total'],1):.0f}%"
+    print(f"{b:>8}{so:>11.4f}{sf:>11.4f}{so-sf:>10.4f}{100*(so-sf)/sf:>8.1f}%   {frac}")
+
+print(f"\nPer-query ON-minus-OFF delta (s), negative = pruning helped")
+print(f"{'query':>7}"+''.join(f"{b:>10}" for b in batches))
+for q in qs:
+    row=[]
+    for b in batches:
+        on,off=runs.get((b,'on')),runs.get((b,'off'))
+        row.append((on['best'].get(q,0)-off['best'].get(q,0)) if on and off else 0.0)
+    if any(abs(x)>0.002 for x in row):
+        print(f"{q:>7}"+''.join(f"{x:>10.4f}" for x in row))

bench/chunk-skipping/run-sweep-sf1000.sh

@@ -0,0 +1,90 @@
+#!/usr/bin/env bash
+# Zone-map pruning sweep at production scale (SF1000).
+#
+# Same experiment as run-sweep.sh, at SF1000 on the clustered dataset, reusing the tuned
+# sf1000-repro config so the only variables are scan_task_batch_size and
+# enable_pinned_zone_map_pruning. Read the ON-vs-OFF delta at a fixed batch size.
+#
+#   /home/nvidia/joost/bench-lock.sh bash bench/chunk-skipping/run-sweep-sf1000.sh
+#
+# MUST run under bench-lock.sh: this config pins 471.2 GB of host memory on a 494 GB box.
+set -euo pipefail
+HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO="$(cd "$HERE/../.." && pwd)"
+
+DATA="${DATA:-/datasets/tpch_sf1000_sorted}"
+PLANS="${PLANS:-$REPO/bench/sf1000-repro/plans}"
+BASE_CFG="${BASE_CFG:-$REPO/bench/sf1000-repro/sirius-sf1000.yaml}"
+BATCHES="${BATCHES:-8GB 2GB}"
+# TIER=host is the experiment that matters for fetch-bound skipping: on a host pin every scan
+# batch pays payload H2D -> sync -> decode on the critical path, so a skipped chunk saves the
+# transfer, not just the decode. On a GPU pin the payload is already resident (ceiling ~2.4%,
+# measured at ~2.4% of suite time on a GPU pin).
+TIER="${TIER:-gpu}"
+ITERS="${ITERS:-3}"
+OUT="${OUT:-$REPO/bench/chunk-skipping/results-sf1000-$TIER}"
+DOWNGRADE_DIR="${DOWNGRADE_DIR:-/datasets/.sirius_disk_memory}"
+
+[ -d "$DATA" ] || { echo "ERROR: no dataset at $DATA (set DATA=)"; exit 1; }
+mkdir -p "$OUT" "$DOWNGRADE_DIR"
+
+export SIRIUS_EXP_LATE_MAT="${SIRIUS_EXP_LATE_MAT:-1}"
+export SIRIUS_EXP_LATE_MAT_PIN_UNIQUE_COLS="${SIRIUS_EXP_LATE_MAT_PIN_UNIQUE_COLS:-c_custkey,n_name,n_nationkey}"
+export SIRIUS_EXP_FUSED_SCAN_FILTER="${SIRIUS_EXP_FUSED_SCAN_FILTER:-1}"
+for t in LINEITEM ORDERS PART CUSTOMER SUPPLIER NATION REGION PARTSUPP; do
+  export "SIRIUS_PIN_TIER_$t=$TIER"
+done
+export SIRIUS_PRE_SQL="SET pin_table_compression = true; \
+SET pin_table_input_compression_plan_dir = '$PLANS'; \
+SET expression_evaluator_strategy = 'ast_jit'"
+
+# Refuse to measure against a busy GPU. bench-lock.sh serialises OUR runs, but anything started
+# outside it (another worktree's test binary, a stray process) still contends -- and a contended
+# run either OOMs at pool init or silently reports inflated times.
+require_idle_gpu() {
+  for _ in $(seq 1 180); do
+    local mem util
+    mem=$(nvidia-smi --query-gpu=memory.used --format=csv,noheader | tr -dc '0-9')
+    util=$(nvidia-smi --query-gpu=utilization.gpu --format=csv,noheader | tr -dc '0-9')
+    if [ "${mem:-9999}" -lt 2048 ] && [ "${util:-100}" -lt 20 ]; then return 0; fi
+    echo "waiting for an idle GPU (used ${mem} MiB, util ${util}%)"
+    sleep 20
+  done
+  echo "ERROR: GPU still busy after 60 min; refusing to measure"; exit 1
+}
+
+cd "$REPO"
+require_idle_gpu
+for batch in $BATCHES; do
+  for prune in on off; do
+    if [ "$prune" = on ]; then flag=true; else flag=false; fi
+    cfg="$OUT/cfg-$batch-$prune.yaml"
+    # Replace the base config's own scan_task_batch_size and append the pruning flag.
+    # Computing the flag BEFORE the sed matters: '&' is special in sed replacement text, so an
+    # inlined $( ... && ... ) silently yields "false" for both arms, which reads as "no benefit".
+    # The base config's downgrade_root_dirs points at a /localhome path that does not exist on
+    # this box; repoint it so a downgrade does not fail on a missing directory.
+    sed -e "s|^        scan_task_batch_size: .*|        scan_task_batch_size: $batch\n        enable_pinned_zone_map_pruning: $flag|" \
+        -e "s|^            downgrade_root_dirs: .*|            downgrade_root_dirs: \"$DOWNGRADE_DIR\"|" \
+      "$BASE_CFG" > "$cfg"
+    grep -q "enable_pinned_zone_map_pruning: $flag" "$cfg" || { echo "ERROR: config injection failed"; exit 1; }
+    grep -q "scan_task_batch_size: $batch" "$cfg" || { echo "ERROR: batch injection failed"; exit 1; }
+
+    require_idle_gpu
+    name="sf1000_$(basename "$DATA")_${TIER}_${batch}_prune-${prune}"
+    echo "################ $name"
+    pixi run python test/tpch_performance/performance_test.py \
+      --input "$DATA" --data-source parquet \
+      --mode grouped --iterations "$ITERS" --engine gpu --pin "$TIER" \
+      --queries 1-22 --config "$cfg" --name "$name" --output "$OUT"
+
+    if [ "$prune" = on ]; then
+      run_dir=$(ls -dt "$OUT"/tpch_*"$name" 2>/dev/null | head -1)
+      if ! grep -rq "zone-map pruning for pinned entry" "$run_dir/log_dir" 2>/dev/null; then
+        echo "ERROR: prune-on arm at $batch pruned nothing — check the config and the pin tier"
+        exit 1
+      fi
+    fi
+  done
+done
+echo "results under $OUT"

bench/chunk-skipping/run-sweep.sh

@@ -0,0 +1,73 @@
+#!/usr/bin/env bash
+# Zone-map pruning: benefit vs. pin-chunk granularity, at SF100.
+#
+# Sweeps scan_task_batch_size (a free granularity dial: smaller batch => smaller pin chunk =>
+# finer zone maps) crossed with enable_pinned_zone_map_pruning on/off, on the clustered SF100
+# dataset. The number that matters is the ON-vs-OFF delta AT A FIXED batch size: smaller batches
+# cost throughput on their own, so absolute times across batch sizes are confounded but the
+# delta is not.
+#
+#   pixi run bash bench/chunk-skipping/run-sweep.sh
+#
+# Env:
+#   DATA      dataset root (default the clustered SF100)
+#   BATCHES   space-separated scan_task_batch_size values
+#   ITERS     iterations per query (default 3, reported best-of)
+set -euo pipefail
+HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO="$(cd "$HERE/../.." && pwd)"
+
+DATA="${DATA:-/datasets/tpch_sf100_sorted}"
+PLANS="${PLANS:-$REPO/bench/sf1000-repro/plans}"
+BATCHES="${BATCHES:-8GB 2GB 512MB 128MB}"
+ITERS="${ITERS:-3}"
+OUT="${OUT:-$REPO/bench/chunk-skipping/results}"
+
+[ -d "$DATA" ] || { echo "ERROR: no dataset at $DATA (set DATA=)"; exit 1; }
+mkdir -p "$OUT" /datasets/.sirius_disk_memory
+
+export SIRIUS_EXP_LATE_MAT="${SIRIUS_EXP_LATE_MAT:-1}"
+export SIRIUS_EXP_LATE_MAT_PIN_UNIQUE_COLS="${SIRIUS_EXP_LATE_MAT_PIN_UNIQUE_COLS:-c_custkey,n_name,n_nationkey}"
+export SIRIUS_EXP_FUSED_SCAN_FILTER="${SIRIUS_EXP_FUSED_SCAN_FILTER:-1}"
+for t in LINEITEM ORDERS PART CUSTOMER SUPPLIER NATION REGION PARTSUPP; do
+  export "SIRIUS_PIN_TIER_$t=gpu"
+done
+
+cd "$REPO"
+for batch in $BATCHES; do
+  for prune in on off; do
+    cfg="$OUT/cfg-$batch-$prune.yaml"
+    # enable_pinned_zone_map_pruning gates BOTH pin-time capture and serve-time pruning, so
+    # 'off' is a statless pin — a clean "feature absent" arm. The pinned DATA is identical
+    # either way (narrowing is driven by pinned_column_types, not by this flag); only the
+    # sidecar differs.
+    # Compute the flag BEFORE the sed. Do not inline a $( ... && ... ) here: '&' is special in
+    # sed's replacement text, escaping it as '\&\&' breaks the shell's AND operator, and the
+    # substitution then silently yields "false" for BOTH arms — which is exactly how the first
+    # run of this sweep produced a meaningless result.
+    if [ "$prune" = on ]; then flag=true; else flag=false; fi
+    sed "s|^    operator_params:|    operator_params:\n        scan_task_batch_size: $batch\n        enable_pinned_zone_map_pruning: $flag|" \
+      "$HERE/sirius-sf100.yaml" > "$cfg"
+    grep -q "enable_pinned_zone_map_pruning: $flag" "$cfg" || { echo "ERROR: config injection failed"; exit 1; }
+    export SIRIUS_PRE_SQL="SET pin_table_compression = true; \
+SET pin_table_input_compression_plan_dir = '$PLANS'; \
+SET expression_evaluator_strategy = 'ast_jit'"
+    name="sf100_$(basename "$DATA")_${batch}_prune-${prune}"
+    echo "################ $name"
+    pixi run python test/tpch_performance/performance_test.py \
+      --input "$DATA" --data-source parquet \
+      --mode grouped --iterations "$ITERS" --engine gpu --pin gpu \
+      --queries 1-22 --config "$cfg" --name "$name" --output "$OUT"
+
+    # Guard: the ON arm MUST show pruning. A silently-misconfigured arm would otherwise be
+    # reported as "the feature does not help", which is the one wrong answer this sweep can give.
+    if [ "$prune" = on ]; then
+      run_dir=$(ls -dt "$OUT"/tpch_*"$name" 2>/dev/null | head -1)
+      if ! grep -rq "zone-map pruning for pinned entry" "$run_dir/log_dir" 2>/dev/null; then
+        echo "ERROR: prune-on arm at $batch pruned nothing — check the config and the pin tier"
+        exit 1
+      fi
+    fi
+  done
+done
+echo "results under $OUT"

bench/chunk-skipping/runtimes/README.md

@@ -0,0 +1,19 @@
+# Retained runtime CSVs
+
+Per-query timings from the zone-map pruning sweeps, kept so results can be re-analysed without
+re-running. Everything else the harness emits (per-query result dumps, Sirius logs, generated
+configs) is regenerable and gitignored.
+
+Directory names mirror the sweep's `OUT`; file names are the run directories. Analyse with
+`analyze.py`, pointing `RES` at a `results*/` directory from a fresh run.
+
+| directory | what it is |
+|---|---|
+| `results/` | SF100, GPU pin, batch sweep 8GB/2GB/512MB/128MB |
+| `results-rep2/`, `results-rep3/` | SF100 repeats of the 8GB and 2GB arms (run-to-run spread ~0.8%) |
+| `results-sf1000/` | SF1000, GPU pin, 8GB and 2GB |
+| `results-sf1000-host/` | SF1000, **host** pin, 8GB and 2GB |
+| `results-sf1000-host-fine/` | SF1000, host pin, 2GB and 512MB |
+| `results-sf1000-host-unsorted/` | partial, aborted — kept only so the directory is not mistaken for a complete run |
+
+All SF1000 runs used `/datasets/tpch_sf1000_sorted` unless the name says otherwise.

bench/chunk-skipping/s3_ceiling.sh

@@ -0,0 +1,27 @@
+#!/usr/bin/env bash
+# Raw ranged-GET ceiling, in C, with no Python in the path.
+#
+# The python probe (s3_range_probe.py) copies every response body through the interpreter, so its
+# ceiling can be the GIL rather than the network. This uses curl's own parallel engine to find the
+# real ceiling; if the two disagree, believe this one and treat the python probe's ABSOLUTE
+# throughput as a floor (its RELATIVE comparisons at equal concurrency are still meaningful).
+#
+# usage: s3_ceiling.sh <presigned-url> <object-size-bytes> [range_mb] [parallel]
+set -euo pipefail
+URL="$1"; SIZE="$2"; RANGE_MB="${3:-16}"; PAR="${4:-64}"
+RANGE=$((RANGE_MB * 1024 * 1024))
+N=$((PAR * 4))                      # 4 waves, enough to amortise ramp-up
+[ $((N * RANGE)) -gt "$SIZE" ] && N=$((SIZE / RANGE))
+
+CFG=$(mktemp); trap 'rm -f "$CFG"' EXIT
+for ((i = 0; i < N; i++)); do
+  off=$(( (i * RANGE) % (SIZE - RANGE) ))
+  printf 'url = "%s"\nrange = "%d-%d"\noutput = "/dev/null"\n' "$URL" "$off" $((off + RANGE - 1)) >> "$CFG"
+done
+
+echo "curl: ${N} x ${RANGE_MB} MB, ${PAR}-way parallel"
+S=$(date +%s.%N)
+curl -sS --parallel --parallel-max "$PAR" --parallel-immediate -K "$CFG"
+E=$(date +%s.%N)
+awk -v s="$S" -v e="$E" -v b="$((N * RANGE))" 'BEGIN{
+  d=e-s; printf "  %.0f MB in %.2fs = %.2f GB/s = %.1f Gb/s\n", b/1e6, d, b/d/1e9, b/d/1e9*8 }'