step one

AdamGS · AdamGS · commit 55a9552cccb2 · 2026-03-18T10:35:01.000Z
Signed-off-by: Adam Gutglick &lt;adam@spiraldb.com&gt;
diff --git a/scripts/run_clickbench_rss.sh b/scripts/run_clickbench_rss.sh
@@ -0,0 +1,57 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$ROOT_DIR"
+
+if ! command -v gtime >/dev/null 2>&1; then
+  echo "error: gtime not found in PATH" >&2
+  exit 1
+fi
+
+BINARY="${BINARY:-$ROOT_DIR/target/release/datafusion-bench}"
+FORMAT="${FORMAT:-vortex}"
+START_QUERY="${START_QUERY:-0}"
+END_QUERY="${END_QUERY:-42}"
+OUTPUT_DIR="${OUTPUT_DIR:-$ROOT_DIR/data/clickbench-rss}"
+
+mkdir -p "$OUTPUT_DIR"
+
+echo "building release benchmark binary..."
+cargo build -p datafusion-bench --release
+
+timestamp="$(date +%Y%m%d-%H%M%S)"
+csv_path="$OUTPUT_DIR/clickbench-rss-$timestamp.csv"
+
+cat >"$csv_path" <<'EOF'
+query,max_rss_kb,elapsed,user_seconds,system_seconds,cpu_percent,exit_status
+EOF
+
+for ((q = START_QUERY; q <= END_QUERY; q++)); do
+  log_path="$OUTPUT_DIR/q${q}.gtime.$timestamp.log"
+  echo "running clickbench query $q ..."
+
+  gtime --verbose \
+    "$BINARY" clickbench --formats "$FORMAT" --queries "$q" \
+    >"$OUTPUT_DIR/q${q}.stdout.$timestamp.log" \
+    2>"$log_path"
+
+  max_rss_kb="$(awk -F': ' '/Maximum resident set size/ {print $2}' "$log_path" | tr -d '[:space:]')"
+  elapsed="$(awk -F': ' '/Elapsed \(wall clock\) time/ {print $2}' "$log_path" | sed 's/^ *//')"
+  user_seconds="$(awk -F': ' '/User time \(seconds\)/ {print $2}' "$log_path" | tr -d '[:space:]')"
+  system_seconds="$(awk -F': ' '/System time \(seconds\)/ {print $2}' "$log_path" | tr -d '[:space:]')"
+  cpu_percent="$(awk -F': ' '/Percent of CPU this job got/ {print $2}' "$log_path" | tr -d '[:space:]')"
+  exit_status="$(awk -F': ' '/Exit status/ {print $2}' "$log_path" | tr -d '[:space:]')"
+
+  printf '%s,%s,%s,%s,%s,%s,%s\n' \
+    "$q" \
+    "${max_rss_kb:-}" \
+    "${elapsed:-}" \
+    "${user_seconds:-}" \
+    "${system_seconds:-}" \
+    "${cpu_percent:-}" \
+    "${exit_status:-}" \
+    >>"$csv_path"
+done
+
+echo "wrote summary to $csv_path"
diff --git a/vortex-datafusion/src/persistent/format.rs b/vortex-datafusion/src/persistent/format.rs
@@ -101,7 +101,7 @@ config_namespace! {
         /// all expressions are evaluated after the scan.
         pub projection_pushdown: bool, default = false
         /// The intra-partition scan concurrency, controlling the number of row splits to process
-        /// concurrently per-thread within each file.
+        /// concurrently within each file.
         ///
         /// This does not affect the overall parallelism
         /// across partitions, which is controlled by DataFusion's execution configuration.
diff --git a/vortex-layout/src/layouts/flat/reader.rs b/vortex-layout/src/layouts/flat/reader.rs
@@ -5,7 +5,6 @@ use std::collections::BTreeSet;
 use std::ops::BitAnd;
 use std::ops::Range;
 use std::sync::Arc;
-use std::sync::OnceLock;
 
 use futures::FutureExt;
 use futures::future::BoxFuture;
@@ -40,7 +39,6 @@ pub struct FlatReader {
     name: Arc<str>,
     segment_source: Arc<dyn SegmentSource>,
     session: VortexSession,
-    array: OnceLock<SharedArrayFuture>,
 }
 
 impl FlatReader {
@@ -55,38 +53,33 @@ impl FlatReader {
             name,
             segment_source,
             session,
-            array: Default::default(),
         }
     }
 
     /// Register the segment request and return a future that would resolve into the deserialised array.
     fn array_future(&self) -> SharedArrayFuture {
-        self.array
-            .get_or_init(|| {
-                let row_count = usize::try_from(self.layout.row_count())
-                    .vortex_expect("row count must fit in usize");
-                let segment_fut = self.segment_source.request(self.layout.segment_id());
-                let ctx = self.layout.array_ctx().clone();
-                let session = self.session.clone();
-                let dtype = self.layout.dtype().clone();
-                let array_tree = self.layout.array_tree().cloned();
-                async move {
-                    let segment = segment_fut.await?;
-                    let parts = if let Some(array_tree) = array_tree {
-                        // Use the pre-stored flatbuffer from layout metadata combined with segment buffers.
-                        ArrayParts::from_flatbuffer_and_segment(array_tree, segment)?
-                    } else {
-                        // Parse the flatbuffer from the segment itself.
-                        ArrayParts::try_from(segment)?
-                    };
-                    parts
-                        .decode(&dtype, row_count, &ctx, &session)
-                        .map_err(Arc::new)
-                }
-                .boxed()
-                .shared()
-            })
-            .clone()
+        let row_count =
+            usize::try_from(self.layout.row_count()).vortex_expect("row count must fit in usize");
+        let segment_fut = self.segment_source.request(self.layout.segment_id());
+        let ctx = self.layout.array_ctx().clone();
+        let session = self.session.clone();
+        let dtype = self.layout.dtype().clone();
+        let array_tree = self.layout.array_tree().cloned();
+        async move {
+            let segment = segment_fut.await?;
+            let parts = if let Some(array_tree) = array_tree {
+                // Use the pre-stored flatbuffer from layout metadata combined with segment buffers.
+                ArrayParts::from_flatbuffer_and_segment(array_tree, segment)?
+            } else {
+                // Parse the flatbuffer from the segment itself.
+                ArrayParts::try_from(segment)?
+            };
+            parts
+                .decode(&dtype, row_count, &ctx, &session)
+                .map_err(Arc::new)
+        }
+        .boxed()
+        .shared()
     }
 }
 
