Pass projection and filter

Author: osipovartem

datafusion/physical-plan/src/joins/grace_hash_join/exec.rs

@@ -684,7 +684,8 @@ impl ExecutionPlan for GraceHashJoinExec {
             spill_right,
             on_left,
             on_right,
-            self.random_state.clone(),
+            self.projection.clone(),
+            self.filter.clone(),
             self.join_type,
             column_indices_after_projection,
             join_metrics,
@@ -854,24 +855,8 @@ impl ExecutionPlan for GraceHashJoinExec {
     }
 }
 
-/// 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,
-}
 
+#[allow(clippy::too_many_arguments)]
 pub async fn partition_and_spill(
     random_state: RandomState,
     on: Vec<(PhysicalExprRef, PhysicalExprRef)>,
@@ -915,6 +900,7 @@ pub async fn partition_and_spill(
     Ok((left_index, right_index))
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn partition_and_spill_one_side(
     input: &mut SendableRecordBatchStream,
     on_exprs: &[PhysicalExprRef],

datafusion/physical-plan/src/joins/grace_hash_join/stream.rs

@@ -34,7 +34,6 @@ use crate::empty::EmptyExec;
 use crate::joins::grace_hash_join::exec::PartitionIndex;
 use crate::joins::{HashJoinExec, PartitionMode};
 use crate::test::TestMemoryExec;
-use ahash::RandomState;
 use arrow::datatypes::{Schema, SchemaRef};
 use arrow::record_batch::RecordBatch;
 use datafusion_common::{JoinType, NullEquality, Result};
@@ -70,7 +69,8 @@ pub struct GraceHashJoinStream {
     spill_right: Arc<SpillManager>,
     on_left: Vec<PhysicalExprRef>,
     on_right: Vec<PhysicalExprRef>,
-    random_state: RandomState,
+    projection: Option<Vec<usize>>,
+    filter: Option<JoinFilter>,
     join_type: JoinType,
     column_indices: Vec<ColumnIndex>,
     join_metrics: Arc<BuildProbeJoinMetrics>,
@@ -113,7 +113,8 @@ impl GraceHashJoinStream {
         spill_right: Arc<SpillManager>,
         on_left: Vec<PhysicalExprRef>,
         on_right: Vec<PhysicalExprRef>,
-        random_state: RandomState,
+        projection: Option<Vec<usize>>,
+        filter: Option<JoinFilter>,
         join_type: JoinType,
         column_indices: Vec<ColumnIndex>,
         join_metrics: Arc<BuildProbeJoinMetrics>,
@@ -127,7 +128,8 @@ impl GraceHashJoinStream {
             spill_right,
             on_left,
             on_right,
-            random_state,
+            projection,
+            filter,
             join_type,
             column_indices,
             join_metrics,
@@ -225,7 +227,8 @@ impl GraceHashJoinStream {
                             right_batches,
                             &self.on_left,
                             &self.on_right,
-                            self.random_state.clone(),
+                            self.projection.clone(),
+                            self.filter.clone(),
                             self.join_type,
                             &self.column_indices,
                             &self.join_metrics,
@@ -283,10 +286,11 @@ fn build_in_memory_join_stream(
     right_batches: Vec<RecordBatch>,
     on_left: &[PhysicalExprRef],
     on_right: &[PhysicalExprRef],
-    random_state: RandomState,
+    projection: Option<Vec<usize>>,
+    filter: Option<JoinFilter>,
     join_type: JoinType,
-    column_indices: &[ColumnIndex],
-    join_metrics: &BuildProbeJoinMetrics,
+    _column_indices: &[ColumnIndex],
+    _join_metrics: &BuildProbeJoinMetrics,
     context: &Arc<TaskContext>,
 ) -> Result<SendableRecordBatchStream> {
     if left_batches.is_empty() && right_batches.is_empty() {
@@ -324,9 +328,9 @@ fn build_in_memory_join_stream(
         left_plan,
         right_plan,
         on,
-        None::<JoinFilter>,
+        filter,
         &join_type,
-        None,
+        projection,
         PartitionMode::CollectLeft,
         NullEquality::NullEqualsNothing,
     )?;

datafusion/physical-plan/src/spill/spill_manager.rs

@@ -173,37 +173,37 @@ impl SpillManager {
     /// Automatically decides whether to spill the given RecordBatch to memory or disk,
     /// depending on available memory pool capacity.
     pub(crate) fn spill_batch_auto(&self, batch: &RecordBatch, request_msg: &str) -> Result<SpillLocation> {
-        let Some(file) = self.spill_record_batch_and_finish(slice::from_ref(batch), request_msg)? else {
-            return Err(DataFusionError::Execution(
-                "failed to spill batch to disk".into(),
-            ));
-        };
-        Ok(SpillLocation::Disk(Arc::new(file)))
-        // //
-        // let size = batch.get_sliced_size()?;
-        //
-        // // Check current memory usage and total limit from the runtime memory pool
-        // let used = self.env.memory_pool.reserved();
-        // let limit = match self.env.memory_pool.memory_limit() {
-        //     datafusion_execution::memory_pool::MemoryLimit::Finite(l) => l,
-        //     _ => usize::MAX,
+        // let Some(file) = self.spill_record_batch_and_finish(slice::from_ref(batch), request_msg)? else {
+        //     return Err(DataFusionError::Execution(
+        //         "failed to spill batch to disk".into(),
+        //     ));
         // };
-        // println!("size {size} used {used}");
-        // // If there's enough memory (with a small safety margin), keep it in memory
-        // if used + size * 3 * 64 / 2  <= limit {
-        //     let buf = Arc::new(InMemorySpillBuffer::from_batch(batch)?);
-        //     self.metrics.spilled_bytes.add(size);
-        //     self.metrics.spilled_rows.add(batch.num_rows());
-        //     Ok(SpillLocation::Memory(buf))
-        // } else {
-        //     // Otherwise spill to disk using the existing SpillManager logic
-        //     let Some(file) = self.spill_record_batch_and_finish(slice::from_ref(batch), request_msg)? else {
-        //         return Err(DataFusionError::Execution(
-        //             "failed to spill batch to disk".into(),
-        //         ));
-        //     };
-        //     Ok(SpillLocation::Disk(Arc::new(file)))
-        // }
+        // Ok(SpillLocation::Disk(Arc::new(file)))
+        // //
+        let size = batch.get_sliced_size()?;
+
+        // Check current memory usage and total limit from the runtime memory pool
+        let used = self.env.memory_pool.reserved();
+        let limit = match self.env.memory_pool.memory_limit() {
+            datafusion_execution::memory_pool::MemoryLimit::Finite(l) => l,
+            _ => usize::MAX,
+        };
+
+        // If there's enough memory (with a safety margin), keep it in memory
+        if used + size * 3 / 2  <= limit {
+            let buf = Arc::new(InMemorySpillBuffer::from_batch(batch)?);
+            self.metrics.spilled_bytes.add(size);
+            self.metrics.spilled_rows.add(batch.num_rows());
+            Ok(SpillLocation::Memory(buf))
+        } else {
+            // Otherwise spill to disk using the existing SpillManager logic
+            let Some(file) = self.spill_record_batch_and_finish(slice::from_ref(batch), request_msg)? else {
+                return Err(DataFusionError::Execution(
+                    "failed to spill batch to disk".into(),
+                ));
+            };
+            Ok(SpillLocation::Disk(Arc::new(file)))
+        }
     }
 
     pub fn spill_batches_auto(