datafusion_physical_plan/joins/hash_join/

exec.rs

// 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::collections::HashSet;
use std::fmt;
use std::mem::size_of;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, OnceLock};
use std::vec;

use crate::execution_plan::{
    EmissionType, boundedness_from_children, has_same_children_properties,
    plan_contains_expression_id, stub_properties,
};
use crate::filter_pushdown::{
    ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase,
    FilterPushdownPropagation,
};
use crate::joins::Map;
use crate::joins::array_map::ArrayMap;
use crate::joins::hash_join::inlist_builder::build_struct_inlist_values;
use crate::joins::hash_join::shared_bounds::{
    ColumnBounds, PartitionBounds, PushdownStrategy, SharedBuildAccumulator,
};
use crate::joins::hash_join::stream::{
    BuildSide, BuildSideInitialState, HashJoinStream, HashJoinStreamState,
};
use crate::joins::join_hash_map::{JoinHashMapU32, JoinHashMapU64};
use crate::joins::utils::{
    OnceAsync, OnceFut, asymmetric_join_output_partitioning, reorder_output_after_swap,
    swap_join_projection, update_hash,
};
use crate::joins::{JoinOn, JoinOnRef, PartitionMode, SharedBitmapBuilder};
use crate::metrics::{Count, MetricBuilder, MetricCategory};
use crate::projection::{
    EmbeddedProjection, JoinData, ProjectionExec, try_embed_projection,
    try_pushdown_through_join_with_column_indices,
};
use crate::repartition::REPARTITION_RANDOM_STATE;
use crate::statistics::{ChildStats, StatisticsArgs};
use crate::{
    ChildrenPropertiesMode, ExecutionPlanProperties, ReplaceChildrenOptions,
    validate_child_count,
};
use crate::{
    DisplayAs, DisplayFormatType, Distribution, ExecutionPlan,
    InputDistributionRequirements, Partitioning, PlanProperties,
    SendableRecordBatchStream, Statistics,
    common::can_project,
    joins::utils::{
        BuildProbeJoinMetrics, ColumnIndex, JoinFilter, JoinHashMapType,
        build_join_schema, check_join_is_valid, estimate_join_statistics,
        need_produce_result_in_final, symmetric_join_output_partitioning,
    },
    metrics::{ExecutionPlanMetricsSet, MetricsSet},
};

use arrow::array::{ArrayRef, BooleanBufferBuilder};
use arrow::compute::concat_batches;
use arrow::datatypes::SchemaRef;
use arrow::record_batch::RecordBatch;
use arrow::util::bit_util;
use arrow_schema::{DataType, Schema};
use datafusion_common::config::ConfigOptions;
use datafusion_common::tree_node::TreeNodeRecursion;
use datafusion_common::utils::memory::{RecordBatchMemoryCounter, estimate_memory_size};
use datafusion_common::{
    JoinSide, JoinType, NullEquality, Result, assert_or_internal_err, internal_err,
    plan_err, project_schema,
};
use datafusion_execution::TaskContext;
use datafusion_execution::memory_pool::{MemoryConsumer, MemoryReservation};
use datafusion_expr::Accumulator;
use datafusion_functions_aggregate_common::min_max::{MaxAccumulator, MinAccumulator};
use datafusion_physical_expr::equivalence::{
    ProjectionMapping, join_equivalence_properties,
};
use datafusion_physical_expr::expressions::{Column, DynamicFilterPhysicalExpr, lit};
use datafusion_physical_expr::projection::{ProjectionRef, combine_projections};
use datafusion_physical_expr::{PhysicalExpr, PhysicalExprRef};

use datafusion_common::hash_utils::RandomState;
use datafusion_physical_expr_common::physical_expr::fmt_sql;
use datafusion_physical_expr_common::utils::evaluate_expressions_to_arrays;
use futures::TryStreamExt;
use parking_lot::Mutex;

use super::partitioned_hash_eval::SeededRandomState;

/// Hard-coded seed to ensure hash values from the hash join differ from `RepartitionExec`, avoiding collisions.
pub(crate) const HASH_JOIN_SEED: SeededRandomState =
    SeededRandomState::with_seed(12210250226015887276);

const ARRAY_MAP_CREATED_COUNT_METRIC_NAME: &str = "array_map_created_count";

#[expect(clippy::too_many_arguments)]
fn try_create_array_map(
    bounds: &Option<PartitionBounds>,
    schema: &SchemaRef,
    batches: &[RecordBatch],
    on_left: &[PhysicalExprRef],
    reservation: &mut MemoryReservation,
    perfect_hash_join_small_build_threshold: usize,
    perfect_hash_join_min_key_density: f64,
    null_equality: NullEquality,
) -> Result<Option<(ArrayMap, RecordBatch, Vec<ArrayRef>)>> {
    if on_left.len() != 1 {
        return Ok(None);
    }

    if null_equality == NullEquality::NullEqualsNull {
        for batch in batches.iter() {
            let arrays = evaluate_expressions_to_arrays(on_left, batch)?;
            if arrays[0].null_count() > 0 {
                return Ok(None);
            }
        }
    }

    let (min_val, max_val) = if let Some(bounds) = bounds {
        let (min_val, max_val) = if let Some(cb) = bounds.get_column_bounds(0) {
            (cb.min.clone(), cb.max.clone())
        } else {
            return Ok(None);
        };

        if min_val.is_null() || max_val.is_null() {
            return Ok(None);
        }

        if min_val > max_val {
            return internal_err!("min_val>max_val");
        }

        if let Some((mi, ma)) =
            ArrayMap::key_to_u64(&min_val).zip(ArrayMap::key_to_u64(&max_val))
        {
            (mi, ma)
        } else {
            return Ok(None);
        }
    } else {
        return Ok(None);
    };

    let range = ArrayMap::calculate_range(min_val, max_val);
    let num_row: usize = batches.iter().map(|x| x.num_rows()).sum();

    // TODO: support create ArrayMap<u64>
    if num_row >= u32::MAX as usize {
        return Ok(None);
    }

    // When the key range spans the full integer domain (e.g. i64::MIN to i64::MAX),
    // range is u64::MAX and `range + 1` below would overflow.
    if range == usize::MAX as u64 {
        return Ok(None);
    }

    let dense_ratio = (num_row as f64) / ((range + 1) as f64);

    if range >= perfect_hash_join_small_build_threshold as u64
        && dense_ratio <= perfect_hash_join_min_key_density
    {
        return Ok(None);
    }

    let mem_size = ArrayMap::estimate_memory_size(min_val, max_val, num_row);
    reservation.try_grow(mem_size)?;

    let batch = concat_batches(schema, batches)?;
    let left_values = evaluate_expressions_to_arrays(on_left, &batch)?;

    let array_map = ArrayMap::try_new(&left_values[0], min_val, max_val)?;

    Ok(Some((array_map, batch, left_values)))
}

/// HashTable and input data for the left (build side) of a join
pub(super) struct JoinLeftData {
    /// The hash table with indices into `batch`
    /// Arc is used to allow sharing with SharedBuildAccumulator for hash map pushdown
    pub(super) map: Arc<Map>,
    /// The input rows for the build side
    batch: RecordBatch,
    /// The build side on expressions values
    values: Vec<ArrayRef>,
    /// Shared bitmap builder for visited left indices
    visited_indices_bitmap: SharedBitmapBuilder,
    /// Counter of running probe-threads, potentially
    /// able to update `visited_indices_bitmap`
    probe_threads_counter: AtomicUsize,
    /// We need to keep this field to maintain accurate memory accounting, even though we don't directly use it.
    /// Without holding onto this reservation, the recorded memory usage would become inconsistent with actual usage.
    /// This could hide potential out-of-memory issues, especially when upstream operators increase their memory consumption.
    /// The MemoryReservation ensures proper tracking of memory resources throughout the join operation's lifecycle.
    _reservation: MemoryReservation,
    /// Bounds computed from the build side for dynamic filter pushdown.
    /// If the partition is empty (no rows) this will be None.
    /// If the partition has some rows this will be Some with the bounds for each join key column.
    pub(super) bounds: Option<PartitionBounds>,
    /// Membership testing strategy for filter pushdown
    /// Contains either InList values for small build sides or hash table reference for large build sides
    pub(super) membership: PushdownStrategy,
    /// Shared atomic flag indicating if any probe partition saw data (for null-aware anti joins)
    /// This is shared across all probe partitions to provide global knowledge
    pub(super) probe_side_non_empty: AtomicBool,
    /// Shared atomic flag indicating if any probe partition saw NULL in join keys (for null-aware anti joins)
    pub(super) probe_side_has_null: AtomicBool,
}

impl JoinLeftData {
    /// return a reference to the map
    pub(super) fn map(&self) -> &Map {
        &self.map
    }

    /// returns a reference to the build side batch
    pub(super) fn batch(&self) -> &RecordBatch {
        &self.batch
    }

    /// Returns `true` if the build side physically contains rows.
    ///
    /// This is distinct from [`Self::has_matchable_build_rows`]: a build side
    /// can hold rows while its hash map is empty (see that method).
    pub(super) fn has_build_rows(&self) -> bool {
        self.batch().num_rows() > 0
    }

    /// Returns `true` if the build-side hash map has any matchable entries.
    ///
    /// Under [`NullEquality::NullEqualsNothing`] build rows whose join key is
    /// NULL are omitted from the map, so this can be `false` even when
    /// [`Self::has_build_rows`] is `true`.
    pub(super) fn has_matchable_build_rows(&self) -> bool {
        !self.map().is_empty()
    }

    /// returns a reference to the build side expressions values
    pub(super) fn values(&self) -> &[ArrayRef] {
        &self.values
    }

    /// returns a reference to the visited indices bitmap
    pub(super) fn visited_indices_bitmap(&self) -> &SharedBitmapBuilder {
        &self.visited_indices_bitmap
    }

    /// returns a reference to the InList values for filter pushdown
    pub(super) fn membership(&self) -> &PushdownStrategy {
        &self.membership
    }

    /// Decrements the counter of running threads, and returns `true`
    /// if caller is the last running thread
    pub(super) fn report_probe_completed(&self) -> bool {
        self.probe_threads_counter.fetch_sub(1, Ordering::Relaxed) == 1
    }
}

/// Helps to build [`HashJoinExec`].
///
/// Builder can be created from an existing [`HashJoinExec`] using [`From::from`].
/// In this case, all its fields are inherited. If a field that affects the node's
/// properties is modified, they will be automatically recomputed during the build.
///
/// # Adding setters
///
/// When adding a new setter, it is necessary to ensure that the `preserve_properties`
/// flag is set to false if modifying the field requires a recomputation of the plan's
/// properties.
///
pub struct HashJoinExecBuilder {
    exec: HashJoinExec,
    preserve_properties: bool,
}

impl HashJoinExecBuilder {
    /// Make a new [`HashJoinExecBuilder`].
    pub fn new(
        left: Arc<dyn ExecutionPlan>,
        right: Arc<dyn ExecutionPlan>,
        on: Vec<(PhysicalExprRef, PhysicalExprRef)>,
        join_type: JoinType,
    ) -> Self {
        Self {
            exec: HashJoinExec {
                left,
                right,
                on,
                filter: None,
                join_type,
                left_fut: Default::default(),
                random_state: HASH_JOIN_SEED,
                mode: PartitionMode::Auto,
                fetch: None,
                metrics: ExecutionPlanMetricsSet::new(),
                projection: None,
                column_indices: vec![],
                null_equality: NullEquality::NullEqualsNothing,
                null_aware: false,
                dynamic_filter: None,
                // Will be computed at when plan will be built.
                cache: stub_properties(),
                join_schema: Arc::new(Schema::empty()),
            },
            // As `exec` is initialized with stub properties,
            // they will be properly computed when plan will be built.
            preserve_properties: false,
        }
    }

    /// Set join type.
    pub fn with_type(mut self, join_type: JoinType) -> Self {
        self.exec.join_type = join_type;
        self.preserve_properties = false;
        self
    }

    /// Set projection from the vector.
    pub fn with_projection(self, projection: Option<Vec<usize>>) -> Self {
        self.with_projection_ref(projection.map(Into::into))
    }

    /// Set projection from the shared reference.
    pub fn with_projection_ref(mut self, projection: Option<ProjectionRef>) -> Self {
        self.exec.projection = projection;
        self.preserve_properties = false;
        self
    }

    /// Set optional filter.
    pub fn with_filter(mut self, filter: Option<JoinFilter>) -> Self {
        self.exec.filter = filter;
        self
    }

    /// Set expressions to join on.
    pub fn with_on(mut self, on: Vec<(PhysicalExprRef, PhysicalExprRef)>) -> Self {
        self.exec.on = on;
        self.preserve_properties = false;
        self
    }

    /// Set partition mode.
    pub fn with_partition_mode(mut self, mode: PartitionMode) -> Self {
        self.exec.mode = mode;
        self.preserve_properties = false;
        self
    }

    /// Set null equality property.
    pub fn with_null_equality(mut self, null_equality: NullEquality) -> Self {
        self.exec.null_equality = null_equality;
        self
    }

    /// Set null aware property.
    pub fn with_null_aware(mut self, null_aware: bool) -> Self {
        self.exec.null_aware = null_aware;
        self
    }

    /// Set fetch property.
    pub fn with_fetch(mut self, fetch: Option<usize>) -> Self {
        self.exec.fetch = fetch;
        self
    }

    /// Require to recompute plan properties.
    pub fn recompute_properties(mut self) -> Self {
        self.preserve_properties = false;
        self
    }

    /// Replace children.
    pub fn with_new_children(
        mut self,
        mut children: Vec<Arc<dyn ExecutionPlan>>,
    ) -> Result<Self> {
        assert_or_internal_err!(
            children.len() == 2,
            "wrong number of children passed into `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);
        Ok(self)
    }

    /// Reset runtime state.
    pub fn reset_state(mut self) -> Self {
        self.exec.left_fut = Default::default();
        self.exec.dynamic_filter = None;
        self.exec.metrics = ExecutionPlanMetricsSet::new();
        self
    }

    /// Build result as a dyn execution plan.
    pub fn build_exec(self) -> Result<Arc<dyn ExecutionPlan>> {
        self.build().map(|p| Arc::new(p) as _)
    }

    /// Build resulting execution plan.
    pub fn build(self) -> Result<HashJoinExec> {
        let Self {
            exec,
            preserve_properties,
        } = self;

        // Validate null_aware flag
        if exec.null_aware {
            let join_type = exec.join_type();
            if !matches!(join_type, JoinType::LeftAnti) {
                return plan_err!(
                    "null_aware can only be true for LeftAnti joins, got {join_type}"
                );
            }
            let on = exec.on();
            if on.len() != 1 {
                return plan_err!(
                    "null_aware anti join only supports single column join key, got {} columns",
                    on.len()
                );
            }
        }

        if preserve_properties {
            return Ok(exec);
        }

        let HashJoinExec {
            left,
            right,
            on,
            filter,
            join_type,
            left_fut,
            random_state,
            mode,
            metrics,
            projection,
            null_equality,
            null_aware,
            dynamic_filter,
            fetch,
            // Recomputed.
            join_schema: _,
            column_indices: _,
            cache: _,
        } = exec;

        let left_schema = left.schema();
        let right_schema = right.schema();
        if on.is_empty() {
            return plan_err!("On constraints in HashJoinExec should be non-empty");
        }

        check_join_is_valid(&left_schema, &right_schema, &on)?;
        let (join_schema, column_indices) =
            build_join_schema(&left_schema, &right_schema, &join_type);

        let join_schema = Arc::new(join_schema);

        // Check if the projection is valid.
        can_project(&join_schema, projection.as_deref())?;

        let cache = HashJoinExec::compute_properties(
            &left,
            &right,
            &join_schema,
            join_type,
            &on,
            mode,
            projection.as_deref(),
        )?;

        Ok(HashJoinExec {
            left,
            right,
            on,
            filter,
            join_type,
            join_schema,
            left_fut,
            random_state,
            mode,
            metrics,
            projection,
            column_indices,
            null_equality,
            null_aware,
            cache: Arc::new(cache),
            dynamic_filter,
            fetch,
        })
    }

    fn with_dynamic_filter(mut self, filter: Option<HashJoinExecDynamicFilter>) -> Self {
        self.exec.dynamic_filter = filter;
        self
    }
}

impl From<&HashJoinExec> for HashJoinExecBuilder {
    fn from(exec: &HashJoinExec) -> Self {
        Self {
            exec: HashJoinExec {
                left: Arc::clone(exec.left()),
                right: Arc::clone(exec.right()),
                on: exec.on.clone(),
                filter: exec.filter.clone(),
                join_type: exec.join_type,
                join_schema: Arc::clone(&exec.join_schema),
                left_fut: Arc::clone(&exec.left_fut),
                random_state: exec.random_state.clone(),
                mode: exec.mode,
                metrics: exec.metrics.clone(),
                projection: exec.projection.clone(),
                column_indices: exec.column_indices.clone(),
                null_equality: exec.null_equality,
                null_aware: exec.null_aware,
                cache: Arc::clone(&exec.cache),
                dynamic_filter: exec.dynamic_filter.clone(),
                fetch: exec.fetch,
            },
            preserve_properties: true,
        }
    }
}

#[expect(rustdoc::private_intra_doc_links)]
/// Join execution plan: Evaluates equijoin predicates in parallel on multiple
/// partitions using a hash table and an optional filter list to apply post
/// join.
///
/// # Join Expressions
///
/// This implementation is optimized for evaluating equijoin predicates  (
/// `<col1> = <col2>`) expressions, which are represented as a list of `Columns`
/// in [`Self::on`].
///
/// Non-equality predicates, which can not pushed down to a join inputs (e.g.
/// `<col1> != <col2>`) are known as "filter expressions" and are evaluated
/// after the equijoin predicates.
///
/// # ArrayMap Optimization
///
/// For joins with a single integer-based join key, `HashJoinExec` may use an [`ArrayMap`]
/// (also known as a "perfect hash join") instead of a general-purpose hash map.
/// This optimization is used when:
/// 1. There is exactly one join key.
/// 2. The join key is an integer type up to 64 bits wide that can be losslessly converted
///    to `u64` (128-bit integer types such as `i128` and `u128` are not supported).
/// 3. The range of keys is small enough (controlled by `perfect_hash_join_small_build_threshold`)
///    OR the keys are sufficiently dense (controlled by `perfect_hash_join_min_key_density`).
/// 4. build_side.num_rows() < u32::MAX
/// 5. NullEqualsNothing || (NullEqualsNull && build side doesn't contain null)
///
/// See [`try_create_array_map`] for more details.
///
/// Note that when using [`PartitionMode::Partitioned`], the build side is split into multiple
/// partitions. This can cause a dense build side to become sparse within each partition,
/// potentially disabling this optimization.
///
/// For example, consider:
/// ```sql
/// SELECT t1.value, t2.value
/// FROM range(10000) AS t1
/// JOIN range(10000) AS t2
///   ON t1.value = t2.value;
/// ```
/// With 24 partitions, each partition will only receive a subset of the 10,000 rows.
/// The first partition might contain values like `3, 10, 18, 39, 43`, which are sparse
/// relative to the original range, even though the overall data set is dense.
///
/// # "Build Side" vs "Probe Side"
///
/// HashJoin takes two inputs, which are referred to as the "build" and the
/// "probe". The build side is the first child, and the probe side is the second
/// child.
///
/// The two inputs are treated differently and it is VERY important that the
/// *smaller* input is placed on the build side to minimize the work of creating
/// the hash table.
///
/// ```text
///          ┌───────────┐
///          │ HashJoin  │
///          │           │
///          └───────────┘
///              │   │
///        ┌─────┘   └─────┐
///        ▼               ▼
/// ┌────────────┐  ┌─────────────┐
/// │   Input    │  │    Input    │
/// │    [0]     │  │     [1]     │
/// └────────────┘  └─────────────┘
///
///  "build side"    "probe side"
/// ```
///
/// Execution proceeds in 2 stages:
///
/// 1. the **build phase** creates a hash table from the tuples of the build side,
///    and single concatenated batch containing data from all fetched record batches.
///    Resulting hash table stores hashed join-key fields for each row as a key, and
///    indices of corresponding rows in concatenated batch.
///
/// When using the standard `JoinHashMap`, hash join uses LIFO data structure as a hash table,
/// and in order to retain original build-side input order while obtaining data during probe phase,
/// hash table is updated by iterating batch sequence in reverse order -- it allows to
/// keep rows with smaller indices "on the top" of hash table, and still maintain
/// correct indexing for concatenated build-side data batch.
///
/// Example of build phase for 3 record batches:
///
///
/// ```text
///
///  Original build-side data   Inserting build-side values into hashmap    Concatenated build-side batch
///                                                                         ┌───────────────────────────┐
///                             hashmap.insert(row-hash, row-idx + offset)  │                      idx  │
///            ┌───────┐                                                    │          ┌───────┐        │
///            │ Row 1 │        1) update_hash for batch 3 with offset 0    │          │ Row 6 │    0   │
///   Batch 1  │       │           - hashmap.insert(Row 7, idx 1)           │ Batch 3  │       │        │
///            │ Row 2 │           - hashmap.insert(Row 6, idx 0)           │          │ Row 7 │    1   │
///            └───────┘                                                    │          └───────┘        │
///                                                                         │                           │
///            ┌───────┐                                                    │          ┌───────┐        │
///            │ Row 3 │        2) update_hash for batch 2 with offset 2    │          │ Row 3 │    2   │
///            │       │           - hashmap.insert(Row 5, idx 4)           │          │       │        │
///   Batch 2  │ Row 4 │           - hashmap.insert(Row 4, idx 3)           │ Batch 2  │ Row 4 │    3   │
///            │       │           - hashmap.insert(Row 3, idx 2)           │          │       │        │
///            │ Row 5 │                                                    │          │ Row 5 │    4   │
///            └───────┘                                                    │          └───────┘        │
///                                                                         │                           │
///            ┌───────┐                                                    │          ┌───────┐        │
///            │ Row 6 │        3) update_hash for batch 1 with offset 5    │          │ Row 1 │    5   │
///   Batch 3  │       │           - hashmap.insert(Row 2, idx 6)           │ Batch 1  │       │        │
///            │ Row 7 │           - hashmap.insert(Row 1, idx 5)           │          │ Row 2 │    6   │
///            └───────┘                                                    │          └───────┘        │
///                                                                         │                           │
///                                                                         └───────────────────────────┘
/// ```
///
/// 2. the **probe phase** where the tuples of the probe side are streamed
///    through, checking for matches of the join keys in the hash table.
///
/// ```text
///                 ┌────────────────┐          ┌────────────────┐
///                 │ ┌─────────┐    │          │ ┌─────────┐    │
///                 │ │  Hash   │    │          │ │  Hash   │    │
///                 │ │  Table  │    │          │ │  Table  │    │
///                 │ │(keys are│    │          │ │(keys are│    │
///                 │ │equi join│    │          │ │equi join│    │  Stage 2: batches from
///  Stage 1: the   │ │columns) │    │          │ │columns) │    │    the probe side are
/// *entire* build  │ │         │    │          │ │         │    │  streamed through, and
///  side is read   │ └─────────┘    │          │ └─────────┘    │   checked against the
/// into the hash   │      ▲         │          │          ▲     │   contents of the hash
///     table       │       HashJoin │          │  HashJoin      │          table
///                 └──────┼─────────┘          └──────────┼─────┘
///             ─ ─ ─ ─ ─ ─                                 ─ ─ ─ ─ ─ ─ ─
///            │                                                         │
///
///            │                                                         │
///     ┌────────────┐                                            ┌────────────┐
///     │RecordBatch │                                            │RecordBatch │
///     └────────────┘                                            └────────────┘
///     ┌────────────┐                                            ┌────────────┐
///     │RecordBatch │                                            │RecordBatch │
///     └────────────┘                                            └────────────┘
///           ...                                                       ...
///     ┌────────────┐                                            ┌────────────┐
///     │RecordBatch │                                            │RecordBatch │
///     └────────────┘                                            └────────────┘
///
///        build side                                                probe side
/// ```
///
/// # Example "Optimal" Plans
///
/// The differences in the inputs means that for classic "Star Schema Query",
/// the optimal plan will be a **"Right Deep Tree"** . A Star Schema Query is
/// one where there is one large table and several smaller "dimension" tables,
/// joined on `Foreign Key = Primary Key` predicates.
///
/// A "Right Deep Tree" looks like this large table as the probe side on the
/// lowest join:
///
/// ```text
///             ┌───────────┐
///             │ HashJoin  │
///             │           │
///             └───────────┘
///                 │   │
///         ┌───────┘   └──────────┐
///         ▼                      ▼
/// ┌───────────────┐        ┌───────────┐
/// │ small table 1 │        │ HashJoin  │
/// │  "dimension"  │        │           │
/// └───────────────┘        └───┬───┬───┘
///                   ┌──────────┘   └───────┐
///                   │                      │
///                   ▼                      ▼
///           ┌───────────────┐        ┌───────────┐
///           │ small table 2 │        │ HashJoin  │
///           │  "dimension"  │        │           │
///           └───────────────┘        └───┬───┬───┘
///                               ┌────────┘   └────────┐
///                               │                     │
///                               ▼                     ▼
///                       ┌───────────────┐     ┌───────────────┐
///                       │ small table 3 │     │  large table  │
///                       │  "dimension"  │     │    "fact"     │
///                       └───────────────┘     └───────────────┘
/// ```
///
/// # Clone / Shared State
///
/// Note this structure includes a [`OnceAsync`] that is used to coordinate the
/// loading of the left side with the processing in each output stream.
/// Therefore it can not be [`Clone`]
pub struct HashJoinExec {
    /// left (build) side which gets hashed
    pub left: Arc<dyn ExecutionPlan>,
    /// right (probe) side which are filtered by the hash table
    pub right: Arc<dyn ExecutionPlan>,
    /// Set of equijoin columns from the relations: `(left_col, right_col)`
    pub on: Vec<(PhysicalExprRef, PhysicalExprRef)>,
    /// Filters which are applied while finding matching rows
    pub filter: Option<JoinFilter>,
    /// How the join is performed (`OUTER`, `INNER`, etc)
    pub join_type: JoinType,
    /// The schema after join. Please be careful when using this schema,
    /// if there is a projection, the schema isn't the same as the output schema.
    join_schema: SchemaRef,
    /// Future that consumes left input and builds the hash table
    ///
    /// For CollectLeft partition mode, this structure is *shared* across all output streams.
    ///
    /// Each output stream waits on the `OnceAsync` to signal the completion of
    /// the hash table creation.
    left_fut: Arc<OnceAsync<JoinLeftData>>,
    /// Shared the `SeededRandomState` for the hashing algorithm (seeds preserved for serialization)
    random_state: SeededRandomState,
    /// Partitioning mode to use
    pub mode: PartitionMode,
    /// Execution metrics
    metrics: ExecutionPlanMetricsSet,
    /// The projection indices of the columns in the output schema of join
    pub projection: Option<ProjectionRef>,
    /// Information of index and left / right placement of columns
    column_indices: Vec<ColumnIndex>,
    /// The equality null-handling behavior of the join algorithm.
    pub null_equality: NullEquality,
    /// Flag to indicate if this is a null-aware anti join
    pub null_aware: bool,
    /// Cache holding plan properties like equivalences, output partitioning etc.
    cache: Arc<PlanProperties>,
    /// Dynamic filter for pushing down to the probe side
    /// Set when dynamic filter pushdown is detected in handle_child_pushdown_result.
    /// HashJoinExec also needs to keep a shared bounds accumulator for coordinating updates.
    dynamic_filter: Option<HashJoinExecDynamicFilter>,
    /// Maximum number of rows to return
    fetch: Option<usize>,
}

#[derive(Clone)]
struct HashJoinExecDynamicFilter {
    /// Dynamic filter that we'll update with the results of the build side once that is done.
    filter: Arc<DynamicFilterPhysicalExpr>,
    /// Build accumulator to collect build-side information (hash maps and/or bounds) from each partition.
    /// It is lazily initialized during execution to make sure we use the actual execution time partition counts.
    build_accumulator: OnceLock<Arc<SharedBuildAccumulator>>,
}

impl fmt::Debug for HashJoinExec {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("HashJoinExec")
            .field("left", &self.left)
            .field("right", &self.right)
            .field("on", &self.on)
            .field("filter", &self.filter)
            .field("join_type", &self.join_type)
            .field("join_schema", &self.join_schema)
            .field("left_fut", &self.left_fut)
            .field("random_state", &self.random_state)
            .field("mode", &self.mode)
            .field("metrics", &self.metrics)
            .field("projection", &self.projection)
            .field("column_indices", &self.column_indices)
            .field("null_equality", &self.null_equality)
            .field("cache", &self.cache)
            // Explicitly exclude dynamic_filter to avoid runtime state differences in tests
            .finish()
    }
}

