2010YOUY01 commented on code in PR #26227: URL: https://github.com/apache/datafusion/pull/26227#discussion_r4369989526
########## 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: good point, updated in [fb2e50e](https://github.com/apache/datafusion/pull/26227/commits/fb2e50e5b80b7eae8c0b80fbb2f36b7c3099935f) -- 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]
