From 07f916c1a10b12f7a39be7be21df4896b9ebe6bc Mon Sep 17 00:00:00 2001 From: hushen <190065939+918154429@users.noreply.github.com> Date: Fri, 11 Sep 2026 00:06:56 +0800 Subject: [PATCH] feat: add opt-in probe selection for partitioned inner hash joins --- .../file_stream_provider.rs | 1 + datafusion/common/src/config.rs | 7 + datafusion/common/src/rounding.rs | 2 +- datafusion/core/tests/sql/joins.rs | 87 +++++ datafusion/physical-plan/Cargo.toml | 5 + .../benches/hash_join_selection.rs | 284 ++++++++++++++ .../physical-plan/src/joins/hash_join/exec.rs | 107 ++++-- .../physical-plan/src/joins/hash_join/mod.rs | 3 + .../src/joins/hash_join/selection.rs | 255 +++++++++++++ .../src/joins/hash_join/selection_tests.rs | 347 ++++++++++++++++++ .../src/joins/hash_join/stream.rs | 67 +++- .../test_files/information_schema.slt | 2 + docs/source/user-guide/configs.md | 1 + 13 files changed, 1136 insertions(+), 32 deletions(-) create mode 100644 datafusion/physical-plan/benches/hash_join_selection.rs create mode 100644 datafusion/physical-plan/src/joins/hash_join/selection.rs create mode 100644 datafusion/physical-plan/src/joins/hash_join/selection_tests.rs diff --git a/datafusion-examples/examples/custom_data_source/file_stream_provider.rs b/datafusion-examples/examples/custom_data_source/file_stream_provider.rs index 5b43072d43f80..e8a92f17e1f96 100644 --- a/datafusion-examples/examples/custom_data_source/file_stream_provider.rs +++ b/datafusion-examples/examples/custom_data_source/file_stream_provider.rs @@ -28,6 +28,7 @@ /// with DataFusion without needing to reload the entire dataset each time. /// /// This example does not work on Windows. +#[cfg_attr(target_os = "windows", expect(clippy::unused_async))] pub async fn file_stream_provider() -> datafusion::error::Result<()> { #[cfg(target_os = "windows")] { diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 360586b0e9bae..2a5931ccf2521 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -912,6 +912,13 @@ config_namespace! { /// Support for build_side.num_rows() >= u32::MAX will be added in the future. pub perfect_hash_join_small_build_threshold: usize, default = 1024 + /// Enable probe-side selection exchange for partitioned inner hash joins. + /// Shares payload batches and copies only selected join keys before lookup. + /// Requires simple column keys, an unordered hash repartition directly on + /// the probe side, no dynamic filter, and an unlimited memory pool. + /// Other plans retain the ordinary spill-capable repartition path. + pub enable_hash_join_probe_selection: bool, default = false + /// The minimum required density of join keys on the build side to consider a /// perfect hash join (see `HashJoinExec` for more details). Density is calculated as: /// `(number of rows) / (max_key - min_key + 1)`. diff --git a/datafusion/common/src/rounding.rs b/datafusion/common/src/rounding.rs index 1796143d7cf1a..b3b514e8631c8 100644 --- a/datafusion/common/src/rounding.rs +++ b/datafusion/common/src/rounding.rs @@ -254,7 +254,7 @@ where } } _ => {} - }; + } Ok(result) } diff --git a/datafusion/core/tests/sql/joins.rs b/datafusion/core/tests/sql/joins.rs index dd32830dd5eb8..1b5c0022ccb6d 100644 --- a/datafusion/core/tests/sql/joins.rs +++ b/datafusion/core/tests/sql/joins.rs @@ -27,6 +27,93 @@ use datafusion_sql::unparser::plan_to_sql; use super::*; +#[tokio::test] +async fn hash_join_probe_selection_sql() -> Result<()> { + use arrow::array::Int64Array; + use datafusion::physical_plan::{ExecutionPlan, collect}; + + fn selected_partitions(plan: &dyn ExecutionPlan) -> usize { + let here = plan + .metrics() + .and_then(|m| m.sum_by_name("probe_selection_partitions")) + .map_or(0, |m| m.as_usize()); + here + plan + .children() + .iter() + .map(|child| selected_partitions(child.as_ref())) + .sum::() + } + + let mut config = SessionConfig::new().with_target_partitions(4); + let options = config.options_mut(); + options.optimizer.hash_join_single_partition_threshold = 0; + options.optimizer.hash_join_single_partition_threshold_rows = 0; + options.optimizer.enable_join_dynamic_filter_pushdown = false; + for key_type in [DataType::Int64, DataType::Utf8, DataType::Utf8View] { + let ctx = SessionContext::new_with_config(config.clone()); + let schema = Arc::new(Schema::new(vec![ + Field::new("key", key_type.clone(), true), + Field::new("id", DataType::Int64, false), + ])); + for (name, keys, ids) in [ + ( + "selection_left", + vec![Some(1), Some(1), Some(2), None], + vec![10, 20, 30, 40], + ), + ( + "selection_right", + vec![Some(1), Some(2), None, Some(3)], + vec![15, 35, 45, 55], + ), + ] { + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + arrow::compute::cast(&Int64Array::from(keys), &key_type)?, + Arc::new(Int64Array::from(ids)), + ], + )?; + ctx.register_table( + name, + Arc::new(MemTable::try_new( + Arc::clone(&schema), + (0..4).map(|row| vec![batch.slice(row, 1)]).collect(), + )?), + )?; + } + for enabled in [false, true] { + ctx.sql(&format!( + "SET datafusion.execution.enable_hash_join_probe_selection = '{enabled}'" + )) + .await? + .collect() + .await?; + let plan = ctx.sql("SELECT l.id AS l, r.id AS r FROM selection_left l JOIN selection_right r ON l.key = r.key AND l.id < r.id ORDER BY l.id") + .await?.create_physical_plan().await?; + let batches = collect(Arc::clone(&plan), ctx.task_ctx()).await?; + assert_batches_eq!( + [ + "+----+----+", + "| l | r |", + "+----+----+", + "| 10 | 15 |", + "| 30 | 35 |", + "+----+----+", + ], + &batches + ); + assert_eq!( + selected_partitions(plan.as_ref()), + if enabled { 4 } else { 0 }, + "{}", + displayable(plan.as_ref()).indent(true) + ); + } + } + Ok(()) +} + #[tokio::test] async fn join_change_in_planner() -> Result<()> { let config = SessionConfig::new().with_target_partitions(8); diff --git a/datafusion/physical-plan/Cargo.toml b/datafusion/physical-plan/Cargo.toml index 534cea8ea9cbb..1f4b349785809 100644 --- a/datafusion/physical-plan/Cargo.toml +++ b/datafusion/physical-plan/Cargo.toml @@ -169,3 +169,8 @@ name = "window_filter" [[bench]] harness = false name = "range_repartition" + +[[bench]] +harness = false +name = "hash_join_selection" +required-features = ["test_utils"] diff --git a/datafusion/physical-plan/benches/hash_join_selection.rs b/datafusion/physical-plan/benches/hash_join_selection.rs new file mode 100644 index 0000000000000..1d71d49fafafd --- /dev/null +++ b/datafusion/physical-plan/benches/hash_join_selection.rs @@ -0,0 +1,284 @@ +// 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. + +//! Paired end-to-end physical-plan experiment. Set SAMPLES (default 21), +//! CASE (e.g. Utf8-8-10), or MODE (true/false) to restrict a run. CSV includes +//! plan construction, both repartitions, build, probe, output and destruction. +//! `peak_reserved` is engine-accounted memory, not process RSS. + +use datafusion_common::instant::Instant; +use std::fmt; +use std::hint::black_box; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use arrow::array::{ArrayRef, Int64Array, StringArray, StringViewArray}; +use arrow::datatypes::{Field, Schema}; +use arrow::record_batch::RecordBatch; +use datafusion_common::{JoinType, Result}; +use datafusion_execution::TaskContext; +use datafusion_execution::config::SessionConfig; +use datafusion_execution::memory_pool::{MemoryLimit, MemoryPool, MemoryReservation}; +use datafusion_execution::runtime_env::RuntimeEnvBuilder; +use datafusion_physical_expr::PhysicalExprRef; +use datafusion_physical_expr::expressions::Column; +use datafusion_physical_plan::joins::{HashJoinExecBuilder, PartitionMode}; +use datafusion_physical_plan::repartition::RepartitionExec; +use datafusion_physical_plan::test::TestMemoryExec; +use datafusion_physical_plan::{ExecutionPlan, Partitioning, collect}; + +#[derive(Debug, Default)] +struct PeakPool { + current: AtomicUsize, + peak: AtomicUsize, +} +impl fmt::Display for PeakPool { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "PeakPool") + } +} +impl MemoryPool for PeakPool { + fn name(&self) -> &str { + "PeakPool" + } + fn grow(&self, _: &MemoryReservation, bytes: usize) { + let current = self.current.fetch_add(bytes, Ordering::SeqCst) + bytes; + self.peak.fetch_max(current, Ordering::SeqCst); + } + fn shrink(&self, _: &MemoryReservation, bytes: usize) { + self.current.fetch_sub(bytes, Ordering::SeqCst); + } + fn try_grow(&self, r: &MemoryReservation, bytes: usize) -> Result<()> { + self.grow(r, bytes); + Ok(()) + } + fn reserved(&self) -> usize { + self.current.load(Ordering::SeqCst) + } + fn memory_limit(&self) -> MemoryLimit { + MemoryLimit::Infinite + } +} + +fn batch( + rows: usize, + kind: &str, + coverage: usize, + build: bool, + skew: bool, +) -> RecordBatch { + let keys: Vec = (0..rows) + .filter(|i| !build || i % 100 < coverage) + .map(|i| { + if !build && skew && i % 10 != 0 { + 1 + } else { + (i % 2048) as i64 + } + }) + .collect(); + let n = keys.len(); + let mut columns: Vec = vec![Arc::new(Int64Array::from(keys))]; + match kind { + "Wide" if !build => { + for column in 0..24 { + columns.push(Arc::new(Int64Array::from_iter_values( + (0..n).map(|i| (i * 31 + column) as i64), + ))); + } + } + "Utf8" | "Utf8View" if !build => { + let strings: Vec<_> = (0..n) + .map(|i| format!("{i:08}-{}", "payload-".repeat(31))) + .collect(); + for _ in 0..4 { + columns.push(if kind == "Utf8" { + Arc::new(StringArray::from_iter_values(&strings)) as ArrayRef + } else { + Arc::new(StringViewArray::from_iter_values(&strings)) as ArrayRef + }); + } + } + _ => columns.push(Arc::new(Int64Array::from_iter_values( + (0..n).map(|i| i as i64), + ))), + } + let schema = Arc::new(Schema::new( + columns + .iter() + .enumerate() + .map(|(i, c)| Field::new(format!("c{i}"), c.data_type().clone(), false)) + .collect::>(), + )); + RecordBatch::try_new(schema, columns).unwrap() +} + +fn source(batches: &[RecordBatch]) -> Arc { + let mut partitions = vec![vec![], vec![], vec![], vec![]]; + for (index, batch) in batches.iter().enumerate() { + partitions[index % 4].push(batch.clone()); + } + TestMemoryExec::try_new_exec(&partitions, batches[0].schema(), None).unwrap() +} + +async fn query( + left: &[RecordBatch], + right: &[RecordBatch], + partitions: usize, + selected: bool, +) -> (usize, usize) { + let pool = Arc::new(PeakPool::default()); + let runtime = RuntimeEnvBuilder::new() + .with_memory_pool(pool.clone()) + .build_arc() + .unwrap(); + let mut config = SessionConfig::new(); + config + .options_mut() + .execution + .enable_hash_join_probe_selection = selected; + let context = Arc::new( + TaskContext::default() + .with_runtime(runtime) + .with_session_config(config), + ); + let key: PhysicalExprRef = Arc::new(Column::new("c0", 0)); + let left = Arc::new( + RepartitionExec::try_new( + source(left), + Partitioning::Hash(vec![Arc::clone(&key)], partitions), + ) + .unwrap(), + ); + let right = Arc::new( + RepartitionExec::try_new( + source(right), + Partitioning::Hash(vec![Arc::clone(&key)], partitions), + ) + .unwrap(), + ); + let join = HashJoinExecBuilder::new( + left, + right, + vec![(Arc::clone(&key), key)], + JoinType::Inner, + ) + .with_partition_mode(PartitionMode::Partitioned) + .build() + .unwrap(); + let join = Arc::new(join); + let output = collect(join.clone(), context).await.unwrap(); + let used = join + .metrics() + .unwrap() + .sum_by_name("probe_selection_partitions") + .map_or(0, |m| m.as_usize()); + assert_eq!(used, if selected { partitions } else { 0 }); + drop(join); + let rows = output.iter().map(RecordBatch::num_rows).sum(); + drop(output); + // Ensure an asynchronous producer destructor is included in measurement. + while pool.reserved() != 0 { + tokio::task::yield_now().await; + } + (rows, pool.peak.load(Ordering::SeqCst)) +} + +fn main() { + if std::env::args().any(|a| a == "--test") { + return; + } + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .build() + .unwrap(); + let samples: usize = std::env::var("SAMPLES") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(21); + let case_filter = std::env::var("CASE").ok(); + let mode_filter = std::env::var("MODE") + .ok() + .map(|s| s.parse::().unwrap()); + println!( + "kind,partitions,hit_percent,skew,sample,selection,total_us,peak_reserved,output_rows" + ); + for kind in ["Narrow", "Wide", "Utf8", "Utf8View"] { + for partitions in [8, 32] { + for hit in [10, 100] { + for skew in [false, true] { + if case_filter + .as_ref() + .is_some_and(|f| *f != format!("{kind}-{partitions}-{hit}")) + { + continue; + } + let left = vec![batch(2048, kind, hit, true, false)]; + // Independent source buffers, not slices of one large buffer: + // otherwise every source batch's memory estimate counts that + // large allocation and obscures the exchange's retention cost. + let right: Vec<_> = (0..4) + .map(|_| batch(8192, kind, hit, false, skew)) + .collect(); + let build_keys: std::collections::HashSet<_> = left[0] + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .values() + .iter() + .copied() + .collect(); + let expected: usize = right + .iter() + .map(|batch| { + batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .values() + .iter() + .filter(|key| build_keys.contains(key)) + .count() + }) + .sum(); + for sample in 0..=samples { + for selected in [sample % 2 == 0, sample % 2 != 0] { + if mode_filter.is_some_and(|m| m != selected) { + continue; + } + let start = Instant::now(); + let (rows, peak) = black_box( + runtime + .block_on(query(&left, &right, partitions, selected)), + ); + let elapsed = start.elapsed().as_secs_f64() * 1e6; + assert_eq!(rows, expected); + if sample > 0 { + println!( + "{kind},{partitions},{hit},{skew},{sample},{selected},{elapsed:.3},{peak},{rows}" + ); + } + } + } + } + } + } + } +} diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 01dec6570bd71..291135367aa06 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -100,6 +100,7 @@ use futures::TryStreamExt; use parking_lot::Mutex; use super::partitioned_hash_eval::SeededRandomState; +use super::selection::SelectionExchange; /// Hard-coded seed to ensure hash values from the hash join differ from `RepartitionExec`, avoiding collisions. pub(crate) const HASH_JOIN_SEED: SeededRandomState = @@ -413,6 +414,7 @@ impl HashJoinExecBuilder { filter: None, join_type, left_fut: Default::default(), + selection_exchange: Default::default(), random_state: HASH_JOIN_SEED, mode: PartitionMode::Auto, fetch: None, @@ -507,12 +509,14 @@ impl HashJoinExecBuilder { self.preserve_properties &= has_same_children_properties(&self.exec, &children)?; self.exec.right = children.swap_remove(1); self.exec.left = children.swap_remove(0); + self.exec.selection_exchange = Default::default(); Ok(self) } /// Reset runtime state. pub fn reset_state(mut self) -> Self { self.exec.left_fut = Default::default(); + self.exec.selection_exchange = Default::default(); self.exec.dynamic_filter = None; self.exec.metrics = ExecutionPlanMetricsSet::new(); self @@ -544,6 +548,7 @@ impl HashJoinExecBuilder { filter, join_type, left_fut, + selection_exchange, random_state, mode, metrics, @@ -591,6 +596,7 @@ impl HashJoinExecBuilder { join_type, join_schema, left_fut, + selection_exchange, random_state, mode, metrics, @@ -621,6 +627,7 @@ impl From<&HashJoinExec> for HashJoinExecBuilder { join_type: exec.join_type, join_schema: Arc::clone(&exec.join_schema), left_fut: Arc::clone(&exec.left_fut), + selection_exchange: Arc::clone(&exec.selection_exchange), random_state: exec.random_state.clone(), mode: exec.mode, metrics: exec.metrics.clone(), @@ -850,6 +857,7 @@ pub struct HashJoinExec { /// Each output stream waits on the `OnceAsync` to signal the completion of /// the hash table creation. left_fut: Arc>, + selection_exchange: Arc, /// Shared the `SeededRandomState` for the hashing algorithm (seeds preserved for serialization) random_state: SeededRandomState, /// Partitioning mode to use @@ -1658,7 +1666,57 @@ impl ExecutionPlan for HashJoinExec { // we have the batches and the hash map with their keys. We can how create a stream // over the right that uses this information to issue new batches. - let right_stream = self.right.execute(partition, context)?; + let selection = if context + .session_config() + .options() + .execution + .enable_hash_join_probe_selection + && matches!( + context.memory_pool().memory_limit(), + datafusion_execution::memory_pool::MemoryLimit::Infinite + ) + && self.mode == PartitionMode::Partitioned + && self.join_type == JoinType::Inner + && !self.null_aware + && self.dynamic_filter.is_none() + && self + .on + .iter() + .all(|(_, key)| key.downcast_ref::().is_some()) + { + self.right + .downcast_ref::() + .filter(|r| !r.preserve_order()) + .and_then(|r| match r.partitioning() { + Partitioning::Hash(keys, n) + if *n > 1 + && keys.len() == self.on.len() + && keys.iter().zip(&self.on).all(|(a, (_, b))| a.eq(b)) => + { + Some(self.selection_exchange.execute( + r.input().as_ref(), + keys, + *n, + partition, + &context, + )) + } + _ => None, + }) + } else { + None + } + .transpose()?; + let right_stream = if selection.is_some() { + MetricBuilder::new(&self.metrics) + .counter("probe_selection_partitions", partition) + .add(1); + Box::pin(crate::stream::EmptyRecordBatchStream::new( + self.right.schema(), + )) as SendableRecordBatchStream + } else { + self.right.execute(partition, context)? + }; // update column indices to reflect the projection let column_indices_after_projection = match self.projection.as_ref() { @@ -1675,27 +1733,30 @@ impl ExecutionPlan for HashJoinExec { .map(|(_, right_expr)| Arc::clone(right_expr)) .collect::>(); - Ok(Box::pin(HashJoinStream::new( - partition, - self.schema(), - on_right, - self.filter.clone(), - self.join_type, - right_stream, - self.random_state.random_state().clone(), - join_metrics, - column_indices_after_projection, - self.null_equality, - HashJoinStreamState::WaitBuildSide, - BuildSide::Initial(BuildSideInitialState { left_fut }), - batch_size, - vec![], - self.right.output_ordering().is_some(), - build_accumulator, - self.mode, - null_aware, - self.fetch, - ))) + Ok(Box::pin( + HashJoinStream::new( + partition, + self.schema(), + on_right, + self.filter.clone(), + self.join_type, + right_stream, + self.random_state.random_state().clone(), + join_metrics, + column_indices_after_projection, + self.null_equality, + HashJoinStreamState::WaitBuildSide, + BuildSide::Initial(BuildSideInitialState { left_fut }), + batch_size, + vec![], + self.right.output_ordering().is_some(), + build_accumulator, + self.mode, + null_aware, + self.fetch, + ) + .with_selection_input(selection), + )) } fn metrics(&self) -> Option { @@ -2000,6 +2061,8 @@ impl ExecutionPlan for HashJoinExec { join_schema: _, // runtime build-side state, not part of the plan left_fut: _, + // runtime probe-side state, not part of the plan + selection_exchange: _, // the fixed `HASH_JOIN_SEED` constant, set identically by the // builder on decode random_state: _, diff --git a/datafusion/physical-plan/src/joins/hash_join/mod.rs b/datafusion/physical-plan/src/joins/hash_join/mod.rs index b915802ea4015..e65a3a01f6afe 100644 --- a/datafusion/physical-plan/src/joins/hash_join/mod.rs +++ b/datafusion/physical-plan/src/joins/hash_join/mod.rs @@ -23,5 +23,8 @@ pub use partitioned_hash_eval::{HashExpr, HashTableLookupExpr, SeededRandomState mod exec; mod inlist_builder; mod partitioned_hash_eval; +mod selection; +#[cfg(test)] +mod selection_tests; mod shared_bounds; mod stream; diff --git a/datafusion/physical-plan/src/joins/hash_join/selection.rs b/datafusion/physical-plan/src/joins/hash_join/selection.rs new file mode 100644 index 0000000000000..45f84ab4e3e34 --- /dev/null +++ b/datafusion/physical-plan/src/joins/hash_join/selection.rs @@ -0,0 +1,255 @@ +// 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. + +//! Private probe-side exchange. Selection messages never implement +//! `RecordBatchStream`: only the inner hash join may consume them. + +use std::pin::Pin; +use std::sync::{Arc, Weak}; + +use arrow::array::{Array, ArrayRef, UInt32Array}; +use arrow::compute::take; +use arrow::record_batch::RecordBatch; +use datafusion_common::hash_utils::create_hashes; +use datafusion_common::{DataFusionError, Result, exec_err}; +use datafusion_common_runtime::SpawnedTask; +use datafusion_execution::TaskContext; +use datafusion_execution::memory_pool::{MemoryConsumer, MemoryReservation}; +use datafusion_physical_expr::PhysicalExprRef; +use datafusion_physical_expr_common::utils::evaluate_expressions_to_arrays; +use futures::{Stream, StreamExt}; +use parking_lot::Mutex; +use tokio::sync::mpsc; + +use crate::repartition::REPARTITION_RANDOM_STATE; +use crate::{ExecutionPlan, ExecutionPlanProperties}; + +pub(super) type SelectionStream = + Pin> + Send>>; + +/// One reservation for the shared payload, irrespective of fan-out. +#[derive(Debug)] +pub(super) struct SharedBatch { + pub batch: RecordBatch, + _reservation: MemoryReservation, +} + +#[derive(Debug)] +pub(super) struct SelectedBatch { + pub owner: Arc, + pub indices: UInt32Array, + /// Compact keys: lookup positions refer to these arrays, not the payload. + pub values: Vec, + _reservation: MemoryReservation, +} + +type ReceiverSlot = Mutex>>>; + +struct Tasks { + _tasks: Vec>, + receivers: Vec>, +} + +impl Drop for Tasks { + fn drop(&mut self) { + // Also release queued messages for partitions never executed, even if + // the caller retains the plan after canceling the query. + for receiver in &self.receivers { + if let Some(receiver) = receiver.upgrade() { + receiver.lock().take(); + } + } + } +} + +struct ExchangeState { + receivers: Vec>, + // Only consumers keep producers alive. Retaining the plan must not keep + // tasks running after all consumers have been dropped. + tasks: Weak, +} + +#[derive(Default)] +pub(super) struct SelectionExchange { + state: Mutex>, +} + +impl SelectionExchange { + pub fn execute( + self: &Arc, + input: &dyn ExecutionPlan, + keys: &[PhysicalExprRef], + partitions: usize, + partition: usize, + context: &Arc, + ) -> Result { + let (receiver, guard) = { + let mut state = self.state.lock(); + let guard = if let Some(state) = state.as_ref() { + state.tasks.upgrade().ok_or_else(|| { + DataFusionError::Execution( + "Selection exchange already stopped".into(), + ) + })? + } else { + let (senders, receivers): (Vec<_>, Vec<_>) = + (0..partitions).map(|_| mpsc::channel(1)).unzip(); + let mut tasks = Vec::new(); + for p in 0..input.output_partitioning().partition_count() { + let stream = input.execute(p, Arc::clone(context))?; + let senders = senders.clone(); + let task = SpawnedTask::spawn(route_input( + stream, + keys.to_vec(), + senders.clone(), + Arc::clone(context), + )); + tasks.push(SpawnedTask::spawn(async move { + let error = match task.join().await { + Ok(Ok(())) => return, + Ok(Err(e)) => e, + Err(e) => DataFusionError::External(Box::new(e)), + }; + let error = Arc::new(error); + // Send errors concurrently: an unpolled output must + // not prevent an active output observing the error. + futures::future::join_all( + senders + .iter() + .map(|tx| tx.send(Err(DataFusionError::from(&error)))), + ) + .await; + })); + } + let receivers: Vec<_> = receivers + .into_iter() + .map(|r| Arc::new(Mutex::new(Some(r)))) + .collect(); + let guard = Arc::new(Tasks { + _tasks: tasks, + receivers: receivers.iter().map(Arc::downgrade).collect(), + }); + *state = Some(ExchangeState { + receivers, + tasks: Arc::downgrade(&guard), + }); + guard + }; + let receiver = state + .as_mut() + .unwrap() + .receivers + .get_mut(partition) + .and_then(|receiver| receiver.lock().take()) + .ok_or_else(|| { + DataFusionError::Execution(format!( + "Selection partition {partition} already consumed or out of range" + )) + })?; + (receiver, guard) + }; + Ok(Box::pin(futures::stream::unfold( + (receiver, guard), + |(mut receiver, guard)| async move { + receiver + .recv() + .await + .map(|batch| (batch, (receiver, guard))) + }, + ))) + } +} + +async fn route_input( + mut input: crate::SendableRecordBatchStream, + keys: Vec, + senders: Vec>>, + context: Arc, +) -> Result<()> { + let scratch = + MemoryConsumer::new("HashJoinSelectionScratch").register(context.memory_pool()); + while let Some(batch) = input.next().await.transpose()? { + if senders.iter().all(mpsc::Sender::is_closed) { + break; + } + let n = batch.num_rows(); + if n == 0 { + continue; + } + if n > u32::MAX as usize { + return exec_err!( + "Selection exchange requires at most u32::MAX rows per batch" + ); + } + let reservation = + MemoryConsumer::new("HashJoinSelectionBatch").register(context.memory_pool()); + reservation.try_grow(batch.get_array_memory_size())?; + // Bound routing scratch before allocation (u64 hashes + u32 indices, + // plus Vec headers). Vec growth is avoided by counting first. + scratch.try_resize(n * 12 + senders.len() * (size_of::>() + 16))?; + let values = evaluate_expressions_to_arrays(&keys, &batch)?; + let mut hashes = vec![0; n]; + create_hashes( + &values, + REPARTITION_RANDOM_STATE.random_state(), + &mut hashes, + )?; + let mut counts = vec![0usize; senders.len()]; + for hash in &hashes { + counts[(*hash % senders.len() as u64) as usize] += 1; + } + let mut indices: Vec> = + counts.iter().map(|&n| Vec::with_capacity(n)).collect(); + for (row, hash) in hashes.iter().enumerate() { + indices[(*hash % senders.len() as u64) as usize].push(row as u32); + } + let owner = Arc::new(SharedBatch { + batch, + _reservation: reservation, + }); + for (tx, rows) in senders.iter().zip(indices) { + if rows.is_empty() || tx.is_closed() { + continue; + } + let indices = UInt32Array::from(rows); + let selected_values = values + .iter() + .map(|v| take(v.as_ref(), &indices, None)) + .collect::, _>>()?; + let bytes = indices.get_array_memory_size() + + selected_values + .iter() + .map(|v| v.get_array_memory_size()) + .sum::(); + let reservation = MemoryConsumer::new("HashJoinSelectionIndices") + .register(context.memory_pool()); + reservation.try_grow(bytes)?; + // A full queue applies backpressure. A dropped consumer simply + // relinquishes its partition; other consumers can still finish. + tx.send(Ok(SelectedBatch { + owner: Arc::clone(&owner), + indices, + values: selected_values, + _reservation: reservation, + })) + .await + .ok(); + } + scratch.free(); + } + Ok(()) +} diff --git a/datafusion/physical-plan/src/joins/hash_join/selection_tests.rs b/datafusion/physical-plan/src/joins/hash_join/selection_tests.rs new file mode 100644 index 0000000000000..1ef65738d6a8a --- /dev/null +++ b/datafusion/physical-plan/src/joins/hash_join/selection_tests.rs @@ -0,0 +1,347 @@ +// 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 std::sync::Arc; +use std::time::Duration; + +use arrow::array::{ArrayRef, Int64Array, StringArray}; +use arrow::datatypes::{DataType, Field, Schema}; +use arrow::record_batch::RecordBatch; +use datafusion_common::test_util::batches_to_sort_string; +use datafusion_common::{JoinSide, JoinType, NullEquality, Result}; +use datafusion_execution::TaskContext; +use datafusion_execution::config::SessionConfig; +use datafusion_execution::memory_pool::GreedyMemoryPool; +use datafusion_execution::runtime_env::RuntimeEnvBuilder; +use datafusion_expr::Operator; +use datafusion_physical_expr::PhysicalExprRef; +use datafusion_physical_expr::expressions::{BinaryExpr, Column}; +use futures::StreamExt; + +use super::HashJoinExecBuilder; +use super::selection::SelectionExchange; +use crate::joins::PartitionMode; +use crate::joins::utils::{ColumnIndex, JoinFilter}; +use crate::repartition::RepartitionExec; +use crate::test::TestMemoryExec; +use crate::{ExecutionPlan, Partitioning, collect}; + +fn source(seed: usize, width: usize) -> Result> { + let schema = Arc::new(Schema::new(vec![ + Field::new("key", DataType::Int64, true), + Field::new("id", DataType::Int64, false), + Field::new("payload", DataType::Utf8, false), + ])); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(Int64Array::from_iter( + (0..137).map(|i| (i % 11 != 0).then_some(((i * 17 + seed) % 23) as i64)), + )) as ArrayRef, + Arc::new(Int64Array::from_iter_values((0..137).map(|i| i as i64))), + Arc::new(StringArray::from_iter_values( + (0..137).map(|i| format!("row-{i}-{}", "x".repeat(width))), + )), + ], + )?; + let mut partitions = vec![vec![], vec![], vec![]]; + let step = 7 + seed; + for start in (0..137).step_by(step) { + partitions[start % 3].push(batch.slice(start, step.min(137 - start))); + } + Ok(TestMemoryExec::try_new_exec(&partitions, schema, None)?) +} + +fn context(batch_size: usize, array_map: bool, bounded: bool) -> Arc { + let mut config = SessionConfig::new().with_batch_size(batch_size); + if !array_map { + config + .options_mut() + .execution + .perfect_hash_join_small_build_threshold = 0; + config + .options_mut() + .execution + .perfect_hash_join_min_key_density = f64::INFINITY; + } + let mut context = TaskContext::default().with_session_config(config); + if bounded { + context = context.with_runtime( + RuntimeEnvBuilder::new() + .with_memory_pool(Arc::new(GreedyMemoryPool::new(16 * 1024 * 1024))) + .build_arc() + .unwrap(), + ); + } + Arc::new(context) +} + +async fn run_join( + enabled: bool, + join_type: JoinType, + seed: usize, + partitions: usize, + filtered: bool, + null_equality: NullEquality, + context: Arc, +) -> Result<(Vec, usize)> { + let mut config = context.session_config().clone(); + config + .options_mut() + .execution + .enable_hash_join_probe_selection = enabled; + let context = Arc::new( + TaskContext::default() + .with_runtime(context.runtime_env()) + .with_session_config(config), + ); + let key: PhysicalExprRef = Arc::new(Column::new("key", 0)); + let left = Arc::new(RepartitionExec::try_new( + source(seed, 64)?, + Partitioning::Hash(vec![Arc::clone(&key)], partitions), + )?); + let right = Arc::new(RepartitionExec::try_new( + source(seed + 1, 128)?, + Partitioning::Hash(vec![Arc::clone(&key)], partitions), + )?); + let filter = filtered.then(|| { + JoinFilter::new( + Arc::new(BinaryExpr::new( + Arc::new(Column::new("l", 0)), + Operator::Lt, + Arc::new(Column::new("r", 1)), + )), + vec![ + ColumnIndex { + index: 1, + side: JoinSide::Left, + }, + ColumnIndex { + index: 1, + side: JoinSide::Right, + }, + ], + Arc::new(Schema::new(vec![ + Field::new("l", DataType::Int64, false), + Field::new("r", DataType::Int64, false), + ])), + ) + }); + let join = Arc::new( + HashJoinExecBuilder::new(left, right, vec![(Arc::clone(&key), key)], join_type) + .with_partition_mode(PartitionMode::Partitioned) + .with_filter(filter) + .with_null_equality(null_equality) + .build()?, + ); + let output = tokio::time::timeout( + Duration::from_secs(10), + collect( + Arc::clone(&join) as Arc, + Arc::clone(&context), + ), + ) + .await + .expect("join must not deadlock")?; + let used = join + .metrics() + .unwrap() + .sum_by_name("probe_selection_partitions") + .map_or(0, |m| m.as_usize()); + drop(join); + // Producers abort asynchronously when the last output is dropped. + tokio::time::timeout(Duration::from_secs(5), async { + while context.memory_pool().reserved() != 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("all reservations released"); + Ok((output, used)) +} + +#[tokio::test] +async fn selected_probe_matches_materialized_join() -> Result<()> { + for seed in 1..=3 { + for partitions in [2, 8, 32] { + for array_map in [false, true] { + for filtered in [false, true] { + for nulls in [ + NullEquality::NullEqualsNothing, + NullEquality::NullEqualsNull, + ] { + let ctx = context(5, array_map, false); + let (expected, ordinary) = run_join( + false, + JoinType::Inner, + seed, + partitions, + filtered, + nulls, + Arc::clone(&ctx), + ) + .await?; + let (actual, selected) = run_join( + true, + JoinType::Inner, + seed, + partitions, + filtered, + nulls, + ctx, + ) + .await?; + assert_eq!(ordinary, 0); + assert_eq!(selected, partitions); + assert_eq!( + batches_to_sort_string(&actual), + batches_to_sort_string(&expected) + ); + assert!(actual.iter().all(|b| b.num_rows() <= 5)); + } + } + } + } + } + Ok(()) +} + +#[tokio::test] +async fn unsupported_selection_uses_ordinary_exchange() -> Result<()> { + for (join_type, bounded, partitions) in [ + (JoinType::Left, false, 4), + (JoinType::Inner, true, 4), + (JoinType::Inner, false, 1), + ] { + let ctx = context(7, false, bounded); + let (expected, _) = run_join( + false, + join_type, + 2, + partitions, + true, + NullEquality::NullEqualsNothing, + Arc::clone(&ctx), + ) + .await?; + let (actual, used) = run_join( + true, + join_type, + 2, + partitions, + true, + NullEquality::NullEqualsNothing, + ctx, + ) + .await?; + assert_eq!(used, 0); + assert_eq!( + batches_to_sort_string(&actual), + batches_to_sort_string(&expected) + ); + } + Ok(()) +} + +#[tokio::test] +async fn selection_cancel_releases_unclaimed_partitions() -> Result<()> { + let ctx = context(5, false, false); + let exchange = Arc::new(SelectionExchange::default()); + let key: PhysicalExprRef = Arc::new(Column::new("key", 0)); + let stream = exchange.execute(source(1, 4096)?.as_ref(), &[key], 8, 0, &ctx)?; + tokio::time::timeout(Duration::from_secs(5), async { + while ctx.memory_pool().reserved() == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("producer started"); + drop(stream); + // Keep the plan/exchange alive deliberately. It must not retain the other + // seven output queues or keep the producer tasks alive. + tokio::time::timeout(Duration::from_secs(5), async { + while ctx.memory_pool().reserved() != 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("canceled exchange leaked reservations"); + drop(exchange); + Ok(()) +} + +#[tokio::test] +async fn selection_propagates_input_failure_without_polling_other_outputs() -> Result<()> +{ + use crate::test::exec::{MockExec, PanicExec}; + let schema = source(1, 0)?.schema(); + let sources: Vec> = vec![ + Arc::new(MockExec::new( + vec![datafusion_common::exec_err!("selection input failure")], + Arc::clone(&schema), + )), + Arc::new(PanicExec::new(schema, 1)), + ]; + for source in sources { + let ctx = context(5, false, false); + let exchange = Arc::new(SelectionExchange::default()); + let key: PhysicalExprRef = Arc::new(Column::new("key", 0)); + let mut stream = exchange.execute(source.as_ref(), &[key], 8, 7, &ctx)?; + let result = tokio::time::timeout(Duration::from_secs(5), stream.next()) + .await + .expect("an idle output must not block error delivery") + .unwrap(); + let error = result + .expect_err("input failure must propagate") + .to_string(); + assert!( + error.contains("selection input failure") || error.contains("panicked"), + "{error}" + ); + drop(stream); + assert_eq!(ctx.memory_pool().reserved(), 0); + } + Ok(()) +} + +#[tokio::test] +async fn selection_reservation_failure_is_reported_and_released() -> Result<()> { + // Exercise the private exchange's error boundary directly. Public execution + // uses the spill-capable ordinary exchange for a finite pool. + let pool = Arc::new(GreedyMemoryPool::new(1)); + let runtime = RuntimeEnvBuilder::new() + .with_memory_pool(pool) + .build_arc()?; + let ctx = Arc::new(TaskContext::default().with_runtime(runtime)); + let exchange = Arc::new(SelectionExchange::default()); + let key: PhysicalExprRef = Arc::new(Column::new("key", 0)); + let mut stream = exchange.execute(source(1, 128)?.as_ref(), &[key], 8, 7, &ctx)?; + let result = tokio::time::timeout(Duration::from_secs(5), stream.next()) + .await + .expect("reservation failure must propagate") + .unwrap(); + assert!( + result + .err() + .unwrap() + .to_string() + .contains("Resources exhausted") + ); + drop(stream); + assert_eq!(ctx.memory_pool().reserved(), 0); + Ok(()) +} diff --git a/datafusion/physical-plan/src/joins/hash_join/stream.rs b/datafusion/physical-plan/src/joins/hash_join/stream.rs index 2c1ad94460540..a9bf46cd5d13f 100644 --- a/datafusion/physical-plan/src/joins/hash_join/stream.rs +++ b/datafusion/physical-plan/src/joins/hash_join/stream.rs @@ -178,6 +178,8 @@ impl HashJoinStreamState { pub(super) struct ProcessProbeBatchState { /// Current probe-side batch batch: RecordBatch, + /// Owns the original payload and reservation until all selected rows finish. + selection: Option>, /// Probe-side on expressions values values: Vec, /// Combined validity of the probe-side key columns, set when NULL keys @@ -341,6 +343,7 @@ pub(super) struct HashJoinStream { join_type: JoinType, /// right (probe) input right: SendableRecordBatchStream, + selection_input: Option, /// Random state used for hashing initialization random_state: RandomState, /// Metrics @@ -545,6 +548,7 @@ impl HashJoinStream { filter, join_type, right, + selection_input: None, random_state, join_metrics, column_indices, @@ -568,6 +572,14 @@ impl HashJoinStream { } } + pub(super) fn with_selection_input( + mut self, + input: Option, + ) -> Self { + self.selection_input = input; + self + } + /// Returns the next state after the build side has been fully collected /// and any required build-side coordination has completed. fn state_after_build_ready( @@ -684,7 +696,10 @@ impl HashJoinStream { // Continue loop to emit the flushed batch continue; } - HashJoinStreamState::Completed => Poll::Ready(None), + HashJoinStreamState::Completed => { + self.selection_input = None; + Poll::Ready(None) + } }; } } @@ -744,24 +759,41 @@ impl HashJoinStream { &mut self, cx: &mut std::task::Context<'_>, ) -> Poll>>> { - match ready!(self.right.poll_next_unpin(cx)) { + let next = if let Some(input) = self.selection_input.as_mut() { + ready!(input.poll_next_unpin(cx)).map(|result| { + result.map(|selected| { + (selected.owner.batch.clone(), Some(Arc::new(selected))) + }) + }) + } else { + ready!(self.right.poll_next_unpin(cx)) + .map(|result| result.map(|batch| (batch, None))) + }; + match next { None => { // Release the probe-side input pipeline's resources. The schema // is preserved so callers that still query `self.right.schema()` // (e.g. for unmatched-build emission) keep working. let right_schema = self.right.schema(); self.right = Box::pin(EmptyRecordBatchStream::new(right_schema)); + self.selection_input = None; self.state = HashJoinStreamState::ExhaustedProbeSide; } - Some(Ok(batch)) => { + Some(Ok((batch, selection))) => { // Precalculate hash values for fetched batch - let keys_values = evaluate_expressions_to_arrays(&self.on_right, &batch)?; + let keys_values = match &selection { + Some(selected) => selected.values.clone(), + None => evaluate_expressions_to_arrays(&self.on_right, &batch)?, + }; + let num_rows = selection + .as_ref() + .map_or_else(|| batch.num_rows(), |s| s.indices.len()); let valid_keys = if let Map::HashMap(_) = self.build_side.try_as_ready()?.left_data.map() { self.hashes_buffer.clear(); - self.hashes_buffer.resize(batch.num_rows(), 0); + self.hashes_buffer.resize(num_rows, 0); create_hashes( &keys_values, &self.random_state, @@ -773,11 +805,12 @@ impl HashJoinStream { }; self.join_metrics.input_batches.add(1); - self.join_metrics.input_rows.add(batch.num_rows()); + self.join_metrics.input_rows.add(num_rows); self.state = HashJoinStreamState::ProcessProbeBatch(ProcessProbeBatchState { batch, + selection, values: keys_values, valid_keys, offset: (0, None), @@ -799,9 +832,12 @@ impl HashJoinStream { let state = self.state.try_as_process_probe_batch_mut()?; let build_side = self.build_side.try_as_ready_mut()?; - self.join_metrics - .probe_hit_rate - .add_total(state.batch.num_rows()); + self.join_metrics.probe_hit_rate.add_total( + state + .selection + .as_ref() + .map_or_else(|| state.batch.num_rows(), |s| s.indices.len()), + ); let timer = self.join_metrics.join_time.timer(); @@ -898,6 +934,19 @@ impl HashJoinStream { .avg_fanout .add_total(distinct_right_indices_count); + // Equality checks used compact keys. Filters and output materialization + // use the original payload, so map back only after collision checking. + let right_indices = if let Some(selected) = &state.selection { + UInt32Array::from_iter_values( + right_indices + .values() + .iter() + .map(|&i| selected.indices.value(i as usize)), + ) + } else { + right_indices + }; + // apply join filter if exists let (left_indices, right_indices) = if let Some(filter) = &self.filter { apply_join_filter_to_indices( diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index 7c2e4951289a4..a17333730f9a9 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -219,6 +219,7 @@ datafusion.execution.coalesce_batches true datafusion.execution.collect_statistics true datafusion.execution.enable_ansi_mode false datafusion.execution.enable_file_stream_work_stealing true +datafusion.execution.enable_hash_join_probe_selection false datafusion.execution.enable_migration_aggregate true datafusion.execution.enable_nlj_coordinated_fallback true datafusion.execution.enable_recursive_ctes true @@ -380,6 +381,7 @@ datafusion.execution.coalesce_batches true When set to true, record batches will datafusion.execution.collect_statistics true Should DataFusion collect statistics when first creating a table. Has no effect after the table is created. Defaults to true. 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.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.enable_hash_join_probe_selection false Enable probe-side selection exchange for partitioned inner hash joins. Shares payload batches and copies only selected join keys before lookup. Requires simple column keys, an unordered hash repartition directly on the probe side, no dynamic filter, and an unlimited memory pool. Other plans retain the ordinary spill-capable repartition path. datafusion.execution.enable_migration_aggregate true Whether aggregation uses the implementation from the major refactor completed in the 56.0.0 release. When set to `false`, aggregation falls back to the implementation used before 55.0.0. The fallback exists only as a workaround for bugs in the new implementation and will be removed, together with this option, after the 56.0.0 release. See for details. datafusion.execution.enable_nlj_coordinated_fallback true Enables the memory-limited fallback for `NestedLoopJoinExec` join types that emit unmatched left rows in the final output (LEFT, LEFT SEMI, LEFT ANTI, LEFT MARK, FULL) when the right side has multiple partitions. This fallback coordinates per-chunk left state (visited bitmap and probe-thread counter) across all right-side partitions, which assumes every partition runs in the same process. Distributed engines that execute each output partition as an independent task (e.g. Ballista, datafusion-distributed) build a separate coordinator per task and poll only one partition, so the cross-partition counter never reaches zero and the fallback would stall. Such engines should set this to `false`: the coordinated fallback is then disabled for left-emitting multi-partition joins, which instead fail with a resource-exhaustion error under memory pressure rather than deadlocking. Single-partition and non-left-emitting joins are unaffected and always keep the fallback. datafusion.execution.enable_recursive_ctes true Should DataFusion support recursive CTEs diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 16750d79d750d..f909c02f2e16c 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -75,6 +75,7 @@ The following configuration settings are available: | datafusion.catalog.newlines_in_values | false | Specifies whether newlines in (quoted) CSV values are supported. This is the default value for `format.newlines_in_values` for `CREATE EXTERNAL TABLE` if not specified explicitly in the statement. Parsing newlines in quoted values may be affected by execution behaviour such as parallel file scanning. Setting this to `true` ensures that newlines in values are parsed successfully, which may reduce performance. | | datafusion.execution.batch_size | 8192 | Default batch size while creating new batches, it's especially useful for buffer-in-memory batches since creating tiny batches would result in too much metadata memory consumption | | datafusion.execution.perfect_hash_join_small_build_threshold | 1024 | A perfect hash join (see `HashJoinExec` for more details) will be considered if the range of keys (max - min) on the build side is < this threshold. This provides a fast path for joins with very small key ranges, bypassing the density check. Currently only supports cases where build_side.num_rows() < u32::MAX. Support for build_side.num_rows() >= u32::MAX will be added in the future. | +| datafusion.execution.enable_hash_join_probe_selection | false | Enable probe-side selection exchange for partitioned inner hash joins. Shares payload batches and copies only selected join keys before lookup. Requires simple column keys, an unordered hash repartition directly on the probe side, no dynamic filter, and an unlimited memory pool. Other plans retain the ordinary spill-capable repartition path. | | datafusion.execution.perfect_hash_join_min_key_density | 0.15 | The minimum required density of join keys on the build side to consider a perfect hash join (see `HashJoinExec` for more details). Density is calculated as: `(number of rows) / (max_key - min_key + 1)`. A perfect hash join may be used if the actual key density > this value. Currently only supports cases where build_side.num_rows() < u32::MAX. Support for build_side.num_rows() >= u32::MAX will be added in the future. | | datafusion.execution.coalesce_batches | true | When set to true, record batches will be examined between each operator and small batches will be coalesced into larger batches. This is helpful when there are highly selective filters or joins that could produce tiny output batches. The target batch size is determined by the configuration setting | | datafusion.execution.collect_statistics | true | Should DataFusion collect statistics when first creating a table. Has no effect after the table is created. Defaults to true. |