impl EmbeddedProjection for HashJoinExec {
    fn with_projection(&self, projection: Option<Vec<usize>>) -> Result<Self> {
        self.with_projection(projection)
    }
}

impl HashJoinExec {
    /// Tries to create a new [`HashJoinExec`].
    ///
    /// # Error
    /// This function errors when it is not possible to join the left and right sides on keys `on`.
    #[expect(clippy::too_many_arguments)]
    pub fn try_new(
        left: Arc<dyn ExecutionPlan>,
        right: Arc<dyn ExecutionPlan>,
        on: JoinOn,
        filter: Option<JoinFilter>,
        join_type: &JoinType,
        projection: Option<Vec<usize>>,
        partition_mode: PartitionMode,
        null_equality: NullEquality,
        null_aware: bool,
    ) -> Result<Self> {
        HashJoinExecBuilder::new(left, right, on, *join_type)
            .with_filter(filter)
            .with_projection(projection)
            .with_partition_mode(partition_mode)
            .with_null_equality(null_equality)
            .with_null_aware(null_aware)
            .build()
    }

    /// Create a builder based on the existing [`HashJoinExec`].
    ///
    /// Returned builder preserves all existing fields. If a field requiring properties
    /// recomputation is modified, this will be done automatically during the node build.
    ///
    pub fn builder(&self) -> HashJoinExecBuilder {
        self.into()
    }

    fn create_dynamic_filter(on: &JoinOn) -> Arc<DynamicFilterPhysicalExpr> {
        // Extract the right-side keys (probe side keys) from the `on` clauses
        // Dynamic filter will be created from build side values (left side) and applied to probe side (right side)
        let right_keys: Vec<_> = on.iter().map(|(_, r)| Arc::clone(r)).collect();
        // Initialize with a placeholder expression (true) that will be updated when the hash table is built
        Arc::new(DynamicFilterPhysicalExpr::new(right_keys, lit(true)))
    }

    fn allow_join_dynamic_filter_pushdown(&self, config: &ConfigOptions) -> bool {
        let (_, probe_preserved) = self.join_type.on_lr_is_preserved();
        if !probe_preserved || !config.optimizer.enable_join_dynamic_filter_pushdown {
            return false;
        }

        // A null-aware anti join emits a build-side NULL only when the probe
        // is truly empty. The pushed filter can empty the probe by pruning
        // every row, which would surface that NULL wrongly. A NOT NULL build
        // key cannot produce such a NULL, so the filter stays there.
        if self.null_aware
            && self.on.iter().any(|(build_key, _)| {
                build_key.nullable(&self.left.schema()).unwrap_or(true)
            })
        {
            return false;
        }

        // `preserve_file_partitions` can report Hive-style file groups as Hash
        // partitioned even though their partition indexes do not follow the
        // hash router used by partitioned dynamic filters. Reject Hash inputs
        // because the metadata cannot distinguish those scans from a real hash
        // repartition. Compatible Range inputs remain safe because matching
        // ordering and split points align each build filter with its probe
        // partition. Other unsupported layouts are rejected.
        // Follow-up work: enable dynamic filtering for preserve_file_partitioned scans (issue #20195).
        // https://github.com/apache/datafusion/issues/20195
        if config.optimizer.preserve_file_partitions > 0
            && self.mode == PartitionMode::Partitioned
            && matches!(
                (
                    self.left.output_partitioning(),
                    self.right.output_partitioning()
                ),
                (Partitioning::Hash(_, _), Partitioning::Hash(_, _))
            )
        {
            return false;
        }

        if self.mode == PartitionMode::Partitioned
            && !self.has_partitioned_dynamic_filter_routing()
        {
            return false;
        }

        true
    }

    fn has_partitioned_dynamic_filter_routing(&self) -> bool {
        match (
            self.left.output_partitioning(),
            self.right.output_partitioning(),
        ) {
            (
                Partitioning::Hash(_, left_partition_count),
                Partitioning::Hash(_, right_partition_count),
            ) => left_partition_count == right_partition_count,
            (Partitioning::Range(_), Partitioning::Range(_)) => {
                let children = [self.left.as_ref(), self.right.as_ref()];
                matches!(
                    self.input_distribution_requirements()
                        .unsatisfied_co_partitioned_children(self.name(), &children),
                    Ok(unsatisfied) if unsatisfied.is_empty()
                )
            }
            (left_partitioning, right_partitioning) => {
                left_partitioning.partition_count() == 1
                    && right_partitioning.partition_count() == 1
            }
        }
    }

    /// left (build) side which gets hashed
    pub fn left(&self) -> &Arc<dyn ExecutionPlan> {
        &self.left
    }

    /// right (probe) side which are filtered by the hash table
    pub fn right(&self) -> &Arc<dyn ExecutionPlan> {
        &self.right
    }

    /// Set of common columns used to join on
    pub fn on(&self) -> &[(PhysicalExprRef, PhysicalExprRef)] {
        &self.on
    }

    /// Filters applied before join output
    pub fn filter(&self) -> Option<&JoinFilter> {
        self.filter.as_ref()
    }

    /// How the join is performed
    pub fn join_type(&self) -> &JoinType {
        &self.join_type
    }

    /// The schema after join. Please be careful when using this schema,
    /// if there is a projection, the schema isn't the same as the output schema.
    pub fn join_schema(&self) -> &SchemaRef {
        &self.join_schema
    }

    /// The partitioning mode of this hash join
    pub fn partition_mode(&self) -> &PartitionMode {
        &self.mode
    }

    /// Get null_equality
    pub fn null_equality(&self) -> NullEquality {
        self.null_equality
    }

    /// Returns the dynamic filter expression produced by this hash join, if set.
    #[deprecated(
        since = "55.0.0",
        note = "Use ExecutionPlan::dynamic_expressions_produced instead"
    )]
    pub fn dynamic_filter_expr(&self) -> Option<&Arc<DynamicFilterPhysicalExpr>> {
        self.dynamic_filter.as_ref().map(|df| &df.filter)
    }

    /// Set the dynamic filter on this hash join.
    ///
    /// Resets any internal state that depends on any existing dynamic filter.
    ///
    /// Validates that the filter's children reference valid columns in
    /// the probe (right) side's schema.
    pub fn with_dynamic_filter_expr(
        mut self,
        filter: Arc<DynamicFilterPhysicalExpr>,
    ) -> Result<Self> {
        let probe_schema = self.right.schema();
        for child in filter.children() {
            child.data_type(&probe_schema)?;
        }
        self.dynamic_filter = Some(HashJoinExecDynamicFilter {
            filter,
            // Initialize with an empty accumulator which will be lazily populated
            // during execution.
            build_accumulator: OnceLock::new(),
        });
        Ok(self)
    }

    /// Calculate order preservation flags for this hash join.
    fn maintains_input_order(join_type: JoinType) -> Vec<bool> {
        vec![
            false,
            matches!(
                join_type,
                JoinType::Inner
                    | JoinType::Right
                    | JoinType::RightAnti
                    | JoinType::RightSemi
                    | JoinType::RightMark
            ),
        ]
    }

    /// Get probe side information for the hash join.
    pub fn probe_side() -> JoinSide {
        // In current implementation right side is always probe side.
        JoinSide::Right
    }

    /// Return whether the join contains a projection
    pub fn contains_projection(&self) -> bool {
        self.projection.is_some()
    }

    /// Return new instance of [HashJoinExec] with the given projection.
    pub fn with_projection(&self, projection: Option<Vec<usize>>) -> Result<Self> {
        let projection = projection.map(Into::into);
        //  check if the projection is valid
        can_project(&self.schema(), projection.as_deref())?;
        let projection =
            combine_projections(projection.as_ref(), self.projection.as_ref())?;
        self.builder().with_projection_ref(projection).build()
    }

    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
    fn compute_properties(
        left: &Arc<dyn ExecutionPlan>,
        right: &Arc<dyn ExecutionPlan>,
        schema: &SchemaRef,
        join_type: JoinType,
        on: JoinOnRef,
        mode: PartitionMode,
        projection: Option<&[usize]>,
    ) -> Result<PlanProperties> {
        // Calculate equivalence properties:
        let mut eq_properties = join_equivalence_properties(
            left.equivalence_properties().clone(),
            right.equivalence_properties().clone(),
            &join_type,
            Arc::clone(schema),
            &Self::maintains_input_order(join_type),
            Some(Self::probe_side()),
            on,
        )?;

        let mut output_partitioning = match mode {
            PartitionMode::CollectLeft => {
                asymmetric_join_output_partitioning(left, right, &join_type)?
            }
            PartitionMode::Auto => Partitioning::UnknownPartitioning(
                right.output_partitioning().partition_count(),
            ),
            PartitionMode::Partitioned => {
                symmetric_join_output_partitioning(left, right, &join_type)?
            }
        };

        let emission_type = if left.boundedness().is_unbounded() {
            EmissionType::Final
        } else if right.pipeline_behavior() == EmissionType::Incremental {
            match join_type {
                // If we only need to generate matched rows from the probe side,
                // we can emit rows incrementally.
                JoinType::Inner
                | JoinType::LeftSemi
                | JoinType::RightSemi
                | JoinType::Right
                | JoinType::RightAnti
                | JoinType::RightMark => EmissionType::Incremental,
                // If we need to generate unmatched rows from the *build side*,
                // we need to emit them at the end.
                JoinType::Left
                | JoinType::LeftAnti
                | JoinType::LeftMark
                | JoinType::Full => EmissionType::Both,
            }
        } else {
            right.pipeline_behavior()
        };

        // If contains projection, update the PlanProperties.
        if let Some(projection) = projection {
            // construct a map from the input expressions to the output expression of the Projection
            let projection_mapping = ProjectionMapping::from_indices(projection, schema)?;
            let out_schema = project_schema(schema, Some(&projection))?;
            output_partitioning =
                output_partitioning.project(&projection_mapping, &eq_properties);
            eq_properties = eq_properties.project(&projection_mapping, out_schema);
        }

        Ok(PlanProperties::new(
            eq_properties,
            output_partitioning,
            emission_type,
            boundedness_from_children([left, right]),
        ))
    }

    /// Returns a new `ExecutionPlan` that computes the same join as this one,
    /// with the left and right inputs swapped using the  specified
    /// `partition_mode`.
    ///
    /// # Notes:
    ///
    /// This function is public so other downstream projects can use it to
    /// construct `HashJoinExec` with right side as the build side.
    ///
    /// For using this interface directly, please refer to below:
    ///
    /// Hash join execution may require specific input partitioning (for example,
    /// the left child may have a single partition while the right child has multiple).
    ///
    /// Calling this function on join nodes whose children have already been repartitioned
    /// (e.g., after a `RepartitionExec` has been inserted) may break the partitioning
    /// requirements of the hash join. Therefore, ensure you call this function
    /// before inserting any repartitioning operators on the join's children.
    ///
    /// In DataFusion's default SQL interface, this function is used by the `JoinSelection`
    /// physical optimizer rule to determine a good join order, which is
    /// executed before the `EnforceDistribution` rule (the rule that may
    /// insert `RepartitionExec` operators).
    pub fn swap_inputs(
        &self,
        partition_mode: PartitionMode,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        assert_or_internal_err!(
            self.dynamic_filter.is_none(),
            "Cannot swap HashJoinExec inputs after dynamic filters have been constructed. \
             Optimizer rules that reorder join inputs must run before optimizer rules `FilterPushdown::new_post_optimization()`"
        );

        let left = self.left();
        let right = self.right();
        let new_join = self
            .builder()
            .with_type(self.join_type.swap())
            .with_new_children(vec![Arc::clone(right), Arc::clone(left)])?
            .with_on(
                self.on()
                    .iter()
                    .map(|(l, r)| (Arc::clone(r), Arc::clone(l)))
                    .collect(),
            )
            .with_filter(self.filter().map(JoinFilter::swap))
            .with_projection(swap_join_projection(
                left.schema().fields().len(),
                right.schema().fields().len(),
                self.projection.as_deref(),
                self.join_type(),
            ))
            .with_partition_mode(partition_mode)
            .build()?;
        // In case of anti / semi joins or if there is embedded projection in HashJoinExec, output column order is preserved, no need to add projection again
        if matches!(
            self.join_type(),
            JoinType::LeftSemi
                | JoinType::RightSemi
                | JoinType::LeftAnti
                | JoinType::RightAnti
                | JoinType::LeftMark
                | JoinType::RightMark
        ) || self.projection.is_some()
        {
            Ok(Arc::new(new_join))
        } else {
            reorder_output_after_swap(Arc::new(new_join), &left.schema(), &right.schema())
        }
    }
}

impl DisplayAs for HashJoinExec {
    fn fmt_as(&self, t: DisplayFormatType, f: &mut fmt::Formatter) -> fmt::Result {
        match t {
            DisplayFormatType::Default | DisplayFormatType::Verbose => {
                let display_filter = self.filter.as_ref().map_or_else(
                    || "".to_string(),
                    |f| format!(", filter={}", f.expression()),
                );
                let display_projections = if self.contains_projection() {
                    format!(
                        ", projection=[{}]",
                        self.projection
                            .as_ref()
                            .unwrap()
                            .iter()
                            .map(|index| format!(
                                "{}@{}",
                                self.join_schema.fields().get(*index).unwrap().name(),
                                index
                            ))
                            .collect::<Vec<_>>()
                            .join(", ")
                    )
                } else {
                    "".to_string()
                };
                let display_null_equality =
                    if self.null_equality() == NullEquality::NullEqualsNull {
                        ", NullsEqual: true"
                    } else {
                        ""
                    };
                let display_fetch = self
                    .fetch
                    .map_or_else(String::new, |f| format!(", fetch={f}"));
                let display_null_aware =
                    if self.null_aware { ", null_aware" } else { "" };
                let on = self
                    .on
                    .iter()
                    .map(|(c1, c2)| format!("({c1}, {c2})"))
                    .collect::<Vec<String>>()
                    .join(", ");
                write!(
                    f,
                    "HashJoinExec: mode={:?}, join_type={:?}, on=[{}]{}{}{}{}{}",
                    self.mode,
                    self.join_type,
                    on,
                    display_filter,
                    display_projections,
                    display_null_equality,
                    display_fetch,
                    display_null_aware,
                )
            }
            DisplayFormatType::TreeRender => {
                let on = self
                    .on
                    .iter()
                    .map(|(c1, c2)| {
                        format!("({} = {})", fmt_sql(c1.as_ref()), fmt_sql(c2.as_ref()))
                    })
                    .collect::<Vec<String>>()
                    .join(", ");

                if *self.join_type() != JoinType::Inner {
                    writeln!(f, "join_type={:?}", self.join_type)?;
                }

                writeln!(f, "on={on}")?;

                if self.null_equality() == NullEquality::NullEqualsNull {
                    writeln!(f, "NullsEqual: true")?;
                }

                if self.null_aware {
                    writeln!(f, "null_aware")?;
                }

                if let Some(filter) = self.filter.as_ref() {
                    writeln!(f, "filter={filter}")?;
                }

                if let Some(fetch) = self.fetch {
                    writeln!(f, "fetch={fetch}")?;
                }

                Ok(())
            }
        }
    }
}

