DO NOT MERGE: diagnose DataFusion memory at TPC-H SF1000

Runs DataFusion Q1, Q2, Q9, Q16, Q18 and Q21 on Parquet and on Vortex under
several scan settings, each with its own memory monitor, to test whether
the Vortex excess scales with in-flight scan splits (scan concurrency x
available parallelism per stream, one stream per DataFusion partition).
Also records page-fault call stacks with perf and EXPLAIN plans for the
queries that run out of memory.

Adds a VORTEX_SCAN_CONCURRENCY env knob to datafusion-bench for the
experiment only.

Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FDifRZhxcVd5gvt3dsDhu5

Author: claude

.github/workflows/tpch-sf1000.yml

@@ -8,16 +8,16 @@
 # OOM-killed or times out is recorded and the rest still run, and system memory is sampled
 # throughout data generation, conversion and every query.
 
-name: TPC-H SF1000
+name: TPC-H SF1000 (OOM diagnosis)
 
 concurrency:
-  group: tpch-sf1000
+  group: tpch-sf1000-oom-diag
   cancel-in-progress: false
 
 on:
   # TEMPORARY: lets the branch run the workflow before it lands; drop this commit before merging.
   push:
-    branches: [ji/tpch-sf1000-weekly]
+    branches: [ji/tpch-sf1000-oom-diag]
   schedule:
     - cron: "17 2 * * 6"  # weekly, Saturday
   workflow_dispatch:
@@ -102,7 +102,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        engine: ${{ fromJSON(inputs.engine == 'datafusion' && '["datafusion"]' || inputs.engine == 'duckdb' && '["duckdb"]' || '["datafusion", "duckdb"]') }}
+        engine: [datafusion]
     runs-on: runs-on=${{ github.run_id }}/runner=bench-dedicated/family=${{ inputs.machine_type || 'c8gd.metal-24xl' }}/image=ubuntu24-full-arm64-pre-v2/tag=tpch-sf1000-${{ matrix.engine }}
     env:
       DATA_DIR: vortex-bench/data/tpch/${{ inputs.scale_factor || '1000.0' }}
