vortex-datafusion: stop scan concurrency multiplying with partitions

A Vortex scan runs `concurrency * available_parallelism` splits per
stream, and DataFusion opens one stream per partition, one per core by
default. Vortex's default concurrency of 4 multiplied with that: TPC-H
SF1000 on 96 cores kept ~37k decoded splits in flight, and DataFusion Q1
on Vortex peaked at 115 GiB against Parquet's 37 GiB. With concurrency 1
it peaked at 31 GiB and ran faster (8.7s vs 10.0s).

Default the DataFusion scan concurrency to 1 when the table options leave
it unset; `vortex.scan_concurrency` still overrides it.

DO NOT MERGE as-is: this commit also points the diagnosis run at the fix,
with and without a memory limit, in its own concurrency group.

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

@@ -11,7 +11,7 @@
 name: TPC-H SF1000 (OOM diagnosis)
 
 concurrency:
-  group: tpch-sf1000-oom-diag
+  group: tpch-sf1000-oom-diag-${{ github.sha }}
   cancel-in-progress: false
 
 on:

scripts/tpch-oom-diag.sh

@@ -3,23 +3,15 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-# DO NOT MERGE: tries DataFusion with a memory limit and spilling at TPC-H SF1000.
-#
-# With DATAFUSION_BENCH_MEMORY_LIMIT_GIB set, aggregations and sorts spill to disk and hash join
-# builds that do not fit fail with a resources-exhausted error instead of growing until the
-# process is killed. prefer_hash_join=false swaps hash joins for sort-merge joins, which spill.
+# DO NOT MERGE: checks whether DataFusion runs TPC-H SF1000 with the Vortex scan concurrency fix
+# (one split per core per stream by default), with and without a memory limit and spilling.
 
 set -Eeuo pipefail
 
 BIN=${BIN:-target/release_debug/datafusion-bench}
 ALL_QUERIES=(1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22)
-JOIN_QUERIES=(9 18 21)
 SPILL_DIR="${GITHUB_WORKSPACE:-$PWD}/df-spill"
-
-# The memory pool does not track scan buffers or batches in flight, so leave headroom below the
-# machine's 188 GiB for them.
-export DATAFUSION_BENCH_MEMORY_LIMIT_GIB=${DATAFUSION_BENCH_MEMORY_LIMIT_GIB:-100}
-export DATAFUSION_BENCH_SPILL_DIR="$SPILL_DIR"
+RUNNER="ec2_${RUNS_ON_INSTANCE_TYPE:-unknown}"
 export RUST_BACKTRACE=0
 
 run_variant() {
@@ -40,12 +32,16 @@ run_variant() {
     rm -rf "$SPILL_DIR"
 }
 
-run_variant parquet-spill parquet "${ALL_QUERIES[*]}"
-run_variant vortex-spill vortex "${ALL_QUERIES[*]}"
-run_variant vortex-spill-scan1 vortex "${ALL_QUERIES[*]}" VORTEX_SCAN_CONCURRENCY=1
-run_variant parquet-spill-smj parquet "${JOIN_QUERIES[*]}" DATAFUSION_OPTIMIZER_PREFER_HASH_JOIN=false
-run_variant vortex-spill-scan1-smj vortex "${JOIN_QUERIES[*]}" \
-    VORTEX_SCAN_CONCURRENCY=1 DATAFUSION_OPTIMIZER_PREFER_HASH_JOIN=false
+run_variant vortex-fix vortex "${ALL_QUERIES[*]}"
+run_variant vortex-fix-spill vortex "${ALL_QUERIES[*]}" \
+    DATAFUSION_BENCH_MEMORY_LIMIT_GIB=100 DATAFUSION_BENCH_SPILL_DIR="$SPILL_DIR"
+run_variant parquet-spill parquet "${ALL_QUERIES[*]}" \
+    DATAFUSION_BENCH_MEMORY_LIMIT_GIB=100 DATAFUSION_BENCH_SPILL_DIR="$SPILL_DIR"
+
+# Q2 and Q16 still used more memory on Vortex than on Parquet with the old scan settings, and
+# only responded to the partition count; compare their plans.
+"$BIN" tpch --explain --formats parquet,vortex --queries 2,16 --runner "$RUNNER" \
+    --opt scale-factor="$SCALE_FACTOR" > "$OUT_DIR/explain-q2-q16.txt" 2>&1 || true
 
 for dir in "$OUT_DIR"/*/; do
     [[ -f "$dir/queries.jsonl" ]] || continue

vortex-datafusion/src/persistent/format.rs

@@ -223,7 +223,8 @@ extensions_options! {
         /// This does not affect the overall parallelism
         /// across partitions, which is controlled by DataFusion's execution configuration.
         ///
-        /// Leave as `None` to use Vortex's scan default. Override per session
+        /// Leave as `None` for one split per core per stream: DataFusion already runs one stream
+        /// per partition, so Vortex's own default would multiply with it. Override per session
         /// with `SET vortex.scan_concurrency = <n>`, or per table with
         /// `OPTIONS(scan_concurrency '<n>')`.
         pub scan_concurrency: Option<usize>, default = None

vortex-datafusion/src/persistent/opener.rs

@@ -70,6 +70,10 @@ use crate::persistent::cache::CachedVortexMetadata;
 use crate::persistent::reader::VortexReaderFactory;
 use crate::persistent::stream::PrunableStream;
 
+/// Per-core scan concurrency when the table options leave it unset. It is lower than the Vortex
+/// scan default because DataFusion already runs one scan stream per partition.
+const DATAFUSION_DEFAULT_SCAN_CONCURRENCY: usize = 1;
+
 #[derive(Clone)]
 pub(crate) struct VortexOpener {
     /// The partition this opener is assigned to. Only used for labeling metrics.
@@ -409,9 +413,13 @@ impl FileOpener for VortexOpener {
                 scan_builder = scan_builder.with_limit(limit);
             }
 
-            if let Some(concurrency) = scan_concurrency {
-                scan_builder = scan_builder.with_concurrency(concurrency);
-            }
+            // The scan runs `concurrency * available_parallelism` splits per stream, and DataFusion
+            // already opens one stream per partition (one per core by default). Vortex's default of
+            // 4 multiplies with that: at SF1000 on 96 cores it kept ~37k decoded splits in flight
+            // and Q1 peaked at 115 GiB against Parquet's 37 GiB. One split per core per stream
+            // keeps every core busy without the second multiplier.
+            scan_builder = scan_builder
+                .with_concurrency(scan_concurrency.unwrap_or(DATAFUSION_DEFAULT_SCAN_CONCURRENCY));
 
             // Set before the byte-range translation below, which computes natural splits for
             // the fields the scan's projection and filter reference.