2010YOUY01 commented on code in PR #26227: URL: https://github.com/apache/datafusion/pull/26227#discussion_r4369992674
########## 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: [8293d23](https://github.com/apache/datafusion/pull/26227/commits/8293d23e4ed886bef451f8612d3b715b8e89c946) ########## 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: [8293d23](https://github.com/apache/datafusion/pull/26227/commits/8293d23e4ed886bef451f8612d3b715b8e89c946) ########## 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: [8293d23](https://github.com/apache/datafusion/pull/26227/commits/8293d23e4ed886bef451f8612d3b715b8e89c946) -- 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]
