Fix errors introduced during rebase

Author: mustafasrepo

datafusion/core/src/physical_optimizer/projection_pushdown.rs

@@ -1295,7 +1295,7 @@ mod tests {
         Distribution, Partitioning, PhysicalExpr, PhysicalSortExpr,
         PhysicalSortRequirement, ScalarFunctionExpr,
     };
-    use datafusion_physical_plan::joins::{SymmetricHashJoinExec, SlidingHashJoinExec};
+    use datafusion_physical_plan::joins::SymmetricHashJoinExec;
     use datafusion_physical_plan::streaming::{PartitionStream, StreamingTableExec};
     use datafusion_physical_plan::union::UnionExec;
 

datafusion/expr/src/interval_arithmetic.rs

@@ -601,6 +601,17 @@ impl Interval {
         }
     }
 
+    /// Decide if this interval is a superset of `other`. If argument `strict`
+    /// is `true`, only returns `true` if this interval is a strict superset.
+    ///
+    /// NOTE: This function only works with intervals of the same data type.
+    ///       Attempting to compare intervals of different data types will lead
+    ///       to an error.
+    pub fn is_superset(&self, other: &Interval, strict: bool) -> Result<bool> {
+        Ok(!(strict && self.eq(other))
+            && (self.contains(other)? == Interval::CERTAINLY_TRUE))
+    }
+
     /// Add the given interval (`other`) to this interval. Say we have intervals
     /// `[a1, b1]` and `[a2, b2]`, then their sum is `[a1 + a2, b1 + b2]`. Note
     /// that this represents all possible values the sum can take if one can

datafusion/physical-expr/src/expressions/negative.rs

@@ -21,7 +21,6 @@ use std::any::Any;
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
 
-use crate::intervals::Interval;
 use crate::physical_expr::down_cast_any_ref;
 use crate::sort_properties::SortProperties;
 use crate::PhysicalExpr;

datafusion/physical-expr/src/intervals/cp_solver.rs

@@ -24,9 +24,7 @@ use std::sync::Arc;
 use super::utils::{
     convert_duration_type_to_interval, convert_interval_type_to_duration, get_inverse_op,
 };
-use super::IntervalBound;
 use crate::expressions::Literal;
-use crate::intervals::interval_aritmetic::{apply_operator, Interval};
 use crate::sort_properties::SortProperties;
 use crate::utils::{build_dag, ExprTreeNode};
 use crate::{PhysicalExpr, PhysicalSortExpr};