diff --git a/vortex-scan/src/multi.rs b/vortex-scan/src/multi.rs
@@ -102,10 +102,6 @@ impl MultiLayoutDataSource {
         session: &VortexSession,
     ) -> Self {
         let dtype = first.dtype().clone();
-        let concurrency = std::thread::available_parallelism()
-            .map(|v| v.get())
-            .unwrap_or(DEFAULT_CONCURRENCY);
-
         let mut children = Vec::with_capacity(1 + remaining.len());
         children.push(MultiLayoutChild::Opened(first));
         children.extend(remaining.into_iter().map(MultiLayoutChild::Deferred));
@@ -114,7 +110,7 @@ impl MultiLayoutDataSource {
             dtype,
             session: session.clone(),
             children,
-            concurrency,
+            concurrency: DEFAULT_CONCURRENCY,
         }
     }
 
@@ -128,25 +124,21 @@ impl MultiLayoutDataSource {
         factories: Vec<Arc<dyn LayoutReaderFactory>>,
         session: &VortexSession,
     ) -> Self {
-        let concurrency = std::thread::available_parallelism()
-            .map(|v| v.get())
-            .unwrap_or(DEFAULT_CONCURRENCY);
-
         Self {
             dtype,
             session: session.clone(),
             children: factories
                 .into_iter()
                 .map(MultiLayoutChild::Deferred)
                 .collect(),
-            concurrency,
+            concurrency: DEFAULT_CONCURRENCY,
         }
     }
 
     /// Sets the concurrency for opening deferred readers.
     ///
     /// Controls how many file opens run in parallel via `buffer_unordered`.
