jayzhan211 commented on code in PR #26227: URL: https://github.com/apache/datafusion/pull/26227#discussion_r4366525738
########## datafusion/physical-plan/src/joins/hash_join/build.rs: ########## @@ -0,0 +1,196 @@ +// 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. + +//! Build preparation and shared data for regular hash joins. +//! +//! Collection is shared across probe partitions in `CollectLeft` mode. Before +//! probing, each partition also waits for the join's dynamic filter to be ready. +//! The prepared result contains only the data and coordination regular joins need. + +use std::sync::Arc; + +use arrow::array::{BooleanBufferBuilder, UInt64Array}; +use arrow::buffer::BooleanBuffer; +use datafusion_common::Result; +use futures::future::poll_fn; +use parking_lot::Mutex; + +use super::exec::JoinBuildData; +use crate::joins::PartitionMode; +use crate::joins::hash_join::shared_bounds::{ + PartitionBounds, PartitionBuildData, SharedBuildAccumulator, +}; +use crate::joins::utils::{BuildProbeJoinMetrics, OnceFut}; + +/// [`JoinBuildData`] plus cross-partition coordination, only one partition is +/// responsible for final build-side unmatched output for Left/Full joins. +pub(super) struct JoinBuildState { + unmatched: Option<Mutex<UnmatchedRows>>, + // Keep the build lease last: its reservation also covers the bitmap. + pub(super) data: Arc<JoinBuildData>, +} + +struct UnmatchedRows { + visited: BooleanBufferBuilder, + remaining_probes: usize, +} + +impl JoinBuildState { + pub(super) fn new( + data: Arc<JoinBuildData>, + visited: Option<BooleanBufferBuilder>, + probe_partitions: usize, + ) -> Self { + Self { + unmatched: visited.map(|visited| { + Mutex::new(UnmatchedRows { + visited, + remaining_probes: probe_partitions, + }) + }), + data, + } + } + + /// Record matches after the ON filter has accepted them. + pub(super) fn record_matches(&self, indices: &UInt64Array) { + if let Some(unmatched) = &self.unmatched { + let mut unmatched = unmatched.lock(); + for index in indices.iter().flatten() { + unmatched.visited.set_bit(index as usize, true); + } + } + } + + /// Report actual probe EOF; a fetch cutoff must not call this. + pub(super) fn finish_probe(&self) -> Option<BooleanBuffer> { + let mut unmatched = self.unmatched.as_ref()?.lock(); + // Counting and snapshotting under the same lock ensures the last probe + // sees every sibling's matches and is the only one to emit unmatched rows. + unmatched.remaining_probes -= 1; + (unmatched.remaining_probes == 0).then(|| unmatched.visited.finish_cloned()) + } +} + +/// Collecting the build side data of hash join. +/// +/// It handles coordination/syncrhonization among all parallel paartitions for: Review Comment: nit: a few typos are still here: "syncrhonization" and "paartitions" on this line, and "coordiation" on line 99. ```diff -/// It handles coordination/syncrhonization among all parallel paartitions for: +/// It handles coordination/synchronization among all parallel partitions for: ``` ```diff - // Shared structures for coordiation among partitions. + // Shared structures for coordination among partitions. ``` ########## datafusion/physical-plan/src/joins/hash_join/hash_join_table.rs: ########## @@ -0,0 +1,640 @@ +// 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. + +//! A regular join's build table, probing, and output materialization. +//! +//! [`HashJoinTable`] accepts probe batches and returns final output batches. +//! Its caller does not manage row indices, lookup cursors, match bitmaps, or +//! output buffering. See the type's example for the input/output contract. + +use std::sync::Arc; + +use arrow::array::{ArrayRef, UInt32Array, UInt64Array}; +use arrow::buffer::{BooleanBuffer, NullBuffer, ScalarBuffer}; +use arrow::datatypes::SchemaRef; +use arrow::record_batch::RecordBatch; +use datafusion_common::hash_utils::RandomState; +use datafusion_common::{ + JoinSide, JoinType, NullEquality, Result, assert_or_internal_err, internal_err, +}; +use datafusion_physical_expr::PhysicalExprRef; +use datafusion_physical_expr_common::utils::evaluate_expressions_to_arrays; + +use super::build::JoinBuildState; +use super::stream::lookup_join_hashmap; +use crate::coalesce::{LimitedBatchCoalescer, PushBatchStatus}; +use crate::hash_utils::create_hashes; +use crate::joins::Map; +use crate::joins::join_hash_map::ProbeOffset; +use crate::joins::utils::{ + BuildProbeJoinMetrics, ColumnIndex, JoinFilter, JoinKeyComparator, + adjust_indices_by_join_type, apply_join_filter_to_indices, + build_batch_empty_build_side, build_batch_from_indices, matchable_join_keys, +}; + +/// Join semantics and output settings for one probe partition. +pub(super) struct HashJoinTableConfig { + /// Join type, which determines whether to keep unmatched build or probe rows. + pub(super) kind: RegularJoinKind, + /// Output schema after projection. + pub(super) schema: SchemaRef, + /// Probe input schema, used for NULL columns in unmatched build output. + pub(super) probe_schema: SchemaRef, + /// For each output column, its source side and column index. + pub(super) column_indices: Vec<ColumnIndex>, + /// Probe equality-key expressions, paired with the build keys by position. + pub(super) on_right: Vec<PhysicalExprRef>, + /// Extra `ON` condition checked after key equality, e.g. `l.x > r.y`. + /// Its input columns are independent of the output projection. + pub(super) filter: Option<JoinFilter>, + /// Hashing state matching the build side, so equal keys hash alike. + pub(super) random_state: RandomState, + /// Whether NULL join keys match other NULLs. + pub(super) null_equality: NullEquality, + /// Target batch size for internal buffering, from configuration. + pub(super) batch_size: usize, + /// Preserve probe order when inserting unmatched rows for a right join. + pub(super) right_side_ordered: bool, + /// Maximum output rows from this table; `None` means no limit. + pub(super) fetch: Option<usize>, +} + +/// Hash join's build-side hash index and state for processing the full join logic +/// for one probe-side batch. It vectorizes the computation steps while keeping +/// computation incremental for memory efficiency: only `batch_size` rows of +/// intermediate batches/indices are kept at a time. See the `Incremental Computation` +/// section for details. +/// +/// # Architecture +/// +/// The key conceptual components inside this struct are: +/// +/// - Build-side batch: the full build side, buffered and concatenated. +/// - Probe-side batch: a single batch for the current iteration. +/// - Hash index: a hash map with the scheme `join key -> indices in build-side batch`. +/// - Runtime state (e.g., the current `(build_index, probe_index)` for incremental +/// processing). +/// - Output buffer for incremental processing. +/// +/// They work together to perform the full join logic of +/// `build_side join one_probe_batch`. +/// +/// # Hash Index +/// +/// Conceptually, the hash index maps +/// `join equality key -> row indices in buffered build-side batches`, for example: +/// +/// ```text +/// Build keys: [1, 2, 2] +/// +/// -- Conceptual hash index +/// Key -> Build row indices +/// 1 -> [0] +/// 2 -> [1, 2] +/// ``` +/// +/// # Incremental Computation +/// +/// It keeps internal state to support incremental computation and coalesce small +/// outputs. Output is aligned to the configured `batch_size`, like: +/// +/// ```text +/// output batch 1: 8192 rows +/// output batch 2: 8192 rows +/// ... +/// last output batch: 1000 rows // Output ends; all remaining rows. +/// ``` +/// +/// - Incremental Computation: even if each probe row has many matches, it will only +/// materialize up to `batch_size` output rows per iteration, for memory efficiency. +/// - Coalescing: if a probe batch produces fewer than `batch_size` output rows, +/// polling will not produce output (unless at the end). The caller must continue +/// providing probe-side batches until there is an output batch of `batch_size` rows. +/// +/// # Key API walkthrough +/// +/// Below are the key APIs; see their comments for details. The following example +/// demonstrates how they are used together in the join logic. +/// - [`Self::push_batch`] +/// - [`Self::next_output_batch`] +/// - [`Self::finish_probe`] +/// +/// For a full join with one probe partition, build keys `[1, 2, 2]`, and +/// `batch_size = 1`: +/// +/// ```text +/// next_output_batch() -> NeedInput +/// push_batch([2]) +/// next_output_batch() -> Output([(2, 2)]) +/// next_output_batch() -> Output([(2, 2)]) +/// next_output_batch() -> NeedInput // This batch is done; the stream may have more. +/// +/// push_batch([3]) +/// next_output_batch() -> Output([(NULL, 3)]) +/// next_output_batch() -> NeedInput +/// +/// // The caller reads the probe stream and observes EOF. +/// finish_probe() +/// next_output_batch() -> Output([(1, NULL)]) +/// next_output_batch() -> Finished // All output from this table is drained. +/// ``` +pub(super) struct HashJoinTable { + context: HashJoinTableConfig, + /// Counters and timings for this join partition. + metrics: BuildProbeJoinMetrics, + + // =================== + // Static data buffers + // =================== + build: Arc<JoinBuildState>, + /// Prepared input, unchanged while this batch is being joined. + probe: Option<ProbeBatch>, + + // ============= + // Runtime state + // ============= + // + // Track lookup progress, such as the last build/probe row examined, + // to produce output incrementally. + /// Current execution phase. + phase: Phase, + /// Where to resume candidate lookup; `None` means this batch is exhausted. + next_offset: Option<ProbeOffset>, Review Comment: nit: `next_offset`, `joined_probe_idx`, `matched_probe_idx` and `key_comparator` all describe progress through the current probe batch, and they're reset together by hand in `push_batch`, `probe_chunk` and `release_probe`. Moving them into `ProbeBatch` makes dropping the batch reset them, so those three blocks go away. ```diff struct ProbeBatch { input: RecordBatch, keys: Vec<ArrayRef>, valid_keys: Option<NullBuffer>, + next_offset: Option<ProbeOffset>, + joined_probe_idx: Option<usize>, + matched_probe_idx: Option<u32>, + key_comparator: Option<JoinKeyComparator>, } ``` ```diff - let Some(probe) = &self.probe else { + let Some(probe) = &mut self.probe else { return Ok(None); }; - let Some(offset) = self.next_offset.take() else { + let Some(offset) = probe.next_offset.take() else { self.probe = None; - self.key_comparator = None; - self.joined_probe_idx = None; - self.matched_probe_idx = None; return Ok(None); }; ``` The rest is renaming `self.<field>` to `probe.<field>` in `probe_chunk`, and deleting the same four resets from `push_batch` and `release_probe`. ########## datafusion/physical-plan/src/joins/hash_join/regular_join.rs: ########## @@ -0,0 +1,182 @@ +// 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. + +//! State machine implementation for the regular hash join execution path. (Inner/ +//! Left/Right/Full join types are regular join) +//! +//! HashJoinExec is getting incrementally migrated to separated streams, see issue +//! for detail: +//! - <https://github.com/apache/datafusion/issues/26361> + +use std::sync::Arc; + +use arrow::datatypes::SchemaRef; +use arrow::record_batch::RecordBatch; +use datafusion_common::{DataFusionError, Result, internal_err}; +use datafusion_execution::{TryEmitter, async_try_stream}; +use futures::TryStreamExt; + +use super::build::HashJoinBuild; +use super::hash_join_table::{HashJoinTable, HashJoinTableConfig, JoinStep}; +use crate::SendableRecordBatchStream; +use crate::joins::utils::BuildProbeJoinMetrics; +use crate::metrics::BaselineMetrics; +use crate::stream::{ObservedStream, RecordBatchStreamAdapter}; + +/// Implementation for regular (Inner/Left/Right/Full) hash join. +/// +/// See [`Self::into_stream`] for execution stages and transitions. +pub(super) struct RegularHashJoinStream { + schema: SchemaRef, + metrics: BaselineMetrics, + stage: ExecutionStage, +} + +/// Execution stages described in [`RegularHashJoinStream::into_stream`]. +enum ExecutionStage { + Building(Building), + Probing(Probing), + Outputting(HashJoinTable), +} + +struct Building { + build: HashJoinBuild, + probe: SendableRecordBatchStream, + config: HashJoinTableConfig, + metrics: BuildProbeJoinMetrics, +} + +struct Probing { + probe: SendableRecordBatchStream, + table: HashJoinTable, +} + +impl RegularHashJoinStream { + pub(super) fn new( + build: HashJoinBuild, + probe: SendableRecordBatchStream, + config: HashJoinTableConfig, + metrics: BuildProbeJoinMetrics, + ) -> Self { + Self { + schema: Arc::clone(&config.schema), + metrics: metrics.baseline.clone(), + stage: ExecutionStage::Building(Building { + build, + probe, + config, + metrics, + }), + } + } + + /// Entry point for the regular hash join execution stages. + /// + /// ```text + /// Building --> Probing --> Outputting --> end + /// ``` + /// + /// # Stages + /// + /// - Building: Load the build-side data, and handle cross-partition synchronization, + /// see [`HashJoinBuild`] for different physical strategies + /// - Probing: Joining probe batch with build data incrementally, until probe + /// side input is exhausted. + /// - Outputting: Actually joining the probe batch and buffered build side, and Review Comment: nit: by the time the stream reaches `Outputting`, probing is done. This stage only drains buffered output and emits the unmatched build rows for left/full joins, so the description reads like the `Probing` stage. ```diff - /// - Outputting: Actually joining the probe batch and buffered build side, and - /// produce final output. + /// - Outputting: Drain buffered output, then emit unmatched build rows for + /// left/full joins. ``` ########## datafusion/physical-plan/src/joins/hash_join/hash_join_table.rs: ########## @@ -0,0 +1,640 @@ +// 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. + +//! A regular join's build table, probing, and output materialization. +//! +//! [`HashJoinTable`] accepts probe batches and returns final output batches. +//! Its caller does not manage row indices, lookup cursors, match bitmaps, or +//! output buffering. See the type's example for the input/output contract. + +use std::sync::Arc; + +use arrow::array::{ArrayRef, UInt32Array, UInt64Array}; +use arrow::buffer::{BooleanBuffer, NullBuffer, ScalarBuffer}; +use arrow::datatypes::SchemaRef; +use arrow::record_batch::RecordBatch; +use datafusion_common::hash_utils::RandomState; +use datafusion_common::{ + JoinSide, JoinType, NullEquality, Result, assert_or_internal_err, internal_err, +}; +use datafusion_physical_expr::PhysicalExprRef; +use datafusion_physical_expr_common::utils::evaluate_expressions_to_arrays; + +use super::build::JoinBuildState; +use super::stream::lookup_join_hashmap; +use crate::coalesce::{LimitedBatchCoalescer, PushBatchStatus}; +use crate::hash_utils::create_hashes; +use crate::joins::Map; +use crate::joins::join_hash_map::ProbeOffset; +use crate::joins::utils::{ + BuildProbeJoinMetrics, ColumnIndex, JoinFilter, JoinKeyComparator, + adjust_indices_by_join_type, apply_join_filter_to_indices, + build_batch_empty_build_side, build_batch_from_indices, matchable_join_keys, +}; + +/// Join semantics and output settings for one probe partition. +pub(super) struct HashJoinTableConfig { + /// Join type, which determines whether to keep unmatched build or probe rows. + pub(super) kind: RegularJoinKind, + /// Output schema after projection. + pub(super) schema: SchemaRef, + /// Probe input schema, used for NULL columns in unmatched build output. + pub(super) probe_schema: SchemaRef, + /// For each output column, its source side and column index. + pub(super) column_indices: Vec<ColumnIndex>, + /// Probe equality-key expressions, paired with the build keys by position. + pub(super) on_right: Vec<PhysicalExprRef>, + /// Extra `ON` condition checked after key equality, e.g. `l.x > r.y`. + /// Its input columns are independent of the output projection. + pub(super) filter: Option<JoinFilter>, + /// Hashing state matching the build side, so equal keys hash alike. + pub(super) random_state: RandomState, + /// Whether NULL join keys match other NULLs. + pub(super) null_equality: NullEquality, + /// Target batch size for internal buffering, from configuration. + pub(super) batch_size: usize, + /// Preserve probe order when inserting unmatched rows for a right join. + pub(super) right_side_ordered: bool, + /// Maximum output rows from this table; `None` means no limit. + pub(super) fetch: Option<usize>, +} + +/// Hash join's build-side hash index and state for processing the full join logic +/// for one probe-side batch. It vectorizes the computation steps while keeping +/// computation incremental for memory efficiency: only `batch_size` rows of +/// intermediate batches/indices are kept at a time. See the `Incremental Computation` +/// section for details. +/// +/// # Architecture +/// +/// The key conceptual components inside this struct are: +/// +/// - Build-side batch: the full build side, buffered and concatenated. +/// - Probe-side batch: a single batch for the current iteration. +/// - Hash index: a hash map with the scheme `join key -> indices in build-side batch`. +/// - Runtime state (e.g., the current `(build_index, probe_index)` for incremental +/// processing). +/// - Output buffer for incremental processing. +/// +/// They work together to perform the full join logic of +/// `build_side join one_probe_batch`. +/// +/// # Hash Index +/// +/// Conceptually, the hash index maps +/// `join equality key -> row indices in buffered build-side batches`, for example: +/// +/// ```text +/// Build keys: [1, 2, 2] +/// +/// -- Conceptual hash index +/// Key -> Build row indices +/// 1 -> [0] +/// 2 -> [1, 2] +/// ``` +/// +/// # Incremental Computation +/// +/// It keeps internal state to support incremental computation and coalesce small +/// outputs. Output is aligned to the configured `batch_size`, like: +/// +/// ```text +/// output batch 1: 8192 rows +/// output batch 2: 8192 rows +/// ... +/// last output batch: 1000 rows // Output ends; all remaining rows. +/// ``` +/// +/// - Incremental Computation: even if each probe row has many matches, it will only +/// materialize up to `batch_size` output rows per iteration, for memory efficiency. +/// - Coalescing: if a probe batch produces fewer than `batch_size` output rows, +/// polling will not produce output (unless at the end). The caller must continue +/// providing probe-side batches until there is an output batch of `batch_size` rows. +/// +/// # Key API walkthrough +/// +/// Below are the key APIs; see their comments for details. The following example +/// demonstrates how they are used together in the join logic. +/// - [`Self::push_batch`] +/// - [`Self::next_output_batch`] +/// - [`Self::finish_probe`] +/// +/// For a full join with one probe partition, build keys `[1, 2, 2]`, and +/// `batch_size = 1`: +/// +/// ```text +/// next_output_batch() -> NeedInput +/// push_batch([2]) +/// next_output_batch() -> Output([(2, 2)]) +/// next_output_batch() -> Output([(2, 2)]) +/// next_output_batch() -> NeedInput // This batch is done; the stream may have more. +/// +/// push_batch([3]) +/// next_output_batch() -> Output([(NULL, 3)]) +/// next_output_batch() -> NeedInput +/// +/// // The caller reads the probe stream and observes EOF. +/// finish_probe() +/// next_output_batch() -> Output([(1, NULL)]) +/// next_output_batch() -> Finished // All output from this table is drained. +/// ``` +pub(super) struct HashJoinTable { + context: HashJoinTableConfig, + /// Counters and timings for this join partition. + metrics: BuildProbeJoinMetrics, + + // =================== + // Static data buffers + // =================== + build: Arc<JoinBuildState>, + /// Prepared input, unchanged while this batch is being joined. + probe: Option<ProbeBatch>, + + // ============= + // Runtime state + // ============= + // + // Track lookup progress, such as the last build/probe row examined, + // to produce output incrementally. + /// Current execution phase. + phase: Phase, + /// Where to resume candidate lookup; `None` means this batch is exhausted. + next_offset: Option<ProbeOffset>, + /// Last probe row passing the full ON condition. Unmatched probe rows + /// are considered after this row, even when its matches span chunks. + joined_probe_idx: Option<usize>, + /// Last equality match, before the ON filter, to count each probe row once + /// in hit-rate and fanout metrics across lookup chunks. + matched_probe_idx: Option<u32>, + + // ================ + // Lookup utilities + // ================ + /// Cached equality checks bound to the build and current probe key arrays. + key_comparator: Option<JoinKeyComparator>, + /// Lookup allocations reused across chunks and input batches. + hashes: Vec<u64>, + + // ============== + // Output buffers + // ============== + // + // Output construction has two steps: + // 1. Probe the hash index for matching row-index pairs. + // 2. Materialize the matched rows, evaluate the residual join filter, and build + // the final joined output. + // + // For `t1 JOIN t2 ON t1.v1 = t2.v1 AND t1.v2 > t2.v2`: + // - Step 1: Row-index pairs satisfying `t1.v1 = t2.v1`. + // - Step 2: Joined rows satisfying `t1.v2 > t2.v2`. + // + // Separate buffers are used for these two steps. + + // Output buffers for step 1. + /// Matched row indices into [`Self::build`]'s data batch. + build_indices: Vec<u64>, + /// Matched row indices into [`Self::probe`]'s data batch. + probe_indices: Vec<u32>, + /// Retain array-map index buffers through materialization, then reclaim + /// them for the next lookup without copying. + array_map_buffers: Option<(ScalarBuffer<u64>, ScalarBuffer<u32>)>, + + // Output buffer for step 2. + /// Materialized output retained across probe batches and join phases. + output: LimitedBatchCoalescer, +} + +/// Prepared input for one probe batch; all mutable join progress lives in +/// [`HashJoinTable`]. +struct ProbeBatch { + input: RecordBatch, + keys: Vec<ArrayRef>, + valid_keys: Option<NullBuffer>, +} + +/// The next action for the caller of [`HashJoinTable::next_output_batch`]. +pub(super) enum JoinStep { + /// A joined output batch is ready. + Output(RecordBatch), + /// The caller should supply another probe batch or report EOF with + /// [`HashJoinTable::finish_probe`]. + NeedInput, + /// All output is drained, possibly before probe EOF. Stop reading input. + Finished, +} + +/// Join work remaining. The output coalescer outlives every phase. +enum Phase { + Probing, + /// Only the last probe partition receives this final match snapshot. + /// Scan it in chunks without holding the shared bitmap lock. + UnmatchedBuild { + visited: BooleanBuffer, + next_row: usize, + empty_probe: RecordBatch, + }, + /// No more rows will be produced; flush and drain buffered output. + DrainingOutput, +} + +impl HashJoinTable { + pub(super) fn new( + build: Arc<JoinBuildState>, + context: HashJoinTableConfig, + metrics: BuildProbeJoinMetrics, + ) -> Self { + // An empty map may still have build rows with NULL keys. Left/full + // joins must retain those rows for final null extension. + let can_skip_probe = (build.data.batch.num_rows() == 0 Review Comment: nit: these two conditions repeat what `JoinType::empty_build_side_produces_empty_result` and `JoinType::empty_map_produces_empty_result` already encode, and the legacy stream uses those helpers. Using them here keeps a single source of truth when existence joins move to this path: ```rs let join_type = context.kind.join_type(); let can_skip_probe = (build.data.batch.num_rows() == 0 && join_type.empty_build_side_produces_empty_result()) || (build.data.map.is_empty() && join_type.empty_map_produces_empty_result()); ``` ########## datafusion/common/src/config.rs: ########## @@ -980,6 +980,12 @@ config_namespace! { /// See <https://github.com/apache/datafusion/issues/22710> for details. pub enable_migration_aggregate: bool, default = true + /// Whether inner, left outer, right outer, and full outer hash joins use Review Comment: nit: `enable_migration_aggregate` just above says the fallback is only a workaround, says when it will be removed, and links its tracking issue. Matching that here tells users not to depend on the option. `configs.md` and `information_schema.slt` would need regenerating after this change. ```diff /// Whether inner, left outer, right outer, and full outer hash joins use /// the refactored implementation. When set to `false`, these joins use /// the existing implementation. Other join types always use the existing /// implementation. + /// + /// The fallback exists only as a workaround for bugs in the new + /// implementation and will be removed, together with this option, once + /// the migration is complete. + /// + /// See <https://github.com/apache/datafusion/issues/26361> for details. pub enable_migration_hash_join: bool, default = true ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