impl ExecutionPlan for HashJoinExec {
    fn name(&self) -> &'static str {
        "HashJoinExec"
    }

    fn properties(&self) -> &Arc<PlanProperties> {
        &self.cache
    }

    fn required_input_distribution(&self) -> Vec<Distribution> {
        self.input_distribution_requirements().into_per_child()
    }

    fn input_distribution_requirements(&self) -> InputDistributionRequirements {
        match self.mode {
            PartitionMode::Partitioned => {
                let (left_expr, right_expr) = self
                    .on
                    .iter()
                    .map(|(l, r)| (Arc::clone(l), Arc::clone(r)))
                    .unzip();
                InputDistributionRequirements::co_partitioned(vec![
                    Distribution::KeyPartitioned(left_expr),
                    Distribution::KeyPartitioned(right_expr),
                ])
            }
            PartitionMode::CollectLeft => InputDistributionRequirements::new(vec![
                Distribution::SinglePartition,
                Distribution::UnspecifiedDistribution,
            ]),
            PartitionMode::Auto => InputDistributionRequirements::new(vec![
                Distribution::UnspecifiedDistribution,
                Distribution::UnspecifiedDistribution,
            ]),
        }
    }

    // For [JoinType::Inner] and [JoinType::RightSemi] in hash joins, the probe phase initiates by
    // applying the hash function to convert the join key(s) in each row into a hash value from the
    // probe side table in the order they're arranged. The hash value is used to look up corresponding
    // entries in the hash table that was constructed from the build side table during the build phase.
    //
    // Because of the immediate generation of result rows once a match is found,
    // the output of the join tends to follow the order in which the rows were read from
    // the probe side table. This is simply due to the sequence in which the rows were processed.
    // Hence, it appears that the hash join is preserving the order of the probe side.
    //
    // Meanwhile, in the case of a [JoinType::RightAnti] hash join,
    // the unmatched rows from the probe side are also kept in order.
    // This is because the **`RightAnti`** join is designed to return rows from the right
    // (probe side) table that have no match in the left (build side) table. Because the rows
    // are processed sequentially in the probe phase, and unmatched rows are directly output
    // as results, these results tend to retain the order of the probe side table.
    fn maintains_input_order(&self) -> Vec<bool> {
        Self::maintains_input_order(self.join_type)
    }

    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
        vec![&self.left, &self.right]
    }

    fn apply_expressions(
        &self,
        f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
    ) -> Result<TreeNodeRecursion> {
        let join_keys = self
            .on
            .iter()
            .flat_map(|(left, right)| [Arc::clone(left), Arc::clone(right)]);
        let filter = self
            .filter
            .iter()
            .map(|filter| Arc::clone(filter.expression()));
        let dynamic_filter = self.dynamic_filter.iter().map(|dynamic_filter| {
            Arc::<DynamicFilterPhysicalExpr>::clone(&dynamic_filter.filter)
                as Arc<dyn PhysicalExpr>
        });
        crate::apply_expression_roots(join_keys.chain(filter).chain(dynamic_filter), f)
    }

    fn dynamic_expressions_produced(&self) -> Vec<Arc<dyn PhysicalExpr>> {
        self.dynamic_filter
            .iter()
            .map(|dynamic_filter| {
                Arc::<DynamicFilterPhysicalExpr>::clone(&dynamic_filter.filter)
                    as Arc<dyn PhysicalExpr>
            })
            .collect()
    }

    /// Creates a new HashJoinExec with different children while preserving configuration.
    ///
    /// This method is called during query optimization when the optimizer creates new
    /// plan nodes. Importantly, it creates a fresh bounds_accumulator via `try_new`
    /// rather than cloning the existing one because partitioning may have changed.
    fn replace_children(
        self: Arc<Self>,
        children: Vec<Arc<dyn ExecutionPlan>>,
        options: ReplaceChildrenOptions,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        validate_child_count!(self, children);
        match options.children_properties {
            ChildrenPropertiesMode::Keep => {
                self.builder().with_new_children(children)?.build_exec()
            }
            ChildrenPropertiesMode::Recompute => self
                .builder()
                .recompute_properties()
                .with_new_children(children)?
                .build_exec(),
        }
    }

    fn with_new_children(
        self: Arc<Self>,
        children: Vec<Arc<dyn ExecutionPlan>>,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        self.replace_children(
            children,
            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
        )
    }

    fn reset_state(self: Arc<Self>) -> Result<Arc<dyn ExecutionPlan>> {
        self.builder().reset_state().build_exec()
    }

    fn execute(
        &self,
        partition: usize,
        context: Arc<TaskContext>,
    ) -> Result<SendableRecordBatchStream> {
        let on_left = self
            .on
            .iter()
            .map(|on| Arc::clone(&on.0))
            .collect::<Vec<_>>();
        let left_partitions = self.left.output_partitioning().partition_count();
        let right_partitions = self.right.output_partitioning().partition_count();

        assert_or_internal_err!(
            self.mode != PartitionMode::Partitioned
                || left_partitions == right_partitions,
            "Invalid HashJoinExec, partition count mismatch {left_partitions}!={right_partitions},\
             consider using RepartitionExec"
        );

        assert_or_internal_err!(
            self.mode != PartitionMode::CollectLeft || left_partitions == 1,
            "Invalid HashJoinExec, the output partition count of the left child must be 1 in CollectLeft mode,\
             consider using CoalescePartitionsExec or the EnforceDistribution rule"
        );

        // Only compute a dynamic filter when the probe subtree contains a consumer.
        // Searching from `self` would always find the producer expression owned by this join.
        let enable_dynamic_filter_pushdown = if self
            .allow_join_dynamic_filter_pushdown(context.session_config().options())
        {
            self.dynamic_filter
                .as_ref()
                .and_then(|df| df.filter.expression_id())
                .map(|id| plan_contains_expression_id(&self.right, id))
                .transpose()?
                .unwrap_or(false)
        } else {
            false
        };

        let join_metrics = BuildProbeJoinMetrics::new(partition, &self.metrics);

        let array_map_created_count = MetricBuilder::new(&self.metrics)
            .with_category(MetricCategory::Rows)
            .counter(ARRAY_MAP_CREATED_COUNT_METRIC_NAME, partition);

        // Initialize build_accumulator lazily with runtime partition counts (only if enabled)
        // Use RepartitionExec's random state (seeds: 0,0,0,0) for partition routing
        let repartition_random_state = REPARTITION_RANDOM_STATE;
        let build_accumulator = enable_dynamic_filter_pushdown
            .then(|| {
                self.dynamic_filter.as_ref().map(|df| {
                    let filter = Arc::clone(&df.filter);
                    let on_right = self
                        .on
                        .iter()
                        .map(|(_, right_expr)| Arc::clone(right_expr))
                        .collect::<Vec<_>>();
                    Some(Arc::clone(df.build_accumulator.get_or_init(|| {
                        Arc::new(SharedBuildAccumulator::new_from_partition_mode(
                            self.mode,
                            self.left.as_ref(),
                            self.right.as_ref(),
                            filter,
                            on_right,
                            repartition_random_state,
                            self.null_equality,
                            self.null_aware,
                        ))
                    })))
                })
            })
            .flatten()
            .flatten();

        let left_fut = match self.mode {
            PartitionMode::CollectLeft => self.left_fut.try_once(|| {
                let left_stream = self.left.execute(0, Arc::clone(&context))?;

                let reservation =
                    MemoryConsumer::new("HashJoinInput").register(context.memory_pool());

                Ok(collect_left_input(
                    self.random_state.random_state().clone(),
                    left_stream,
                    on_left.clone(),
                    join_metrics.clone(),
                    reservation,
                    need_produce_result_in_final(self.join_type),
                    self.right().output_partitioning().partition_count(),
                    enable_dynamic_filter_pushdown,
                    Arc::clone(context.session_config().options()),
                    self.null_equality,
                    array_map_created_count,
                ))
            })?,
            PartitionMode::Partitioned => {
                let left_stream = self.left.execute(partition, Arc::clone(&context))?;

                let reservation =
                    MemoryConsumer::new(format!("HashJoinInput[{partition}]"))
                        .register(context.memory_pool());
                OnceFut::new(collect_left_input(
                    self.random_state.random_state().clone(),
                    left_stream,
                    on_left.clone(),
                    join_metrics.clone(),
                    reservation,
                    need_produce_result_in_final(self.join_type),
                    1,
                    enable_dynamic_filter_pushdown,
                    Arc::clone(context.session_config().options()),
                    self.null_equality,
                    array_map_created_count,
                ))
            }
            PartitionMode::Auto => {
                return plan_err!(
                    "Invalid HashJoinExec, unsupported PartitionMode {:?} in execute()",
                    PartitionMode::Auto
                );
            }
        };

        let batch_size = context.session_config().batch_size();

        // 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)?;

        // update column indices to reflect the projection
        let column_indices_after_projection = match self.projection.as_ref() {
            Some(projection) => projection
                .iter()
                .map(|i| self.column_indices[*i].clone())
                .collect(),
            None => self.column_indices.clone(),
        };

        let on_right = self
            .on
            .iter()
            .map(|(_, right_expr)| Arc::clone(right_expr))
            .collect::<Vec<_>>();

        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,
            self.null_aware,
            self.fetch,
        )))
    }

    fn metrics(&self) -> Option<MetricsSet> {
        Some(self.metrics.clone_inner())
    }

    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
        match (partition, self.mode) {
            // Left side is broadcast, so it always needs overall stats
            // Right side is partitioned, so it needs per-partition stats
            (Some(_), PartitionMode::CollectLeft) => {
                vec![ChildStats::At(None), ChildStats::At(partition)]
            }
            // For Partitioned mode, both sides are hash-partitioned symmetrically,
            // so each output partition uses the matching partition from both sides.
            (Some(_), PartitionMode::Partitioned) => {
                vec![ChildStats::At(partition), ChildStats::At(partition)]
            }
            // Overall stats requested, look up overall child stats.
            (None, _) => vec![ChildStats::At(None), ChildStats::At(None)],
            // Auto mode hasn't decided partitioning yet, so it needs
            // overall stats from both sides.
            (Some(_), PartitionMode::Auto) => {
                vec![ChildStats::At(None), ChildStats::At(None)]
            }
        }
    }

    fn statistics_from_inputs(
        &self,
        input_stats: &[Arc<Statistics>],
        _args: &StatisticsArgs,
    ) -> Result<Arc<Statistics>> {
        let left_stats = Arc::clone(&input_stats[0]);
        let right_stats = Arc::clone(&input_stats[1]);
        let stats = estimate_join_statistics(
            Arc::unwrap_or_clone(left_stats),
            Arc::unwrap_or_clone(right_stats),
            &self.on,
            self.null_equality,
            &self.join_type,
            &self.join_schema,
        )?;
        // Project statistics if there is a projection
        let stats = stats.project(self.projection.as_ref());
        // Apply fetch limit to statistics
        Ok(Arc::new(stats.with_fetch(self.fetch, 0, 1)?))
    }

    /// Tries to push `projection` down through `hash_join`. If possible, performs the
    /// pushdown and returns a new [`HashJoinExec`] as the top plan which has projections
    /// as its children. Otherwise, returns `None`.
    fn try_swapping_with_projection(
        &self,
        projection: &ProjectionExec,
    ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
        // TODO: currently if there is projection in HashJoinExec, we can't push down projection to left or right input. Maybe we can pushdown the mixed projection later.
        if self.contains_projection() {
            return Ok(None);
        }

        let schema = self.schema();
        if let Some(JoinData {
            projected_left_child,
            projected_right_child,
            join_filter,
            join_on,
        }) = try_pushdown_through_join_with_column_indices(
            projection,
            self.left(),
            self.right(),
            self.on(),
            &schema,
            self.filter(),
            self.column_indices.as_slice(),
        )? {
            self.builder()
                .with_new_children(vec![
                    Arc::new(projected_left_child),
                    Arc::new(projected_right_child),
                ])?
                .with_on(join_on)
                .with_filter(join_filter)
                // Returned early if projection is not None
                .with_projection(None)
                .build_exec()
                .map(Some)
        } else {
            try_embed_projection(projection, self)
        }
    }

    fn gather_filters_for_pushdown(
        &self,
        phase: FilterPushdownPhase,
        parent_filters: Vec<Arc<dyn PhysicalExpr>>,
        config: &ConfigOptions,
    ) -> Result<FilterDescription> {
        // This is the physical-plan equivalent of `push_down_all_join` in
        // `datafusion/optimizer/src/push_down_filter.rs`. That function uses `lr_is_preserved`
        // to decide which parent predicates can be pushed past a logical join to its children,
        // then checks column references to route each predicate to the correct side.
        //
        // We apply the same two-level logic here:
        // 1. `lr_is_preserved` gates whether a side is eligible at all.
        // 2. For each filter, we check that all column references belong to the
        //    target child (using `column_indices` to map output column positions
        //    to join sides). This is critical for correctness: name-based matching
        //    alone (as done by `ChildFilterDescription::from_child`) can incorrectly
        //    push filters when different join sides have columns with the same name
        //    (e.g. nested mark joins both producing "mark" columns).
        let (left_preserved, right_preserved) = lr_is_preserved(self.join_type);

        // Build the set of allowed column indices for each side
        let column_indices: Vec<ColumnIndex> = match self.projection.as_ref() {
            Some(projection) => projection
                .iter()
                .map(|i| self.column_indices[*i].clone())
                .collect(),
            None => self.column_indices.clone(),
        };

        let (mut left_allowed, mut right_allowed) = (HashSet::new(), HashSet::new());
        column_indices
            .iter()
            .enumerate()
            .for_each(|(output_idx, ci)| {
                match ci.side {
                    JoinSide::Left => left_allowed.insert(output_idx),
                    JoinSide::Right => right_allowed.insert(output_idx),
                    // Mark columns - don't allow pushdown to either side
                    JoinSide::None => false,
                };
            });

        // For semi joins, filters on output join keys can also be pushed to the
        // non-output side: every emitted row has an equal key there. This is not
        // true for anti joins, whose emitted rows have no match.
        match self.join_type {
            JoinType::LeftSemi => {
                let left_key_indices: HashSet<usize> = self
                    .on
                    .iter()
                    .filter_map(|(left_key, _)| {
                        left_key.downcast_ref::<Column>().map(|c| c.index())
                    })
                    .collect();
                for (output_idx, ci) in column_indices.iter().enumerate() {
                    if ci.side == JoinSide::Left && left_key_indices.contains(&ci.index) {
                        right_allowed.insert(output_idx);
                    }
                }
            }
            JoinType::RightSemi => {
                let right_key_indices: HashSet<usize> = self
                    .on
                    .iter()
                    .filter_map(|(_, right_key)| {
                        right_key.downcast_ref::<Column>().map(|c| c.index())
                    })
                    .collect();
                for (output_idx, ci) in column_indices.iter().enumerate() {
                    if ci.side == JoinSide::Right && right_key_indices.contains(&ci.index)
                    {
                        left_allowed.insert(output_idx);
                    }
                }
            }
            _ => {}
        }

        let left_child = if left_preserved {
            ChildFilterDescription::from_child_with_allowed_indices(
                &parent_filters,
                left_allowed,
                self.left(),
            )?
        } else {
            ChildFilterDescription::all_unsupported(&parent_filters)
        };

        let mut right_child = if right_preserved {
            ChildFilterDescription::from_child_with_allowed_indices(
                &parent_filters,
                right_allowed,
                self.right(),
            )?
        } else {
            ChildFilterDescription::all_unsupported(&parent_filters)
        };

        // Add dynamic filters in Post phase if enabled. Skip when this join
        // already carries a dynamic filter from a previous pass — the shared
        // `Arc<DynamicFilterPhysicalExpr>` is still wired into the probe-side
        // scan's predicate, and re-creating it would AND a fresh duplicate
        // onto every Post-phase invocation (apache/datafusion-ballista#1359
        // surfaces this in AQE replan loops).
        if phase == FilterPushdownPhase::Post
            && self.dynamic_filter.is_none()
            && self.allow_join_dynamic_filter_pushdown(config)
        {
            // Add actual dynamic filter to right side (probe side)
            let dynamic_filter = Self::create_dynamic_filter(&self.on);
            right_child = right_child.with_self_filter(dynamic_filter);
        }

        Ok(FilterDescription::new()
            .with_child(left_child)
            .with_child(right_child))
    }

    fn handle_child_pushdown_result(
        &self,
        _phase: FilterPushdownPhase,
        child_pushdown_result: ChildPushdownResult,
        _config: &ConfigOptions,
    ) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
        let mut result = FilterPushdownPropagation::if_any(child_pushdown_result.clone());
        assert_eq!(child_pushdown_result.self_filters.len(), 2); // Should always be 2, we have 2 children
        let right_child_self_filters = &child_pushdown_result.self_filters[1]; // We only push down filters to the right child
        // We expect 0 or 1 self filters
        if let Some(filter) = right_child_self_filters.first() {
            // Note that we don't check PushdDownPredicate::discrimnant because even if nothing said
            // "yes, I can fully evaluate this filter" things might still use it for statistics -> it's worth updating
            let predicate = Arc::clone(&filter.predicate);
            if let Ok(dynamic_filter) =
                Arc::downcast::<DynamicFilterPhysicalExpr>(predicate)
            {
                // We successfully pushed down our self filter - we need to make a new node with the dynamic filter
                let new_node = self
                    .builder()
                    .with_dynamic_filter(Some(HashJoinExecDynamicFilter {
                        filter: dynamic_filter,
                        build_accumulator: OnceLock::new(),
                    }))
                    .build_exec()?;
                result = result.with_updated_node(new_node);
            }
        }
        Ok(result)
    }

    fn supports_limit_pushdown(&self) -> bool {
        // Hash join execution plan does not support pushing limit down through to children
        // because the children don't know about the join condition and can't
        // determine how many rows to produce
        false
    }

    fn fetch(&self) -> Option<usize> {
        self.fetch
    }

    fn with_fetch(&self, limit: Option<usize>) -> Option<Arc<dyn ExecutionPlan>> {
        self.builder()
            .with_fetch(limit)
            .build()
            .ok()
            .map(|exec| Arc::new(exec) as _)
    }
    #[cfg(feature = "proto")]
    fn try_to_proto(
        &self,
        ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
        use datafusion_proto_models::protobuf;

        let left = ctx.encode_child(self.left())?;
        let right = ctx.encode_child(self.right())?;

        let on = self
            .on()
            .iter()
            .map(|(l, r)| -> Result<protobuf::JoinOn> {
                Ok(protobuf::JoinOn {
                    left: Some(ctx.encode_expr(l)?),
                    right: Some(ctx.encode_expr(r)?),
                })
            })
            .collect::<Result<Vec<_>>>()?;

        let join_type = crate::joins::proto::join_type_to_proto(*self.join_type());
        let null_equality =
            crate::joins::proto::null_equality_to_proto(self.null_equality());
        // `PartitionMode` is specific to `HashJoinExec`, so its conversion stays
        // inline (by-name on purpose: the enums are numbered differently).
        let partition_mode = match self.partition_mode() {
            PartitionMode::CollectLeft => protobuf::PartitionMode::CollectLeft,
            PartitionMode::Partitioned => protobuf::PartitionMode::Partitioned,
            PartitionMode::Auto => protobuf::PartitionMode::Auto,
        };

        let filter = self
            .filter()
            .map(|f| crate::joins::proto::join_filter_to_proto(f, ctx))
            .transpose()?;

        let dynamic_filter = self
            .dynamic_expressions_produced()
            .into_iter()
            .next()
            .map(|expr| ctx.encode_expr(&expr))
            .transpose()?;

        Ok(Some(protobuf::PhysicalPlanNode {
            physical_plan_type: Some(
                protobuf::physical_plan_node::PhysicalPlanType::HashJoin(Box::new(
                    protobuf::HashJoinExecNode {
                        left: Some(Box::new(left)),
                        right: Some(Box::new(right)),
                        on,
                        join_type: join_type.into(),
                        partition_mode: partition_mode.into(),
                        null_equality: null_equality.into(),
                        filter,
                        // Proto3 `repeated` cannot distinguish `None` from
                        // `Some(vec![])`. `Some(vec![])` (reachable via
                        // `try_embed_projection` for e.g. `SELECT count(1) … JOIN …`)
                        // changes the output schema, so it is encoded with the
                        // single-element sentinel `[u32::MAX]` (never a valid column
                        // index); every other state is sent as-is. See
                        // `try_from_proto` for the matching decoder.
                        projection: match self.projection.as_ref() {
                            None => Vec::new(),
                            Some(v) if v.is_empty() => vec![u32::MAX],
                            Some(v) => v.iter().map(|x| *x as u32).collect(),
                        },
                        null_aware: self.null_aware,
                        dynamic_filter,
                        fetch: self.fetch.map(|f| f as u64),
                    },
                )),
            ),
        }))
    }
}

#[cfg(feature = "proto")]
impl HashJoinExec {
    /// Reconstruct a [`HashJoinExec`] from its protobuf representation.
    pub fn try_from_proto(
        node: &datafusion_proto_models::protobuf::PhysicalPlanNode,
        ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        use datafusion_common::{internal_datafusion_err, plan_datafusion_err};
        use datafusion_proto_models::protobuf;
        use std::any::Any;

        let hashjoin = crate::expect_plan_variant!(
            node,
            protobuf::physical_plan_node::PhysicalPlanType::HashJoin,
            "HashJoinExec",
        );

        let left =
            ctx.decode_required_child(hashjoin.left.as_deref(), "HashJoinExec", "left")?;
        let right = ctx.decode_required_child(
            hashjoin.right.as_deref(),
            "HashJoinExec",
            "right",
        )?;
        let left_schema = left.schema();
        let right_schema = right.schema();

        let on: Vec<(PhysicalExprRef, PhysicalExprRef)> = hashjoin
            .on
            .iter()
            .map(|col| {
                let l = ctx.decode_required_expr(
                    col.left.as_ref(),
                    left_schema.as_ref(),
                    "HashJoinExec",
                    "on.left",
                )?;
                let r = ctx.decode_required_expr(
                    col.right.as_ref(),
                    right_schema.as_ref(),
                    "HashJoinExec",
                    "on.right",
                )?;
                Ok((l, r))
            })
            .collect::<Result<_>>()?;

        let join_type = crate::joins::proto::join_type_from_proto(
            hashjoin.join_type,
            "HashJoinExec",
        )?;
        let null_equality = crate::joins::proto::null_equality_from_proto(
            hashjoin.null_equality,
            "HashJoinExec",
        )?;
        // `PartitionMode` is specific to `HashJoinExec`, so its conversion stays
        // inline (by-name on purpose: the enums are numbered differently).
        let partition_mode = match protobuf::PartitionMode::try_from(
            hashjoin.partition_mode,
        )
        .map_err(|_| {
            internal_datafusion_err!(
                "HashJoinExec: unknown PartitionMode {}",
                hashjoin.partition_mode
            )
        })? {
            protobuf::PartitionMode::CollectLeft => PartitionMode::CollectLeft,
            protobuf::PartitionMode::Partitioned => PartitionMode::Partitioned,
            protobuf::PartitionMode::Auto => PartitionMode::Auto,
        };

        let filter = hashjoin
            .filter
            .as_ref()
            .map(|f| crate::joins::proto::join_filter_from_proto(f, ctx, "HashJoinExec"))
            .transpose()?;

        // Preserve the empty-projection sentinel written by `try_to_proto`.
        let projection = match hashjoin.projection.as_slice() {
            [] => None,
            [u32::MAX] => Some(Vec::new()),
            indices => Some(indices.iter().map(|i| *i as usize).collect()),
        };

        // Restore the row limit that `limit_pushdown` may have pushed into the
        // join. The field is presence-tracked, so a message written before it
        // existed decodes to `None` (no limit) rather than to `Some(0)`.
        //
        // The conversion is checked, not `as usize`: `fetch` is a `u64` on the
        // wire but a `usize` in the plan, and on a 32-bit target `as usize`
        // truncates. A fetch of `1 << 32` would become `0` -- not merely a
        // wrong limit but the worst one, silently turning the query into an
        // empty result. Report the out-of-range value instead. Please do not
        // "simplify" this back to `as usize`.
        let fetch = hashjoin
            .fetch
            .map(|f| {
                usize::try_from(f).map_err(|_| {
                    plan_datafusion_err!(
                        "HashJoinExec: fetch value {f} cannot be represented as usize on this target"
                    )
                })
            })
            .transpose()?;

        let mut hash_join = HashJoinExecBuilder::new(left, right, on, join_type)
            .with_filter(filter)
            .with_projection(projection)
            .with_partition_mode(partition_mode)
            .with_null_equality(null_equality)
            .with_null_aware(hashjoin.null_aware)
            .with_fetch(fetch)
            .build()?;

        if let Some(dynamic_filter_proto) = &hashjoin.dynamic_filter {
            // The dynamic filter is a `DynamicFilterPhysicalExpr` over the probe
            // (right) side; decode against the right schema then downcast.
            let dynamic_filter_expr =
                ctx.decode_expr(dynamic_filter_proto, right_schema.as_ref())?;
            let df = (dynamic_filter_expr as Arc<dyn Any + Send + Sync>)
                .downcast::<DynamicFilterPhysicalExpr>()
                .map_err(|_| {
                    internal_datafusion_err!(
                        "HashJoinExec dynamic_filter did not decode to a DynamicFilterPhysicalExpr"
                    )
                })?;
            hash_join = hash_join.with_dynamic_filter_expr(df)?;
        }

        Ok(Arc::new(hash_join))
    }
}

/// Determines which sides of a join are "preserved" for filter pushdown.
///
/// A preserved side means filters on that side's columns can be safely pushed
/// below the join. This mostly mirrors the logical optimizer's `lr_is_preserved`;
/// semi joins additionally allow join-key filters on the non-output side.
fn lr_is_preserved(join_type: JoinType) -> (bool, bool) {
    match join_type {
        JoinType::Inner => (true, true),
        JoinType::Left => (true, false),
        JoinType::Right => (false, true),
        JoinType::Full => (false, false),
        // Callers restrict the non-output side of semi joins to join-key columns.
        JoinType::LeftSemi | JoinType::RightSemi => (true, true),
        JoinType::LeftAnti | JoinType::LeftMark => (true, false),
        JoinType::RightAnti | JoinType::RightMark => (false, true),
    }
}

/// Accumulator for collecting min/max bounds from build-side data during hash join.
///
/// This struct encapsulates the logic for progressively computing column bounds
/// (minimum and maximum values) for a specific join key expression as batches
/// are processed during the build phase of a hash join.
///
/// The bounds are used for dynamic filter pushdown optimization, where filters
/// based on the actual data ranges can be pushed down to the probe side to
/// eliminate unnecessary data early.
struct CollectLeftAccumulator {
    /// The physical expression to evaluate for each batch
    expr: Arc<dyn PhysicalExpr>,
    /// Accumulator for tracking the minimum value across all batches
    min: MinAccumulator,
    /// Accumulator for tracking the maximum value across all batches
    max: MaxAccumulator,
}

impl CollectLeftAccumulator {
    /// Creates a new accumulator for tracking bounds of a join key expression.
    ///
    /// # Arguments
    /// * `expr` - The physical expression to track bounds for
    /// * `schema` - The schema of the input data
    ///
    /// # Returns
    /// A new `CollectLeftAccumulator` instance configured for the expression's data type
    fn try_new(expr: Arc<dyn PhysicalExpr>, schema: &SchemaRef) -> Result<Self> {
        /// Recursively unwraps dictionary types to get the underlying value type.
        fn dictionary_value_type(data_type: &DataType) -> DataType {
            match data_type {
                DataType::Dictionary(_, value_type) => {
                    dictionary_value_type(value_type.as_ref())
                }
                _ => data_type.clone(),
            }
        }

        let data_type = expr
            .data_type(schema)
            // Min/Max can operate on dictionary data but expect to be initialized with the underlying value type
            .map(|dt| dictionary_value_type(&dt))?;
        Ok(Self {
            expr,
            min: MinAccumulator::try_new(&data_type)?,
            max: MaxAccumulator::try_new(&data_type)?,
        })
    }

    /// Updates the accumulators with values from a new batch.
    ///
    /// Evaluates the expression on the batch and updates both min and max
    /// accumulators with the resulting values.
    ///
    /// # Arguments
    /// * `batch` - The record batch to process
    ///
    /// # Returns
    /// Ok(()) if the update succeeds, or an error if expression evaluation fails
    fn update_batch(&mut self, batch: &RecordBatch) -> Result<()> {
        let array = self.expr.evaluate(batch)?.into_array(batch.num_rows())?;
        self.min.update_batch(std::slice::from_ref(&array))?;
        self.max.update_batch(std::slice::from_ref(&array))?;
        Ok(())
    }

    /// Finalizes the accumulation and returns the computed bounds.
    ///
    /// Consumes self to extract the final min and max values from the accumulators.
    ///
    /// # Returns
    /// The `ColumnBounds` containing the minimum and maximum values observed
    fn evaluate(mut self) -> Result<ColumnBounds> {
        Ok(ColumnBounds::new(
            self.min.evaluate()?,
            self.max.evaluate()?,
        ))
    }
}

/// State for collecting the build-side data during hash join
struct BuildSideState {
    batches: Vec<RecordBatch>,
    num_rows: usize,
    metrics: BuildProbeJoinMetrics,
    reservation: MemoryReservation,
    bounds_accumulators: Option<Vec<CollectLeftAccumulator>>,
    /// Counts the memory of `batches` for `reservation`. Batches can share
    /// underlying buffers (e.g. when the input emits zero-copy slices of one
    /// larger batch), so each buffer must be reserved only once.
    memory_counter: RecordBatchMemoryCounter,
}

impl BuildSideState {
    /// Create a new BuildSideState with optional accumulators for bounds computation
    fn try_new(
        metrics: BuildProbeJoinMetrics,
        reservation: MemoryReservation,
        on_left: Vec<Arc<dyn PhysicalExpr>>,
        schema: &SchemaRef,
        should_compute_dynamic_filters: bool,
    ) -> Result<Self> {
        Ok(Self {
            batches: Vec::new(),
            num_rows: 0,
            metrics,
            reservation,
            memory_counter: RecordBatchMemoryCounter::new(),
            bounds_accumulators: should_compute_dynamic_filters
                .then(|| {
                    on_left
                        .into_iter()
                        .map(|expr| CollectLeftAccumulator::try_new(expr, schema))
                        .collect::<Result<Vec<_>>>()
                })
                .transpose()?,
        })
    }
}

fn should_collect_min_max_for_perfect_hash(
    on_left: &[PhysicalExprRef],
    schema: &SchemaRef,
) -> Result<bool> {
    if on_left.len() != 1 {
        return Ok(false);
    }

    let expr = &on_left[0];
    let data_type = expr.data_type(schema)?;
    Ok(ArrayMap::is_supported_type(&data_type))
}