@@ -167,21 +165,25 @@ impl Display for ExprIntervalGraphNode {
 impl ExprIntervalGraphNode {
     /// Constructs a new DAEG node with an [-∞, ∞] range.
     pub fn new_unbounded(expr: Arc<dyn PhysicalExpr>, dt: &DataType) -> Result<Self> {
-        Interval::make_unbounded(dt)
-            .map(|interval| ExprIntervalGraphNode { expr, interval })
+        Interval::make_unbounded(dt).map(|interval| ExprIntervalGraphNode {
+            expr,
+            interval,
+            is_strict_subset: false,
+            sort_properties: SortProperties::Unordered,
+        })
     }
 
     /// Constructs a new DAEG node with the given range.
     pub fn new_with_interval(
         expr: Arc<dyn PhysicalExpr>,
         interval: Interval,
-        sort_option: SortProperties,
+        sort_properties: SortProperties,
     ) -> Self {
         ExprIntervalGraphNode {
             expr,
             interval,
             is_strict_subset: false,
-            sort_properties: sort_option,
+            sort_properties,
         }
     }
 
@@ -197,8 +199,9 @@ impl ExprIntervalGraphNode {
         let expr = node.expression().clone();
         if let Some(literal) = expr.as_any().downcast_ref::<Literal>() {
             let value = literal.value();
-            Interval::try_new(value.clone(), value.clone())
-                .map(|interval| Self::new_with_interval(expr, interval))
+            Interval::try_new(value.clone(), value.clone()).map(|interval| {
+                Self::new_with_interval(expr, interval, SortProperties::Singleton)
+            })
         } else {
             expr.data_type(schema)
                 .and_then(|dt| Self::new_unbounded(expr, &dt))
@@ -659,7 +662,10 @@ impl ExprIntervalGraph {
                 .propagate_constraints(node_interval, &children_intervals)?;
             if let Some(propagated_intervals) = propagated_intervals {
                 for (child, interval) in children.into_iter().zip(propagated_intervals) {
-                    self.graph[child].interval = interval;
+                    let child_node = &mut self.graph[child];
+                    let inside = child_node.interval.is_superset(&interval, true)?;
+                    child_node.interval = interval;
+                    child_node.is_strict_subset = inside;
                 }
             } else {
                 // The constraint is infeasible, report:
@@ -669,25 +675,6 @@ impl ExprIntervalGraph {
         Ok(PropagationResult::Success)
     }
 
-    /// Updates intervals for all expressions in the DAEG by successive
-    /// bottom-up and top-down traversals.
-    pub fn update_ranges(
-        &mut self,
-        leaf_bounds: &mut [(usize, Interval)],
-    ) -> Result<PropagationResult> {
-        self.assign_intervals(leaf_bounds);
-        let bounds = self.evaluate_bounds()?;
-        if bounds == &Interval::CERTAINLY_FALSE {
-            Ok(PropagationResult::Infeasible)
-        } else if bounds == &Interval::UNCERTAIN {
-            let result = self.propagate_constraints();
-            self.update_intervals(leaf_bounds);
-            result
-        } else {
-            Ok(PropagationResult::CannotPropagate)
-        }
-    }
-
     /// Returns the deepest pruning expressions from the graph.
     ///
     /// This function traverses the graph in a depth-first search post-order manner. For each node, it checks its neighbors

datafusion/physical-plan/src/joins/partitioned_hash_join.rs

@@ -14,6 +14,11 @@ use crate::joins::sliding_window_join_utils::{
     partitioned_join_output_partitioning, EagerWindowJoinOperations, LazyJoinStream,
     LazyJoinStreamState, ProbeBuffer,
 };
+use crate::joins::stream_join_utils::{
+    get_filter_representation_of_join_side, prepare_sorted_exprs, EagerJoinStream,
+    EagerJoinStreamState, SortedFilterExpr, StreamJoinStateResult,
+};
+use crate::joins::symmetric_hash_join::StreamJoinMetrics;
 use crate::joins::utils::{
     apply_join_filter_to_indices, build_batch_from_indices, build_join_schema,
     calculate_join_output_ordering, check_join_is_valid, ColumnIndex, JoinFilter, JoinOn,
@@ -29,6 +34,7 @@ use arrow::compute::concat_batches;
 use arrow_array::builder::{UInt32Builder, UInt64Builder};
 use arrow_array::{ArrayRef, RecordBatch, UInt32Array, UInt64Array};
 use arrow_schema::{Field, Schema, SchemaRef};
+use datafusion_common::hash_utils::create_hashes;
 use datafusion_common::utils::{
     get_record_batch_at_indices, get_row_at_idx, linear_search,
 };
@@ -37,22 +43,17 @@ use datafusion_common::{
 };
 use datafusion_execution::memory_pool::MemoryConsumer;
 use datafusion_execution::TaskContext;
+use datafusion_expr::interval_arithmetic::Interval;
+use datafusion_physical_expr::equivalence::join_equivalence_properties;
 use datafusion_physical_expr::expressions::Column;
-use datafusion_physical_expr::intervals::{ExprIntervalGraph, Interval};
+use datafusion_physical_expr::intervals::cp_solver::ExprIntervalGraph;
 use datafusion_physical_expr::window::PartitionKey;
 use datafusion_physical_expr::{
     EquivalenceProperties, PhysicalExpr, PhysicalSortExpr, PhysicalSortRequirement,
 };
 
-use crate::joins::stream_join_utils::{
-    get_filter_representation_of_join_side, prepare_sorted_exprs, EagerJoinStream,
-    EagerJoinStreamState, SortedFilterExpr, StreamJoinStateResult,
-};
-use crate::joins::symmetric_hash_join::StreamJoinMetrics;
 use ahash::RandomState;
 use async_trait::async_trait;
-use datafusion_common::hash_utils::create_hashes;
-use datafusion_physical_expr::equivalence::join_equivalence_properties;
 use futures::Stream;
 use hashbrown::raw::RawTable;
 use parking_lot::Mutex;
@@ -604,9 +605,9 @@ impl BuildBuffer {
 
                                 // Get the lower or upper interval based on the sort direction:
                                 let target = if options.descending {
-                                    &interval.lower.value
+                                    interval.lower()
                                 } else {
-                                    &interval.upper.value
+                                    interval.upper()
                                 }
                                 .clone();
 

datafusion/physical-plan/src/joins/sliding_hash_join.rs

@@ -30,9 +30,14 @@ use crate::joins::{
         adjust_probe_side_indices_by_join_type,
         calculate_build_outer_indices_by_join_type,
         calculate_the_necessary_build_side_range_helper, joinable_probe_batch_helper,
-        EagerWindowJoinOperations, LazyJoinStream, LazyJoinStreamState, ProbeBuffer,
+        partitioned_join_output_partitioning, EagerWindowJoinOperations, LazyJoinStream,
+        LazyJoinStreamState, ProbeBuffer,
     },
-    symmetric_hash_join::OneSideHashJoiner,
+    stream_join_utils::{
+        combine_two_batches, prepare_sorted_exprs, record_visited_indices,
+        EagerJoinStream, EagerJoinStreamState, SortedFilterExpr, StreamJoinStateResult,
+    },
+    symmetric_hash_join::{OneSideHashJoiner, StreamJoinMetrics},
     utils::{
         build_batch_from_indices, build_join_schema, calculate_join_output_ordering,
         check_join_is_valid, swap_filter, swap_join_on, swap_join_type,
@@ -52,18 +57,13 @@ use arrow::record_batch::RecordBatch;
 use datafusion_common::{internal_err, plan_err, JoinSide, JoinType};
 use datafusion_execution::memory_pool::MemoryConsumer;
 use datafusion_execution::TaskContext;
-use datafusion_physical_expr::intervals::{ExprIntervalGraph, Interval};
+use datafusion_expr::interval_arithmetic::Interval;
+use datafusion_physical_expr::equivalence::join_equivalence_properties;
+use datafusion_physical_expr::intervals::cp_solver::ExprIntervalGraph;
 use datafusion_physical_expr::PhysicalSortRequirement;
 
-use crate::joins::sliding_window_join_utils::partitioned_join_output_partitioning;
-use crate::joins::stream_join_utils::{
-    combine_two_batches, prepare_sorted_exprs, record_visited_indices, EagerJoinStream,
-    EagerJoinStreamState, SortedFilterExpr, StreamJoinStateResult,
-};
-use crate::joins::symmetric_hash_join::StreamJoinMetrics;
 use ahash::RandomState;
 use async_trait::async_trait;
-use datafusion_physical_expr::equivalence::join_equivalence_properties;
 use futures::Stream;
 use parking_lot::Mutex;
 

datafusion/physical-plan/src/joins/sliding_nested_loop_join.rs

@@ -19,6 +19,12 @@ use crate::joins::sliding_window_join_utils::{
     partitioned_join_output_partitioning, EagerWindowJoinOperations, LazyJoinStream,
     LazyJoinStreamState, ProbeBuffer,
 };
+use crate::joins::stream_join_utils::{
+    calculate_side_prune_length_helper, combine_two_batches, prepare_sorted_exprs,
+    record_visited_indices, EagerJoinStream, EagerJoinStreamState, SortedFilterExpr,
+    StreamJoinStateResult,
+};
+use crate::joins::symmetric_hash_join::StreamJoinMetrics;
 use crate::joins::utils::{
     apply_join_filter_to_indices, build_batch_from_indices, build_join_schema,
     calculate_join_output_ordering, estimate_join_statistics, swap_filter,
@@ -36,23 +42,18 @@ use crate::{
 use arrow::array::{UInt32Array, UInt32Builder, UInt64Array, UInt64Builder};
 use arrow::datatypes::{Schema, SchemaRef};
 use arrow::record_batch::RecordBatch;
-use async_trait::async_trait;
 use datafusion_common::{internal_err, DataFusionError, JoinSide, Result, Statistics};
 use datafusion_execution::memory_pool::MemoryConsumer;
 use datafusion_execution::TaskContext;
+use datafusion_expr::interval_arithmetic::Interval;
 use datafusion_expr::JoinType;
-use datafusion_physical_expr::intervals::{ExprIntervalGraph, Interval};
+use datafusion_physical_expr::equivalence::join_equivalence_properties;
+use datafusion_physical_expr::intervals::cp_solver::ExprIntervalGraph;
 use datafusion_physical_expr::{
     EquivalenceProperties, PhysicalSortExpr, PhysicalSortRequirement,
 };
 
-use crate::joins::stream_join_utils::{
-    calculate_side_prune_length_helper, combine_two_batches, prepare_sorted_exprs,
-    record_visited_indices, EagerJoinStream, EagerJoinStreamState, SortedFilterExpr,
-    StreamJoinStateResult,
-};
-use crate::joins::symmetric_hash_join::StreamJoinMetrics;
-use datafusion_physical_expr::equivalence::join_equivalence_properties;
+use async_trait::async_trait;
 use futures::Stream;
 use hashbrown::HashSet;
 use parking_lot::Mutex;

datafusion/physical-plan/src/joins/sliding_window_join_utils.rs

@@ -1,40 +1,37 @@
 // Copyright (C) Synnada, Inc. - All Rights Reserved.
 // This file does not contain any Apache Software Foundation copyrighted code.
 
+use std::task::Poll;
+
 use crate::joins::{
     stream_join_utils::{
-        get_pruning_anti_indices, get_pruning_semi_indices, SortedFilterExpr,
-        StreamJoinStateResult,
+        calculate_side_prune_length_helper, get_build_side_pruned_exprs,
+        get_filter_representation_of_join_side,
+        get_filter_representation_schema_of_build_side, get_pruning_anti_indices,
+        get_pruning_semi_indices, SortedFilterExpr, StreamJoinStateResult,
     },
-    utils,
-    utils::{append_right_indices, get_anti_indices, get_semi_indices, JoinFilter},
+    symmetric_hash_join::StreamJoinMetrics,
+    utils::{self, append_right_indices, get_anti_indices, get_semi_indices, JoinFilter},
 };
-use std::task::Poll;
+use crate::{handle_async_state, handle_state};
 
+use arrow::compute::concat_batches;
 use arrow_array::{
     builder::{PrimitiveBuilder, UInt32Builder, UInt64Builder},
     types::{UInt32Type, UInt64Type},
     ArrowPrimitiveType, NativeAdapter, PrimitiveArray, RecordBatch, UInt32Array,
     UInt64Array,
 };
 use arrow_schema::SchemaRef;
-use async_trait::async_trait;
 use datafusion_common::{DataFusionError, JoinSide, JoinType, Result, ScalarValue};
 use datafusion_execution::SendableRecordBatchStream;
+use datafusion_expr::interval_arithmetic::Interval;
 use datafusion_physical_expr::expressions::{Column, PhysicalSortExpr};
-use datafusion_physical_expr::intervals::{ExprIntervalGraph, Interval, IntervalBound};
-use futures::ready;
-
-use crate::joins::stream_join_utils::{
-    calculate_side_prune_length_helper, get_build_side_pruned_exprs,
-    get_filter_representation_of_join_side,
-    get_filter_representation_schema_of_build_side,
-};
-use crate::joins::symmetric_hash_join::StreamJoinMetrics;
-use crate::{handle_async_state, handle_state};
-use arrow::compute::concat_batches;
+use datafusion_physical_expr::intervals::cp_solver::ExprIntervalGraph;
 use datafusion_physical_expr::Partitioning;
-use futures::{FutureExt, StreamExt};
+
+use async_trait::async_trait;
+use futures::{ready, FutureExt, StreamExt};
 use hashbrown::HashSet;
 use std::{mem, usize};
 
@@ -937,7 +934,7 @@ pub fn calculate_the_necessary_build_side_range_helper(
         .collect::<Vec<_>>();
 
     // Update the physical expression graph using the join filter intervals:
-    graph.update_ranges(&mut filter_intervals)?;
+    graph.update_ranges(&mut filter_intervals, Interval::CERTAINLY_TRUE)?;
 
     let intermediate_schema = get_filter_representation_schema_of_build_side(
         filter.schema(),
@@ -995,12 +992,12 @@ pub fn check_if_sliding_window_condition_is_met(
                 // Data is sorted in descending order, so check if latest value is less
                 // than the lower bound of the interval. If it is, we must have processed
                 // all rows that are needed from the build side for this window.
-                latest_value < interval.lower.value
+                &latest_value < interval.lower()
             } else {
                 // Data is sorted in ascending order, so check if latest value is greater
                 // than the upper bound of the interval. If it is, we must have processed
                 // all rows that are needed from the build side for this window.
-                latest_value > interval.upper.value
+                &latest_value > interval.upper()
             })
         })
         .collect::<Result<Vec<_>>>()?;
@@ -1301,18 +1298,13 @@ pub(crate) fn update_filter_expr_bounds(
         .intermediate_batch_filter_expr()
         .data_type(&build_inner_buffer.schema())?;
 
-    // Create a null scalar value with the obtained datatype:
-    let null_scalar = ScalarValue::try_from(build_order_datatype)?;
     // Create a null interval using the null scalar value:
-    let null_interval = Interval::new(
-        IntervalBound::new(null_scalar.clone(), true),
-        IntervalBound::new(null_scalar, true),
-    );
+    let unbounded_interval = Interval::make_unbounded(&build_order_datatype)?;
 
     build_sorted_filter_exprs
         .iter_mut()
         .for_each(|sorted_filter_expr| {
-            sorted_filter_expr.set_interval(null_interval.clone());
+            sorted_filter_expr.set_interval(unbounded_interval.clone());
         });
 
     let first_probe_intermediate_batch = get_filter_representation_of_join_side(
@@ -1349,15 +1341,9 @@ pub(crate) fn update_filter_expr_bounds(
             let right_value = ScalarValue::try_from_array(&last_array, 0)?;
             // Determine the interval bounds based on sort options:
             let interval = if sorted_filter_expr.order().descending {
-                Interval::new(
-                    IntervalBound::new(right_value, false),
-                    IntervalBound::new(left_value, false),
-                )
+                Interval::try_new(right_value, left_value)?
             } else {
-                Interval::new(
-                    IntervalBound::new(left_value, false),
-                    IntervalBound::new(right_value, false),
-                )
+                Interval::try_new(left_value, right_value)?
             };
             // Set the calculated interval for the sorted filter expression:
             sorted_filter_expr.set_interval(interval);
@@ -1368,6 +1354,7 @@ pub(crate) fn update_filter_expr_bounds(
 #[cfg(test)]
 mod tests {
     use crate::joins::sliding_window_join_utils::append_probe_indices_in_order;
+
     use arrow_array::{UInt32Array, UInt64Array};
 
     #[test]