@@ -185,32 +185,11 @@ jobs:
           echo idle > "$PHASE_FILE"
           du -sh "$DATA_DIR"/vortex-file-compressed/*
 
-      - name: Run ${{ matrix.engine }} queries
-        id: queries
+      - name: Diagnose DataFusion memory
         shell: bash
         env:
           RUST_BACKTRACE: "1"
-          QUERIES: ${{ inputs.queries }}
-          ITERATIONS: ${{ inputs.iterations || '1' }}
-          QUERY_TIMEOUT_MINUTES: ${{ inputs.query_timeout_minutes || '45' }}
-        run: |
-          set -Eeuo pipefail
-          # Leave the job time for the summary and uploads once the queries have used 9 hours.
-          # Word splitting of $QUERIES is intended.
-          # shellcheck disable=SC2086
-          python3 scripts/bench-memory.py run-queries \
-            --binary "target/release_debug/${{ matrix.engine }}-bench" \
-            --engine "${{ matrix.engine }}" \
-            --formats parquet vortex \
-            ${QUERIES:+--queries $QUERIES} \
-            --iterations "$ITERATIONS" \
-            --timeout "$(( QUERY_TIMEOUT_MINUTES * 60 ))" \
-            --budget "$(( 9 * 3600 ))" \
-            --opt scale-factor="$SCALE_FACTOR" \
-            --out-dir "$OUT_DIR" \
-            --phase-file "$PHASE_FILE" \
-            --min-available-gib 8 \
-            --fail-on-error
+        run: bash scripts/tpch-oom-diag.sh
 
       - name: Stop memory monitor
         if: always()

benchmarks/datafusion-bench/src/lib.rs

@@ -116,6 +116,13 @@ fn vortex_table_options() -> VortexTableOptions {
 
     opts.predicate_pushdown = true;
     opts.projection_pushdown = true;
+    // DO NOT MERGE: diagnostic knob for the SF1000 OOM investigation.
+    if let Some(concurrency) = std::env::var("VORTEX_SCAN_CONCURRENCY")
+        .ok()
+        .and_then(|v| v.parse().ok())
+    {
+        opts.scan_concurrency = Some(concurrency);
+    }
 
     opts
 }

scripts/tpch-oom-diag.sh

@@ -0,0 +1,78 @@
+#!/usr/bin/env bash
+
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+# DO NOT MERGE: diagnoses DataFusion memory at TPC-H SF1000.
+#
+# 1. Runs the heaviest queries under several scan settings, each with its own memory monitor, to
+#    test whether the Vortex excess scales with in-flight scan splits
+#    (VORTEX_SCAN_CONCURRENCY x available parallelism per stream, one stream per partition).
+# 2. Records page-fault call stacks with perf to show which code touches the new memory.
+# 3. Saves EXPLAIN plans for the queries that run out of memory.
+
+set -Eeuo pipefail
+
+BIN=${BIN:-target/release_debug/datafusion-bench}
+QUERIES=(1 2 9 16 18 21)
+RUNNER="ec2_${RUNS_ON_INSTANCE_TYPE:-unknown}"
+
+run_variant() {
+    local name=$1 formats=$2
+    shift 2
+    local dir="$OUT_DIR/$name"
+    mkdir -p "$dir"
+    python3 scripts/bench-memory.py monitor --output "$dir/memory.csv" --phase-file "$PHASE_FILE" --interval 1 &
+    local monitor=$!
+    # shellcheck disable=SC2086
+    env "$@" python3 scripts/bench-memory.py run-queries \
+        --binary "$BIN" --engine datafusion --formats $formats --queries "${QUERIES[@]}" \
+        --timeout 1800 --opt scale-factor="$SCALE_FACTOR" --out-dir "$dir" \
+        --phase-file "$PHASE_FILE" --min-available-gib 8 || true
+    kill "$monitor" || true
+    wait "$monitor" || true
+}
+
+run_variant parquet-default parquet
+run_variant vortex-default vortex
+run_variant vortex-scan1 vortex VORTEX_SCAN_CONCURRENCY=1
+run_variant vortex-tp24 vortex DATAFUSION_EXECUTION_TARGET_PARTITIONS=24
+run_variant vortex-scan1-tp24 vortex VORTEX_SCAN_CONCURRENCY=1 DATAFUSION_EXECUTION_TARGET_PARTITIONS=24
+
+echo "::group::EXPLAIN plans"
+"$BIN" tpch --explain --formats parquet,vortex --queries 9,18,21 --runner "$RUNNER" \
+    --opt scale-factor="$SCALE_FACTOR" > "$OUT_DIR/explain.txt" 2>&1 || true
+echo "::endgroup::"
+
+# Page faults mark the first touch of each new page, so their call stacks show what grows RSS.
+sudo -n apt-get install -y -qq linux-tools-common "linux-tools-$(uname -r)" > /dev/null 2>&1 \
+    || sudo -n apt-get install -y -qq linux-tools-generic > /dev/null 2>&1 || true
+PERF=$(command -v perf || find /usr/lib/linux-tools* -name perf -type f 2>/dev/null | head -1 || true)
+if [[ -n "$PERF" ]] && "$PERF" --version > /dev/null 2>&1; then
+    sudo -n sysctl -w kernel.perf_event_paranoid=-1 > /dev/null || true
+    profile() {
+        local name=$1 format=$2 query=$3
+        shift 3
+        env "$@" "$PERF" record -e page-faults -c 2000 --call-graph fp -o "$OUT_DIR/$name.perf" -- \
+            "$BIN" tpch --formats "$format" --queries "$query" --iterations 1 --hide-progress-bar \
+            --runner "$RUNNER" --opt scale-factor="$SCALE_FACTOR" > /dev/null 2>&1 || true
+        "$PERF" report -i "$OUT_DIR/$name.perf" --stdio --no-children --sort sym -g none \
+            --percent-limit 0.5 > "$OUT_DIR/$name.flat.txt" 2>/dev/null || true
+        "$PERF" report -i "$OUT_DIR/$name.perf" --stdio --no-children --sort sym \
+            -g caller,1,callee,function,percent --percent-limit 3 2>/dev/null \
+            | head -1500 > "$OUT_DIR/$name.graph.txt" || true
+        rm -f "$OUT_DIR/$name.perf"
+    }
+    profile pagefaults-vortex-q01 vortex 1
+    profile pagefaults-parquet-q01 parquet 1
+    profile pagefaults-vortex-q16 vortex 16
+    profile pagefaults-vortex-scan1-q01 vortex 1 VORTEX_SCAN_CONCURRENCY=1
+else
+    echo "::warning::perf unavailable; skipping page-fault profiles"
+fi
+
+for dir in "$OUT_DIR"/*/; do
+    [[ -f "$dir/queries.jsonl" ]] || continue
+    python3 scripts/bench-memory.py summarize --monitor-csv "$dir/memory.csv" \
+        --queries-jsonl "$dir/queries.jsonl" --title "$(basename "$dir")" >> "$GITHUB_STEP_SUMMARY"
+done