From cbde19d42ebe849b21d506c92d980ed071b3b46c Mon Sep 17 00:00:00 2001 From: Denys Tsomenko Date: Thu, 10 Sep 2026 09:11:28 +0300 Subject: [PATCH 1/3] datasource: opt-in open-ahead so FileStream overlaps the next file's open with scanning The morsel-driven FileStream (55.0) opens files strictly one after another per partition: the next file's footer/page-index/bloom-filter I/O starts only once the active reader is exhausted. Before the rewrite the next file's open future was polled while the current file streamed, hiding per-file open latency. On many-file scans over object storage that latency is now exposed on every file (TPC-H SF100 Q1: DataSourceExec time_elapsed_opening 2.2 s -> 82.3 s summed over 5 partitions, ~16 s of wall). Add `datafusion.execution.file_stream_open_ahead` (default false). When set, ScanState claims the next file while a reader is active and drives its planning until it either yields a ready morsel or blocks on its single outstanding I/O, which poll_scan resolves ahead of polling the reader. At most one file is in flight ahead of the reader, so the extra footprint is one file's metadata per partition. The opening timer then measures only the exposed wait (no reader to drain). Default off keeps the existing behaviour and snapshots byte-identical. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_012m5Yx6yEpZsacqAhZotgkT --- datafusion/common/src/config.rs | 10 ++ .../datasource/src/file_scan_config/mod.rs | 7 ++ .../datasource/src/file_stream/builder.rs | 12 ++ .../datasource/src/file_stream/metrics.rs | 7 ++ datafusion/datasource/src/file_stream/mod.rs | 83 ++++++++++++ .../datasource/src/file_stream/scan_state.rs | 118 +++++++++++++++++- .../test_files/information_schema.slt | 2 + docs/source/user-guide/configs.md | 1 + 8 files changed, 234 insertions(+), 6 deletions(-) diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index bbdbd4f2dcaf6..e3798f389dcb5 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1039,6 +1039,16 @@ config_namespace! { /// runtime reassignment occurs. pub enable_file_stream_work_stealing: bool, default = true + /// When `true`, each file-scan partition starts opening its next file + /// (footer, page index and bloom filter I/O) while the current file is + /// still being scanned, keeping at most one file in flight ahead of + /// the active reader. This overlaps per-file open latency with data + /// reads, which matters for scans over many files on high-latency + /// object stores. The cost is holding one extra file's metadata per + /// partition. When `false` (the default) a partition opens the next + /// file only after the current one is fully consumed. + pub file_stream_open_ahead: bool, default = false + /// Aggregation ratio (number of distinct groups / number of input rows) /// threshold for skipping partial aggregation. If the value is greater /// then partial aggregation will skip aggregation for further input diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs index 91dcd5b76fc46..8a4687e7450d1 100644 --- a/datafusion/datasource/src/file_scan_config/mod.rs +++ b/datafusion/datasource/src/file_scan_config/mod.rs @@ -720,11 +720,18 @@ impl DataSource for FileScanConfig { .and_then(|state| state.downcast_ref::()) .cloned(); + let open_ahead = context + .session_config() + .options() + .execution + .file_stream_open_ahead; + let stream = FileStreamBuilder::new(self) .with_partition(partition) .with_shared_work_source(shared_work_source) .with_morselizer(morselizer) .with_metrics(source.metrics()) + .with_open_ahead(open_ahead) .build()?; Ok(Box::pin(cooperative(stream))) } diff --git a/datafusion/datasource/src/file_stream/builder.rs b/datafusion/datasource/src/file_stream/builder.rs index 7034e902550a9..1a4a27f364b5b 100644 --- a/datafusion/datasource/src/file_stream/builder.rs +++ b/datafusion/datasource/src/file_stream/builder.rs @@ -35,6 +35,7 @@ pub struct FileStreamBuilder<'a> { metrics: Option<&'a ExecutionPlanMetricsSet>, on_error: OnError, shared_work_source: Option, + open_ahead: bool, } impl<'a> FileStreamBuilder<'a> { @@ -47,6 +48,7 @@ impl<'a> FileStreamBuilder<'a> { metrics: None, on_error: OnError::Fail, shared_work_source: None, + open_ahead: false, } } @@ -93,6 +95,14 @@ impl<'a> FileStreamBuilder<'a> { self } + /// Configure whether the stream opens its next file while the current + /// file is still being scanned (see + /// `datafusion.execution.file_stream_open_ahead`). + pub fn with_open_ahead(mut self, open_ahead: bool) -> Self { + self.open_ahead = open_ahead; + self + } + /// Build the configured [`FileStream`]. pub fn build(self) -> Result { let Self { @@ -102,6 +112,7 @@ impl<'a> FileStreamBuilder<'a> { metrics, on_error, shared_work_source, + open_ahead, } = self; let Some(partition) = partition else { @@ -130,6 +141,7 @@ impl<'a> FileStreamBuilder<'a> { config.limit, morselizer, on_error, + open_ahead, file_stream_metrics, )); diff --git a/datafusion/datasource/src/file_stream/metrics.rs b/datafusion/datasource/src/file_stream/metrics.rs index 5f3894404f408..6e2d06316a5b9 100644 --- a/datafusion/datasource/src/file_stream/metrics.rs +++ b/datafusion/datasource/src/file_stream/metrics.rs @@ -33,6 +33,13 @@ impl StartableTime { self.start = Some(Instant::now()); } + /// Start the timer unless it is already running. + pub fn start_if_stopped(&mut self) { + if self.start.is_none() { + self.start = Some(Instant::now()); + } + } + pub fn stop(&mut self) { if let Some(start) = self.start.take() { self.metrics.add_elapsed(start); diff --git a/datafusion/datasource/src/file_stream/mod.rs b/datafusion/datasource/src/file_stream/mod.rs index 6daed7c338022..5afb393b27f1e 100644 --- a/datafusion/datasource/src/file_stream/mod.rs +++ b/datafusion/datasource/src/file_stream/mod.rs @@ -763,6 +763,77 @@ mod tests { Ok(()) } + /// Verifies that with `open_ahead` the next file is morselized and its + /// I/O is issued while the previous file's morsel is still producing + /// batches, so per-file open latency overlaps with reading. + #[tokio::test] + async fn morsel_open_ahead_overlaps_next_file_io() -> Result<()> { + let test = FileStreamMorselTest::new() + .with_open_ahead(true) + .with_file( + MockPlanner::builder("file1.parquet") + .add_plan( + PendingPlannerBuilder::new(IoFutureId(1)) + .with_polls_to_resolve(PollsToResolve(1)), + ) + .add_plan( + MockPlanBuilder::new() + .with_morsel_batches(MorselId(10), vec![42, 43]), + ) + .return_none(), + ) + .with_file( + MockPlanner::builder("file2.parquet") + .add_plan( + PendingPlannerBuilder::new(IoFutureId(2)) + .with_polls_to_resolve(PollsToResolve(1)), + ) + .add_plan( + MockPlanBuilder::new() + .with_morsel_batches(MorselId(20), vec![44]), + ) + .return_none(), + ); + + // file2 is morselized and its I/O issued while morsel 10 is still + // producing batches; the I/O resolves before morsel 10 finishes. + insta::assert_snapshot!(test.run().await.unwrap(), @r" + ----- Output Stream ----- + Batch: 42 + Batch: 43 + Batch: 44 + Done + ----- File Stream Events ----- + morselize_file: file1.parquet + planner_created: file1.parquet + planner_called: file1.parquet + io_future_created: file1.parquet, IoFutureId(1) + io_future_polled: file1.parquet, IoFutureId(1) + io_future_polled: file1.parquet, IoFutureId(1) + io_future_resolved: file1.parquet, IoFutureId(1) + planner_called: file1.parquet + morsel_produced: file1.parquet, MorselId(10) + morsel_stream_started: MorselId(10) + morselize_file: file2.parquet + planner_created: file2.parquet + planner_called: file2.parquet + io_future_created: file2.parquet, IoFutureId(2) + morsel_stream_batch_produced: MorselId(10), BatchId(42) + io_future_polled: file2.parquet, IoFutureId(2) + morsel_stream_batch_produced: MorselId(10), BatchId(43) + io_future_polled: file2.parquet, IoFutureId(2) + io_future_resolved: file2.parquet, IoFutureId(2) + planner_called: file2.parquet + morsel_produced: file2.parquet, MorselId(20) + morsel_stream_finished: MorselId(10) + morsel_stream_started: MorselId(20) + morsel_stream_batch_produced: MorselId(20), BatchId(44) + morsel_stream_finished: MorselId(20) + "); + + Ok(()) + } + /// Verifies that a planner can traverse two sequential I/O phases before /// producing one batch, similar to Parquet. #[tokio::test] @@ -1368,6 +1439,7 @@ mod tests { preserve_order: bool, declared_output_partitioning: bool, enable_file_stream_work_stealing: bool, + open_ahead: bool, file_stream_events: bool, build_streams_on_first_read: bool, reads: Vec, @@ -1383,6 +1455,7 @@ mod tests { preserve_order: false, declared_output_partitioning: false, enable_file_stream_work_stealing: true, + open_ahead: false, file_stream_events: true, build_streams_on_first_read: false, reads: vec![], @@ -1436,6 +1509,14 @@ mod tests { self } + /// Sets `datafusion.execution.file_stream_open_ahead`: when enabled, + /// each stream starts planning its next file while the active reader + /// is still producing batches. + fn with_open_ahead(mut self, open_ahead: bool) -> Self { + self.open_ahead = open_ahead; + self + } + /// Controls whether scheduler events are included in the snapshot. /// /// When disabled, `run()` still includes the event section header but @@ -1527,6 +1608,7 @@ mod tests { .with_shared_work_source(shared_work_source.clone()) .with_morselizer(Box::new(self.morselizer.clone())) .with_metrics(&metrics_set) + .with_open_ahead(self.open_ahead) .build()?; partitions[partition].set_stream(stream); } @@ -1554,6 +1636,7 @@ mod tests { .with_shared_work_source(shared_work_source.clone()) .with_morselizer(Box::new(self.morselizer.clone())) .with_metrics(&metrics_set) + .with_open_ahead(self.open_ahead) .build()?; partition_state.set_stream(stream); } diff --git a/datafusion/datasource/src/file_stream/scan_state.rs b/datafusion/datasource/src/file_stream/scan_state.rs index 21125cd08896c..630b75a0b4811 100644 --- a/datafusion/datasource/src/file_stream/scan_state.rs +++ b/datafusion/datasource/src/file_stream/scan_state.rs @@ -81,6 +81,9 @@ pub(super) struct ScanState { /// Once the I/O completes, yields the next planner and is pushed back /// onto `ready_planners`. pending_planner: Option, + /// Whether to start opening the next file while the active reader is + /// still producing batches. See [`Self::advance_open_ahead`]. + open_ahead: bool, /// Metrics for the active scan queues. metrics: FileStreamMetrics, } @@ -91,6 +94,7 @@ impl ScanState { remain: Option, morselizer: Box, on_error: OnError, + open_ahead: bool, metrics: FileStreamMetrics, ) -> Self { Self { @@ -102,6 +106,7 @@ impl ScanState { ready_morsels: Default::default(), reader: None, pending_planner: None, + open_ahead, metrics, } } @@ -115,16 +120,20 @@ impl ScanState { /// /// Work is attempted in this order: /// 1. resolve any pending planner I/O - /// 2. poll the active reader - /// 3. turn a ready morsel into the active reader - /// 4. run CPU planning on a ready planner - /// 5. morselize the next unopened file + /// 2. with `open_ahead`, start planning the next file behind the active + /// reader (see [`Self::advance_open_ahead`]) + /// 3. poll the active reader + /// 4. turn a ready morsel into the active reader + /// 5. run CPU planning on a ready planner + /// 6. morselize the next unopened file /// /// The return [`ScanAndReturn`] tells `poll_inner` how to update the /// outer `FileStreamState`. pub(super) fn poll_scan(&mut self, cx: &mut Context<'_>) -> ScanAndReturn { - let _processing_timer: ScopedTimerGuard<'_> = - self.metrics.time_processing.timer(); + // Guard a clone of the shared timer so the borrow does not pin + // `self.metrics` for the whole poll (the look-ahead needs `&mut self`). + let time_processing = self.metrics.time_processing.clone(); + let _processing_timer: ScopedTimerGuard<'_> = time_processing.timer(); // Try and resolve outstanding IO first. If it is still pending, check // the current reader or ready morsels before yielding. New planning @@ -154,6 +163,16 @@ impl ScanState { } } + // With open-ahead enabled, use the time the active reader is busy to + // get the next file's planning (and its single outstanding I/O) + // under way, so per-file open latency overlaps with data reads. + if self.open_ahead + && self.reader.is_some() + && let Some(ret) = self.advance_open_ahead() + { + return ret; + } + // Next try and get the next batch from the active reader, if any. if let Some(reader) = self.reader.as_mut() { match reader.poll_next_unpin(cx) { @@ -225,6 +244,12 @@ impl ScanState { // still outstanding because they may need additional IO and ScanState // currently only permits a single outstanding IO if self.pending_planner.is_some() { + if self.open_ahead { + // The next file was claimed while a reader was active without + // starting the opening timer; only the wait that is actually + // exposed (no reader to drain) counts as opening time. + self.metrics.time_opening.start_if_stopped(); + } return ScanAndReturn::Return(Poll::Pending); } @@ -291,6 +316,87 @@ impl ScanState { } } +impl ScanState { + /// Drives planning of the *next* file while a reader is active, keeping at + /// most one file in flight ahead of it. + /// + /// Claims the next unopened file once nothing else is queued and runs its + /// CPU planning until it either yields a ready morsel or blocks on I/O. + /// The single-outstanding-I/O rule still holds: the pending planner is + /// polled by [`Self::poll_scan`] before the reader on every iteration, so + /// the file's open latency overlaps with the reader's data reads instead + /// of following them. + /// + /// The opening timer is deliberately not started here: time spent opening + /// a file behind an active reader is not exposed to the query. It is + /// started by `poll_scan` only if the stream later has to wait for that + /// open with no reader left to drain. + /// + /// Returns `Some` when the outer loop must return (an error surfaced by + /// the look-ahead), `None` to continue with the active reader. + fn advance_open_ahead(&mut self) -> Option { + loop { + // The next file is already either fully open or waiting on I/O. + if self.pending_planner.is_some() || !self.ready_morsels.is_empty() { + return None; + } + + if let Some(planner) = self.ready_planners.pop_front() { + match planner.plan() { + Ok(Some(mut plan)) => { + self.ready_morsels.extend(plan.take_morsels()); + self.ready_planners.extend(plan.take_ready_planners()); + if let Some(pending_planner) = plan.take_pending_planner() { + if self.pending_planner.is_some() { + return Some(ScanAndReturn::Error( + internal_datafusion_err!( + "Conflicting pending planner state in FileStream ScanState" + ), + )); + } + self.pending_planner = Some(pending_planner); + } + } + // Nothing to read from this file (e.g. pruned late). + Ok(None) => { + self.metrics.files_processed.add(1); + } + Err(err) => { + self.metrics.file_open_errors.add(1); + match self.on_error { + OnError::Skip => { + self.metrics.files_processed.add(1); + } + OnError::Fail => return Some(ScanAndReturn::Error(err)), + } + } + } + continue; + } + + // Nothing queued for the next file: claim it. + let Some(part_file) = self.work_source.pop_front() else { + return None; + }; + match self.morselizer.plan_file(part_file) { + Ok(planner) => { + self.metrics.files_opened.add(1); + self.ready_planners.push_back(planner); + } + Err(err) => { + self.metrics.file_open_errors.add(1); + match self.on_error { + OnError::Skip => { + self.metrics.files_processed.add(1); + } + OnError::Fail => return Some(ScanAndReturn::Error(err)), + } + } + } + } + } +} + /// What should be done on the next iteration of [`ScanState::poll_scan`]? pub(super) enum ScanAndReturn { /// Poll again. diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index 399f81b907807..18f9e5c62d435 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -223,6 +223,7 @@ datafusion.execution.enable_hash_join_spill false datafusion.execution.enable_migration_aggregate true datafusion.execution.enable_recursive_ctes true datafusion.execution.enforce_batch_size_in_joins false +datafusion.execution.file_stream_open_ahead false datafusion.execution.hash_join_buffering_capacity 0 datafusion.execution.hash_join_spill_headroom_bytes 33554432 datafusion.execution.hash_join_spill_max_recursion_depth 2 @@ -388,6 +389,7 @@ datafusion.execution.enable_hash_join_spill false When enabled, a hash join whos datafusion.execution.enable_migration_aggregate true Temporary switch for aggregate stream implementations that are being migrated from `GroupedHashAggregateStream`. When set to true, DataFusion tries the migrated implementations when their preconditions are satisfied. When set to false, grouped aggregation falls back to `GroupedHashAggregateStream`. This option will be removed after the migration is finished. See for details. datafusion.execution.enable_recursive_ctes true Should DataFusion support recursive CTEs datafusion.execution.enforce_batch_size_in_joins false Should DataFusion enforce batch size in joins or not. By default, DataFusion will not enforce batch size in joins. Enforcing batch size in joins can reduce memory usage when joining large tables with a highly-selective join filter, but is also slightly slower. +datafusion.execution.file_stream_open_ahead false When `true`, each file-scan partition starts opening its next file (footer, page index and bloom filter I/O) while the current file is still being scanned, keeping at most one file in flight ahead of the active reader. This overlaps per-file open latency with data reads, which matters for scans over many files on high-latency object stores. The cost is holding one extra file's metadata per partition. When `false` (the default) a partition opens the next file only after the current one is fully consumed. datafusion.execution.hash_join_buffering_capacity 0 How many bytes to buffer in the probe side of hash joins while the build side is concurrently being built. Without this, hash joins will wait until the full materialization of the build side before polling the probe side. This is useful in scenarios where the query is not completely CPU bounded, allowing to do some early work concurrently and reducing the latency of the query. Note that when hash join buffering is enabled, the probe side will start eagerly polling data, not giving time for the producer side of dynamic filters to produce any meaningful predicate. Queries with dynamic filters might see performance degradation. Disabled by default, set to a number greater than 0 for enabling it. datafusion.execution.hash_join_spill_headroom_bytes 33554432 Reserved memory headroom for a spilling hash join's scatter scratch space and per-partition write buffers (analogous to `sort_spill_reservation_bytes`). datafusion.execution.hash_join_spill_max_recursion_depth 2 Maximum number of recursive repartition passes a spilling hash join applies to a partition whose build side still exceeds the memory budget (key skew). When the limit is reached the join falls back to a chunked build where supported, or reports a clean resources-exhausted error. diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index aeecf2772b869..57d01fa5d4356 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -141,6 +141,7 @@ The following configuration settings are available: | datafusion.execution.split_file_groups_by_statistics | false | Attempt to eliminate sorts by packing & sorting files with non-overlapping statistics into the same file groups. Currently experimental | | datafusion.execution.keep_partition_by_columns | false | Should DataFusion keep the columns used for partition_by in the output RecordBatches | | datafusion.execution.enable_file_stream_work_stealing | true | When `true` (the default), DataFusion's built-in file scans dynamically rebalance files across partitions at query execution time: a partition that goes idle reads files (or byte-range morsels) originally assigned to a sibling partition, which keeps all partitions busy in a single process. Executors that depend on the plan-time partition assignment — such as Ballista and datafusion-distributed, which run each partition as an isolated task and never poll the siblings — should set this to `false` so each partition reads only its own file group and no runtime reassignment occurs. | +| datafusion.execution.file_stream_open_ahead | false | When `true`, each file-scan partition starts opening its next file (footer, page index and bloom filter I/O) while the current file is still being scanned, keeping at most one file in flight ahead of the active reader. This overlaps per-file open latency with data reads, which matters for scans over many files on high-latency object stores. The cost is holding one extra file's metadata per partition. When `false` (the default) a partition opens the next file only after the current one is fully consumed. | | datafusion.execution.skip_partial_aggregation_probe_ratio_threshold | 0.8 | Aggregation ratio (number of distinct groups / number of input rows) threshold for skipping partial aggregation. If the value is greater then partial aggregation will skip aggregation for further input | | datafusion.execution.skip_partial_aggregation_probe_rows_threshold | 100000 | Number of input rows partial aggregation partition should process, before aggregation ratio check and trying to switch to skipping aggregation mode | | datafusion.execution.use_row_number_estimates_to_optimize_partitioning | false | Should DataFusion use row number estimates at the input to decide whether increasing parallelism is beneficial or not. By default, only exact row numbers (not estimates) are used for this decision. Setting this flag to `true` will likely produce better plans. if the source of statistics is accurate. We plan to make this the default in the future. | From 0e64db5f254b96279f47f30adc67568541ac7687 Mon Sep 17 00:00:00 2001 From: Denys Tsomenko Date: Thu, 10 Sep 2026 09:27:03 +0300 Subject: [PATCH 2/3] datasource: clippy question_mark in advance_open_ahead (no behaviour change) Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_012m5Yx6yEpZsacqAhZotgkT --- datafusion/datasource/src/file_stream/scan_state.rs | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/datafusion/datasource/src/file_stream/scan_state.rs b/datafusion/datasource/src/file_stream/scan_state.rs index 630b75a0b4811..530fa97c15795 100644 --- a/datafusion/datasource/src/file_stream/scan_state.rs +++ b/datafusion/datasource/src/file_stream/scan_state.rs @@ -374,10 +374,9 @@ impl ScanState { continue; } - // Nothing queued for the next file: claim it. - let Some(part_file) = self.work_source.pop_front() else { - return None; - }; + // Nothing queued for the next file: claim it (`None` once the + // work source is drained: keep serving the active reader). + let part_file = self.work_source.pop_front()?; match self.morselizer.plan_file(part_file) { Ok(planner) => { self.metrics.files_opened.add(1); From 7b0416eb04ba1bfb7abe8d1cb4f48af89fa83d27 Mon Sep 17 00:00:00 2001 From: Denys Tsomenko Date: Thu, 10 Sep 2026 17:16:19 +0300 Subject: [PATCH 3/3] docs: prettier formatting for the file_stream_open_ahead row Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_012m5Yx6yEpZsacqAhZotgkT --- docs/source/user-guide/configs.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 57d01fa5d4356..1b793549daf02 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -141,7 +141,7 @@ The following configuration settings are available: | datafusion.execution.split_file_groups_by_statistics | false | Attempt to eliminate sorts by packing & sorting files with non-overlapping statistics into the same file groups. Currently experimental | | datafusion.execution.keep_partition_by_columns | false | Should DataFusion keep the columns used for partition_by in the output RecordBatches | | datafusion.execution.enable_file_stream_work_stealing | true | When `true` (the default), DataFusion's built-in file scans dynamically rebalance files across partitions at query execution time: a partition that goes idle reads files (or byte-range morsels) originally assigned to a sibling partition, which keeps all partitions busy in a single process. Executors that depend on the plan-time partition assignment — such as Ballista and datafusion-distributed, which run each partition as an isolated task and never poll the siblings — should set this to `false` so each partition reads only its own file group and no runtime reassignment occurs. | -| datafusion.execution.file_stream_open_ahead | false | When `true`, each file-scan partition starts opening its next file (footer, page index and bloom filter I/O) while the current file is still being scanned, keeping at most one file in flight ahead of the active reader. This overlaps per-file open latency with data reads, which matters for scans over many files on high-latency object stores. The cost is holding one extra file's metadata per partition. When `false` (the default) a partition opens the next file only after the current one is fully consumed. | +| datafusion.execution.file_stream_open_ahead | false | When `true`, each file-scan partition starts opening its next file (footer, page index and bloom filter I/O) while the current file is still being scanned, keeping at most one file in flight ahead of the active reader. This overlaps per-file open latency with data reads, which matters for scans over many files on high-latency object stores. The cost is holding one extra file's metadata per partition. When `false` (the default) a partition opens the next file only after the current one is fully consumed. | | datafusion.execution.skip_partial_aggregation_probe_ratio_threshold | 0.8 | Aggregation ratio (number of distinct groups / number of input rows) threshold for skipping partial aggregation. If the value is greater then partial aggregation will skip aggregation for further input | | datafusion.execution.skip_partial_aggregation_probe_rows_threshold | 100000 | Number of input rows partial aggregation partition should process, before aggregation ratio check and trying to switch to skipping aggregation mode | | datafusion.execution.use_row_number_estimates_to_optimize_partitioning | false | Should DataFusion use row number estimates at the input to decide whether increasing parallelism is beneficial or not. By default, only exact row numbers (not estimates) are used for this decision. Setting this flag to `true` will likely produce better plans. if the source of statistics is accurate. We plan to make this the default in the future. |