-    /// Defaults to the number of available CPU cores.
+    /// Defaults to a fixed budget.
     pub fn with_concurrency(mut self, concurrency: usize) -> Self {
         self.concurrency = concurrency;
         self
diff --git a/vortex-scan/src/repeated_scan.rs b/vortex-scan/src/repeated_scan.rs
@@ -71,7 +71,7 @@ pub struct RepeatedScan<A: 'static + Send> {
     selection: Selection,
     /// The natural splits of the file.
     splits: Splits,
-    /// The number of splits to make progress on concurrently **per-thread**.
+    /// The total number of splits to make progress on concurrently.
     concurrency: usize,
     /// Function to apply to each [`ArrayRef`] within the spawned split tasks.
     map_fn: Arc<dyn Fn(ArrayRef) -> VortexResult<A> + Send + Sync>,
@@ -148,11 +148,12 @@ impl<A: 'static + Send> RepeatedScan<A> {
         &self,
         row_range: Option<Range<u64>>,
     ) -> VortexResult<BoxStream<'static, VortexResult<A>>> {
-        let num_workers = std::thread::available_parallelism()
-            .map(|n| n.get())
-            .unwrap_or(1);
-        let concurrency = self.concurrency * num_workers;
-        self.execute_stream(row_range, concurrency, self.ordered, self.session.handle())
+        self.execute_stream(
+            row_range,
+            self.concurrency,
+            self.ordered,
+            self.session.handle(),
+        )
     }
 
     fn legacy_stream_from_ranges(
diff --git a/vortex-scan/src/scan_builder.rs b/vortex-scan/src/scan_builder.rs
@@ -61,7 +61,7 @@ pub struct ScanBuilder<A> {
     selection: Selection,
     /// How to split the file for concurrent processing.
     split_by: SplitBy,
-    /// The number of splits to make progress on concurrently **per-thread**.
+    /// The total number of splits to make progress on concurrently.
     concurrency: usize,
     /// Function to apply to each [`ArrayRef`] within the spawned split tasks.
     map_fn: Arc<dyn Fn(ArrayRef) -> VortexResult<A> + Send + Sync>,
@@ -80,6 +80,7 @@ pub struct ScanBuilder<A> {
 }
 
 const DEFAULT_TARGET_OUTPUT_ROWS: usize = DEFAULT_TARGET_OUTPUT_ROWS_HINT;
+const DEFAULT_SCAN_CONCURRENCY: usize = 16;
 const MIN_TARGET_OUTPUT_BYTES: usize = 1 << 20;
 const MAX_TARGET_OUTPUT_BYTES: usize = 8 << 20;
 const DEFAULT_VARIABLE_ROW_BYTES: usize = 32;
@@ -96,9 +97,9 @@ impl ScanBuilder<ArrayRef> {
             row_range: None,
             selection: Default::default(),
             split_by: SplitBy::Layout,
-            // We default to four tasks per worker thread, which allows for some I/O lookahead
-            // without too much impact on work-stealing.
-            concurrency: 4,
+            // Keep one explicit scan-concurrency budget rather than multiplying by the
+            // runtime's worker count later.
+            concurrency: DEFAULT_SCAN_CONCURRENCY,
             map_fn: Arc::new(Ok),
             metrics_registry: None,
             file_stats: None,
@@ -186,8 +187,7 @@ impl<A: 'static + Send> ScanBuilder<A> {
         self.concurrency
     }
 
-    /// The number of row splits to make progress on concurrently per-thread, must
-    /// be greater than 0.
+    /// The total number of row splits to make progress on concurrently.
     pub fn with_concurrency(mut self, concurrency: usize) -> Self {
         assert!(concurrency > 0);
         self.concurrency = concurrency;
@@ -400,10 +400,7 @@ impl<A: 'static + Send> Stream for LazyScanStream<A> {
                 LazyScanState::Builder(builder) => {
                     let builder = builder.take().vortex_expect("polled after completion");
                     let ordered = builder.ordered;
-                    let num_workers = std::thread::available_parallelism()
-                        .map(|n| n.get())
-                        .unwrap_or(1);
-                    let concurrency = builder.concurrency * num_workers;
+                    let concurrency = builder.concurrency;
                     let handle = builder.session.handle();
                     let execute_handle = handle.clone();
                     let task = handle.spawn_blocking(move || {