/// Collects all batches from the left (build) side stream and creates a hash map for joining.
///
/// This function is responsible for:
/// 1. Consuming the entire left stream and collecting all batches into memory
/// 2. Building a hash map from the join key columns for efficient probe operations
/// 3. Computing bounds for dynamic filter pushdown (if enabled)
/// 4. Preparing visited indices bitmap for certain join types
///
/// # Parameters
/// * `random_state` - Random state for consistent hashing across partitions
/// * `left_stream` - Stream of record batches from the build side
/// * `on_left` - Physical expressions for the left side join keys
/// * `metrics` - Metrics collector for tracking memory usage and row counts
/// * `reservation` - Memory reservation tracker for the hash table and data
/// * `with_visited_indices_bitmap` - Whether to track visited indices (for outer joins)
/// * `probe_threads_count` - Number of threads that will probe this hash table
/// * `should_compute_dynamic_filters` - Whether to compute min/max bounds for dynamic filtering
///
/// # Dynamic Filter Coordination
/// When `should_compute_dynamic_filters` is true, this function computes the min/max bounds
/// for each join key column but does NOT update the dynamic filter. Instead, the
/// bounds are stored in the returned `JoinLeftData` and later coordinated by
/// `SharedBuildAccumulator` to ensure all partitions contribute their bounds
/// before updating the filter exactly once.
///
/// # Returns
/// `JoinLeftData` containing the hash map, consolidated batch, join key values,
/// visited indices bitmap, and computed bounds (if requested).
#[expect(clippy::too_many_arguments)]
async fn collect_left_input(
    random_state: RandomState,
    left_stream: SendableRecordBatchStream,
    on_left: Vec<PhysicalExprRef>,
    metrics: BuildProbeJoinMetrics,
    reservation: MemoryReservation,
    with_visited_indices_bitmap: bool,
    probe_threads_count: usize,
    should_compute_dynamic_filters: bool,
    config: Arc<ConfigOptions>,
    null_equality: NullEquality,
    array_map_created_count: Count,
) -> Result<JoinLeftData> {
    let schema = left_stream.schema();

    let should_collect_min_max_for_phj =
        should_collect_min_max_for_perfect_hash(&on_left, &schema)?;

    let initial = BuildSideState::try_new(
        metrics,
        reservation,
        on_left.clone(),
        &schema,
        should_compute_dynamic_filters || should_collect_min_max_for_phj,
    )?;

    let state = left_stream
        .try_fold(initial, |mut state, batch| async move {
            // Update accumulators if computing bounds
            if let Some(ref mut accumulators) = state.bounds_accumulators {
                for accumulator in accumulators {
                    accumulator.update_batch(&batch)?;
                }
            }

            // Decide if we spill or not
            let batch_size = state.memory_counter.count_batch(&batch);
            // Reserve memory for incoming batch
            state.reservation.try_grow(batch_size)?;
            // Update metrics
            state.metrics.build_mem_used.add(batch_size);
            state.metrics.build_input_batches.add(1);
            state.metrics.build_input_rows.add(batch.num_rows());
            // Update row count
            state.num_rows += batch.num_rows();
            // Push batch to output
            state.batches.push(batch);
            Ok(state)
        })
        .await?;

    // Extract fields from state
    let BuildSideState {
        batches,
        num_rows,
        metrics,
        mut reservation,
        bounds_accumulators,
        memory_counter: _,
    } = state;

    // Compute bounds
    let mut bounds = match bounds_accumulators {
        Some(accumulators) if num_rows > 0 => {
            let bounds = accumulators
                .into_iter()
                .map(CollectLeftAccumulator::evaluate)
                .collect::<Result<Vec<_>>>()?;
            Some(PartitionBounds::new(bounds))
        }
        _ => None,
    };

    let (join_hash_map, batch, left_values) =
        if let Some((array_map, batch, left_value)) = try_create_array_map(
            &bounds,
            &schema,
            &batches,
            &on_left,
            &mut reservation,
            config.execution.perfect_hash_join_small_build_threshold,
            config.execution.perfect_hash_join_min_key_density,
            null_equality,
        )? {
            array_map_created_count.add(1);
            metrics.build_mem_used.add(array_map.size());

            (Map::ArrayMap(array_map), batch, left_value)
        } else {
            // Estimation of memory size, required for hashtable, prior to allocation.
            // Final result can be verified using `RawTable.allocation_info()`
            let fixed_size_u32 = size_of::<JoinHashMapU32>();
            let fixed_size_u64 = size_of::<JoinHashMapU64>();

            // Use `u32` indices for the JoinHashMap when num_rows ≤ u32::MAX, otherwise use the
            // `u64` indice variant
            // Arc is used instead of Box to allow sharing with SharedBuildAccumulator for hash map pushdown
            let mut hashmap: Box<dyn JoinHashMapType> = if num_rows > u32::MAX as usize {
                let estimated_hashtable_size =
                    estimate_memory_size::<(u64, u64)>(num_rows, fixed_size_u64)?;
                reservation.try_grow(estimated_hashtable_size)?;
                metrics.build_mem_used.add(estimated_hashtable_size);
                Box::new(JoinHashMapU64::with_capacity(num_rows))
            } else {
                let estimated_hashtable_size =
                    estimate_memory_size::<(u32, u64)>(num_rows, fixed_size_u32)?;
                reservation.try_grow(estimated_hashtable_size)?;
                metrics.build_mem_used.add(estimated_hashtable_size);
                Box::new(JoinHashMapU32::with_capacity(num_rows))
            };

            let mut hashes_buffer = Vec::new();
            let mut offset = 0;

            let batches_iter = batches.iter().rev();

            // Updating hashmap starting from the last batch
            for batch in batches_iter.clone() {
                hashes_buffer.clear();
                hashes_buffer.resize(batch.num_rows(), 0);
                update_hash(
                    &on_left,
                    batch,
                    &mut *hashmap,
                    offset,
                    &random_state,
                    &mut hashes_buffer,
                    0,
                    true,
                    null_equality,
                )?;
                offset += batch.num_rows();
            }

            // Merge all batches into a single batch, so we can directly index into the arrays
            let batch = concat_batches(&schema, batches_iter.clone())?;

            let left_values = evaluate_expressions_to_arrays(&on_left, &batch)?;

            (Map::HashMap(hashmap), batch, left_values)
        };

    // Reserve additional memory for visited indices bitmap and create shared builder
    let visited_indices_bitmap = if with_visited_indices_bitmap {
        let bitmap_size = bit_util::ceil(batch.num_rows(), 8);
        reservation.try_grow(bitmap_size)?;
        metrics.build_mem_used.add(bitmap_size);

        let mut bitmap_buffer = BooleanBufferBuilder::new(batch.num_rows());
        bitmap_buffer.append_n(num_rows, false);
        bitmap_buffer
    } else {
        BooleanBufferBuilder::new(0)
    };

    let map = Arc::new(join_hash_map);

    let membership = if num_rows == 0 {
        PushdownStrategy::Empty
    } else {
        // If the build side is small enough we can use IN list pushdown.
        // If it's too big we fall back to pushing down a reference to the hash table.
        // See `PushdownStrategy` for more details.
        let estimated_size = left_values
            .iter()
            .map(|arr| arr.get_array_memory_size())
            .sum::<usize>();
        if left_values.is_empty()
            || left_values[0].is_empty()
            || estimated_size > config.optimizer.hash_join_inlist_pushdown_max_size
            || map.num_of_distinct_key()
                > config
                    .optimizer
                    .hash_join_inlist_pushdown_max_distinct_values
        {
            PushdownStrategy::Map(Arc::clone(&map))
        } else if let Some(in_list_values) = build_struct_inlist_values(&left_values)? {
            PushdownStrategy::InList(in_list_values)
        } else {
            PushdownStrategy::Map(Arc::clone(&map))
        }
    };

    if should_collect_min_max_for_phj && !should_compute_dynamic_filters {
        bounds = None;
    }

    let data = JoinLeftData {
        map,
        batch,
        values: left_values,
        visited_indices_bitmap: Mutex::new(visited_indices_bitmap),
        probe_threads_counter: AtomicUsize::new(probe_threads_count),
        _reservation: reservation,
        bounds,
        membership,
        probe_side_non_empty: AtomicBool::new(false),
        probe_side_has_null: AtomicBool::new(false),
    };

    Ok(data)
}

#[cfg(test)]
mod tests {
    use super::*;

    fn assert_phj_used(metrics: &MetricsSet, use_phj: bool) {
        if use_phj {
            assert!(
                metrics
                    .sum_by_name(ARRAY_MAP_CREATED_COUNT_METRIC_NAME)
                    .expect("should have array_map_created_count metrics")
                    .as_usize()
                    >= 1
            );
        } else {
            assert_eq!(
                metrics
                    .sum_by_name(ARRAY_MAP_CREATED_COUNT_METRIC_NAME)
                    .map(|v| v.as_usize())
                    .unwrap_or(0),
                0
            )
        }
    }

