From ca08b4592825896fe323cd19edda86b081444a9b Mon Sep 17 00:00:00 2001 From: Tobias Schwarzinger Date: Thu, 3 Sep 2026 13:01:46 +0200 Subject: [PATCH 1/5] Some improvements --- datafusion/common/src/config.rs | 11 ++++++ datafusion/datasource/src/source.rs | 52 ++++++++++++++++++++++++++++- 2 files changed, 62 insertions(+), 1 deletion(-) diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 4258ce6f003a9..15173245bbaf7 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1162,6 +1162,17 @@ config_namespace! { /// /// Disabled by default, set to a number greater than 0 for enabling it. pub hash_join_buffering_capacity: usize, default = 0 + + /// Sets the threshold for a scan to be considered small. + /// + /// If the statistics suggest that scanning a partition requires reading fewer bytes than + /// this threshold, DataFusion may eagerly evaluate the scan to cut down latency. This + /// approach is not applied to large scans (> `small_scan_partition_threshold`) as eagerly + /// evaluated scans do not have access to the final dynamic filters, which may significantly + /// reduce the number of rows scanned. + /// + /// Set this threshold to 0 to disable eager evaluation of scans. + pub data_source_small_scan_partition_threshold: usize, default = 1024 * 1024 } } diff --git a/datafusion/datasource/src/source.rs b/datafusion/datasource/src/source.rs index 741010c595197..95e92842930cf 100644 --- a/datafusion/datasource/src/source.rs +++ b/datafusion/datasource/src/source.rs @@ -31,7 +31,7 @@ use datafusion_physical_plan::metrics::{ BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet, }; use datafusion_physical_plan::projection::ProjectionExec; -use datafusion_physical_plan::stream::BatchSplitStream; +use datafusion_physical_plan::stream::{BatchSplitStream, RecordBatchStreamAdapter}; use datafusion_physical_plan::{ ChildrenPropertiesMode, DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, ReplaceChildrenOptions, @@ -43,11 +43,13 @@ use crate::file_scan_config::FileScanConfig; use datafusion_common::config::ConfigOptions; use datafusion_common::tree_node::TreeNodeRecursion; use datafusion_common::{Constraints, Result, Statistics}; +use datafusion_execution::memory_pool::MemoryConsumer; use datafusion_execution::{SendableRecordBatchStream, TaskContext}; use datafusion_physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr}; use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr}; use datafusion_physical_plan::SortOrderPushdownResult; use datafusion_physical_plan::StatisticsArgs; +use datafusion_physical_plan::buffer::MemoryBufferedStream; use datafusion_physical_plan::filter_pushdown::{ ChildPushdownResult, FilterPushdownPhase, FilterPushdownPropagation, PushedDown, }; @@ -468,6 +470,8 @@ impl ExecutionPlan for DataSourceExec { let args = OpenArgs::new(partition, Arc::clone(&context)) .with_shared_state(shared_state); let stream = self.data_source.open_with_args(args)?; + let stream = self.try_buffer_stream(partition, &context, stream)?; + let batch_size = context.session_config().batch_size(); log::debug!( @@ -694,6 +698,52 @@ impl DataSourceExec { .map(|source| (file_scan_conf, source)) }) } + + /// Tries to evaluate the stream in a separate task, depending on whether the statistics show + /// that this is a small scan. + /// + /// For some scenarios, this can significantly improve the latency, as object store requests + /// are triggered earlier. For example, while a join is building its build side, the eager poll + /// can already trigger network requests, cutting down on the latency of the overall join. + /// + /// Note that this can pessimize queries if the dynamic filters have not yet been computed by + /// other operators and can therefore not be used for pruning. To alleviate this problem, + /// the eager fetches are only executed if the scan is "small" and thus the savings from dynamic + /// filter pushdown are negligible. Nevertheless, inaccurate or wrong statistics could pessimize + /// queries. + fn try_buffer_stream( + &self, + partition: usize, + context: &Arc, + stream: SendableRecordBatchStream, + ) -> Result { + let small_scan_threshold = context + .session_config() + .options() + .execution + .data_source_small_scan_partition_threshold; + if small_scan_threshold == 0 { + return Ok(stream); + } + + let partition_stats = self.data_source.partition_statistics(Some(partition))?; + let Some(actual_size) = partition_stats.total_byte_size.get_value() else { + return Ok(stream); + }; + + let stream = if *actual_size <= small_scan_threshold { + let mem_reservation = + MemoryConsumer::new(format!("DataSourceExecEagerExecution[{partition}]")) + .register(context.memory_pool()); + Box::pin(RecordBatchStreamAdapter::new( + stream.schema(), + MemoryBufferedStream::new(stream, *actual_size, mem_reservation), + )) + } else { + stream + }; + Ok(stream) + } } /// Create a new `DataSourceExec` from a `DataSource` From c359de665203e63347adadfd485aa6ad172875f1 Mon Sep 17 00:00:00 2001 From: Tobias Schwarzinger Date: Tue, 8 Sep 2026 17:39:02 +0200 Subject: [PATCH 2/5] rewrite based on BufferExec --- datafusion/common/src/config.rs | 16 +- .../core/src/optimizer_rule_reference.md | 10 +- datafusion/datasource/src/source.rs | 51 +--- .../physical-optimizer/src/buffering.rs | 265 ++++++++++++++++++ .../src/hash_join_buffering.rs | 98 +------ datafusion/physical-optimizer/src/lib.rs | 1 + .../physical-optimizer/src/optimizer.rs | 9 +- .../sqllogictest/test_files/buffering.slt | 136 +++++++++ .../sqllogictest/test_files/explain.slt | 8 +- .../test_files/information_schema.slt | 2 + 10 files changed, 432 insertions(+), 164 deletions(-) create mode 100644 datafusion/physical-optimizer/src/buffering.rs create mode 100644 datafusion/sqllogictest/test_files/buffering.slt diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index d4d67ca404a0f..4b91cfc09e51f 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1182,16 +1182,16 @@ config_namespace! { /// Disabled by default, set to a number greater than 0 for enabling it. pub hash_join_buffering_capacity: usize, default = 0 - /// Sets the threshold for a scan to be considered small. + /// Sets the threshold for scans that should be buffered. /// - /// If the statistics suggest that scanning a partition requires reading fewer bytes than - /// this threshold, DataFusion may eagerly evaluate the scan to cut down latency. This - /// approach is not applied to large scans (> `small_scan_partition_threshold`) as eagerly - /// evaluated scans do not have access to the final dynamic filters, which may significantly - /// reduce the number of rows scanned. + /// If the statistics suggest that a scan requires reading fewer bytes than this threshold, + /// DataFusion may eagerly evaluate the scan to cut down latency. This approach is not + /// applied to large scans (> `small_scan_buffering_threshold`) as eagerly evaluated scans + /// do not have access to the final dynamic filters, which may significantly reduce the + /// number of bytes scanned. /// - /// Set this threshold to 0 to disable eager evaluation of scans. - pub data_source_small_scan_partition_threshold: usize, default = 1024 * 1024 + /// Disabled by default, set to a number greater than 0 for enabling it. + pub small_scan_buffering_threshold: usize, default = 0 } } diff --git a/datafusion/core/src/optimizer_rule_reference.md b/datafusion/core/src/optimizer_rule_reference.md index 1367ed0843c59..bf93d3d3b2e2a 100644 --- a/datafusion/core/src/optimizer_rule_reference.md +++ b/datafusion/core/src/optimizer_rule_reference.md @@ -83,11 +83,11 @@ in multiple phases. | 11 | `OutputRequirements` | remove phase | Removes the temporary output-requirement helper nodes after requirement-sensitive planning is done. | | 12 | `LimitAggregation` | - | Passes a limit hint into eligible aggregations so they can keep fewer accumulator buckets. | | 13 | `LimitPushPastWindows` | - | Pushes fetch limits through bounded window operators when doing so keeps the result correct. | -| 14 | `HashJoinBuffering` | - | Adds buffering on the probe side of hash joins so probing can start before build completion. | -| 15 | `LimitPushdown` | - | Moves physical limits into child operators or fetch-enabled variants to cut data early. | -| 16 | `TopKRepartition` | - | Pushes TopK below hash repartition when the partition key is a prefix of the sort key. | -| 17 | `ProjectionPushdown` | late pass | Runs projection pushdown again after limit and TopK rewrites expose new pruning opportunities. | -| 18 | `PushdownSort` | - | Pushes sort requirements into data sources that can already return sorted output. | +| 14 | `LimitPushdown` | - | Moves physical limits into child operators or fetch-enabled variants to cut data early. | +| 15 | `TopKRepartition` | - | Pushes TopK below hash repartition when the partition key is a prefix of the sort key. | +| 16 | `ProjectionPushdown` | late pass | Runs projection pushdown again after limit and TopK rewrites expose new pruning opportunities. | +| 17 | `PushdownSort` | - | Pushes sort requirements into data sources that can already return sorted output. | +| 18 | `BufferInsertion` | - | Adds buffering on the probe side of hash joins and for small scans to cut down latency. | | 19 | `EnsureCooperative` | - | Wraps non-cooperative plan parts so long-running tasks yield fairly. | | 20 | `FilterPushdown(Post)` | post-optimization phase | Pushes dynamic filters at the end of optimization, after plan references stop moving. | | 21 | `SanityCheckPlan` | - | Validates that the final physical plan meets ordering, distribution, and infinite-input safety requirements. | diff --git a/datafusion/datasource/src/source.rs b/datafusion/datasource/src/source.rs index 95e92842930cf..085d64148eff3 100644 --- a/datafusion/datasource/src/source.rs +++ b/datafusion/datasource/src/source.rs @@ -31,7 +31,7 @@ use datafusion_physical_plan::metrics::{ BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet, }; use datafusion_physical_plan::projection::ProjectionExec; -use datafusion_physical_plan::stream::{BatchSplitStream, RecordBatchStreamAdapter}; +use datafusion_physical_plan::stream::BatchSplitStream; use datafusion_physical_plan::{ ChildrenPropertiesMode, DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, ReplaceChildrenOptions, @@ -43,13 +43,11 @@ use crate::file_scan_config::FileScanConfig; use datafusion_common::config::ConfigOptions; use datafusion_common::tree_node::TreeNodeRecursion; use datafusion_common::{Constraints, Result, Statistics}; -use datafusion_execution::memory_pool::MemoryConsumer; use datafusion_execution::{SendableRecordBatchStream, TaskContext}; use datafusion_physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr}; use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr}; use datafusion_physical_plan::SortOrderPushdownResult; use datafusion_physical_plan::StatisticsArgs; -use datafusion_physical_plan::buffer::MemoryBufferedStream; use datafusion_physical_plan::filter_pushdown::{ ChildPushdownResult, FilterPushdownPhase, FilterPushdownPropagation, PushedDown, }; @@ -470,7 +468,6 @@ impl ExecutionPlan for DataSourceExec { let args = OpenArgs::new(partition, Arc::clone(&context)) .with_shared_state(shared_state); let stream = self.data_source.open_with_args(args)?; - let stream = self.try_buffer_stream(partition, &context, stream)?; let batch_size = context.session_config().batch_size(); @@ -698,52 +695,6 @@ impl DataSourceExec { .map(|source| (file_scan_conf, source)) }) } - - /// Tries to evaluate the stream in a separate task, depending on whether the statistics show - /// that this is a small scan. - /// - /// For some scenarios, this can significantly improve the latency, as object store requests - /// are triggered earlier. For example, while a join is building its build side, the eager poll - /// can already trigger network requests, cutting down on the latency of the overall join. - /// - /// Note that this can pessimize queries if the dynamic filters have not yet been computed by - /// other operators and can therefore not be used for pruning. To alleviate this problem, - /// the eager fetches are only executed if the scan is "small" and thus the savings from dynamic - /// filter pushdown are negligible. Nevertheless, inaccurate or wrong statistics could pessimize - /// queries. - fn try_buffer_stream( - &self, - partition: usize, - context: &Arc, - stream: SendableRecordBatchStream, - ) -> Result { - let small_scan_threshold = context - .session_config() - .options() - .execution - .data_source_small_scan_partition_threshold; - if small_scan_threshold == 0 { - return Ok(stream); - } - - let partition_stats = self.data_source.partition_statistics(Some(partition))?; - let Some(actual_size) = partition_stats.total_byte_size.get_value() else { - return Ok(stream); - }; - - let stream = if *actual_size <= small_scan_threshold { - let mem_reservation = - MemoryConsumer::new(format!("DataSourceExecEagerExecution[{partition}]")) - .register(context.memory_pool()); - Box::pin(RecordBatchStreamAdapter::new( - stream.schema(), - MemoryBufferedStream::new(stream, *actual_size, mem_reservation), - )) - } else { - stream - }; - Ok(stream) - } } /// Create a new `DataSourceExec` from a `DataSource` diff --git a/datafusion/physical-optimizer/src/buffering.rs b/datafusion/physical-optimizer/src/buffering.rs new file mode 100644 index 0000000000000..e3ceb391f149b --- /dev/null +++ b/datafusion/physical-optimizer/src/buffering.rs @@ -0,0 +1,265 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::PhysicalOptimizerRule; +use datafusion_common::config::ConfigOptions; +use datafusion_common::tree_node::{Transformed, TreeNode}; +use datafusion_common::{JoinSide, Result, Statistics}; +use datafusion_physical_plan::ExecutionPlan; +use datafusion_physical_plan::buffer::BufferExec; +use datafusion_physical_plan::empty::EmptyExec; +use datafusion_physical_plan::execution_plan::replace_children_if_necessary; +use datafusion_physical_plan::joins::HashJoinExec; +use datafusion_physical_plan::placeholder_row::PlaceholderRowExec; +use datafusion_physical_plan::statistics::{StatisticsArgs, StatisticsContext}; +use std::sync::Arc; + +/// The [`BufferInsertion`] optimizer rule places [`BufferExec`] nodes in the physical plan to +/// compute multiple parts of a single partition in parallel to cut down on latency. This is done +/// in two scenarios: +/// +/// 1. **Hash Join Probe Buffering**: For [`HashJoinExec`], buffers the probe side with +/// capacity `config.execution.hash_join_buffering_capacity` so that the probe side can +/// be eagerly polled while the build side is concurrently being built. +/// 2. **Small Scan Buffering**: For data source scans whose statistics indicate that they +/// are smaller than `config.execution.small_scan_partition_threshold`, we wrap them in +/// a [`BufferExec`]. +/// +/// If the probe side of a hash join is buffered by (1), scans within that probe +/// side are not double-buffered. +/// +/// ## `HashJoinExec` Buffering +/// +/// Looks for all the [HashJoinExec]s in the plan and places a [BufferExec] node with the +/// configured capacity in the probe side: +/// +/// ```text +/// ┌───────────────────┐ +/// │ HashJoinExec │ +/// └─────▲────────▲────┘ +/// ┌───────┘ └─────────┐ +/// │ │ +/// ┌────────────────┐ ┌─────────────────┐ +/// │ Build side │ + │ BufferExec │ +/// └────────────────┘ └────────▲────────┘ +/// │ +/// ┌────────┴────────┐ +/// │ Probe side │ +/// └─────────────────┘ +/// ``` +/// +/// Which allows eagerly pulling it even before the build side has completely finished. +/// +/// ## Small Scan Buffering +/// +/// The buffering of small scans works independently of the hash join buffering. Whenever a small +/// scan is detected, a [`BufferExec`] is inserted with the goal of reducing query latency as the +/// I/O of the small scan is triggered eagerly. As the scan is considered small, dynamic filters may +/// not yield significant improvements that warrant delaying the I/O until they are fully available. +/// +/// ```text +/// ┌───────────────────┐ +/// │ MyOtherExec │ +/// └─────▲────────▲────┘ +/// ┌───────┘ └─────────┐ +/// │ │ +/// ┌────────────────┐ ┌─────────────────┐ +/// │ Scan (200 MiB) │ + │ BufferExec │ +/// └────────────────┘ └────────▲────────┘ +/// │ +/// ┌────────┴────────┐ +/// │ Scan (1 MiB) │ +/// └─────────────────┘ +/// ``` +#[derive(Debug, Default)] +pub struct BufferInsertion {} + +impl BufferInsertion { + pub fn new() -> Self { + Self::default() + } +} + +impl PhysicalOptimizerRule for BufferInsertion { + fn optimize( + &self, + plan: Arc, + config: &ConfigOptions, + ) -> Result> { + let hash_join_capacity = config.execution.hash_join_buffering_capacity; + let small_scan_threshold = config.execution.small_scan_buffering_threshold; + + if hash_join_capacity == 0 && small_scan_threshold == 0 { + return Ok(plan); + } + + let stats_ctx = StatisticsContext::new(); + + transform_plan( + plan, + hash_join_capacity, + small_scan_threshold, + &stats_ctx, + false, + ) + .map(|t| t.data) + } + + fn name(&self) -> &str { + "BufferInsertion" + } + + fn schema_check(&self) -> bool { + true + } +} + +fn transform_plan( + plan: Arc, + hash_join_capacity: usize, + small_scan_threshold: usize, + stats_ctx: &StatisticsContext, + in_buffer: bool, +) -> Result>> { + if plan.is::() { + // Prevent stacking BufferExec nodes together and avoid double-buffering scans within it. + return plan.map_children(|child| { + transform_plan( + child, + hash_join_capacity, + small_scan_threshold, + stats_ctx, + true, + ) + }); + } + + if let Some(join) = plan.downcast_ref::() { + let probe_is_left = HashJoinExec::probe_side() == JoinSide::Left; + let probe_child = if probe_is_left { + &join.left + } else { + &join.right + }; + let build_child = if probe_is_left { + &join.right + } else { + &join.left + }; + + let transformed_build = transform_plan( + Arc::clone(build_child), + hash_join_capacity, + small_scan_threshold, + stats_ctx, + in_buffer, + )?; + + let (transformed_probe, probe_transformed) = + if hash_join_capacity > 0 && !probe_child.is::() { + // Buffer the probe side. Since the probe child is wrapped in BufferExec, + // scans within that probe side must NOT be double-buffered (`in_buffer: true`). + let probe_inner = transform_plan( + Arc::clone(probe_child), + hash_join_capacity, + small_scan_threshold, + stats_ctx, + true, + )?; + let buffered: Arc = + Arc::new(BufferExec::new(probe_inner.data, hash_join_capacity)); + (buffered, true) + } else { + let probe_res = transform_plan( + Arc::clone(probe_child), + hash_join_capacity, + small_scan_threshold, + stats_ctx, + in_buffer, + )?; + (probe_res.data, probe_res.transformed) + }; + + if transformed_build.transformed || probe_transformed { + let (new_left, new_right) = if probe_is_left { + (transformed_probe, transformed_build.data) + } else { + (transformed_build.data, transformed_probe) + }; + let new_plan = + replace_children_if_necessary(plan, vec![new_left, new_right])?; + return Ok(Transformed::yes(new_plan)); + } else { + return Ok(Transformed::no(plan)); + } + } + + if !in_buffer + && is_scan(&plan) + && is_small_scan(&plan, small_scan_threshold, stats_ctx)? + { + let buffered: Arc = + Arc::new(BufferExec::new(plan, small_scan_threshold)); + return Ok(Transformed::yes(buffered)); + } + + plan.map_children(|child| { + transform_plan( + child, + hash_join_capacity, + small_scan_threshold, + stats_ctx, + in_buffer, + ) + }) +} + +fn is_scan(plan: &Arc) -> bool { + plan.children().is_empty() + && !plan.is::() + && !plan.is::() + && !plan.is::() +} + +fn is_small_scan( + plan: &Arc, + threshold: usize, + stats_ctx: &StatisticsContext, +) -> Result { + if threshold == 0 { + return Ok(false); + } + + let overall_stats = stats_ctx.compute(plan.as_ref(), &StatisticsArgs::default())?; + Ok(get_total_byte_size(&overall_stats) + .is_some_and(|total_size| total_size > 0 && total_size <= threshold)) +} + +fn get_total_byte_size(stats: &Statistics) -> Option { + if let Some(&size) = stats.total_byte_size.get_value() { + return Some(size); + } + if !stats.column_statistics.is_empty() { + let mut sum = 0usize; + for col in &stats.column_statistics { + let bytes = col.byte_size.get_value()?; + sum = sum.saturating_add(*bytes); + } + return Some(sum); + } + None +} diff --git a/datafusion/physical-optimizer/src/hash_join_buffering.rs b/datafusion/physical-optimizer/src/hash_join_buffering.rs index dbdfd34a9a01e..60d997c388b8c 100644 --- a/datafusion/physical-optimizer/src/hash_join_buffering.rs +++ b/datafusion/physical-optimizer/src/hash_join_buffering.rs @@ -15,96 +15,10 @@ // specific language governing permissions and limitations // under the License. -use crate::PhysicalOptimizerRule; -use datafusion_common::JoinSide; -use datafusion_common::config::ConfigOptions; -use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode}; -use datafusion_physical_plan::ExecutionPlan; -use datafusion_physical_plan::buffer::BufferExec; -use datafusion_physical_plan::execution_plan::replace_children_if_necessary; -use datafusion_physical_plan::joins::HashJoinExec; -use std::sync::Arc; +use crate::buffering::BufferInsertion; -/// Looks for all the [HashJoinExec]s in the plan and places a [BufferExec] node with the -/// configured capacity in the probe side: -/// -/// ```text -/// ┌───────────────────┐ -/// │ HashJoinExec │ -/// └─────▲────────▲────┘ -/// ┌───────┘ └─────────┐ -/// │ │ -/// ┌────────────────┐ ┌─────────────────┐ -/// │ Build side │ + │ BufferExec │ -/// └────────────────┘ └────────▲────────┘ -/// │ -/// ┌────────┴────────┐ -/// │ Probe side │ -/// └─────────────────┘ -/// ``` -/// -/// Which allows eagerly pulling it even before the build side has completely finished. -#[derive(Debug, Default)] -pub struct HashJoinBuffering {} - -impl HashJoinBuffering { - pub fn new() -> Self { - Self::default() - } -} - -impl PhysicalOptimizerRule for HashJoinBuffering { - fn optimize( - &self, - plan: Arc, - config: &ConfigOptions, - ) -> datafusion_common::Result> { - let capacity = config.execution.hash_join_buffering_capacity; - if capacity == 0 { - return Ok(plan); - } - - plan.transform_down(|plan| { - let Some(node) = plan.downcast_ref::() else { - return Ok(Transformed::no(plan)); - }; - let plan = Arc::clone(&plan); - Ok(Transformed::yes( - if HashJoinExec::probe_side() == JoinSide::Left { - // Do not stack BufferExec nodes together. - if node.left.is::() { - return Ok(Transformed::no(plan)); - } - replace_children_if_necessary( - plan, - vec![ - Arc::new(BufferExec::new(Arc::clone(&node.left), capacity)), - Arc::clone(&node.right), - ], - )? - } else { - // Do not stack BufferExec nodes together. - if node.right.is::() { - return Ok(Transformed::no(plan)); - } - replace_children_if_necessary( - plan, - vec![ - Arc::clone(&node.left), - Arc::new(BufferExec::new(Arc::clone(&node.right), capacity)), - ], - )? - }, - )) - }) - .data() - } - - fn name(&self) -> &str { - "HashJoinBuffering" - } - - fn schema_check(&self) -> bool { - true - } -} +#[deprecated( + since = "56.0.0", + note = "Use the more general BufferingRule instead. This is only an alias." +)] +pub type HashJoinBuffering = BufferInsertion; diff --git a/datafusion/physical-optimizer/src/lib.rs b/datafusion/physical-optimizer/src/lib.rs index b9eb248f6e843..759eac399b1c4 100644 --- a/datafusion/physical-optimizer/src/lib.rs +++ b/datafusion/physical-optimizer/src/lib.rs @@ -42,6 +42,7 @@ pub mod optimizer; pub mod output_requirements; pub mod projection_pushdown; pub use datafusion_pruning as pruning; +pub mod buffering; pub mod hash_join_buffering; pub mod pushdown_sort; pub mod sanity_checker; diff --git a/datafusion/physical-optimizer/src/optimizer.rs b/datafusion/physical-optimizer/src/optimizer.rs index aed25546cd09b..9fcebf0209b11 100644 --- a/datafusion/physical-optimizer/src/optimizer.rs +++ b/datafusion/physical-optimizer/src/optimizer.rs @@ -35,7 +35,7 @@ use crate::topk_aggregation::TopKAggregation; use crate::topk_repartition::TopKRepartition; use crate::update_aggr_exprs::OptimizeAggregateOrder; -use crate::hash_join_buffering::HashJoinBuffering; +use crate::buffering::BufferInsertion; use crate::limit_pushdown_past_window::LimitPushPastWindows; use crate::pushdown_sort::PushdownSort; use crate::window_topn::WindowTopN; @@ -152,10 +152,6 @@ impl PhysicalOptimizer { // This can possibly be combined with [LimitPushdown] // It needs to come after [EnsureRequirements] (which handles sort enforcement) Arc::new(LimitPushPastWindows::new()), - // The HashJoinBuffering rule adds a BufferExec node with the configured capacity - // in the prob side of hash joins. That way, the probe side gets eagerly polled before - // the build side is completely finished. - Arc::new(HashJoinBuffering::new()), // The LimitPushdown rule tries to push limits down as far as possible, // replacing operators with fetching variants, or adding limits // past operators that support limit pushdown. @@ -174,6 +170,9 @@ impl PhysicalOptimizer { Arc::new(ProjectionPushdown::new()), // PushdownSort: Detect sorts that can be pushed down to data sources. Arc::new(PushdownSort::new()), + // The BufferingRule adds BufferExec nodes to eagerly buffer the probe side of hash joins + // and/or small data source scans. + Arc::new(BufferInsertion::new()), Arc::new(EnsureCooperative::new()), // This FilterPushdown handles dynamic filters that may have references to the source ExecutionPlan. // Therefore, it should be run at the end of the optimization process since any changes to the plan may break the dynamic filter's references. diff --git a/datafusion/sqllogictest/test_files/buffering.slt b/datafusion/sqllogictest/test_files/buffering.slt new file mode 100644 index 0000000000000..d69494d5d5ea7 --- /dev/null +++ b/datafusion/sqllogictest/test_files/buffering.slt @@ -0,0 +1,136 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +statement ok +CREATE EXTERNAL TABLE t1 STORED AS PARQUET LOCATION '../../parquet-testing/data/alltypes_plain.parquet'; + +statement ok +CREATE EXTERNAL TABLE t2 STORED AS PARQUET LOCATION '../../parquet-testing/data/alltypes_plain.parquet'; + +statement ok +CREATE EXTERNAL TABLE t3 STORED AS PARQUET LOCATION '../../parquet-testing/data/alltypes_plain.parquet'; + +statement ok +set datafusion.explain.physical_plan_only = true; + +# No BufferExec should be present (disabled by default). +query TT +EXPLAIN SELECT * FROM t1; +---- +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col], file_type=parquet + +# Small scan should be wrapped with BufferExec. +statement ok +set datafusion.execution.small_scan_buffering_threshold = 1048576; + + +query TT +EXPLAIN SELECT * FROM t1; +---- +physical_plan +01)BufferExec: capacity=1048576 +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col], file_type=parquet + +# Verify execution produces correct results through BufferExec +query I +SELECT count(*) FROM t1; +---- +8 + +# Scan size smaller than threshold should not be buffered +statement ok +set datafusion.execution.small_scan_buffering_threshold = 10; + + +query TT +EXPLAIN SELECT * FROM t1; +---- +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col], file_type=parquet + +# Hash Join probe buffering with hash_join_buffering_capacity > 0 (small scans disabled) +statement ok +set datafusion.execution.small_scan_buffering_threshold = 0; + +statement ok +set datafusion.execution.hash_join_buffering_capacity = 65536; + +query TT +EXPLAIN SELECT t1.id, t2.id FROM t1 JOIN t2 ON t1.id = t2.id; +---- +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id], file_type=parquet +03)--BufferExec: capacity=65536 +04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible + +# Hash join probe buffering and small scan buffering. The probe side is buffered by hash join buffering. Its inner small +# scan must NOT be double-buffered. The build side (if small) gets small scan buffering. +statement ok +set datafusion.execution.hash_join_buffering_capacity = 65536; + +statement ok +set datafusion.execution.small_scan_buffering_threshold = 1048576; + +query TT +EXPLAIN SELECT t1.id, t2.id FROM t1 JOIN t2 ON t1.id = t2.id; +---- +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)] +02)--BufferExec: capacity=1048576 +03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id], file_type=parquet +04)--BufferExec: capacity=65536 +05)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible + +# nested joins: Ensure optimizer buffers multiple nested hash joins +statement ok +set datafusion.execution.hash_join_buffering_capacity = 65536; + +statement ok +set datafusion.execution.small_scan_buffering_threshold = 1048576; + +query TT +EXPLAIN SELECT t1.id, t2.id, t3.id FROM t1 JOIN t2 ON t1.id = t2.id JOIN t3 ON t1.id = t3.id; +---- +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)] +02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)] +03)----BufferExec: capacity=1048576 +04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id], file_type=parquet +05)----BufferExec: capacity=65536 +06)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible +07)--BufferExec: capacity=65536 +08)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible + +# Clean up + +statement ok +reset datafusion.explain.physical_plan_only; + +statement ok +reset datafusion.execution.hash_join_buffering_capacity; + +statement ok +reset datafusion.execution.small_scan_buffering_threshold; + +statement ok +DROP TABLE t1; + +statement ok +DROP TABLE t2; + +statement ok +DROP TABLE t3; diff --git a/datafusion/sqllogictest/test_files/explain.slt b/datafusion/sqllogictest/test_files/explain.slt index b6837002086ad..a9c553daa21f6 100644 --- a/datafusion/sqllogictest/test_files/explain.slt +++ b/datafusion/sqllogictest/test_files/explain.slt @@ -245,11 +245,11 @@ physical_plan after ProjectionPushdown SAME TEXT AS ABOVE physical_plan after OutputRequirements DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/example.csv]]}, projection=[a, b, c], file_type=csv, has_header=true physical_plan after LimitAggregation SAME TEXT AS ABOVE physical_plan after LimitPushPastWindows SAME TEXT AS ABOVE -physical_plan after HashJoinBuffering SAME TEXT AS ABOVE physical_plan after LimitPushdown SAME TEXT AS ABOVE physical_plan after TopKRepartition SAME TEXT AS ABOVE physical_plan after ProjectionPushdown SAME TEXT AS ABOVE physical_plan after PushdownSort SAME TEXT AS ABOVE +physical_plan after BufferInsertion SAME TEXT AS ABOVE physical_plan after EnsureCooperative SAME TEXT AS ABOVE physical_plan after FilterPushdown(Post) SAME TEXT AS ABOVE physical_plan after SanityCheckPlan SAME TEXT AS ABOVE @@ -327,11 +327,11 @@ physical_plan after OutputRequirements 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col], limit=10, file_type=parquet, statistics=[Rows=Exact(8), Bytes=Absent, [(Col[0]: ScanBytes=Exact(32)),(Col[1]: ScanBytes=Inexact(24)),(Col[2]: ScanBytes=Exact(32)),(Col[3]: ScanBytes=Exact(32)),(Col[4]: ScanBytes=Exact(32)),(Col[5]: ScanBytes=Exact(64)),(Col[6]: ScanBytes=Exact(32)),(Col[7]: ScanBytes=Exact(64)),(Col[8]: ScanBytes=Inexact(88)),(Col[9]: ScanBytes=Inexact(49)),(Col[10]: ScanBytes=Exact(64))]] physical_plan after LimitAggregation SAME TEXT AS ABOVE physical_plan after LimitPushPastWindows SAME TEXT AS ABOVE -physical_plan after HashJoinBuffering SAME TEXT AS ABOVE physical_plan after LimitPushdown DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col], limit=10, file_type=parquet, statistics=[Rows=Exact(8), Bytes=Absent, [(Col[0]: ScanBytes=Exact(32)),(Col[1]: ScanBytes=Inexact(24)),(Col[2]: ScanBytes=Exact(32)),(Col[3]: ScanBytes=Exact(32)),(Col[4]: ScanBytes=Exact(32)),(Col[5]: ScanBytes=Exact(64)),(Col[6]: ScanBytes=Exact(32)),(Col[7]: ScanBytes=Exact(64)),(Col[8]: ScanBytes=Inexact(88)),(Col[9]: ScanBytes=Inexact(49)),(Col[10]: ScanBytes=Exact(64))]] physical_plan after TopKRepartition SAME TEXT AS ABOVE physical_plan after ProjectionPushdown SAME TEXT AS ABOVE physical_plan after PushdownSort SAME TEXT AS ABOVE +physical_plan after BufferInsertion SAME TEXT AS ABOVE physical_plan after EnsureCooperative SAME TEXT AS ABOVE physical_plan after FilterPushdown(Post) SAME TEXT AS ABOVE physical_plan after SanityCheckPlan SAME TEXT AS ABOVE @@ -373,11 +373,11 @@ physical_plan after OutputRequirements 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col], limit=10, file_type=parquet physical_plan after LimitAggregation SAME TEXT AS ABOVE physical_plan after LimitPushPastWindows SAME TEXT AS ABOVE -physical_plan after HashJoinBuffering SAME TEXT AS ABOVE physical_plan after LimitPushdown DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col], limit=10, file_type=parquet physical_plan after TopKRepartition SAME TEXT AS ABOVE physical_plan after ProjectionPushdown SAME TEXT AS ABOVE physical_plan after PushdownSort SAME TEXT AS ABOVE +physical_plan after BufferInsertion SAME TEXT AS ABOVE physical_plan after EnsureCooperative SAME TEXT AS ABOVE physical_plan after FilterPushdown(Post) SAME TEXT AS ABOVE physical_plan after SanityCheckPlan SAME TEXT AS ABOVE @@ -624,11 +624,11 @@ physical_plan after ProjectionPushdown SAME TEXT AS ABOVE physical_plan after OutputRequirements DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/example.csv]]}, projection=[a, b, c], file_type=csv, has_header=true physical_plan after LimitAggregation SAME TEXT AS ABOVE physical_plan after LimitPushPastWindows SAME TEXT AS ABOVE -physical_plan after HashJoinBuffering SAME TEXT AS ABOVE physical_plan after LimitPushdown SAME TEXT AS ABOVE physical_plan after TopKRepartition SAME TEXT AS ABOVE physical_plan after ProjectionPushdown SAME TEXT AS ABOVE physical_plan after PushdownSort SAME TEXT AS ABOVE +physical_plan after BufferInsertion SAME TEXT AS ABOVE physical_plan after EnsureCooperative SAME TEXT AS ABOVE physical_plan after FilterPushdown(Post) SAME TEXT AS ABOVE physical_plan after SanityCheckPlan SAME TEXT AS ABOVE diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index f2a587ed72ae0..5f1dd03353af0 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -277,6 +277,7 @@ datafusion.execution.planning_concurrency 13 datafusion.execution.skip_partial_aggregation_probe_ratio_threshold 0.8 datafusion.execution.skip_partial_aggregation_probe_rows_threshold 100000 datafusion.execution.skip_physical_aggregate_schema_check false +datafusion.execution.small_scan_buffering_threshold 0 datafusion.execution.soft_max_rows_per_output_file 50000000 datafusion.execution.sort_in_place_threshold_bytes 1048576 datafusion.execution.sort_pushdown_buffer_capacity 1073741824 @@ -438,6 +439,7 @@ datafusion.execution.planning_concurrency 13 Fan-out during initial physical pla 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.skip_physical_aggregate_schema_check false When set to true, skips verifying that the schema produced by planning the input of `LogicalPlan::Aggregate` exactly matches the schema of the input plan. When set to false, if the schema does not match exactly (including nullability and metadata), a planning error will be raised. This is used to workaround bugs in the planner that are now caught by the new schema verification step. +datafusion.execution.small_scan_buffering_threshold 0 Sets the threshold for scans that should be buffered. If the statistics suggest that a scan requires reading fewer bytes than this threshold, DataFusion may eagerly evaluate the scan to cut down latency. This approach is not applied to large scans (> `small_scan_buffering_threshold`) as eagerly evaluated scans do not have access to the final dynamic filters, which may significantly reduce the number of bytes scanned. Disabled by default, set to a number greater than 0 for enabling it. datafusion.execution.soft_max_rows_per_output_file 50000000 Target number of rows in output files when writing multiple. This is a soft max, so it can be exceeded slightly. There also will be one file smaller than the limit if the total number of rows written is not roughly divisible by the soft max datafusion.execution.sort_in_place_threshold_bytes 1048576 When sorting, below what size should data be concatenated and sorted in a single RecordBatch rather than sorted in batches and merged. datafusion.execution.sort_pushdown_buffer_capacity 1073741824 Maximum buffer capacity (in bytes) per partition for BufferExec inserted during sort pushdown optimization. When PushdownSort eliminates a SortExec under SortPreservingMergeExec, a BufferExec is inserted to replace SortExec's buffering role. This prevents I/O stalls by allowing the scan to run ahead of the merge. This uses strictly less memory than the SortExec it replaces (which buffers the entire partition). The buffer respects the global memory pool limit. Setting this to a large value is safe — actual memory usage is bounded by partition size and global memory limits. From ee7aa97e68c1827d176884e961f979f2881723e2 Mon Sep 17 00:00:00 2001 From: Tobias Schwarzinger Date: Tue, 8 Sep 2026 18:15:33 +0200 Subject: [PATCH 3/5] move rule back to old position --- datafusion/core/src/optimizer_rule_reference.md | 10 +++++----- datafusion/physical-optimizer/src/optimizer.rs | 6 +++--- datafusion/sqllogictest/test_files/explain.slt | 8 ++++---- 3 files changed, 12 insertions(+), 12 deletions(-) diff --git a/datafusion/core/src/optimizer_rule_reference.md b/datafusion/core/src/optimizer_rule_reference.md index bf93d3d3b2e2a..7767c46bac3bc 100644 --- a/datafusion/core/src/optimizer_rule_reference.md +++ b/datafusion/core/src/optimizer_rule_reference.md @@ -83,11 +83,11 @@ in multiple phases. | 11 | `OutputRequirements` | remove phase | Removes the temporary output-requirement helper nodes after requirement-sensitive planning is done. | | 12 | `LimitAggregation` | - | Passes a limit hint into eligible aggregations so they can keep fewer accumulator buckets. | | 13 | `LimitPushPastWindows` | - | Pushes fetch limits through bounded window operators when doing so keeps the result correct. | -| 14 | `LimitPushdown` | - | Moves physical limits into child operators or fetch-enabled variants to cut data early. | -| 15 | `TopKRepartition` | - | Pushes TopK below hash repartition when the partition key is a prefix of the sort key. | -| 16 | `ProjectionPushdown` | late pass | Runs projection pushdown again after limit and TopK rewrites expose new pruning opportunities. | -| 17 | `PushdownSort` | - | Pushes sort requirements into data sources that can already return sorted output. | -| 18 | `BufferInsertion` | - | Adds buffering on the probe side of hash joins and for small scans to cut down latency. | +| 14 | `BufferInsertion` | - | Adds buffering on the probe side of hash joins and for small scans to cut down latency. | +| 15 | `LimitPushdown` | - | Moves physical limits into child operators or fetch-enabled variants to cut data early. | +| 16 | `TopKRepartition` | - | Pushes TopK below hash repartition when the partition key is a prefix of the sort key. | +| 17 | `ProjectionPushdown` | late pass | Runs projection pushdown again after limit and TopK rewrites expose new pruning opportunities. | +| 18 | `PushdownSort` | - | Pushes sort requirements into data sources that can already return sorted output. | | 19 | `EnsureCooperative` | - | Wraps non-cooperative plan parts so long-running tasks yield fairly. | | 20 | `FilterPushdown(Post)` | post-optimization phase | Pushes dynamic filters at the end of optimization, after plan references stop moving. | | 21 | `SanityCheckPlan` | - | Validates that the final physical plan meets ordering, distribution, and infinite-input safety requirements. | diff --git a/datafusion/physical-optimizer/src/optimizer.rs b/datafusion/physical-optimizer/src/optimizer.rs index 9fcebf0209b11..587174a721de9 100644 --- a/datafusion/physical-optimizer/src/optimizer.rs +++ b/datafusion/physical-optimizer/src/optimizer.rs @@ -152,6 +152,9 @@ impl PhysicalOptimizer { // This can possibly be combined with [LimitPushdown] // It needs to come after [EnsureRequirements] (which handles sort enforcement) Arc::new(LimitPushPastWindows::new()), + // The BufferInsertion rule adds BufferExec nodes to eagerly buffer the probe side of hash joins + // and/or small data source scans. + Arc::new(BufferInsertion::new()), // The LimitPushdown rule tries to push limits down as far as possible, // replacing operators with fetching variants, or adding limits // past operators that support limit pushdown. @@ -170,9 +173,6 @@ impl PhysicalOptimizer { Arc::new(ProjectionPushdown::new()), // PushdownSort: Detect sorts that can be pushed down to data sources. Arc::new(PushdownSort::new()), - // The BufferingRule adds BufferExec nodes to eagerly buffer the probe side of hash joins - // and/or small data source scans. - Arc::new(BufferInsertion::new()), Arc::new(EnsureCooperative::new()), // This FilterPushdown handles dynamic filters that may have references to the source ExecutionPlan. // Therefore, it should be run at the end of the optimization process since any changes to the plan may break the dynamic filter's references. diff --git a/datafusion/sqllogictest/test_files/explain.slt b/datafusion/sqllogictest/test_files/explain.slt index a9c553daa21f6..cc73833311508 100644 --- a/datafusion/sqllogictest/test_files/explain.slt +++ b/datafusion/sqllogictest/test_files/explain.slt @@ -245,11 +245,11 @@ physical_plan after ProjectionPushdown SAME TEXT AS ABOVE physical_plan after OutputRequirements DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/example.csv]]}, projection=[a, b, c], file_type=csv, has_header=true physical_plan after LimitAggregation SAME TEXT AS ABOVE physical_plan after LimitPushPastWindows SAME TEXT AS ABOVE +physical_plan after BufferInsertion SAME TEXT AS ABOVE physical_plan after LimitPushdown SAME TEXT AS ABOVE physical_plan after TopKRepartition SAME TEXT AS ABOVE physical_plan after ProjectionPushdown SAME TEXT AS ABOVE physical_plan after PushdownSort SAME TEXT AS ABOVE -physical_plan after BufferInsertion SAME TEXT AS ABOVE physical_plan after EnsureCooperative SAME TEXT AS ABOVE physical_plan after FilterPushdown(Post) SAME TEXT AS ABOVE physical_plan after SanityCheckPlan SAME TEXT AS ABOVE @@ -327,11 +327,11 @@ physical_plan after OutputRequirements 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col], limit=10, file_type=parquet, statistics=[Rows=Exact(8), Bytes=Absent, [(Col[0]: ScanBytes=Exact(32)),(Col[1]: ScanBytes=Inexact(24)),(Col[2]: ScanBytes=Exact(32)),(Col[3]: ScanBytes=Exact(32)),(Col[4]: ScanBytes=Exact(32)),(Col[5]: ScanBytes=Exact(64)),(Col[6]: ScanBytes=Exact(32)),(Col[7]: ScanBytes=Exact(64)),(Col[8]: ScanBytes=Inexact(88)),(Col[9]: ScanBytes=Inexact(49)),(Col[10]: ScanBytes=Exact(64))]] physical_plan after LimitAggregation SAME TEXT AS ABOVE physical_plan after LimitPushPastWindows SAME TEXT AS ABOVE +physical_plan after BufferInsertion SAME TEXT AS ABOVE physical_plan after LimitPushdown DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col], limit=10, file_type=parquet, statistics=[Rows=Exact(8), Bytes=Absent, [(Col[0]: ScanBytes=Exact(32)),(Col[1]: ScanBytes=Inexact(24)),(Col[2]: ScanBytes=Exact(32)),(Col[3]: ScanBytes=Exact(32)),(Col[4]: ScanBytes=Exact(32)),(Col[5]: ScanBytes=Exact(64)),(Col[6]: ScanBytes=Exact(32)),(Col[7]: ScanBytes=Exact(64)),(Col[8]: ScanBytes=Inexact(88)),(Col[9]: ScanBytes=Inexact(49)),(Col[10]: ScanBytes=Exact(64))]] physical_plan after TopKRepartition SAME TEXT AS ABOVE physical_plan after ProjectionPushdown SAME TEXT AS ABOVE physical_plan after PushdownSort SAME TEXT AS ABOVE -physical_plan after BufferInsertion SAME TEXT AS ABOVE physical_plan after EnsureCooperative SAME TEXT AS ABOVE physical_plan after FilterPushdown(Post) SAME TEXT AS ABOVE physical_plan after SanityCheckPlan SAME TEXT AS ABOVE @@ -373,11 +373,11 @@ physical_plan after OutputRequirements 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col], limit=10, file_type=parquet physical_plan after LimitAggregation SAME TEXT AS ABOVE physical_plan after LimitPushPastWindows SAME TEXT AS ABOVE +physical_plan after BufferInsertion SAME TEXT AS ABOVE physical_plan after LimitPushdown DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/parquet-testing/data/alltypes_plain.parquet]]}, projection=[id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col], limit=10, file_type=parquet physical_plan after TopKRepartition SAME TEXT AS ABOVE physical_plan after ProjectionPushdown SAME TEXT AS ABOVE physical_plan after PushdownSort SAME TEXT AS ABOVE -physical_plan after BufferInsertion SAME TEXT AS ABOVE physical_plan after EnsureCooperative SAME TEXT AS ABOVE physical_plan after FilterPushdown(Post) SAME TEXT AS ABOVE physical_plan after SanityCheckPlan SAME TEXT AS ABOVE @@ -624,11 +624,11 @@ physical_plan after ProjectionPushdown SAME TEXT AS ABOVE physical_plan after OutputRequirements DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/example.csv]]}, projection=[a, b, c], file_type=csv, has_header=true physical_plan after LimitAggregation SAME TEXT AS ABOVE physical_plan after LimitPushPastWindows SAME TEXT AS ABOVE +physical_plan after BufferInsertion SAME TEXT AS ABOVE physical_plan after LimitPushdown SAME TEXT AS ABOVE physical_plan after TopKRepartition SAME TEXT AS ABOVE physical_plan after ProjectionPushdown SAME TEXT AS ABOVE physical_plan after PushdownSort SAME TEXT AS ABOVE -physical_plan after BufferInsertion SAME TEXT AS ABOVE physical_plan after EnsureCooperative SAME TEXT AS ABOVE physical_plan after FilterPushdown(Post) SAME TEXT AS ABOVE physical_plan after SanityCheckPlan SAME TEXT AS ABOVE From 3d443e9deeae79a1059bb87930b7c0e653050483 Mon Sep 17 00:00:00 2001 From: Tobias Schwarzinger Date: Tue, 8 Sep 2026 18:28:59 +0200 Subject: [PATCH 4/5] Smaller changes --- datafusion/datasource/src/source.rs | 1 - datafusion/physical-optimizer/src/buffering.rs | 15 ++++++--------- .../physical-optimizer/src/hash_join_buffering.rs | 2 +- 3 files changed, 7 insertions(+), 11 deletions(-) diff --git a/datafusion/datasource/src/source.rs b/datafusion/datasource/src/source.rs index 085d64148eff3..741010c595197 100644 --- a/datafusion/datasource/src/source.rs +++ b/datafusion/datasource/src/source.rs @@ -468,7 +468,6 @@ impl ExecutionPlan for DataSourceExec { let args = OpenArgs::new(partition, Arc::clone(&context)) .with_shared_state(shared_state); let stream = self.data_source.open_with_args(args)?; - let batch_size = context.session_config().batch_size(); log::debug!( diff --git a/datafusion/physical-optimizer/src/buffering.rs b/datafusion/physical-optimizer/src/buffering.rs index e3ceb391f149b..77bb692a1df3f 100644 --- a/datafusion/physical-optimizer/src/buffering.rs +++ b/datafusion/physical-optimizer/src/buffering.rs @@ -36,12 +36,9 @@ use std::sync::Arc; /// capacity `config.execution.hash_join_buffering_capacity` so that the probe side can /// be eagerly polled while the build side is concurrently being built. /// 2. **Small Scan Buffering**: For data source scans whose statistics indicate that they -/// are smaller than `config.execution.small_scan_partition_threshold`, we wrap them in +/// are smaller than `config.execution.small_scan_buffering_threshold`, we wrap them in /// a [`BufferExec`]. /// -/// If the probe side of a hash join is buffered by (1), scans within that probe -/// side are not double-buffered. -/// /// ## `HashJoinExec` Buffering /// /// Looks for all the [HashJoinExec]s in the plan and places a [BufferExec] node with the @@ -66,10 +63,10 @@ use std::sync::Arc; /// /// ## Small Scan Buffering /// -/// The buffering of small scans works independently of the hash join buffering. Whenever a small -/// scan is detected, a [`BufferExec`] is inserted with the goal of reducing query latency as the -/// I/O of the small scan is triggered eagerly. As the scan is considered small, dynamic filters may -/// not yield significant improvements that warrant delaying the I/O until they are fully available. +/// Whenever a small "scan" (leaf node) is detected, a [`BufferExec`] is inserted with the goal of +/// reducing query latency as the I/O of the small scan is executed eagerly. As the scan is +/// considered small, dynamic filters may not yield significant improvements that warrant delaying +/// the I/O until they are fully available. /// /// ```text /// ┌───────────────────┐ @@ -172,7 +169,7 @@ fn transform_plan( let (transformed_probe, probe_transformed) = if hash_join_capacity > 0 && !probe_child.is::() { // Buffer the probe side. Since the probe child is wrapped in BufferExec, - // scans within that probe side must NOT be double-buffered (`in_buffer: true`). + // scans within that probe side should NOT be double-buffered (`in_buffer: true`). let probe_inner = transform_plan( Arc::clone(probe_child), hash_join_capacity, diff --git a/datafusion/physical-optimizer/src/hash_join_buffering.rs b/datafusion/physical-optimizer/src/hash_join_buffering.rs index 60d997c388b8c..b5bcb71558c45 100644 --- a/datafusion/physical-optimizer/src/hash_join_buffering.rs +++ b/datafusion/physical-optimizer/src/hash_join_buffering.rs @@ -19,6 +19,6 @@ use crate::buffering::BufferInsertion; #[deprecated( since = "56.0.0", - note = "Use the more general BufferingRule instead. This is only an alias." + note = "Use the more general BufferInsertion rule instead. This is only an alias." )] pub type HashJoinBuffering = BufferInsertion; From 7bb6e5de14089c3e075298e5486e1ce9ab031a74 Mon Sep 17 00:00:00 2001 From: Tobias Schwarzinger Date: Tue, 8 Sep 2026 18:58:27 +0200 Subject: [PATCH 5/5] Update configs.md --- docs/source/user-guide/configs.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index ab344028cdab4..00e753db96ea2 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -145,6 +145,7 @@ The following configuration settings are available: | datafusion.execution.objectstore_writer_buffer_size | 10485760 | Size (bytes) of data buffer DataFusion uses when writing output files. This affects the size of the data chunks that are uploaded to remote object stores (e.g. AWS S3). If very large (>= 100 GiB) output files are being written, it may be necessary to increase this size to avoid errors from the remote end point. | | datafusion.execution.enable_ansi_mode | false | Whether to enable ANSI SQL mode. The flag is experimental and relevant only for DataFusion Spark built-in functions When `enable_ansi_mode` is set to `true`, the query engine follows ANSI SQL semantics for expressions, casting, and error handling. This means: - **Strict type coercion rules:** implicit casts between incompatible types are disallowed. - **Standard SQL arithmetic behavior:** operations such as division by zero, numeric overflow, or invalid casts raise runtime errors rather than returning `NULL` or adjusted values. - **Consistent ANSI behavior** for string concatenation, comparisons, and `NULL` handling. When `enable_ansi_mode` is `false` (the default), the engine uses a more permissive, non-ANSI mode designed for user convenience and backward compatibility. In this mode: - Implicit casts between types are allowed (e.g., string to integer when possible). - Arithmetic operations are more lenient — for example, `abs()` on the minimum representable integer value returns the input value instead of raising overflow. - Division by zero or invalid casts may return `NULL` instead of failing. # Default `false` — ANSI SQL mode is disabled by default. | | 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.small_scan_buffering_threshold | 0 | Sets the threshold for scans that should be buffered. If the statistics suggest that a scan requires reading fewer bytes than this threshold, DataFusion may eagerly evaluate the scan to cut down latency. This approach is not applied to large scans (> `small_scan_buffering_threshold`) as eagerly evaluated scans do not have access to the final dynamic filters, which may significantly reduce the number of bytes scanned. Disabled by default, set to a number greater than 0 for enabling it. | | datafusion.optimizer.enable_distinct_aggregation_soft_limit | true | When set to true, the optimizer will push a limit operation into grouped aggregations which have no aggregate expressions, as a soft limit, emitting groups once the limit is reached, before all rows in the group are read. | | datafusion.optimizer.enable_round_robin_repartition | true | When set to true, the physical plan optimizer will try to add round robin repartitioning to increase parallelism to leverage more CPU cores | | datafusion.optimizer.enable_topk_aggregation | true | When set to true, the optimizer will attempt to perform limit operations during aggregations, if possible |