diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 347721c43d7cf..4b91cfc09e51f 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1181,6 +1181,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 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. + 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..7767c46bac3bc 100644 --- a/datafusion/core/src/optimizer_rule_reference.md +++ b/datafusion/core/src/optimizer_rule_reference.md @@ -83,7 +83,7 @@ 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. | +| 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. | diff --git a/datafusion/physical-optimizer/src/buffering.rs b/datafusion/physical-optimizer/src/buffering.rs new file mode 100644 index 0000000000000..77bb692a1df3f --- /dev/null +++ b/datafusion/physical-optimizer/src/buffering.rs @@ -0,0 +1,262 @@ +// 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_buffering_threshold`, we wrap them in +/// a [`BufferExec`]. +/// +/// ## `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 +/// +/// 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 +/// ┌───────────────────┐ +/// │ 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 should 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..b5bcb71558c45 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 BufferInsertion rule 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..587174a721de9 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,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 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 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. 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..cc73833311508 100644 --- a/datafusion/sqllogictest/test_files/explain.slt +++ b/datafusion/sqllogictest/test_files/explain.slt @@ -245,7 +245,7 @@ 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 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 @@ -327,7 +327,7 @@ 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 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 @@ -373,7 +373,7 @@ 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 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 @@ -624,7 +624,7 @@ 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 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 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. 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 |