    fn build_schema_and_on() -> Result<(SchemaRef, SchemaRef, JoinOn)> {
        let left_schema = Arc::new(Schema::new(vec![
            Field::new("a1", DataType::Int32, true),
            Field::new("b1", DataType::Int32, true),
        ]));
        let right_schema = Arc::new(Schema::new(vec![
            Field::new("a2", DataType::Int32, true),
            Field::new("b1", DataType::Int32, true),
        ]));
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left_schema)?) as _,
            Arc::new(Column::new_with_schema("b1", &right_schema)?) as _,
        )];
        Ok((left_schema, right_schema, on))
    }

    use crate::coalesce_partitions::CoalescePartitionsExec;
    use crate::execution_plan::Boundedness;
    use crate::filter::FilterExecBuilder;
    use crate::joins::hash_join::stream::lookup_join_hashmap;
    use crate::test::{TestMemoryExec, assert_join_metrics};
    use crate::{ChildrenPropertiesMode, ReplaceChildrenOptions};
    use crate::{
        common, expressions::Column, repartition::RepartitionExec, test::build_table_i32,
        test::exec::MockExec,
    };

    use arrow::array::{
        Date32Array, Int32Array, Int64Array, StructArray, UInt32Array, UInt64Array,
    };
    use arrow::buffer::NullBuffer;
    use arrow::datatypes::{DataType, Field};
    use datafusion_common::hash_utils::create_hashes;
    use datafusion_common::test_util::{batches_to_sort_string, batches_to_string};
    use datafusion_common::{
        ScalarValue, assert_batches_eq, assert_batches_sorted_eq, assert_contains,
        exec_err, internal_err,
    };
    use datafusion_execution::config::SessionConfig;
    use datafusion_execution::runtime_env::RuntimeEnvBuilder;
    use datafusion_expr::Operator;
    use datafusion_physical_expr::expressions::{BinaryExpr, Literal};
    use datafusion_physical_expr::{
        EquivalenceProperties, PhysicalSortExpr, RangePartitioning, SplitPoint,
    };
    use hashbrown::HashTable;
    use insta::{allow_duplicates, assert_snapshot};
    use rstest::*;
    use rstest_reuse::*;

    #[derive(Debug)]
    struct PartitionedTestExec {
        cache: Arc<PlanProperties>,
    }

    impl PartitionedTestExec {
        fn try_new(schema: SchemaRef, partitioning: Partitioning) -> Result<Self> {
            Ok(Self {
                cache: Arc::new(PlanProperties::new(
                    EquivalenceProperties::new(Arc::clone(&schema)),
                    partitioning,
                    EmissionType::Incremental,
                    Boundedness::Bounded,
                )),
            })
        }
    }

    impl DisplayAs for PartitionedTestExec {
        fn fmt_as(&self, _t: DisplayFormatType, f: &mut fmt::Formatter) -> fmt::Result {
            write!(f, "PartitionedTestExec")
        }
    }

    impl ExecutionPlan for PartitionedTestExec {
        fn name(&self) -> &'static str {
            "PartitionedTestExec"
        }

        fn properties(&self) -> &Arc<PlanProperties> {
            &self.cache
        }

        fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
            vec![]
        }

        fn apply_expressions(
            &self,
            _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
        ) -> Result<TreeNodeRecursion> {
            Ok(TreeNodeRecursion::Continue)
        }

        fn replace_children(
            self: Arc<Self>,
            _: Vec<Arc<dyn ExecutionPlan>>,
            _: ReplaceChildrenOptions,
        ) -> Result<Arc<dyn ExecutionPlan>> {
            Ok(self)
        }

        fn with_new_children(
            self: Arc<Self>,
            children: Vec<Arc<dyn ExecutionPlan>>,
        ) -> Result<Arc<dyn ExecutionPlan>> {
            self.replace_children(
                children,
                ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
            )
        }

        fn execute(
            &self,
            _partition: usize,
            _context: Arc<TaskContext>,
        ) -> Result<SendableRecordBatchStream> {
            unreachable!()
        }
    }

    fn div_ceil(a: usize, b: usize) -> usize {
        a.div_ceil(b)
    }

    #[template]
    #[rstest]
    fn hash_join_exec_configs(
        #[values(8192, 10, 5, 2, 1)] batch_size: usize,
        #[values(true, false)] use_perfect_hash_join_as_possible: bool,
    ) {
    }

    fn prepare_task_ctx(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Arc<TaskContext> {
        let mut session_config = SessionConfig::default().with_batch_size(batch_size);

        if use_perfect_hash_join_as_possible {
            session_config
                .options_mut()
                .execution
                .perfect_hash_join_small_build_threshold = 819200;
            session_config
                .options_mut()
                .execution
                .perfect_hash_join_min_key_density = 0.0;
        } else {
            session_config
                .options_mut()
                .execution
                .perfect_hash_join_small_build_threshold = 0;
            session_config
                .options_mut()
                .execution
                .perfect_hash_join_min_key_density = f64::INFINITY;
        }
        Arc::new(TaskContext::default().with_session_config(session_config))
    }

    fn build_table(
        a: (&str, &Vec<i32>),
        b: (&str, &Vec<i32>),
        c: (&str, &Vec<i32>),
    ) -> Arc<dyn ExecutionPlan> {
        let batch = build_table_i32(a, b, c);
        let schema = batch.schema();
        TestMemoryExec::try_new_exec(&[vec![batch]], schema, None).unwrap()
    }

    /// Build a table with two columns supporting nullable values
    fn build_table_two_cols(
        a: (&str, &Vec<Option<i32>>),
        b: (&str, &Vec<Option<i32>>),
    ) -> Arc<dyn ExecutionPlan> {
        let schema = Arc::new(Schema::new(vec![
            Field::new(a.0, DataType::Int32, true),
            Field::new(b.0, DataType::Int32, true),
        ]));
        let batch = RecordBatch::try_new(
            Arc::clone(&schema),
            vec![
                Arc::new(Int32Array::from(a.1.clone())),
                Arc::new(Int32Array::from(b.1.clone())),
            ],
        )
        .unwrap();
        TestMemoryExec::try_new_exec(&[vec![batch]], schema, None).unwrap()
    }

    fn join(
        left: Arc<dyn ExecutionPlan>,
        right: Arc<dyn ExecutionPlan>,
        on: JoinOn,
        join_type: &JoinType,
        null_equality: NullEquality,
    ) -> Result<HashJoinExec> {
        HashJoinExec::try_new(
            left,
            right,
            on,
            None,
            join_type,
            None,
            PartitionMode::CollectLeft,
            null_equality,
            false,
        )
    }

    fn join_with_filter(
        left: Arc<dyn ExecutionPlan>,
        right: Arc<dyn ExecutionPlan>,
        on: JoinOn,
        filter: JoinFilter,
        join_type: &JoinType,
        null_equality: NullEquality,
    ) -> Result<HashJoinExec> {
        HashJoinExec::try_new(
            left,
            right,
            on,
            Some(filter),
            join_type,
            None,
            PartitionMode::CollectLeft,
            null_equality,
            false,
        )
    }

    fn empty_build_with_probe_error_inputs()
    -> (Arc<dyn ExecutionPlan>, Arc<dyn ExecutionPlan>, JoinOn) {
        let left_batch =
            build_table_i32(("a1", &vec![]), ("b1", &vec![]), ("c1", &vec![]));
        let left_schema = left_batch.schema();
        let left: Arc<dyn ExecutionPlan> = TestMemoryExec::try_new_exec(
            &[vec![left_batch]],
            Arc::clone(&left_schema),
            None,
        )
        .unwrap();

        let err = exec_err!("bad data error");
        let right_batch =
            build_table_i32(("a2", &vec![]), ("b1", &vec![]), ("c2", &vec![]));
        let right_schema = right_batch.schema();
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left_schema).unwrap()) as _,
            Arc::new(Column::new_with_schema("b1", &right_schema).unwrap()) as _,
        )];
        let right: Arc<dyn ExecutionPlan> = Arc::new(
            MockExec::new(vec![Ok(right_batch), err], right_schema)
                .with_use_task(false)
                // The planted error must only surface if the probe side is
                // polled, not when a parent node computes statistics during
                // planning.
                .with_unknown_statistics(),
        );

        (left, right, on)
    }

    async fn assert_empty_build_probe_behavior(
        join_types: &[JoinType],
        expect_probe_error: bool,
        with_filter: bool,
    ) {
        let (left, right, on) = empty_build_with_probe_error_inputs();
        let filter = prepare_join_filter();

        for join_type in join_types {
            let join = if with_filter {
                join_with_filter(
                    Arc::clone(&left),
                    Arc::clone(&right),
                    on.clone(),
                    filter.clone(),
                    join_type,
                    NullEquality::NullEqualsNothing,
                )
                .unwrap()
            } else {
                join(
                    Arc::clone(&left),
                    Arc::clone(&right),
                    on.clone(),
                    join_type,
                    NullEquality::NullEqualsNothing,
                )
                .unwrap()
            };

            let result = common::collect(
                join.execute(0, Arc::new(TaskContext::default())).unwrap(),
            )
            .await;

            if expect_probe_error {
                let result_string = result.unwrap_err().to_string();
                assert!(
                    result_string.contains("bad data error"),
                    "actual: {result_string}"
                );
            } else {
                let batches = result.unwrap();
                assert!(
                    batches.is_empty(),
                    "expected no output batches for {join_type}, got {batches:?}"
                );
            }
        }
    }

    fn hash_join_with_dynamic_filter(
        left: Arc<dyn ExecutionPlan>,
        right: Arc<dyn ExecutionPlan>,
        on: JoinOn,
        join_type: JoinType,
    ) -> Result<(HashJoinExec, Arc<DynamicFilterPhysicalExpr>)> {
        hash_join_with_dynamic_filter_and_mode(
            left,
            right,
            on,
            join_type,
            PartitionMode::CollectLeft,
        )
    }

    fn hash_join_with_dynamic_filter_and_mode(
        left: Arc<dyn ExecutionPlan>,
        right: Arc<dyn ExecutionPlan>,
        on: JoinOn,
        join_type: JoinType,
        mode: PartitionMode,
    ) -> Result<(HashJoinExec, Arc<DynamicFilterPhysicalExpr>)> {
        let dynamic_filter = HashJoinExec::create_dynamic_filter(&on);
        let consumer: Arc<dyn PhysicalExpr> = Arc::clone(&dynamic_filter) as _;
        let right = Arc::new(FilterExecBuilder::new(consumer, right).build()?);
        let mut join = HashJoinExec::try_new(
            left,
            right,
            on,
            None,
            &join_type,
            None,
            mode,
            NullEquality::NullEqualsNothing,
            false,
        )?;
        join.dynamic_filter = Some(HashJoinExecDynamicFilter {
            filter: Arc::clone(&dynamic_filter),
            build_accumulator: OnceLock::new(),
        });

        Ok((join, dynamic_filter))
    }

    async fn join_collect(
        left: Arc<dyn ExecutionPlan>,
        right: Arc<dyn ExecutionPlan>,
        on: JoinOn,
        join_type: &JoinType,
        null_equality: NullEquality,
        context: Arc<TaskContext>,
    ) -> Result<(Vec<String>, Vec<RecordBatch>, MetricsSet)> {
        let join = join(left, right, on, join_type, null_equality)?;
        let columns_header = columns(&join.schema());

        let stream = join.execute(0, context)?;
        let batches = common::collect(stream).await?;
        let metrics = join.metrics().unwrap();

        Ok((columns_header, batches, metrics))
    }

    async fn partitioned_join_collect(
        left: Arc<dyn ExecutionPlan>,
        right: Arc<dyn ExecutionPlan>,
        on: JoinOn,
        join_type: &JoinType,
        null_equality: NullEquality,
        context: Arc<TaskContext>,
    ) -> Result<(Vec<String>, Vec<RecordBatch>, MetricsSet)> {
        join_collect_with_partition_mode(
            left,
            right,
            on,
            join_type,
            PartitionMode::Partitioned,
            null_equality,
            context,
        )
        .await
    }

    async fn join_collect_with_partition_mode(
        left: Arc<dyn ExecutionPlan>,
        right: Arc<dyn ExecutionPlan>,
        on: JoinOn,
        join_type: &JoinType,
        partition_mode: PartitionMode,
        null_equality: NullEquality,
        context: Arc<TaskContext>,
    ) -> Result<(Vec<String>, Vec<RecordBatch>, MetricsSet)> {
        let partition_count = 4;

        let (left_expr, right_expr) = on
            .iter()
            .map(|(l, r)| (Arc::clone(l), Arc::clone(r)))
            .unzip();

        let left_repartitioned: Arc<dyn ExecutionPlan> = match partition_mode {
            PartitionMode::CollectLeft => Arc::new(CoalescePartitionsExec::new(left)),
            PartitionMode::Partitioned => Arc::new(RepartitionExec::try_new(
                left,
                Partitioning::Hash(left_expr, partition_count),
            )?),
            PartitionMode::Auto => {
                return internal_err!("Unexpected PartitionMode::Auto in join tests");
            }
        };

        let right_repartitioned: Arc<dyn ExecutionPlan> = match partition_mode {
            PartitionMode::CollectLeft => {
                let partition_column_name = right.schema().field(0).name().clone();
                let partition_expr = vec![Arc::new(Column::new_with_schema(
                    &partition_column_name,
                    &right.schema(),
                )?) as _];
                Arc::new(RepartitionExec::try_new(
                    right,
                    Partitioning::Hash(partition_expr, partition_count),
                )?) as _
            }
            PartitionMode::Partitioned => Arc::new(RepartitionExec::try_new(
                right,
                Partitioning::Hash(right_expr, partition_count),
            )?),
            PartitionMode::Auto => {
                return internal_err!("Unexpected PartitionMode::Auto in join tests");
            }
        };

        let join = HashJoinExec::try_new(
            left_repartitioned,
            right_repartitioned,
            on,
            None,
            join_type,
            None,
            partition_mode,
            null_equality,
            false,
        )?;

        let columns = columns(&join.schema());

        let mut batches = vec![];
        for i in 0..partition_count {
            let stream = join.execute(i, Arc::clone(&context))?;
            let more_batches = common::collect(stream).await?;
            batches.extend(
                more_batches
                    .into_iter()
                    .filter(|b| b.num_rows() > 0)
                    .collect::<Vec<_>>(),
            );
        }
        let metrics = join.metrics().unwrap();

        Ok((columns, batches, metrics))
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_inner_one(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 5]), // this has a repetition
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30]),
            ("b1", &vec![4, 5, 6]),
            ("c2", &vec![70, 80, 90]),
        );

        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
        )];

        let (columns, batches, metrics) = join_collect(
            Arc::clone(&left),
            Arc::clone(&right),
            on.clone(),
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b1", "c2"]);

        allow_duplicates! {
            // Inner join output is expected to preserve both inputs order
            assert_snapshot!(batches_to_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b1 | c2 |
            +----+----+----+----+----+----+
            | 1  | 4  | 7  | 10 | 4  | 70 |
            | 2  | 5  | 8  | 20 | 5  | 80 |
            | 3  | 5  | 9  | 20 | 5  | 80 |
            +----+----+----+----+----+----+
            ");
        }

        assert_join_metrics!(metrics, 3);
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn partitioned_join_inner_one(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 5]), // this has a repetition
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30]),
            ("b1", &vec![4, 5, 6]),
            ("c2", &vec![70, 80, 90]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
        )];

        let (columns, batches, metrics) = partitioned_join_collect(
            Arc::clone(&left),
            Arc::clone(&right),
            on.clone(),
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b1", "c2"]);

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b1 | c2 |
            +----+----+----+----+----+----+
            | 1  | 4  | 7  | 10 | 4  | 70 |
            | 2  | 5  | 8  | 20 | 5  | 80 |
            | 3  | 5  | 9  | 20 | 5  | 80 |
            +----+----+----+----+----+----+
            ");
        }

        assert_join_metrics!(metrics, 3);
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[tokio::test]
    async fn join_inner_one_no_shared_column_names() -> Result<()> {
        let task_ctx = Arc::new(TaskContext::default());
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 5]), // this has a repetition
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30]),
            ("b2", &vec![4, 5, 6]),
            ("c2", &vec![70, 80, 90]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
        )];

        let (columns, batches, metrics) = join_collect(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b2", "c2"]);

        // Inner join output is expected to preserve both inputs order
        allow_duplicates! {
            assert_snapshot!(batches_to_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b2 | c2 |
            +----+----+----+----+----+----+
            | 1  | 4  | 7  | 10 | 4  | 70 |
            | 2  | 5  | 8  | 20 | 5  | 80 |
            | 3  | 5  | 9  | 20 | 5  | 80 |
            +----+----+----+----+----+----+
            ");
        }

        assert_join_metrics!(metrics, 3);

        Ok(())
    }

    #[tokio::test]
    async fn join_inner_one_randomly_ordered() -> Result<()> {
        let task_ctx = Arc::new(TaskContext::default());
        let left = build_table(
            ("a1", &vec![0, 3, 2, 1]),
            ("b1", &vec![4, 5, 5, 4]),
            ("c1", &vec![6, 9, 8, 7]),
        );
        let right = build_table(
            ("a2", &vec![20, 30, 10]),
            ("b2", &vec![5, 6, 4]),
            ("c2", &vec![80, 90, 70]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
        )];

        let (columns, batches, metrics) = join_collect(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b2", "c2"]);

        // Inner join output is expected to preserve both inputs order
        allow_duplicates! {
            assert_snapshot!(batches_to_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b2 | c2 |
            +----+----+----+----+----+----+
            | 3  | 5  | 9  | 20 | 5  | 80 |
            | 2  | 5  | 8  | 20 | 5  | 80 |
            | 0  | 4  | 6  | 10 | 4  | 70 |
            | 1  | 4  | 7  | 10 | 4  | 70 |
            +----+----+----+----+----+----+
            ");
        }

        assert_join_metrics!(metrics, 4);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_inner_two(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 2]),
            ("b2", &vec![1, 2, 2]),
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b2", &vec![1, 2, 2]),
            ("c2", &vec![70, 80, 90]),
        );
        let on = vec![
            (
                Arc::new(Column::new_with_schema("a1", &left.schema())?) as _,
                Arc::new(Column::new_with_schema("a1", &right.schema())?) as _,
            ),
            (
                Arc::new(Column::new_with_schema("b2", &left.schema())?) as _,
                Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
            ),
        ];

        let (columns, batches, metrics) = join_collect(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b2", "c1", "a1", "b2", "c2"]);

        let expected_batch_count = if cfg!(not(feature = "force_hash_collisions")) {
            // Expected number of hash table matches = 3
            // in case batch_size is 1 - additional empty batch for remaining 3-2 row
            let mut expected_batch_count = div_ceil(3, batch_size);
            if batch_size == 1 {
                expected_batch_count += 1;
            }
            expected_batch_count
        } else {
            // With hash collisions enabled, all records will match each other
            // and filtered later.
            div_ceil(9, batch_size)
        };

        // With batch coalescing, we may have fewer batches than expected
        assert!(
            batches.len() <= expected_batch_count,
            "expected at most {expected_batch_count} batches, got {}",
            batches.len()
        );

        // Inner join output is expected to preserve both inputs order
        allow_duplicates! {
            assert_snapshot!(batches_to_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b2 | c1 | a1 | b2 | c2 |
            +----+----+----+----+----+----+
            | 1  | 1  | 7  | 1  | 1  | 70 |
            | 2  | 2  | 8  | 2  | 2  | 80 |
            | 2  | 2  | 9  | 2  | 2  | 80 |
            +----+----+----+----+----+----+
            ");
        }

        assert_join_metrics!(metrics, 3);

        Ok(())
    }

    /// Test where the left has 2 parts, the right with 1 part => 1 part
    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_inner_one_two_parts_left(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let batch1 = build_table_i32(
            ("a1", &vec![1, 2]),
            ("b2", &vec![1, 2]),
            ("c1", &vec![7, 8]),
        );
        let batch2 =
            build_table_i32(("a1", &vec![2]), ("b2", &vec![2]), ("c1", &vec![9]));
        let schema = batch1.schema();
        let left =
            TestMemoryExec::try_new_exec(&[vec![batch1], vec![batch2]], schema, None)
                .unwrap();
        let left = Arc::new(CoalescePartitionsExec::new(left));

        let right = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b2", &vec![1, 2, 2]),
            ("c2", &vec![70, 80, 90]),
        );
        let on = vec![
            (
                Arc::new(Column::new_with_schema("a1", &left.schema())?) as _,
                Arc::new(Column::new_with_schema("a1", &right.schema())?) as _,
            ),
            (
                Arc::new(Column::new_with_schema("b2", &left.schema())?) as _,
                Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
            ),
        ];

        let (columns, batches, metrics) = join_collect(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b2", "c1", "a1", "b2", "c2"]);

        let expected_batch_count = if cfg!(not(feature = "force_hash_collisions")) {
            // Expected number of hash table matches = 3
            // in case batch_size is 1 - additional empty batch for remaining 3-2 row
            let mut expected_batch_count = div_ceil(3, batch_size);
            if batch_size == 1 {
                expected_batch_count += 1;
            }
            expected_batch_count
        } else {
            // With hash collisions enabled, all records will match each other
            // and filtered later.
            div_ceil(9, batch_size)
        };

        // With batch coalescing, we may have fewer batches than expected
        assert!(
            batches.len() <= expected_batch_count,
            "expected at most {expected_batch_count} batches, got {}",
            batches.len()
        );

        // Inner join output is expected to preserve both inputs order
        allow_duplicates! {
            assert_snapshot!(batches_to_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b2 | c1 | a1 | b2 | c2 |
            +----+----+----+----+----+----+
            | 1  | 1  | 7  | 1  | 1  | 70 |
            | 2  | 2  | 8  | 2  | 2  | 80 |
            | 2  | 2  | 9  | 2  | 2  | 80 |
            +----+----+----+----+----+----+
            ");
        }

        assert_join_metrics!(metrics, 3);

        Ok(())
    }

    #[tokio::test]
    async fn join_inner_one_two_parts_left_randomly_ordered() -> Result<()> {
        let task_ctx = Arc::new(TaskContext::default());
        let batch1 = build_table_i32(
            ("a1", &vec![0, 3]),
            ("b1", &vec![4, 5]),
            ("c1", &vec![6, 9]),
        );
        let batch2 = build_table_i32(
            ("a1", &vec![2, 1]),
            ("b1", &vec![5, 4]),
            ("c1", &vec![8, 7]),
        );
        let schema = batch1.schema();

        let left =
            TestMemoryExec::try_new_exec(&[vec![batch1], vec![batch2]], schema, None)
                .unwrap();
        let left = Arc::new(CoalescePartitionsExec::new(left));
        let right = build_table(
            ("a2", &vec![20, 30, 10]),
            ("b2", &vec![5, 6, 4]),
            ("c2", &vec![80, 90, 70]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
        )];

        let (columns, batches, metrics) = join_collect(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b2", "c2"]);

        // Inner join output is expected to preserve both inputs order
        allow_duplicates! {
            assert_snapshot!(batches_to_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b2 | c2 |
            +----+----+----+----+----+----+
            | 3  | 5  | 9  | 20 | 5  | 80 |
            | 2  | 5  | 8  | 20 | 5  | 80 |
            | 0  | 4  | 6  | 10 | 4  | 70 |
            | 1  | 4  | 7  | 10 | 4  | 70 |
            +----+----+----+----+----+----+
            ");
        }

        assert_join_metrics!(metrics, 4);

        Ok(())
    }

    /// Test where the left has 1 part, the right has 2 parts => 2 parts
    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_inner_one_two_parts_right(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 5]), // this has a repetition
            ("c1", &vec![7, 8, 9]),
        );

        let batch1 = build_table_i32(
            ("a2", &vec![10, 20]),
            ("b1", &vec![4, 6]),
            ("c2", &vec![70, 80]),
        );
        let batch2 =
            build_table_i32(("a2", &vec![30]), ("b1", &vec![5]), ("c2", &vec![90]));
        let schema = batch1.schema();
        let right =
            TestMemoryExec::try_new_exec(&[vec![batch1], vec![batch2]], schema, None)
                .unwrap();

        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
        )];

        let join = join(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
        )?;

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b1", "c2"]);

        // first part
        let stream = join.execute(0, Arc::clone(&task_ctx))?;
        let batches = common::collect(stream).await?;

        let expected_batch_count = if cfg!(not(feature = "force_hash_collisions")) {
            // Expected number of hash table matches for first right batch = 1
            // and additional empty batch for non-joined 20-6-80
            let mut expected_batch_count = div_ceil(1, batch_size);
            if batch_size == 1 {
                expected_batch_count += 1;
            }
            expected_batch_count
        } else {
            // With hash collisions enabled, all records will match each other
            // and filtered later.
            div_ceil(6, batch_size)
        };
        // With batch coalescing, we may have fewer batches than expected
        assert!(
            batches.len() <= expected_batch_count,
            "expected at most {expected_batch_count} batches, got {}",
            batches.len()
        );

        // Inner join output is expected to preserve both inputs order
        allow_duplicates! {
            assert_snapshot!(batches_to_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b1 | c2 |
            +----+----+----+----+----+----+
            | 1  | 4  | 7  | 10 | 4  | 70 |
            +----+----+----+----+----+----+
            ");
        }

        // second part
        let stream = join.execute(1, Arc::clone(&task_ctx))?;
        let batches = common::collect(stream).await?;

        let expected_batch_count = if cfg!(not(feature = "force_hash_collisions")) {
            // Expected number of hash table matches for second right batch = 2
            div_ceil(2, batch_size)
        } else {
            // With hash collisions enabled, all records will match each other
            // and filtered later.
            div_ceil(3, batch_size)
        };
        // With batch coalescing, we may have fewer batches than expected
        assert!(
            batches.len() <= expected_batch_count,
            "expected at most {expected_batch_count} batches, got {}",
            batches.len()
        );

        // Inner join output is expected to preserve both inputs order
        allow_duplicates! {
            assert_snapshot!(batches_to_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b1 | c2 |
            +----+----+----+----+----+----+
            | 2  | 5  | 8  | 30 | 5  | 90 |
            | 3  | 5  | 9  | 30 | 5  | 90 |
            +----+----+----+----+----+----+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    fn build_table_two_batches(
        a: (&str, &Vec<i32>),
        b: (&str, &Vec<i32>),
        c: (&str, &Vec<i32>),
    ) -> Arc<dyn ExecutionPlan> {
        let batch = build_table_i32(a, b, c);
        let schema = batch.schema();
        TestMemoryExec::try_new_exec(&[vec![batch.clone(), batch]], schema, None).unwrap()
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_left_multi_batch(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]), // 7 does not exist on the right
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table_two_batches(
            ("a2", &vec![10, 20, 30]),
            ("b1", &vec![4, 5, 6]),
            ("c2", &vec![70, 80, 90]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema()).unwrap()) as _,
        )];

        let join = join(
            Arc::clone(&left),
            Arc::clone(&right),
            on.clone(),
            &JoinType::Left,
            NullEquality::NullEqualsNothing,
        )
        .unwrap();

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b1", "c2"]);

        let (_, batches, metrics) = join_collect(
            Arc::clone(&left),
            Arc::clone(&right),
            on.clone(),
            &JoinType::Left,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b1 | c2 |
            +----+----+----+----+----+----+
            | 1  | 4  | 7  | 10 | 4  | 70 |
            | 1  | 4  | 7  | 10 | 4  | 70 |
            | 2  | 5  | 8  | 20 | 5  | 80 |
            | 2  | 5  | 8  | 20 | 5  | 80 |
            | 3  | 7  | 9  |    |    |    |
            +----+----+----+----+----+----+
            ");
        }

        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);
        return Ok(());
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_full_multi_batch(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]), // 7 does not exist on the right
            ("c1", &vec![7, 8, 9]),
        );
        // create two identical batches for the right side
        let right = build_table_two_batches(
            ("a2", &vec![10, 20, 30]),
            ("b2", &vec![4, 5, 6]),
            ("c2", &vec![70, 80, 90]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema()).unwrap()) as _,
        )];

        let join = join(
            left,
            right,
            on,
            &JoinType::Full,
            NullEquality::NullEqualsNothing,
        )
        .unwrap();

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b2", "c2"]);

        let stream = join.execute(0, task_ctx).unwrap();
        let batches = common::collect(stream).await.unwrap();
        let metrics = join.metrics().unwrap();

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b2 | c2 |
            +----+----+----+----+----+----+
            |    |    |    | 30 | 6  | 90 |
            |    |    |    | 30 | 6  | 90 |
            | 1  | 4  | 7  | 10 | 4  | 70 |
            | 1  | 4  | 7  | 10 | 4  | 70 |
            | 2  | 5  | 8  | 20 | 5  | 80 |
            | 2  | 5  | 8  | 20 | 5  | 80 |
            | 3  | 7  | 9  |    |    |    |
            +----+----+----+----+----+----+
            ");
        }

        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_left_empty_right(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]),
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table_i32(("a2", &vec![]), ("b1", &vec![]), ("c2", &vec![]));
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema()).unwrap()) as _,
        )];
        let schema = right.schema();
        let right = TestMemoryExec::try_new_exec(&[vec![right]], schema, None).unwrap();
        let join = join(
            left,
            right,
            on,
            &JoinType::Left,
            NullEquality::NullEqualsNothing,
        )
        .unwrap();

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b1", "c2"]);

        let stream = join.execute(0, task_ctx).unwrap();
        let batches = common::collect(stream).await.unwrap();
        let metrics = join.metrics().unwrap();

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b1 | c2 |
            +----+----+----+----+----+----+
            | 1  | 4  | 7  |    |    |    |
            | 2  | 5  | 8  |    |    |    |
            | 3  | 7  | 9  |    |    |    |
            +----+----+----+----+----+----+
            ");
        }

        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_full_empty_right(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]),
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table_i32(("a2", &vec![]), ("b2", &vec![]), ("c2", &vec![]));
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema()).unwrap()) as _,
        )];
        let schema = right.schema();
        let right = TestMemoryExec::try_new_exec(&[vec![right]], schema, None).unwrap();
        let join = join(
            left,
            right,
            on,
            &JoinType::Full,
            NullEquality::NullEqualsNothing,
        )
        .unwrap();

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b2", "c2"]);

        let stream = join.execute(0, task_ctx).unwrap();
        let batches = common::collect(stream).await.unwrap();
        let metrics = join.metrics().unwrap();

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b2 | c2 |
            +----+----+----+----+----+----+
            | 1  | 4  | 7  |    |    |    |
            | 2  | 5  | 8  |    |    |    |
            | 3  | 7  | 9  |    |    |    |
            +----+----+----+----+----+----+
            ");
        }

        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_left_one(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]), // 7 does not exist on the right
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30]),
            ("b1", &vec![4, 5, 6]),
            ("c2", &vec![70, 80, 90]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
        )];

        let (columns, batches, metrics) = join_collect(
            Arc::clone(&left),
            Arc::clone(&right),
            on.clone(),
            &JoinType::Left,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b1", "c2"]);

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b1 | c2 |
            +----+----+----+----+----+----+
            | 1  | 4  | 7  | 10 | 4  | 70 |
            | 2  | 5  | 8  | 20 | 5  | 80 |
            | 3  | 7  | 9  |    |    |    |
            +----+----+----+----+----+----+
            ");
        }

        assert_join_metrics!(metrics, 3);
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    /// Under NullEqualsNothing, NULL join keys are not inserted into the hash
    /// map, so a build side whose keys are all NULL produces an empty map even
    /// though it contains rows. Join types that emit unmatched build rows must
    /// still produce them from the visited bitmap.
    #[rstest]
    #[tokio::test]
    async fn join_all_null_build_keys(
        #[values(PartitionMode::CollectLeft, PartitionMode::Partitioned)]
        partition_mode: PartitionMode,
    ) -> Result<()> {
        let left = build_table_two_cols(
            ("a1", &vec![Some(1), Some(2)]),
            ("b1", &vec![None, None]), // all build-side join keys are NULL
        );
        let right = build_table_two_cols(
            ("a2", &vec![Some(10), Some(20), Some(30)]),
            ("b1", &vec![Some(4), None, Some(6)]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
        )];

        for join_type in [
            JoinType::Inner,
            JoinType::Left,
            JoinType::Right,
            JoinType::Full,
            JoinType::LeftSemi,
            JoinType::LeftAnti,
            JoinType::RightSemi,
            JoinType::RightAnti,
            JoinType::LeftMark,
            JoinType::RightMark,
        ] {
            let (_, batches, metrics) = join_collect_with_partition_mode(
                Arc::clone(&left),
                Arc::clone(&right),
                on.clone(),
                &join_type,
                partition_mode,
                NullEquality::NullEqualsNothing,
                Arc::new(TaskContext::default()),
            )
            .await?;

            // For join types whose output requires a build-side match, an
            // empty map guarantees an empty result, so `state_after_build_ready`
            // completes the stream without ever fetching a probe batch (probe
            // `input_rows` stays 0). All other join types must still scan the
            // probe side. `input_rows` is summed across every partition.
            let probe_rows = metrics
                .sum_by_name("input_rows")
                .map(|v| v.as_usize())
                .unwrap_or(0);
            if join_type.empty_map_produces_empty_result() {
                assert_eq!(
                    probe_rows, 0,
                    "{join_type} should skip the probe side for an all-NULL build"
                );
            } else {
                assert!(probe_rows > 0, "{join_type} must scan the probe side");
            }

            match join_type {
                JoinType::Inner | JoinType::LeftSemi | JoinType::RightSemi => {
                    let num_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
                    assert_eq!(num_rows, 0, "unexpected rows for {join_type}");
                }
                JoinType::Left => {
                    allow_duplicates! {
                        assert_snapshot!(batches_to_sort_string(&batches), @r"
                        +----+----+----+----+
                        | a1 | b1 | a2 | b1 |
                        +----+----+----+----+
                        | 1  |    |    |    |
                        | 2  |    |    |    |
                        +----+----+----+----+
                        ");
                    }
                }
                JoinType::Right => {
                    allow_duplicates! {
                        assert_snapshot!(batches_to_sort_string(&batches), @r"
                        +----+----+----+----+
                        | a1 | b1 | a2 | b1 |
                        +----+----+----+----+
                        |    |    | 10 | 4  |
                        |    |    | 20 |    |
                        |    |    | 30 | 6  |
                        +----+----+----+----+
                        ");
                    }
                }
                JoinType::Full => {
                    allow_duplicates! {
                        assert_snapshot!(batches_to_sort_string(&batches), @r"
                        +----+----+----+----+
                        | a1 | b1 | a2 | b1 |
                        +----+----+----+----+
                        |    |    | 10 | 4  |
                        |    |    | 20 |    |
                        |    |    | 30 | 6  |
                        | 1  |    |    |    |
                        | 2  |    |    |    |
                        +----+----+----+----+
                        ");
                    }
                }
                JoinType::LeftAnti => {
                    allow_duplicates! {
                        assert_snapshot!(batches_to_sort_string(&batches), @r"
                        +----+----+
                        | a1 | b1 |
                        +----+----+
                        | 1  |    |
                        | 2  |    |
                        +----+----+
                        ");
                    }
                }
                JoinType::RightAnti => {
                    allow_duplicates! {
                        assert_snapshot!(batches_to_sort_string(&batches), @r"
                        +----+----+
                        | a2 | b1 |
                        +----+----+
                        | 10 | 4  |
                        | 20 |    |
                        | 30 | 6  |
                        +----+----+
                        ");
                    }
                }
                JoinType::LeftMark => {
                    allow_duplicates! {
                        assert_snapshot!(batches_to_sort_string(&batches), @r"
                        +----+----+-------+
                        | a1 | b1 | mark  |
                        +----+----+-------+
                        | 1  |    | false |
                        | 2  |    | false |
                        +----+----+-------+
                        ");
                    }
                }
                JoinType::RightMark => {
                    allow_duplicates! {
                        assert_snapshot!(batches_to_sort_string(&batches), @r"
                        +----+----+-------+
                        | a2 | b1 | mark  |
                        +----+----+-------+
                        | 10 | 4  | false |
                        | 20 |    | false |
                        | 30 | 6  | false |
                        +----+----+-------+
                        ");
                    }
                }
            }
        }

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn partitioned_join_left_one(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]), // 7 does not exist on the right
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30]),
            ("b1", &vec![4, 5, 6]),
            ("c2", &vec![70, 80, 90]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
        )];

        let (columns, batches, metrics) = partitioned_join_collect(
            Arc::clone(&left),
            Arc::clone(&right),
            on.clone(),
            &JoinType::Left,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b1", "c2"]);

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b1 | c2 |
            +----+----+----+----+----+----+
            | 1  | 4  | 7  | 10 | 4  | 70 |
            | 2  | 5  | 8  | 20 | 5  | 80 |
            | 3  | 7  | 9  |    |    |    |
            +----+----+----+----+----+----+
            ");
        }

        assert_join_metrics!(metrics, 3);
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    fn build_semi_anti_left_table() -> Arc<dyn ExecutionPlan> {
        // just two line match
        // b1 = 10
        build_table(
            ("a1", &vec![1, 3, 5, 7, 9, 11, 13]),
            ("b1", &vec![1, 3, 5, 7, 8, 8, 10]),
            ("c1", &vec![10, 30, 50, 70, 90, 110, 130]),
        )
    }

    fn build_semi_anti_right_table() -> Arc<dyn ExecutionPlan> {
        // just two line match
        // b2 = 10
        build_table(
            ("a2", &vec![8, 12, 6, 2, 10, 4]),
            ("b2", &vec![8, 10, 6, 2, 10, 4]),
            ("c2", &vec![20, 40, 60, 80, 100, 120]),
        )
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_left_semi(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_semi_anti_left_table();
        let right = build_semi_anti_right_table();
        // left_table left semi join right_table on left_table.b1 = right_table.b2
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
        )];

        let join = join(
            left,
            right,
            on,
            &JoinType::LeftSemi,
            NullEquality::NullEqualsNothing,
        )?;

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a1", "b1", "c1"]);

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        // ignore the order
        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+-----+
            | a1 | b1 | c1  |
            +----+----+-----+
            | 11 | 8  | 110 |
            | 13 | 10 | 130 |
            | 9  | 8  | 90  |
            +----+----+-----+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_left_semi_with_filter(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_semi_anti_left_table();
        let right = build_semi_anti_right_table();

        // left_table left semi join right_table on left_table.b1 = right_table.b2 and right_table.a2 != 10
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
        )];

        let column_indices = vec![ColumnIndex {
            index: 0,
            side: JoinSide::Right,
        }];
        let intermediate_schema =
            Schema::new(vec![Field::new("x", DataType::Int32, true)]);

        let filter_expression = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("x", 0)),
            Operator::NotEq,
            Arc::new(Literal::new(ScalarValue::Int32(Some(10)))),
        )) as Arc<dyn PhysicalExpr>;

        let filter = JoinFilter::new(
            filter_expression,
            column_indices.clone(),
            Arc::new(intermediate_schema.clone()),
        );

        let join = join_with_filter(
            Arc::clone(&left),
            Arc::clone(&right),
            on.clone(),
            filter,
            &JoinType::LeftSemi,
            NullEquality::NullEqualsNothing,
        )?;

        let columns_header = columns(&join.schema());
        assert_eq!(columns_header.clone(), vec!["a1", "b1", "c1"]);

        let stream = join.execute(0, Arc::clone(&task_ctx))?;
        let batches = common::collect(stream).await?;

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+-----+
            | a1 | b1 | c1  |
            +----+----+-----+
            | 11 | 8  | 110 |
            | 13 | 10 | 130 |
            | 9  | 8  | 90  |
            +----+----+-----+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        // left_table left semi join right_table on left_table.b1 = right_table.b2 and right_table.a2 > 10
        let filter_expression = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("x", 0)),
            Operator::Gt,
            Arc::new(Literal::new(ScalarValue::Int32(Some(10)))),
        )) as Arc<dyn PhysicalExpr>;
        let filter = JoinFilter::new(
            filter_expression,
            column_indices,
            Arc::new(intermediate_schema),
        );

        let join = join_with_filter(
            left,
            right,
            on,
            filter,
            &JoinType::LeftSemi,
            NullEquality::NullEqualsNothing,
        )?;

        let columns_header = columns(&join.schema());
        assert_eq!(columns_header, vec!["a1", "b1", "c1"]);

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+-----+
            | a1 | b1 | c1  |
            +----+----+-----+
            | 13 | 10 | 130 |
            +----+----+-----+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_right_semi(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_semi_anti_left_table();
        let right = build_semi_anti_right_table();

        // left_table right semi join right_table on left_table.b1 = right_table.b2
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
        )];

        let join = join(
            left,
            right,
            on,
            &JoinType::RightSemi,
            NullEquality::NullEqualsNothing,
        )?;

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a2", "b2", "c2"]);

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        // RightSemi join output is expected to preserve right input order
        allow_duplicates! {
            assert_snapshot!(batches_to_string(&batches), @r"
            +----+----+-----+
            | a2 | b2 | c2  |
            +----+----+-----+
            | 8  | 8  | 20  |
            | 12 | 10 | 40  |
            | 10 | 10 | 100 |
            +----+----+-----+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_right_semi_with_filter(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_semi_anti_left_table();
        let right = build_semi_anti_right_table();

        // left_table right semi join right_table on left_table.b1 = right_table.b2 on left_table.a1!=9
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
        )];

        let column_indices = vec![ColumnIndex {
            index: 0,
            side: JoinSide::Left,
        }];
        let intermediate_schema =
            Schema::new(vec![Field::new("x", DataType::Int32, true)]);

        let filter_expression = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("x", 0)),
            Operator::NotEq,
            Arc::new(Literal::new(ScalarValue::Int32(Some(9)))),
        )) as Arc<dyn PhysicalExpr>;

        let filter = JoinFilter::new(
            filter_expression,
            column_indices.clone(),
            Arc::new(intermediate_schema.clone()),
        );

        let join = join_with_filter(
            Arc::clone(&left),
            Arc::clone(&right),
            on.clone(),
            filter,
            &JoinType::RightSemi,
            NullEquality::NullEqualsNothing,
        )?;

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a2", "b2", "c2"]);

        let stream = join.execute(0, Arc::clone(&task_ctx))?;
        let batches = common::collect(stream).await?;

        // RightSemi join output is expected to preserve right input order
        allow_duplicates! {
            assert_snapshot!(batches_to_string(&batches), @r"
            +----+----+-----+
            | a2 | b2 | c2  |
            +----+----+-----+
            | 8  | 8  | 20  |
            | 12 | 10 | 40  |
            | 10 | 10 | 100 |
            +----+----+-----+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        // left_table right semi join right_table on left_table.b1 = right_table.b2 on left_table.a1!=9
        let filter_expression = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("x", 0)),
            Operator::Gt,
            Arc::new(Literal::new(ScalarValue::Int32(Some(11)))),
        )) as Arc<dyn PhysicalExpr>;

        let filter = JoinFilter::new(
            filter_expression,
            column_indices,
            Arc::new(intermediate_schema.clone()),
        );

        let join = join_with_filter(
            left,
            right,
            on,
            filter,
            &JoinType::RightSemi,
            NullEquality::NullEqualsNothing,
        )?;
        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        // RightSemi join output is expected to preserve right input order
        allow_duplicates! {
            assert_snapshot!(batches_to_string(&batches), @r"
            +----+----+-----+
            | a2 | b2 | c2  |
            +----+----+-----+
            | 12 | 10 | 40  |
            | 10 | 10 | 100 |
            +----+----+-----+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_left_anti(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_semi_anti_left_table();
        let right = build_semi_anti_right_table();
        // left_table left anti join right_table on left_table.b1 = right_table.b2
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
        )];

        let join = join(
            left,
            right,
            on,
            &JoinType::LeftAnti,
            NullEquality::NullEqualsNothing,
        )?;

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a1", "b1", "c1"]);

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+----+
            | a1 | b1 | c1 |
            +----+----+----+
            | 1  | 1  | 10 |
            | 3  | 3  | 30 |
            | 5  | 5  | 50 |
            | 7  | 7  | 70 |
            +----+----+----+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_left_anti_with_filter(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_semi_anti_left_table();
        let right = build_semi_anti_right_table();
        // left_table left anti join right_table on left_table.b1 = right_table.b2 and right_table.a2!=8
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
        )];

        let column_indices = vec![ColumnIndex {
            index: 0,
            side: JoinSide::Right,
        }];
        let intermediate_schema =
            Schema::new(vec![Field::new("x", DataType::Int32, true)]);
        let filter_expression = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("x", 0)),
            Operator::NotEq,
            Arc::new(Literal::new(ScalarValue::Int32(Some(8)))),
        )) as Arc<dyn PhysicalExpr>;

        let filter = JoinFilter::new(
            filter_expression,
            column_indices.clone(),
            Arc::new(intermediate_schema.clone()),
        );

        let join = join_with_filter(
            Arc::clone(&left),
            Arc::clone(&right),
            on.clone(),
            filter,
            &JoinType::LeftAnti,
            NullEquality::NullEqualsNothing,
        )?;

        let columns_header = columns(&join.schema());
        assert_eq!(columns_header, vec!["a1", "b1", "c1"]);

        let stream = join.execute(0, Arc::clone(&task_ctx))?;
        let batches = common::collect(stream).await?;

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+-----+
            | a1 | b1 | c1  |
            +----+----+-----+
            | 1  | 1  | 10  |
            | 11 | 8  | 110 |
            | 3  | 3  | 30  |
            | 5  | 5  | 50  |
            | 7  | 7  | 70  |
            | 9  | 8  | 90  |
            +----+----+-----+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        // left_table left anti join right_table on left_table.b1 = right_table.b2 and right_table.a2 != 13
        let filter_expression = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("x", 0)),
            Operator::NotEq,
            Arc::new(Literal::new(ScalarValue::Int32(Some(8)))),
        )) as Arc<dyn PhysicalExpr>;

        let filter = JoinFilter::new(
            filter_expression,
            column_indices,
            Arc::new(intermediate_schema),
        );

        let join = join_with_filter(
            left,
            right,
            on,
            filter,
            &JoinType::LeftAnti,
            NullEquality::NullEqualsNothing,
        )?;

        let columns_header = columns(&join.schema());
        assert_eq!(columns_header, vec!["a1", "b1", "c1"]);

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+-----+
            | a1 | b1 | c1  |
            +----+----+-----+
            | 1  | 1  | 10  |
            | 11 | 8  | 110 |
            | 3  | 3  | 30  |
            | 5  | 5  | 50  |
            | 7  | 7  | 70  |
            | 9  | 8  | 90  |
            +----+----+-----+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_right_anti(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_semi_anti_left_table();
        let right = build_semi_anti_right_table();
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
        )];

        let join = join(
            left,
            right,
            on,
            &JoinType::RightAnti,
            NullEquality::NullEqualsNothing,
        )?;

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a2", "b2", "c2"]);

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        // RightAnti join output is expected to preserve right input order
        allow_duplicates! {
            assert_snapshot!(batches_to_string(&batches), @r"
            +----+----+-----+
            | a2 | b2 | c2  |
            +----+----+-----+
            | 6  | 6  | 60  |
            | 2  | 2  | 80  |
            | 4  | 4  | 120 |
            +----+----+-----+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_right_anti_with_filter(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_semi_anti_left_table();
        let right = build_semi_anti_right_table();
        // left_table right anti join right_table on left_table.b1 = right_table.b2 and left_table.a1!=13
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
        )];

        let column_indices = vec![ColumnIndex {
            index: 0,
            side: JoinSide::Left,
        }];
        let intermediate_schema =
            Schema::new(vec![Field::new("x", DataType::Int32, true)]);

        let filter_expression = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("x", 0)),
            Operator::NotEq,
            Arc::new(Literal::new(ScalarValue::Int32(Some(13)))),
        )) as Arc<dyn PhysicalExpr>;

        let filter = JoinFilter::new(
            filter_expression,
            column_indices,
            Arc::new(intermediate_schema.clone()),
        );

        let join = join_with_filter(
            Arc::clone(&left),
            Arc::clone(&right),
            on.clone(),
            filter,
            &JoinType::RightAnti,
            NullEquality::NullEqualsNothing,
        )?;

        let columns_header = columns(&join.schema());
        assert_eq!(columns_header, vec!["a2", "b2", "c2"]);

        let stream = join.execute(0, Arc::clone(&task_ctx))?;
        let batches = common::collect(stream).await?;

        // RightAnti join output is expected to preserve right input order
        allow_duplicates! {
            assert_snapshot!(batches_to_string(&batches), @r"
            +----+----+-----+
            | a2 | b2 | c2  |
            +----+----+-----+
            | 12 | 10 | 40  |
            | 6  | 6  | 60  |
            | 2  | 2  | 80  |
            | 10 | 10 | 100 |
            | 4  | 4  | 120 |
            +----+----+-----+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        // left_table right anti join right_table on left_table.b1 = right_table.b2 and right_table.b2!=8
        let column_indices = vec![ColumnIndex {
            index: 1,
            side: JoinSide::Right,
        }];
        let filter_expression = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("x", 0)),
            Operator::NotEq,
            Arc::new(Literal::new(ScalarValue::Int32(Some(8)))),
        )) as Arc<dyn PhysicalExpr>;

        let filter = JoinFilter::new(
            filter_expression,
            column_indices,
            Arc::new(intermediate_schema),
        );

        let join = join_with_filter(
            left,
            right,
            on,
            filter,
            &JoinType::RightAnti,
            NullEquality::NullEqualsNothing,
        )?;

        let columns_header = columns(&join.schema());
        assert_eq!(columns_header, vec!["a2", "b2", "c2"]);

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        // RightAnti join output is expected to preserve right input order
        allow_duplicates! {
            assert_snapshot!(batches_to_string(&batches), @r"
            +----+----+-----+
            | a2 | b2 | c2  |
            +----+----+-----+
            | 8  | 8  | 20  |
            | 6  | 6  | 60  |
            | 2  | 2  | 80  |
            | 4  | 4  | 120 |
            +----+----+-----+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_right_one(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]),
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30]),
            ("b1", &vec![4, 5, 6]), // 6 does not exist on the left
            ("c2", &vec![70, 80, 90]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
        )];

        let (columns, batches, metrics) = join_collect(
            left,
            right,
            on,
            &JoinType::Right,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b1", "c2"]);

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b1 | c2 |
            +----+----+----+----+----+----+
            |    |    |    | 30 | 6  | 90 |
            | 1  | 4  | 7  | 10 | 4  | 70 |
            | 2  | 5  | 8  | 20 | 5  | 80 |
            +----+----+----+----+----+----+
            ");
        }

        assert_join_metrics!(metrics, 3);
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn partitioned_join_right_one(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]),
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30]),
            ("b1", &vec![4, 5, 6]), // 6 does not exist on the left
            ("c2", &vec![70, 80, 90]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
        )];

        let (columns, batches, metrics) = partitioned_join_collect(
            left,
            right,
            on,
            &JoinType::Right,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b1", "c2"]);

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b1 | c2 |
            +----+----+----+----+----+----+
            |    |    |    | 30 | 6  | 90 |
            | 1  | 4  | 7  | 10 | 4  | 70 |
            | 2  | 5  | 8  | 20 | 5  | 80 |
            +----+----+----+----+----+----+
            ");
        }

        assert_join_metrics!(metrics, 3);
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_full_one(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]), // 7 does not exist on the right
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30]),
            ("b2", &vec![4, 5, 6]),
            ("c2", &vec![70, 80, 90]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema()).unwrap()) as _,
        )];

        let join = join(
            left,
            right,
            on,
            &JoinType::Full,
            NullEquality::NullEqualsNothing,
        )?;

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b2", "c2"]);

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+----+----+----+----+
            | a1 | b1 | c1 | a2 | b2 | c2 |
            +----+----+----+----+----+----+
            |    |    |    | 30 | 6  | 90 |
            | 1  | 4  | 7  | 10 | 4  | 70 |
            | 2  | 5  | 8  | 20 | 5  | 80 |
            | 3  | 7  | 9  |    |    |    |
            +----+----+----+----+----+----+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_left_mark(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]), // 7 does not exist on the right
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30]),
            ("b1", &vec![4, 5, 6]),
            ("c2", &vec![70, 80, 90]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
        )];

        let (columns, batches, metrics) = join_collect(
            Arc::clone(&left),
            Arc::clone(&right),
            on.clone(),
            &JoinType::LeftMark,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b1", "c1", "mark"]);

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+----+-------+
            | a1 | b1 | c1 | mark  |
            +----+----+----+-------+
            | 1  | 4  | 7  | true  |
            | 2  | 5  | 8  | true  |
            | 3  | 7  | 9  | false |
            +----+----+----+-------+
            ");
        }

        assert_join_metrics!(metrics, 3);
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn partitioned_join_left_mark(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]), // 7 does not exist on the right
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30, 40]),
            ("b1", &vec![4, 4, 5, 6]),
            ("c2", &vec![60, 70, 80, 90]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
        )];

        let (columns, batches, metrics) = partitioned_join_collect(
            Arc::clone(&left),
            Arc::clone(&right),
            on.clone(),
            &JoinType::LeftMark,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b1", "c1", "mark"]);

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+----+----+-------+
            | a1 | b1 | c1 | mark  |
            +----+----+----+-------+
            | 1  | 4  | 7  | true  |
            | 2  | 5  | 8  | true  |
            | 3  | 7  | 9  | false |
            +----+----+----+-------+
            ");
        }

        assert_join_metrics!(metrics, 3);
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_right_mark(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]), // 7 does not exist on the right
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30]),
            ("b1", &vec![4, 5, 6]), // 6 does not exist on the left
            ("c2", &vec![70, 80, 90]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
        )];

        let (columns, batches, metrics) = join_collect(
            Arc::clone(&left),
            Arc::clone(&right),
            on.clone(),
            &JoinType::RightMark,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a2", "b1", "c2", "mark"]);

        let expected = [
            "+----+----+----+-------+",
            "| a2 | b1 | c2 | mark  |",
            "+----+----+----+-------+",
            "| 10 | 4  | 70 | true  |",
            "| 20 | 5  | 80 | true  |",
            "| 30 | 6  | 90 | false |",
            "+----+----+----+-------+",
        ];
        assert_batches_sorted_eq!(expected, &batches);

        assert_join_metrics!(metrics, 3);
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn partitioned_join_right_mark(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]), // 7 does not exist on the right
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30, 40]),
            ("b1", &vec![4, 4, 5, 6]), // 6 does not exist on the left
            ("c2", &vec![60, 70, 80, 90]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
        )];

        let (columns, batches, metrics) = partitioned_join_collect(
            Arc::clone(&left),
            Arc::clone(&right),
            on.clone(),
            &JoinType::RightMark,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a2", "b1", "c2", "mark"]);

        let expected = [
            "+----+----+----+-------+",
            "| a2 | b1 | c2 | mark  |",
            "+----+----+----+-------+",
            "| 10 | 4  | 60 | true  |",
            "| 20 | 4  | 70 | true  |",
            "| 30 | 5  | 80 | true  |",
            "| 40 | 6  | 90 | false |",
            "+----+----+----+-------+",
        ];
        assert_batches_sorted_eq!(expected, &batches);

        assert_join_metrics!(metrics, 4);
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[test]
    fn join_with_hash_collisions_64() -> Result<()> {
        let mut hashmap_left = HashTable::with_capacity(4);
        let left = build_table_i32(
            ("a", &vec![10, 20]),
            ("x", &vec![100, 200]),
            ("y", &vec![200, 300]),
        );

        let random_state = RandomState::with_seed(0);
        let hashes_buff = &mut vec![0; left.num_rows()];
        let hashes = create_hashes([&left.columns()[0]], &random_state, hashes_buff)?;

        // Maps both values to both indices (1 and 2, representing input 0 and 1)
        // 0 -> (0, 1)
        // 1 -> (0, 2)
        // The equality check will make sure only hashes[0] maps to 0 and hashes[1] maps to 1
        hashmap_left.insert_unique(hashes[0], (hashes[0], 1), |(h, _)| *h);
        hashmap_left.insert_unique(hashes[0], (hashes[0], 2), |(h, _)| *h);

        hashmap_left.insert_unique(hashes[1], (hashes[1], 1), |(h, _)| *h);
        hashmap_left.insert_unique(hashes[1], (hashes[1], 2), |(h, _)| *h);

        let next = vec![2, 0];

        let right = build_table_i32(
            ("a", &vec![10, 20]),
            ("b", &vec![0, 0]),
            ("c", &vec![30, 40]),
        );

        // Join key column for both join sides
        let key_column: PhysicalExprRef = Arc::new(Column::new("a", 0)) as _;

        let join_hash_map = JoinHashMapU64::new(hashmap_left, next);

        let left_keys_values = key_column.evaluate(&left)?.into_array(left.num_rows())?;
        let right_keys_values =
            key_column.evaluate(&right)?.into_array(right.num_rows())?;
        let mut hashes_buffer = vec![0; right.num_rows()];
        create_hashes([&right_keys_values], &random_state, &mut hashes_buffer)?;

        let mut probe_indices_buffer = Vec::new();
        let mut build_indices_buffer = Vec::new();
        let (l, r, _) = lookup_join_hashmap(
            &join_hash_map,
            &[left_keys_values],
            &[right_keys_values],
            NullEquality::NullEqualsNothing,
            &hashes_buffer,
            None,
            8192,
            (0, None),
            &mut probe_indices_buffer,
            &mut build_indices_buffer,
        )?;

        let left_ids: UInt64Array = vec![0, 1].into();

        let right_ids: UInt32Array = vec![0, 1].into();

        assert_eq!(left_ids, l);

        assert_eq!(right_ids, r);

        Ok(())
    }

    #[test]
    fn join_with_hash_collisions_u32() -> Result<()> {
        let mut hashmap_left = HashTable::with_capacity(4);
        let left = build_table_i32(
            ("a", &vec![10, 20]),
            ("x", &vec![100, 200]),
            ("y", &vec![200, 300]),
        );

        let random_state = RandomState::with_seed(0);
        let hashes_buff = &mut vec![0; left.num_rows()];
        let hashes = create_hashes([&left.columns()[0]], &random_state, hashes_buff)?;

        hashmap_left.insert_unique(hashes[0], (hashes[0], 1u32), |(h, _)| *h);
        hashmap_left.insert_unique(hashes[0], (hashes[0], 2u32), |(h, _)| *h);
        hashmap_left.insert_unique(hashes[1], (hashes[1], 1u32), |(h, _)| *h);
        hashmap_left.insert_unique(hashes[1], (hashes[1], 2u32), |(h, _)| *h);

        let next: Vec<u32> = vec![2, 0];

        let right = build_table_i32(
            ("a", &vec![10, 20]),
            ("b", &vec![0, 0]),
            ("c", &vec![30, 40]),
        );

        let key_column: PhysicalExprRef = Arc::new(Column::new("a", 0)) as _;

        let join_hash_map = JoinHashMapU32::new(hashmap_left, next);

        let left_keys_values = key_column.evaluate(&left)?.into_array(left.num_rows())?;
        let right_keys_values =
            key_column.evaluate(&right)?.into_array(right.num_rows())?;
        let mut hashes_buffer = vec![0; right.num_rows()];
        create_hashes([&right_keys_values], &random_state, &mut hashes_buffer)?;

        let mut probe_indices_buffer = Vec::new();
        let mut build_indices_buffer = Vec::new();
        let (l, r, _) = lookup_join_hashmap(
            &join_hash_map,
            &[left_keys_values],
            &[right_keys_values],
            NullEquality::NullEqualsNothing,
            &hashes_buffer,
            None,
            8192,
            (0, None),
            &mut probe_indices_buffer,
            &mut build_indices_buffer,
        )?;

        // We still expect to match rows 0 and 1 on both sides
        let left_ids: UInt64Array = vec![0, 1].into();
        let right_ids: UInt32Array = vec![0, 1].into();

        assert_eq!(left_ids, l);
        assert_eq!(right_ids, r);

        Ok(())
    }

    #[tokio::test]
    async fn join_with_duplicated_column_names() -> Result<()> {
        let task_ctx = Arc::new(TaskContext::default());
        let left = build_table(
            ("a", &vec![1, 2, 3]),
            ("b", &vec![4, 5, 7]),
            ("c", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a", &vec![10, 20, 30]),
            ("b", &vec![1, 2, 7]),
            ("c", &vec![70, 80, 90]),
        );
        let on = vec![(
            // join on a=b so there are duplicate column names on unjoined columns
            Arc::new(Column::new_with_schema("a", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("b", &right.schema()).unwrap()) as _,
        )];

        let join = join(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
        )?;

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a", "b", "c", "a", "b", "c"]);

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +---+---+---+----+---+----+
            | a | b | c | a  | b | c  |
            +---+---+---+----+---+----+
            | 1 | 4 | 7 | 10 | 1 | 70 |
            | 2 | 5 | 8 | 20 | 2 | 80 |
            +---+---+---+----+---+----+
            ");
        }

        Ok(())
    }

    fn prepare_join_filter() -> JoinFilter {
        let column_indices = vec![
            ColumnIndex {
                index: 2,
                side: JoinSide::Left,
            },
            ColumnIndex {
                index: 2,
                side: JoinSide::Right,
            },
        ];
        let intermediate_schema = Schema::new(vec![
            Field::new("c", DataType::Int32, true),
            Field::new("c", DataType::Int32, true),
        ]);
        let filter_expression = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("c", 0)),
            Operator::Gt,
            Arc::new(Column::new("c", 1)),
        )) as Arc<dyn PhysicalExpr>;

        JoinFilter::new(
            filter_expression,
            column_indices,
            Arc::new(intermediate_schema),
        )
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_inner_with_filter(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a", &vec![0, 1, 2, 2]),
            ("b", &vec![4, 5, 7, 8]),
            ("c", &vec![7, 8, 9, 1]),
        );
        let right = build_table(
            ("a", &vec![10, 20, 30, 40]),
            ("b", &vec![2, 2, 3, 4]),
            ("c", &vec![7, 5, 6, 4]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("a", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("b", &right.schema()).unwrap()) as _,
        )];
        let filter = prepare_join_filter();

        let join = join_with_filter(
            left,
            right,
            on,
            filter,
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
        )?;

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a", "b", "c", "a", "b", "c"]);

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +---+---+---+----+---+---+
            | a | b | c | a  | b | c |
            +---+---+---+----+---+---+
            | 2 | 7 | 9 | 10 | 2 | 7 |
            | 2 | 7 | 9 | 20 | 2 | 5 |
            +---+---+---+----+---+---+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_left_with_filter(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a", &vec![0, 1, 2, 2]),
            ("b", &vec![4, 5, 7, 8]),
            ("c", &vec![7, 8, 9, 1]),
        );
        let right = build_table(
            ("a", &vec![10, 20, 30, 40]),
            ("b", &vec![2, 2, 3, 4]),
            ("c", &vec![7, 5, 6, 4]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("a", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("b", &right.schema()).unwrap()) as _,
        )];
        let filter = prepare_join_filter();

        let join = join_with_filter(
            left,
            right,
            on,
            filter,
            &JoinType::Left,
            NullEquality::NullEqualsNothing,
        )?;

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a", "b", "c", "a", "b", "c"]);

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +---+---+---+----+---+---+
            | a | b | c | a  | b | c |
            +---+---+---+----+---+---+
            | 0 | 4 | 7 |    |   |   |
            | 1 | 5 | 8 |    |   |   |
            | 2 | 7 | 9 | 10 | 2 | 7 |
            | 2 | 7 | 9 | 20 | 2 | 5 |
            | 2 | 8 | 1 |    |   |   |
            +---+---+---+----+---+---+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_right_with_filter(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a", &vec![0, 1, 2, 2]),
            ("b", &vec![4, 5, 7, 8]),
            ("c", &vec![7, 8, 9, 1]),
        );
        let right = build_table(
            ("a", &vec![10, 20, 30, 40]),
            ("b", &vec![2, 2, 3, 4]),
            ("c", &vec![7, 5, 6, 4]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("a", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("b", &right.schema()).unwrap()) as _,
        )];
        let filter = prepare_join_filter();

        let join = join_with_filter(
            left,
            right,
            on,
            filter,
            &JoinType::Right,
            NullEquality::NullEqualsNothing,
        )?;

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a", "b", "c", "a", "b", "c"]);

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +---+---+---+----+---+---+
            | a | b | c | a  | b | c |
            +---+---+---+----+---+---+
            |   |   |   | 30 | 3 | 6 |
            |   |   |   | 40 | 4 | 4 |
            | 2 | 7 | 9 | 10 | 2 | 7 |
            | 2 | 7 | 9 | 20 | 2 | 5 |
            +---+---+---+----+---+---+
            ");
        }

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn join_full_with_filter(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let left = build_table(
            ("a", &vec![0, 1, 2, 2]),
            ("b", &vec![4, 5, 7, 8]),
            ("c", &vec![7, 8, 9, 1]),
        );
        let right = build_table(
            ("a", &vec![10, 20, 30, 40]),
            ("b", &vec![2, 2, 3, 4]),
            ("c", &vec![7, 5, 6, 4]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("a", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("b", &right.schema()).unwrap()) as _,
        )];
        let filter = prepare_join_filter();

        let join = join_with_filter(
            left,
            right,
            on,
            filter,
            &JoinType::Full,
            NullEquality::NullEqualsNothing,
        )?;

        let columns = columns(&join.schema());
        assert_eq!(columns, vec!["a", "b", "c", "a", "b", "c"]);

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        let expected = [
            "+---+---+---+----+---+---+",
            "| a | b | c | a  | b | c |",
            "+---+---+---+----+---+---+",
            "|   |   |   | 30 | 3 | 6 |",
            "|   |   |   | 40 | 4 | 4 |",
            "| 2 | 7 | 9 | 10 | 2 | 7 |",
            "| 2 | 7 | 9 | 20 | 2 | 5 |",
            "| 0 | 4 | 7 |    |   |   |",
            "| 1 | 5 | 8 |    |   |   |",
            "| 2 | 8 | 1 |    |   |   |",
            "+---+---+---+----+---+---+",
        ];
        assert_batches_sorted_eq!(expected, &batches);

        let metrics = join.metrics().unwrap();
        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        // THIS MIGRATION HALTED DUE TO ISSUE #15312
        //allow_duplicates! {
        //    assert_snapshot!(batches_to_sort_string(&batches), @r#"
        //    +---+---+---+----+---+---+
        //    | a | b | c | a  | b | c |
        //    +---+---+---+----+---+---+
        //    |   |   |   | 30 | 3 | 6 |
        //    |   |   |   | 40 | 4 | 4 |
        //    | 2 | 7 | 9 | 10 | 2 | 7 |
        //    | 2 | 7 | 9 | 20 | 2 | 5 |
        //    | 0 | 4 | 7 |    |   |   |
        //    | 1 | 5 | 8 |    |   |   |
        //    | 2 | 8 | 1 |    |   |   |
        //    +---+---+---+----+---+---+
        //        "#)
        //}

        Ok(())
    }

    /// Test for parallelized HashJoinExec with PartitionMode::CollectLeft
    #[tokio::test]
    async fn test_collect_left_multiple_partitions_join() -> Result<()> {
        let task_ctx = Arc::new(TaskContext::default());
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]),
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30]),
            ("b2", &vec![4, 5, 6]),
            ("c2", &vec![70, 80, 90]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema()).unwrap()) as _,
        )];

        let expected_inner = vec![
            "+----+----+----+----+----+----+",
            "| a1 | b1 | c1 | a2 | b2 | c2 |",
            "+----+----+----+----+----+----+",
            "| 1  | 4  | 7  | 10 | 4  | 70 |",
            "| 2  | 5  | 8  | 20 | 5  | 80 |",
            "+----+----+----+----+----+----+",
        ];
        let expected_left = vec![
            "+----+----+----+----+----+----+",
            "| a1 | b1 | c1 | a2 | b2 | c2 |",
            "+----+----+----+----+----+----+",
            "| 1  | 4  | 7  | 10 | 4  | 70 |",
            "| 2  | 5  | 8  | 20 | 5  | 80 |",
            "| 3  | 7  | 9  |    |    |    |",
            "+----+----+----+----+----+----+",
        ];
        let expected_right = vec![
            "+----+----+----+----+----+----+",
            "| a1 | b1 | c1 | a2 | b2 | c2 |",
            "+----+----+----+----+----+----+",
            "|    |    |    | 30 | 6  | 90 |",
            "| 1  | 4  | 7  | 10 | 4  | 70 |",
            "| 2  | 5  | 8  | 20 | 5  | 80 |",
            "+----+----+----+----+----+----+",
        ];
        let expected_full = vec![
            "+----+----+----+----+----+----+",
            "| a1 | b1 | c1 | a2 | b2 | c2 |",
            "+----+----+----+----+----+----+",
            "|    |    |    | 30 | 6  | 90 |",
            "| 1  | 4  | 7  | 10 | 4  | 70 |",
            "| 2  | 5  | 8  | 20 | 5  | 80 |",
            "| 3  | 7  | 9  |    |    |    |",
            "+----+----+----+----+----+----+",
        ];
        let expected_left_semi = vec![
            "+----+----+----+",
            "| a1 | b1 | c1 |",
            "+----+----+----+",
            "| 1  | 4  | 7  |",
            "| 2  | 5  | 8  |",
            "+----+----+----+",
        ];
        let expected_left_anti = vec![
            "+----+----+----+",
            "| a1 | b1 | c1 |",
            "+----+----+----+",
            "| 3  | 7  | 9  |",
            "+----+----+----+",
        ];
        let expected_right_semi = vec![
            "+----+----+----+",
            "| a2 | b2 | c2 |",
            "+----+----+----+",
            "| 10 | 4  | 70 |",
            "| 20 | 5  | 80 |",
            "+----+----+----+",
        ];
        let expected_right_anti = vec![
            "+----+----+----+",
            "| a2 | b2 | c2 |",
            "+----+----+----+",
            "| 30 | 6  | 90 |",
            "+----+----+----+",
        ];
        let expected_left_mark = vec![
            "+----+----+----+-------+",
            "| a1 | b1 | c1 | mark  |",
            "+----+----+----+-------+",
            "| 1  | 4  | 7  | true  |",
            "| 2  | 5  | 8  | true  |",
            "| 3  | 7  | 9  | false |",
            "+----+----+----+-------+",
        ];
        let expected_right_mark = vec![
            "+----+----+----+-------+",
            "| a2 | b2 | c2 | mark  |",
            "+----+----+----+-------+",
            "| 10 | 4  | 70 | true  |",
            "| 20 | 5  | 80 | true  |",
            "| 30 | 6  | 90 | false |",
            "+----+----+----+-------+",
        ];

        let test_cases = vec![
            (JoinType::Inner, expected_inner),
            (JoinType::Left, expected_left),
            (JoinType::Right, expected_right),
            (JoinType::Full, expected_full),
            (JoinType::LeftSemi, expected_left_semi),
            (JoinType::LeftAnti, expected_left_anti),
            (JoinType::RightSemi, expected_right_semi),
            (JoinType::RightAnti, expected_right_anti),
            (JoinType::LeftMark, expected_left_mark),
            (JoinType::RightMark, expected_right_mark),
        ];

        for (join_type, expected) in test_cases {
            let (_, batches, metrics) = join_collect_with_partition_mode(
                Arc::clone(&left),
                Arc::clone(&right),
                on.clone(),
                &join_type,
                PartitionMode::CollectLeft,
                NullEquality::NullEqualsNothing,
                Arc::clone(&task_ctx),
            )
            .await?;
            assert_batches_sorted_eq!(expected, &batches);
            assert_join_metrics!(metrics, expected.len() - 4);
        }

        Ok(())
    }

    #[tokio::test]
    async fn join_date32() -> Result<()> {
        let schema = Arc::new(Schema::new(vec![
            Field::new("date", DataType::Date32, false),
            Field::new("n", DataType::Int32, false),
        ]));

        let dates: ArrayRef = Arc::new(Date32Array::from(vec![19107, 19108, 19109]));
        let n: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3]));
        let batch = RecordBatch::try_new(Arc::clone(&schema), vec![dates, n])?;
        let left =
            TestMemoryExec::try_new_exec(&[vec![batch]], Arc::clone(&schema), None)
                .unwrap();
        let dates: ArrayRef = Arc::new(Date32Array::from(vec![19108, 19108, 19109]));
        let n: ArrayRef = Arc::new(Int32Array::from(vec![4, 5, 6]));
        let batch = RecordBatch::try_new(Arc::clone(&schema), vec![dates, n])?;
        let right = TestMemoryExec::try_new_exec(&[vec![batch]], schema, None).unwrap();
        let on = vec![(
            Arc::new(Column::new_with_schema("date", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("date", &right.schema()).unwrap()) as _,
        )];

        let join = join(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
        )?;

        let task_ctx = Arc::new(TaskContext::default());
        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +------------+---+------------+---+
            | date       | n | date       | n |
            +------------+---+------------+---+
            | 2022-04-26 | 2 | 2022-04-26 | 4 |
            | 2022-04-26 | 2 | 2022-04-26 | 5 |
            | 2022-04-27 | 3 | 2022-04-27 | 6 |
            +------------+---+------------+---+
            ");
        }

        Ok(())
    }

    #[tokio::test]
    async fn join_with_error_right() {
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 7]),
            ("c1", &vec![7, 8, 9]),
        );

        // right input stream returns one good batch and then one error.
        // The error should be returned.
        let err = exec_err!("bad data error");
        let right = build_table_i32(("a2", &vec![]), ("b1", &vec![]), ("c2", &vec![]));

        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema()).unwrap()) as _,
        )];
        let schema = right.schema();
        let right = build_table_i32(("a2", &vec![]), ("b1", &vec![]), ("c2", &vec![]));
        let right_input = Arc::new(MockExec::new(vec![Ok(right), err], schema));

        let join_types = vec![
            JoinType::Inner,
            JoinType::Left,
            JoinType::Right,
            JoinType::Full,
            JoinType::LeftSemi,
            JoinType::LeftAnti,
            JoinType::RightSemi,
            JoinType::RightAnti,
        ];

        for join_type in join_types {
            let join = join(
                Arc::clone(&left),
                Arc::clone(&right_input) as Arc<dyn ExecutionPlan>,
                on.clone(),
                &join_type,
                NullEquality::NullEqualsNothing,
            )
            .unwrap();
            let task_ctx = Arc::new(TaskContext::default());

            let stream = join.execute(0, task_ctx).unwrap();

            // Expect that an error is returned
            let result_string = common::collect(stream).await.unwrap_err().to_string();
            assert!(
                result_string.contains("bad data error"),
                "actual: {result_string}"
            );
        }
    }

    #[tokio::test]
    async fn join_does_not_consume_probe_when_empty_build_fixes_output() {
        assert_empty_build_probe_behavior(
            &[
                JoinType::Inner,
                JoinType::Left,
                JoinType::LeftSemi,
                JoinType::LeftAnti,
                JoinType::LeftMark,
                JoinType::RightSemi,
            ],
            false,
            false,
        )
        .await;
    }

    #[tokio::test]
    async fn join_does_not_consume_probe_when_empty_build_fixes_output_with_filter() {
        assert_empty_build_probe_behavior(
            &[
                JoinType::Inner,
                JoinType::Left,
                JoinType::LeftSemi,
                JoinType::LeftAnti,
                JoinType::LeftMark,
                JoinType::RightSemi,
            ],
            false,
            true,
        )
        .await;
    }

    #[tokio::test]
    async fn join_still_consumes_probe_when_empty_build_needs_probe_rows() {
        assert_empty_build_probe_behavior(
            &[
                JoinType::Right,
                JoinType::Full,
                JoinType::RightAnti,
                JoinType::RightMark,
            ],
            true,
            false,
        )
        .await;
    }

    #[tokio::test]
    async fn join_still_consumes_probe_when_empty_build_needs_probe_rows_with_filter() {
        assert_empty_build_probe_behavior(
            &[
                JoinType::Right,
                JoinType::Full,
                JoinType::RightAnti,
                JoinType::RightMark,
            ],
            true,
            true,
        )
        .await;
    }

    #[tokio::test]
    async fn join_split_batch() {
        let left = build_table(
            ("a1", &vec![1, 2, 3, 4]),
            ("b1", &vec![1, 1, 1, 1]),
            ("c1", &vec![0, 0, 0, 0]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30, 40, 50]),
            ("b2", &vec![1, 1, 1, 1, 1]),
            ("c2", &vec![0, 0, 0, 0, 0]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema()).unwrap()) as _,
        )];

        let join_types = vec![
            JoinType::Inner,
            JoinType::Left,
            JoinType::Right,
            JoinType::Full,
            JoinType::RightSemi,
            JoinType::RightAnti,
            JoinType::LeftSemi,
            JoinType::LeftAnti,
        ];
        let expected_resultset_records = 20;
        let common_result = [
            "+----+----+----+----+----+----+",
            "| a1 | b1 | c1 | a2 | b2 | c2 |",
            "+----+----+----+----+----+----+",
            "| 1  | 1  | 0  | 10 | 1  | 0  |",
            "| 2  | 1  | 0  | 10 | 1  | 0  |",
            "| 3  | 1  | 0  | 10 | 1  | 0  |",
            "| 4  | 1  | 0  | 10 | 1  | 0  |",
            "| 1  | 1  | 0  | 20 | 1  | 0  |",
            "| 2  | 1  | 0  | 20 | 1  | 0  |",
            "| 3  | 1  | 0  | 20 | 1  | 0  |",
            "| 4  | 1  | 0  | 20 | 1  | 0  |",
            "| 1  | 1  | 0  | 30 | 1  | 0  |",
            "| 2  | 1  | 0  | 30 | 1  | 0  |",
            "| 3  | 1  | 0  | 30 | 1  | 0  |",
            "| 4  | 1  | 0  | 30 | 1  | 0  |",
            "| 1  | 1  | 0  | 40 | 1  | 0  |",
            "| 2  | 1  | 0  | 40 | 1  | 0  |",
            "| 3  | 1  | 0  | 40 | 1  | 0  |",
            "| 4  | 1  | 0  | 40 | 1  | 0  |",
            "| 1  | 1  | 0  | 50 | 1  | 0  |",
            "| 2  | 1  | 0  | 50 | 1  | 0  |",
            "| 3  | 1  | 0  | 50 | 1  | 0  |",
            "| 4  | 1  | 0  | 50 | 1  | 0  |",
            "+----+----+----+----+----+----+",
        ];
        let left_batch = [
            "+----+----+----+",
            "| a1 | b1 | c1 |",
            "+----+----+----+",
            "| 1  | 1  | 0  |",
            "| 2  | 1  | 0  |",
            "| 3  | 1  | 0  |",
            "| 4  | 1  | 0  |",
            "+----+----+----+",
        ];
        let right_batch = [
            "+----+----+----+",
            "| a2 | b2 | c2 |",
            "+----+----+----+",
            "| 10 | 1  | 0  |",
            "| 20 | 1  | 0  |",
            "| 30 | 1  | 0  |",
            "| 40 | 1  | 0  |",
            "| 50 | 1  | 0  |",
            "+----+----+----+",
        ];
        let right_empty = [
            "+----+----+----+",
            "| a2 | b2 | c2 |",
            "+----+----+----+",
            "+----+----+----+",
        ];
        let left_empty = [
            "+----+----+----+",
            "| a1 | b1 | c1 |",
            "+----+----+----+",
            "+----+----+----+",
        ];

        // validation of partial join results output for different batch_size setting
        for join_type in join_types {
            for batch_size in (1..21).rev() {
                let task_ctx = prepare_task_ctx(batch_size, true);

                let join = join(
                    Arc::clone(&left),
                    Arc::clone(&right),
                    on.clone(),
                    &join_type,
                    NullEquality::NullEqualsNothing,
                )
                .unwrap();

                let stream = join.execute(0, task_ctx).unwrap();
                let batches = common::collect(stream).await.unwrap();

                // For inner/right join expected batch count equals dev_ceil result,
                // as there is no need to append non-joined build side data.
                // For other join types it'll be div_ceil + 1 -- for additional batch
                // containing not visited build side rows (empty in this test case).
                let expected_batch_count = match join_type {
                    JoinType::Inner
                    | JoinType::Right
                    | JoinType::RightSemi
                    | JoinType::RightAnti => {
                        div_ceil(expected_resultset_records, batch_size)
                    }
                    _ => div_ceil(expected_resultset_records, batch_size) + 1,
                };
                // With batch coalescing, we may have fewer batches than expected
                assert!(
                    batches.len() <= expected_batch_count,
                    "expected at most {expected_batch_count} output batches for {join_type} join with batch_size = {batch_size}, got {}",
                    batches.len()
                );

                let expected = match join_type {
                    JoinType::RightSemi => right_batch.to_vec(),
                    JoinType::RightAnti => right_empty.to_vec(),
                    JoinType::LeftSemi => left_batch.to_vec(),
                    JoinType::LeftAnti => left_empty.to_vec(),
                    _ => common_result.to_vec(),
                };
                // For anti joins with empty results, we may get zero batches
                // (with coalescing) instead of one empty batch with schema
                if batches.is_empty() {
                    // Verify this is an expected empty result case
                    assert!(
                        matches!(join_type, JoinType::RightAnti | JoinType::LeftAnti),
                        "Unexpected empty result for {join_type} join"
                    );
                } else {
                    assert_batches_eq!(expected, &batches);
                }
            }
        }
    }

    #[tokio::test]
    async fn single_partition_join_overallocation() -> Result<()> {
        let left = build_table(
            ("a1", &vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 0]),
            ("b1", &vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 0]),
            ("c1", &vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 0]),
        );
        let right = build_table(
            ("a2", &vec![10, 11]),
            ("b2", &vec![12, 13]),
            ("c2", &vec![14, 15]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("a1", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("b2", &right.schema()).unwrap()) as _,
        )];

        let join_types = vec![
            JoinType::Inner,
            JoinType::Left,
            JoinType::Right,
            JoinType::Full,
            JoinType::LeftSemi,
            JoinType::LeftAnti,
            JoinType::RightSemi,
            JoinType::RightAnti,
            JoinType::LeftMark,
            JoinType::RightMark,
        ];

        for join_type in join_types {
            let runtime = RuntimeEnvBuilder::new()
                .with_memory_limit(100, 1.0)
                .build_arc()?;
            let task_ctx = TaskContext::default().with_runtime(runtime);
            let task_ctx = Arc::new(task_ctx);

            let join = join(
                Arc::clone(&left),
                Arc::clone(&right),
                on.clone(),
                &join_type,
                NullEquality::NullEqualsNothing,
            )?;

            let stream = join.execute(0, task_ctx)?;
            let err = common::collect(stream).await.unwrap_err();

            // Asserting that operator-level reservation attempting to overallocate
            assert_contains!(
                err.to_string(),
                "Resources exhausted: Additional allocation failed for HashJoinInput with top memory consumers (across reservations) as:\n  HashJoinInput"
            );

            assert_contains!(
                err.to_string(),
                "Failed to allocate additional 120.0 B for HashJoinInput"
            );
        }

        Ok(())
    }

    #[tokio::test]
    async fn build_side_sliced_batches_memory_accounting() -> Result<()> {
        // The build side emits zero-copy slices of one large batch, as e.g. an
        // aggregate emitting its output in batch_size chunks does. The buffers
        // shared by the slices must be reserved once in total, not once per
        // slice: per-slice accounting reserves number_of_slices x parent size
        // and aborts queries that fit in memory with room to spare.
        let n = 4096;
        let v: Vec<i32> = (0..n).collect();
        let parent = build_table_i32(("a1", &v), ("b1", &v), ("c1", &v));
        let slices: Vec<RecordBatch> =
            (0..16).map(|i| parent.slice(i * 256, 256)).collect();
        let left =
            TestMemoryExec::try_new_exec(&[slices], parent.schema(), None).unwrap();

        let right_batch = build_table_i32(
            ("a2", &vec![10, 11]),
            ("b2", &vec![0, 1]),
            ("c2", &vec![14, 15]),
        );
        let right = TestMemoryExec::try_new_exec(
            &[vec![right_batch.clone()]],
            right_batch.schema(),
            None,
        )
        .unwrap();
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &parent.schema())?) as _,
            Arc::new(Column::new_with_schema("b2", &right_batch.schema())?) as _,
        )];

        // Enough for the parent batch (~48KB) plus the join hash table, but far
        // below the ~768KB that per-slice accounting would reserve
        let runtime = RuntimeEnvBuilder::new()
            .with_memory_limit(400_000, 1.0)
            .build_arc()?;
        let task_ctx = TaskContext::default().with_runtime(runtime);
        let task_ctx = Arc::new(task_ctx);

        let join = join(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
        )?;

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;
        let num_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
        assert_eq!(num_rows, 2);

        Ok(())
    }

    #[tokio::test]
    async fn partitioned_join_overallocation() -> Result<()> {
        // Prepare partitioned inputs for HashJoinExec
        // No need to adjust partitioning, as execution should fail with `Resources exhausted` error
        let left_batch = build_table_i32(
            ("a1", &vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 0]),
            ("b1", &vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 0]),
            ("c1", &vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 0]),
        );
        let left = TestMemoryExec::try_new_exec(
            &[vec![left_batch.clone()], vec![left_batch.clone()]],
            left_batch.schema(),
            None,
        )
        .unwrap();
        let right_batch = build_table_i32(
            ("a2", &vec![10, 11]),
            ("b2", &vec![12, 13]),
            ("c2", &vec![14, 15]),
        );
        let right = TestMemoryExec::try_new_exec(
            &[vec![right_batch.clone()], vec![right_batch.clone()]],
            right_batch.schema(),
            None,
        )
        .unwrap();
        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left_batch.schema())?) as _,
            Arc::new(Column::new_with_schema("b2", &right_batch.schema())?) as _,
        )];

        let join_types = vec![
            JoinType::Inner,
            JoinType::Left,
            JoinType::Right,
            JoinType::Full,
            JoinType::LeftSemi,
            JoinType::LeftAnti,
            JoinType::RightSemi,
            JoinType::RightAnti,
        ];

        for join_type in join_types {
            let runtime = RuntimeEnvBuilder::new()
                .with_memory_limit(100, 1.0)
                .build_arc()?;
            let session_config = SessionConfig::default().with_batch_size(50);
            let task_ctx = TaskContext::default()
                .with_session_config(session_config)
                .with_runtime(runtime);
            let task_ctx = Arc::new(task_ctx);

            let join = HashJoinExec::try_new(
                Arc::clone(&left) as Arc<dyn ExecutionPlan>,
                Arc::clone(&right) as Arc<dyn ExecutionPlan>,
                on.clone(),
                None,
                &join_type,
                None,
                PartitionMode::Partitioned,
                NullEquality::NullEqualsNothing,
                false,
            )?;

            let stream = join.execute(1, task_ctx)?;
            let err = common::collect(stream).await.unwrap_err();

            // Asserting that stream-level reservation attempting to overallocate
            assert_contains!(
                err.to_string(),
                "Resources exhausted: Additional allocation failed for HashJoinInput[1] with top memory consumers (across reservations) as:\n  HashJoinInput[1]"
            );

            assert_contains!(
                err.to_string(),
                "Failed to allocate additional 120.0 B for HashJoinInput[1]"
            );
        }

        Ok(())
    }

    fn build_table_struct(
        struct_name: &str,
        field_name_and_values: (&str, &Vec<Option<i32>>),
        nulls: Option<NullBuffer>,
    ) -> Arc<dyn ExecutionPlan> {
        let (field_name, values) = field_name_and_values;
        let inner_fields = vec![Field::new(field_name, DataType::Int32, true)];
        let schema = Schema::new(vec![Field::new(
            struct_name,
            DataType::Struct(inner_fields.clone().into()),
            nulls.is_some(),
        )]);

        let batch = RecordBatch::try_new(
            Arc::new(schema),
            vec![Arc::new(StructArray::new(
                inner_fields.into(),
                vec![Arc::new(Int32Array::from(values.clone()))],
                nulls,
            ))],
        )
        .unwrap();
        let schema_ref = batch.schema();
        TestMemoryExec::try_new_exec(&[vec![batch]], schema_ref, None).unwrap()
    }

    #[tokio::test]
    async fn join_on_struct() -> Result<()> {
        let task_ctx = Arc::new(TaskContext::default());
        let left =
            build_table_struct("n1", ("a", &vec![None, Some(1), Some(2), Some(3)]), None);
        let right =
            build_table_struct("n2", ("a", &vec![None, Some(1), Some(2), Some(4)]), None);
        let on = vec![(
            Arc::new(Column::new_with_schema("n1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("n2", &right.schema())?) as _,
        )];

        let (columns, batches, metrics) = join_collect(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["n1", "n2"]);

        allow_duplicates! {
            assert_snapshot!(batches_to_string(&batches), @r"
            +--------+--------+
            | n1     | n2     |
            +--------+--------+
            | {a: }  | {a: }  |
            | {a: 1} | {a: 1} |
            | {a: 2} | {a: 2} |
            +--------+--------+
            ");
        }

        assert_join_metrics!(metrics, 3);

        Ok(())
    }

    #[tokio::test]
    async fn join_on_struct_with_nulls() -> Result<()> {
        let task_ctx = Arc::new(TaskContext::default());
        let left =
            build_table_struct("n1", ("a", &vec![None]), Some(NullBuffer::new_null(1)));
        let right =
            build_table_struct("n2", ("a", &vec![None]), Some(NullBuffer::new_null(1)));
        let on = vec![(
            Arc::new(Column::new_with_schema("n1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("n2", &right.schema())?) as _,
        )];

        let (_, batches_null_eq, metrics) = join_collect(
            Arc::clone(&left),
            Arc::clone(&right),
            on.clone(),
            &JoinType::Inner,
            NullEquality::NullEqualsNull,
            Arc::clone(&task_ctx),
        )
        .await?;

        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches_null_eq), @r"
            +----+----+
            | n1 | n2 |
            +----+----+
            |    |    |
            +----+----+
            ");
        }

        assert_join_metrics!(metrics, 1);

        let (_, batches_null_neq, metrics) = join_collect(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_join_metrics!(metrics, 0);

        // With batch coalescing, empty results may not emit any batches
        // Check that either we have no batches, or an empty batch with proper schema
        if batches_null_neq.is_empty() {
            // This is fine - no output rows
        } else {
            let expected_null_neq =
                ["+----+----+", "| n1 | n2 |", "+----+----+", "+----+----+"];
            assert_batches_eq!(expected_null_neq, &batches_null_neq);
        }

        Ok(())
    }

    /// Returns the column names on the schema
    fn columns(schema: &Schema) -> Vec<String> {
        schema.fields().iter().map(|f| f.name().clone()).collect()
    }

    /// This test verifies that the dynamic filter is marked as complete after HashJoinExec finishes building the hash table.
    #[tokio::test]
    async fn test_hash_join_marks_filter_complete() -> Result<()> {
        let task_ctx = Arc::new(TaskContext::default());
        let left = build_table(
            ("a1", &vec![1, 2, 3]),
            ("b1", &vec![4, 5, 6]),
            ("c1", &vec![7, 8, 9]),
        );
        let right = build_table(
            ("a2", &vec![10, 20, 30]),
            ("b1", &vec![4, 5, 6]),
            ("c2", &vec![70, 80, 90]),
        );

        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
        )];

        let (join, dynamic_filter) =
            hash_join_with_dynamic_filter(left, right, on, JoinType::Inner)?;

        // Execute the join
        let stream = join.execute(0, task_ctx)?;
        let _batches = common::collect(stream).await?;

        // After the join completes, the dynamic filter should be marked as complete
        // wait_complete() should return immediately
        dynamic_filter.wait_complete().await;

        Ok(())
    }

    /// This test verifies that the dynamic filter is marked as complete even when the build side is empty.
    #[tokio::test]
    async fn test_hash_join_marks_filter_complete_empty_build_side() -> Result<()> {
        let task_ctx = Arc::new(TaskContext::default());
        // Empty left side (build side)
        let left = build_table(("a1", &vec![]), ("b1", &vec![]), ("c1", &vec![]));
        let right = build_table(
            ("a2", &vec![10, 20, 30]),
            ("b1", &vec![4, 5, 6]),
            ("c2", &vec![70, 80, 90]),
        );

        let on = vec![(
            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
        )];

        let (join, dynamic_filter) =
            hash_join_with_dynamic_filter(left, right, on, JoinType::Inner)?;

        // Execute the join
        let stream = join.execute(0, task_ctx)?;
        let _batches = common::collect(stream).await?;

        // Even with empty build side, the dynamic filter should be marked as complete
        // wait_complete() should return immediately
        dynamic_filter.wait_complete().await;

        Ok(())
    }

    #[tokio::test]
    async fn test_partitioned_dynamic_filter_reports_empty_canceled_partitions()
    -> Result<()> {
        let mut session_config = SessionConfig::default();
        session_config
            .options_mut()
            .optimizer
            .enable_dynamic_filter_pushdown = true;
        let task_ctx =
            Arc::new(TaskContext::default().with_session_config(session_config));

        let child_left_schema = Arc::new(Schema::new(vec![
            Field::new("child_left_payload", DataType::Int32, false),
            Field::new("child_key", DataType::Int32, false),
            Field::new("child_left_extra", DataType::Int32, false),
        ]));
        let child_right_schema = Arc::new(Schema::new(vec![
            Field::new("child_right_payload", DataType::Int32, false),
            Field::new("child_right_key", DataType::Int32, false),
            Field::new("child_right_extra", DataType::Int32, false),
        ]));
        let parent_left_schema = Arc::new(Schema::new(vec![
            Field::new("parent_payload", DataType::Int32, false),
            Field::new("parent_key", DataType::Int32, false),
            Field::new("parent_extra", DataType::Int32, false),
        ]));

        let child_left: Arc<dyn ExecutionPlan> = TestMemoryExec::try_new_exec(
            &[
                vec![build_table_i32(
                    ("child_left_payload", &vec![10]),
                    ("child_key", &vec![0]),
                    ("child_left_extra", &vec![100]),
                )],
                vec![build_table_i32(
                    ("child_left_payload", &vec![11]),
                    ("child_key", &vec![1]),
                    ("child_left_extra", &vec![101]),
                )],
                vec![build_table_i32(
                    ("child_left_payload", &vec![12]),
                    ("child_key", &vec![2]),
                    ("child_left_extra", &vec![102]),
                )],
                vec![build_table_i32(
                    ("child_left_payload", &vec![13]),
                    ("child_key", &vec![3]),
                    ("child_left_extra", &vec![103]),
                )],
            ],
            Arc::clone(&child_left_schema),
            None,
        )?;
        let child_right: Arc<dyn ExecutionPlan> = TestMemoryExec::try_new_exec(
            &[
                vec![build_table_i32(
                    ("child_right_payload", &vec![20]),
                    ("child_right_key", &vec![0]),
                    ("child_right_extra", &vec![200]),
                )],
                vec![build_table_i32(
                    ("child_right_payload", &vec![21]),
                    ("child_right_key", &vec![1]),
                    ("child_right_extra", &vec![201]),
                )],
                vec![build_table_i32(
                    ("child_right_payload", &vec![22]),
                    ("child_right_key", &vec![2]),
                    ("child_right_extra", &vec![202]),
                )],
                vec![build_table_i32(
                    ("child_right_payload", &vec![23]),
                    ("child_right_key", &vec![3]),
                    ("child_right_extra", &vec![203]),
                )],
            ],
            Arc::clone(&child_right_schema),
            None,
        )?;
        let parent_left: Arc<dyn ExecutionPlan> = TestMemoryExec::try_new_exec(
            &[
                vec![build_table_i32(
                    ("parent_payload", &vec![30]),
                    ("parent_key", &vec![0]),
                    ("parent_extra", &vec![300]),
                )],
                vec![RecordBatch::new_empty(Arc::clone(&parent_left_schema))],
                vec![build_table_i32(
                    ("parent_payload", &vec![32]),
                    ("parent_key", &vec![2]),
                    ("parent_extra", &vec![302]),
                )],
                vec![RecordBatch::new_empty(Arc::clone(&parent_left_schema))],
            ],
            Arc::clone(&parent_left_schema),
            None,
        )?;

        let child_on = vec![(
            Arc::new(Column::new_with_schema("child_key", &child_left_schema)?) as _,
            Arc::new(Column::new_with_schema(
                "child_right_key",
                &child_right_schema,
            )?) as _,
        )];
        let (child_join, _child_dynamic_filter) = hash_join_with_dynamic_filter_and_mode(
            child_left,
            child_right,
            child_on,
            JoinType::Inner,
            PartitionMode::Partitioned,
        )?;
        let child_join: Arc<dyn ExecutionPlan> = Arc::new(child_join);

        let parent_on = vec![(
            Arc::new(Column::new_with_schema("parent_key", &parent_left_schema)?) as _,
            Arc::new(Column::new_with_schema("child_key", &child_join.schema())?) as _,
        )];
        let parent_join = HashJoinExec::try_new(
            parent_left,
            child_join,
            parent_on,
            None,
            &JoinType::RightSemi,
            None,
            PartitionMode::Partitioned,
            NullEquality::NullEqualsNothing,
            false,
        )?;

        let batches = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            crate::execution_plan::collect(Arc::new(parent_join), task_ctx),
        )
        .await
        .expect("partitioned right-semi join should not hang")?;

        assert_batches_sorted_eq!(
            [
                "+--------------------+-----------+------------------+---------------------+-----------------+-------------------+",
                "| child_left_payload | child_key | child_left_extra | child_right_payload | child_right_key | child_right_extra |",
                "+--------------------+-----------+------------------+---------------------+-----------------+-------------------+",
                "| 10                 | 0         | 100              | 20                  | 0               | 200               |",
                "| 12                 | 2         | 102              | 22                  | 2               | 202               |",
                "+--------------------+-----------+------------------+---------------------+-----------------+-------------------+",
            ],
            &batches
        );

        Ok(())
    }

    #[tokio::test]
    async fn test_hash_join_skips_probe_on_empty_build_after_partition_bounds_report()
    -> Result<()> {
        let task_ctx = Arc::new(TaskContext::default());
        let (left, right, on) = empty_build_with_probe_error_inputs();

        // Keep an extra consumer reference so execute() enables dynamic filter pushdown
        // and enters the WaitPartitionBoundsReport path before deciding whether to poll
        // the probe side.
        let (join, dynamic_filter) =
            hash_join_with_dynamic_filter(left, right, on, JoinType::Inner)?;

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;
        assert!(batches.is_empty());

        dynamic_filter.wait_complete().await;

        Ok(())
    }

    #[tokio::test]
    async fn test_perfect_hash_join_with_negative_numbers() -> Result<()> {
        let task_ctx = prepare_task_ctx(8192, true);
        let (left_schema, right_schema, on) = build_schema_and_on()?;

        let left_batch = RecordBatch::try_new(
            Arc::clone(&left_schema),
            vec![
                Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef,
                Arc::new(Int32Array::from(vec![-1, 0, 1])) as ArrayRef,
            ],
        )?;
        let left = TestMemoryExec::try_new_exec(&[vec![left_batch]], left_schema, None)?;

        let right_batch = RecordBatch::try_new(
            Arc::clone(&right_schema),
            vec![
                Arc::new(Int32Array::from(vec![10, 20, 30, 40])) as ArrayRef,
                Arc::new(Int32Array::from(vec![1, -1, 0, 2])) as ArrayRef,
            ],
        )?;
        let right =
            TestMemoryExec::try_new_exec(&[vec![right_batch]], right_schema, None)?;

        let (columns, batches, metrics) = join_collect(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b1", "a2", "b1"]);

        assert_batches_sorted_eq!(
            [
                "+----+----+----+----+",
                "| a1 | b1 | a2 | b1 |",
                "+----+----+----+----+",
                "| 1  | -1 | 20 | -1 |",
                "| 2  | 0  | 30 | 0  |",
                "| 3  | 1  | 10 | 1  |",
                "+----+----+----+----+",
            ],
            &batches
        );

        assert_phj_used(&metrics, true);

        Ok(())
    }

    #[tokio::test]
    async fn test_perfect_hash_join_overflow_full_int64_range() -> Result<()> {
        let task_ctx = prepare_task_ctx(8192, true);
        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, true)]));
        let batch = RecordBatch::try_new(
            Arc::clone(&schema),
            vec![Arc::new(Int64Array::from(vec![i64::MIN, i64::MAX]))],
        )?;
        let left = TestMemoryExec::try_new_exec(
            &[vec![batch.clone()]],
            Arc::clone(&schema),
            None,
        )?;
        let right = TestMemoryExec::try_new_exec(&[vec![batch]], schema, None)?;
        let on: JoinOn = vec![(
            Arc::new(Column::new_with_schema("a", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("a", &right.schema())?) as _,
        )];
        let (_columns, batches, _metrics) = join_collect(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;
        let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
        assert_eq!(total_rows, 2);
        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn test_phj_null_equals_null_build_no_nulls_probe_has_nulls(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let (left_schema, right_schema, on) = build_schema_and_on()?;

        let left_batch = RecordBatch::try_new(
            Arc::clone(&left_schema),
            vec![
                Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef,
                Arc::new(Int32Array::from(vec![10, 20])) as ArrayRef,
            ],
        )?;
        let left = TestMemoryExec::try_new_exec(&[vec![left_batch]], left_schema, None)?;

        let right_batch = RecordBatch::try_new(
            Arc::clone(&right_schema),
            vec![
                Arc::new(Int32Array::from(vec![3, 4])) as ArrayRef,
                Arc::new(Int32Array::from(vec![Some(10), None])) as ArrayRef,
            ],
        )?;
        let right =
            TestMemoryExec::try_new_exec(&[vec![right_batch]], right_schema, None)?;

        let (columns, batches, metrics) = join_collect(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNull,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b1", "a2", "b1"]);
        assert_batches_sorted_eq!(
            [
                "+----+----+----+----+",
                "| a1 | b1 | a2 | b1 |",
                "+----+----+----+----+",
                "| 1  | 10 | 3  | 10 |",
                "+----+----+----+----+",
            ],
            &batches
        );

        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn test_phj_null_equals_nothing_build_probe_all_have_nulls(
        batch_size: usize,
        use_perfect_hash_join_as_possible: bool,
    ) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, use_perfect_hash_join_as_possible);
        let (left_schema, right_schema, on) = build_schema_and_on()?;

        let left_batch = RecordBatch::try_new(
            Arc::clone(&left_schema),
            vec![
                Arc::new(Int32Array::from(vec![Some(1), Some(2)])) as ArrayRef,
                Arc::new(Int32Array::from(vec![Some(10), None])) as ArrayRef,
            ],
        )?;
        let left = TestMemoryExec::try_new_exec(&[vec![left_batch]], left_schema, None)?;

        let right_batch = RecordBatch::try_new(
            Arc::clone(&right_schema),
            vec![
                Arc::new(Int32Array::from(vec![Some(3), Some(4)])) as ArrayRef,
                Arc::new(Int32Array::from(vec![Some(10), None])) as ArrayRef,
            ],
        )?;
        let right =
            TestMemoryExec::try_new_exec(&[vec![right_batch]], right_schema, None)?;

        let (columns, batches, metrics) = join_collect(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNothing,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b1", "a2", "b1"]);
        assert_batches_sorted_eq!(
            [
                "+----+----+----+----+",
                "| a1 | b1 | a2 | b1 |",
                "+----+----+----+----+",
                "| 1  | 10 | 3  | 10 |",
                "+----+----+----+----+",
            ],
            &batches
        );

        assert_phj_used(&metrics, use_perfect_hash_join_as_possible);

        Ok(())
    }

    #[tokio::test]
    async fn test_phj_null_equals_null_build_have_nulls() -> Result<()> {
        let task_ctx = prepare_task_ctx(8192, true);
        let (left_schema, right_schema, on) = build_schema_and_on()?;

        let left_batch = RecordBatch::try_new(
            Arc::clone(&left_schema),
            vec![
                Arc::new(Int32Array::from(vec![Some(1), Some(2), Some(3)])) as ArrayRef,
                Arc::new(Int32Array::from(vec![Some(10), Some(20), None])) as ArrayRef,
            ],
        )?;
        let left = TestMemoryExec::try_new_exec(&[vec![left_batch]], left_schema, None)?;

        let right_batch = RecordBatch::try_new(
            Arc::clone(&right_schema),
            vec![
                Arc::new(Int32Array::from(vec![Some(3), Some(4)])) as ArrayRef,
                Arc::new(Int32Array::from(vec![Some(10), Some(30)])) as ArrayRef,
            ],
        )?;
        let right =
            TestMemoryExec::try_new_exec(&[vec![right_batch]], right_schema, None)?;

        let (columns, batches, metrics) = join_collect(
            left,
            right,
            on,
            &JoinType::Inner,
            NullEquality::NullEqualsNull,
            task_ctx,
        )
        .await?;

        assert_eq!(columns, vec!["a1", "b1", "a2", "b1"]);
        assert_batches_sorted_eq!(
            [
                "+----+----+----+----+",
                "| a1 | b1 | a2 | b1 |",
                "+----+----+----+----+",
                "| 1  | 10 | 3  | 10 |",
                "+----+----+----+----+",
            ],
            &batches
        );

        assert_phj_used(&metrics, false);

        Ok(())
    }

    /// Test null-aware anti join when probe side (right) contains NULL
    /// Expected: no rows should be output (NULL in subquery means all results are unknown)
    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn test_null_aware_anti_join_probe_null(batch_size: usize) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, false);

        // Build left table (rows to potentially output)
        let left = build_table_two_cols(
            ("c1", &vec![Some(1), Some(2), Some(3), Some(4)]),
            ("dummy", &vec![Some(10), Some(20), Some(30), Some(40)]),
        );

        // Build right table (subquery with NULL)
        let right = build_table_two_cols(
            ("c2", &vec![Some(1), Some(2), Some(3), None]),
            ("dummy", &vec![Some(100), Some(200), Some(300), Some(400)]),
        );

        let on = vec![(
            Arc::new(Column::new_with_schema("c1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("c2", &right.schema())?) as _,
        )];

        // Create null-aware anti join
        let join = HashJoinExec::try_new(
            left,
            right,
            on,
            None,
            &JoinType::LeftAnti,
            None,
            PartitionMode::CollectLeft,
            NullEquality::NullEqualsNothing,
            true, // null_aware = true
        )?;

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        // Expected: empty result (probe side has NULL, so no rows should be output)
        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            ++
            ++
            ");
        }
        Ok(())
    }

    /// Test null-aware anti join when build side (left) contains NULL keys
    /// Expected: rows with NULL keys should not be output
    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn test_null_aware_anti_join_build_null(batch_size: usize) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, false);

        // Build left table with NULL key (this row should not be output)
        let left = build_table_two_cols(
            ("c1", &vec![Some(1), Some(4), None]),
            ("dummy", &vec![Some(10), Some(40), Some(0)]),
        );

        // Build right table (no NULL, so probe-side check passes)
        let right = build_table_two_cols(
            ("c2", &vec![Some(1), Some(2), Some(3)]),
            ("dummy", &vec![Some(100), Some(200), Some(300)]),
        );

        let on = vec![(
            Arc::new(Column::new_with_schema("c1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("c2", &right.schema())?) as _,
        )];

        // Create null-aware anti join
        let join = HashJoinExec::try_new(
            left,
            right,
            on,
            None,
            &JoinType::LeftAnti,
            None,
            PartitionMode::CollectLeft,
            NullEquality::NullEqualsNothing,
            true, // null_aware = true
        )?;

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        // Expected: only c1=4 (not c1=1 which matches, not c1=NULL)
        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+-------+
            | c1 | dummy |
            +----+-------+
            | 4  | 40    |
            +----+-------+
            ");
        }
        Ok(())
    }

    /// Test null-aware anti join with no NULLs (should work like regular anti join)
    #[apply(hash_join_exec_configs)]
    #[tokio::test]
    async fn test_null_aware_anti_join_no_nulls(batch_size: usize) -> Result<()> {
        let task_ctx = prepare_task_ctx(batch_size, false);

        // Build left table (no NULLs)
        let left = build_table_two_cols(
            ("c1", &vec![Some(1), Some(2), Some(4), Some(5)]),
            ("dummy", &vec![Some(10), Some(20), Some(40), Some(50)]),
        );

        // Build right table (no NULLs)
        let right = build_table_two_cols(
            ("c2", &vec![Some(1), Some(2), Some(3)]),
            ("dummy", &vec![Some(100), Some(200), Some(300)]),
        );

        let on = vec![(
            Arc::new(Column::new_with_schema("c1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("c2", &right.schema())?) as _,
        )];

        // Create null-aware anti join
        let join = HashJoinExec::try_new(
            left,
            right,
            on,
            None,
            &JoinType::LeftAnti,
            None,
            PartitionMode::CollectLeft,
            NullEquality::NullEqualsNothing,
            true, // null_aware = true
        )?;

        let stream = join.execute(0, task_ctx)?;
        let batches = common::collect(stream).await?;

        // Expected: c1=4 and c1=5 (they don't match anything in right)
        allow_duplicates! {
            assert_snapshot!(batches_to_sort_string(&batches), @r"
            +----+-------+
            | c1 | dummy |
            +----+-------+
            | 4  | 40    |
            | 5  | 50    |
            +----+-------+
            ");
        }
        Ok(())
    }

    /// Test that null_aware validation rejects non-LeftAnti join types
    #[tokio::test]
    async fn test_null_aware_validation_wrong_join_type() {
        let left =
            build_table_two_cols(("c1", &vec![Some(1)]), ("dummy", &vec![Some(10)]));
        let right =
            build_table_two_cols(("c2", &vec![Some(1)]), ("dummy", &vec![Some(100)]));

        let on = vec![(
            Arc::new(Column::new_with_schema("c1", &left.schema()).unwrap()) as _,
            Arc::new(Column::new_with_schema("c2", &right.schema()).unwrap()) as _,
        )];

        // Try to create null-aware Inner join (should fail)
        let result = HashJoinExec::try_new(
            left,
            right,
            on,
            None,
            &JoinType::Inner,
            None,
            PartitionMode::CollectLeft,
            NullEquality::NullEqualsNothing,
            true, // null_aware = true (invalid for Inner join)
        );

        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("null_aware can only be true for LeftAnti joins")
        );
    }

    /// Test that null_aware validation rejects multi-column joins
    #[tokio::test]
    async fn test_null_aware_validation_multi_column() {
        let left = build_table(("a", &vec![1]), ("b", &vec![2]), ("c", &vec![3]));
        let right = build_table(("x", &vec![1]), ("y", &vec![2]), ("z", &vec![3]));

        // Try multi-column join
        let on = vec![
            (
                Arc::new(Column::new_with_schema("a", &left.schema()).unwrap()) as _,
                Arc::new(Column::new_with_schema("x", &right.schema()).unwrap()) as _,
            ),
            (
                Arc::new(Column::new_with_schema("b", &left.schema()).unwrap()) as _,
                Arc::new(Column::new_with_schema("y", &right.schema()).unwrap()) as _,
            ),
        ];

        // Try to create null-aware anti join with 2 columns (should fail)
        let result = HashJoinExec::try_new(
            left,
            right,
            on,
            None,
            &JoinType::LeftAnti,
            None,
            PartitionMode::CollectLeft,
            NullEquality::NullEqualsNothing,
            true, // null_aware = true (invalid for multi-column)
        );

        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("null_aware anti join only supports single column join key")
        );
    }

    #[test]
    fn test_lr_is_preserved() {
        assert_eq!(lr_is_preserved(JoinType::Inner), (true, true));
        assert_eq!(lr_is_preserved(JoinType::Left), (true, false));
        assert_eq!(lr_is_preserved(JoinType::Right), (false, true));
        assert_eq!(lr_is_preserved(JoinType::Full), (false, false));
        assert_eq!(lr_is_preserved(JoinType::LeftSemi), (true, true));
        assert_eq!(lr_is_preserved(JoinType::LeftAnti), (true, false));
        assert_eq!(lr_is_preserved(JoinType::LeftMark), (true, false));
        assert_eq!(lr_is_preserved(JoinType::RightSemi), (true, true));
        assert_eq!(lr_is_preserved(JoinType::RightAnti), (false, true));
        assert_eq!(lr_is_preserved(JoinType::RightMark), (false, true));
    }

    #[test]
    fn test_with_dynamic_filter() -> Result<()> {
        let (_, _, on) = build_schema_and_on()?;
        let left = build_table(("a1", &vec![1]), ("b1", &vec![1]), ("c1", &vec![1]));
        let right = build_table(("a2", &vec![1]), ("b1", &vec![1]), ("c2", &vec![1]));

        let join = HashJoinExec::try_new(
            left,
            right,
            on,
            None,
            &JoinType::Inner,
            None,
            PartitionMode::CollectLeft,
            NullEquality::NullEqualsNothing,
            false,
        )?;
        assert!(join.dynamic_expressions_produced().is_empty());

        let df = Arc::new(DynamicFilterPhysicalExpr::new(
            vec![Arc::new(Column::new("b1", 1)) as _],
            lit(true),
        ));
        let join = join.with_dynamic_filter_expr(Arc::clone(&df))?;

        let produced = join.dynamic_expressions_produced();
        assert_eq!(produced.len(), 1);
        assert_eq!(
            produced[0]
                .expression_id()
                .expect("DynamicFilterPhysicalExpr always has an expression_id"),
            df.expression_id()
                .expect("DynamicFilterPhysicalExpr always has an expression_id"),
        );
        Ok(())
    }

    #[test]
    fn test_swap_inputs_rejects_dynamic_filter() -> Result<()> {
        let left = build_table(
            ("l_key", &vec![1]),
            ("l_payload", &vec![10]),
            ("l_other", &vec![100]),
        );
        let right = build_table(
            ("r_payload", &vec![20]),
            ("r_key", &vec![1]),
            ("r_other", &vec![200]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("l_key", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("r_key", &right.schema())?) as _,
        )];

        let dynamic_filter = HashJoinExec::create_dynamic_filter(&on);
        let join = HashJoinExec::try_new(
            left,
            right,
            on,
            None,
            &JoinType::LeftSemi,
            None,
            PartitionMode::CollectLeft,
            NullEquality::NullEqualsNothing,
            false,
        )?
        .with_dynamic_filter_expr(dynamic_filter)?;

        let err = join.swap_inputs(PartitionMode::CollectLeft).unwrap_err();
        assert_contains!(
            err.to_string(),
            "Cannot swap HashJoinExec inputs after dynamic filters have been constructed"
        );
        Ok(())
    }

    #[test]
    fn test_dynamic_filter_pushdown_allowed_for_null_equal_join() -> Result<()> {
        let (_, _, on) = build_schema_and_on()?;
        let left = build_table(("a1", &vec![1]), ("b1", &vec![1]), ("c1", &vec![1]));
        let right = build_table(("a2", &vec![1]), ("b1", &vec![1]), ("c2", &vec![1]));

        let mut session_config = SessionConfig::default();
        session_config
            .options_mut()
            .optimizer
            .enable_join_dynamic_filter_pushdown = true;

        let join = HashJoinExec::try_new(
            left,
            right,
            on,
            None,
            &JoinType::RightSemi,
            None,
            PartitionMode::CollectLeft,
            NullEquality::NullEqualsNull,
            false,
        )?;

        // Null-equal joins keep dynamic filter pushdown: the pushed predicate carries an
        // `IS NULL` disjunct so a probe-side NULL still reaches the join.
        assert!(join.allow_join_dynamic_filter_pushdown(session_config.options()));

        Ok(())
    }

    #[test]
    fn test_dynamic_filter_pushdown_rejects_null_aware_nullable_build_key() -> Result<()>
    {
        let left = build_table_two_cols(
            ("a1", &vec![Some(1), None]),
            ("b1", &vec![Some(1), Some(2)]),
        );
        let right = build_table_two_cols(
            ("a2", &vec![Some(2), Some(3)]),
            ("b2", &vec![Some(1), Some(2)]),
        );
        let on = vec![(
            Arc::new(Column::new_with_schema("a1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("a2", &right.schema())?) as _,
        )];

        let mut session_config = SessionConfig::default();
        session_config
            .options_mut()
            .optimizer
            .enable_join_dynamic_filter_pushdown = true;

        let join = HashJoinExec::try_new(
            left,
            right,
            on,
            None,
            &JoinType::LeftAnti,
            None,
            PartitionMode::CollectLeft,
            NullEquality::NullEqualsNothing,
            true,
        )?;

        assert!(!join.allow_join_dynamic_filter_pushdown(session_config.options()));

        Ok(())
    }

    #[test]
    fn test_dynamic_filter_pushdown_allows_null_aware_non_null_build_key() -> Result<()> {
        // A NOT NULL build key cannot surface a build-side NULL, so the
        // pushdown must stay enabled.
        let left = build_table(("a1", &vec![1]), ("b1", &vec![1]), ("c1", &vec![1]));
        let right = build_table(("a2", &vec![2]), ("b2", &vec![2]), ("c2", &vec![2]));
        let on = vec![(
            Arc::new(Column::new_with_schema("a1", &left.schema())?) as _,
            Arc::new(Column::new_with_schema("a2", &right.schema())?) as _,
        )];

        let mut session_config = SessionConfig::default();
        session_config
            .options_mut()
            .optimizer
            .enable_join_dynamic_filter_pushdown = true;

        let join = HashJoinExec::try_new(
            left,
            right,
            on,
            None,
            &JoinType::LeftAnti,
            None,
            PartitionMode::CollectLeft,
            NullEquality::NullEqualsNothing,
            true,
        )?;

        assert!(join.allow_join_dynamic_filter_pushdown(session_config.options()));

        Ok(())
    }

    fn range_partitioned_dynamic_filter_test_join(
        left_split: i32,
        right_split: i32,
    ) -> Result<(HashJoinExec, JoinOn)> {
        let (left_schema, right_schema, on) = build_schema_and_on()?;
        let left_partitioning = Partitioning::Range(RangePartitioning::try_new(
            [PhysicalSortExpr {
                expr: Arc::clone(&on[0].0),
                options: Default::default(),
            }]
            .into(),
            vec![SplitPoint::new(vec![ScalarValue::Int32(Some(left_split))])],
        )?);
        let right_partitioning = Partitioning::Range(RangePartitioning::try_new(
            [PhysicalSortExpr {
                expr: Arc::clone(&on[0].1),
                options: Default::default(),
            }]
            .into(),
            vec![SplitPoint::new(vec![ScalarValue::Int32(Some(right_split))])],
        )?);
        let left = Arc::new(PartitionedTestExec::try_new(
            left_schema,
            left_partitioning,
        )?);
        let right = Arc::new(PartitionedTestExec::try_new(
            right_schema,
            right_partitioning,
        )?);

        let join = HashJoinExec::try_new(
            left,
            right,
            on.clone(),
            None,
            &JoinType::Inner,
            None,
            PartitionMode::Partitioned,
            NullEquality::NullEqualsNothing,
            false,
        )?;
        Ok((join, on))
    }

    fn with_hash_partitioned_children(
        join: &HashJoinExec,
        on: &JoinOn,
    ) -> Result<HashJoinExec> {
        join.builder()
            .with_new_children(vec![
                Arc::new(PartitionedTestExec::try_new(
                    join.left().schema(),
                    Partitioning::Hash(vec![Arc::clone(&on[0].0)], 2),
                )?),
                Arc::new(PartitionedTestExec::try_new(
                    join.right().schema(),
                    Partitioning::Hash(vec![Arc::clone(&on[0].1)], 2),
                )?),
            ])?
            .build()
    }

    #[test]
    fn test_partitioned_dynamic_filter_pushdown_allows_supported_partitioning()
    -> Result<()> {
        let (range_join, on) = range_partitioned_dynamic_filter_test_join(10, 10)?;
        let hash_join = with_hash_partitioned_children(&range_join, &on)?;
        let mut session_config = SessionConfig::default();
        session_config
            .options_mut()
            .optimizer
            .enable_join_dynamic_filter_pushdown = true;

        assert!(range_join.allow_join_dynamic_filter_pushdown(session_config.options()));
        assert!(hash_join.allow_join_dynamic_filter_pushdown(session_config.options()));

        session_config
            .options_mut()
            .optimizer
            .preserve_file_partitions = 1;
        assert!(range_join.allow_join_dynamic_filter_pushdown(session_config.options()));

        Ok(())
    }

    #[test]
    fn test_partitioned_dynamic_filter_pushdown_rejects_unsupported_partitioning()
    -> Result<()> {
        let (range_join, on) = range_partitioned_dynamic_filter_test_join(10, 10)?;
        let hash_join = with_hash_partitioned_children(&range_join, &on)?;
        let (mismatched_range_join, _) =
            range_partitioned_dynamic_filter_test_join(10, 11)?;
        let mut session_config = SessionConfig::default();
        session_config
            .options_mut()
            .optimizer
            .enable_join_dynamic_filter_pushdown = true;

        assert!(
            !mismatched_range_join
                .allow_join_dynamic_filter_pushdown(session_config.options())
        );

        session_config
            .options_mut()
            .optimizer
            .preserve_file_partitions = 1;
        assert!(!hash_join.allow_join_dynamic_filter_pushdown(session_config.options()));

        Ok(())
    }

    #[test]
    fn test_with_dynamic_filter_rejects_invalid_columns() -> Result<()> {
        let (_, _, on) = build_schema_and_on()?;
        let left = build_table(("a1", &vec![1]), ("b1", &vec![1]), ("c1", &vec![1]));
        let right = build_table(("a2", &vec![1]), ("b1", &vec![1]), ("c2", &vec![1]));

        let join = HashJoinExec::try_new(
            left,
            right,
            on,
            None,
            &JoinType::Inner,
            None,
            PartitionMode::CollectLeft,
            NullEquality::NullEqualsNothing,
            false,
        )?;

        // Column index 99 is out of bounds for the right (probe) side schema.
        let df = Arc::new(DynamicFilterPhysicalExpr::new(
            vec![Arc::new(Column::new("bad", 99)) as _],
            lit(true),
        ));
        assert!(join.with_dynamic_filter_expr(df).is_err());
        Ok(())
    }
}