Merge remote-tracking branch 'apache/main' into askalt/improve-hash-join-exec-builder

Author: alamb

Cargo.lock

@@ -185,7 +185,7 @@ serde_json = "1"
 sha2 = "^0.10.9"
 sqlparser = { version = "0.61.0", default-features = false, features = ["std", "visitor"] }
 strum = "0.28.0"
-strum_macros = "0.27.2"
+strum_macros = "0.28.0"
 tempfile = "3"
 testcontainers-modules = { version = "0.15" }
 tokio = { version = "1.48", features = ["macros", "rt", "sync"] }

Cargo.toml

@@ -37,6 +37,16 @@ impl AliasGenerator {
         Self::default()
     }
 
+    /// Advance the counter to at least `min_id`, ensuring future aliases
+    /// won't collide with already-existing ones.
+    ///
+    /// For example, if the query already contains an alias `alias_42`, then calling
+    /// `update_min_id(42)` will ensure that future aliases generated by this
+    /// [`AliasGenerator`] will start from `alias_43`.
+    pub fn update_min_id(&self, min_id: usize) {
+        self.next_id.fetch_max(min_id + 1, Ordering::Relaxed);
+    }
+
     /// Return a unique alias with the provided prefix
     pub fn next(&self, prefix: &str) -> String {
         let id = self.next_id.fetch_add(1, Ordering::Relaxed);

datafusion/common/src/alias.rs

@@ -1265,7 +1265,7 @@ mod tests {
             col_stats.min_value,
             Precision::Inexact(ScalarValue::Int32(Some(-10)))
         );
-        assert!(matches!(col_stats.sum_value, Precision::Absent));
+        assert_eq!(col_stats.sum_value, Precision::Absent);
     }
 
     #[test]

datafusion/common/src/stats.rs

@@ -499,7 +499,7 @@ impl NativeType {
 
     #[inline]
     pub fn is_date(&self) -> bool {
-        matches!(self, NativeType::Date)
+        *self == NativeType::Date
     }
 
     #[inline]
@@ -524,7 +524,7 @@ impl NativeType {
 
     #[inline]
     pub fn is_null(&self) -> bool {
-        matches!(self, NativeType::Null)
+        *self == NativeType::Null
     }
 
     #[inline]

datafusion/common/src/types/native.rs

@@ -66,10 +66,9 @@ fn generate_spm_for_round_robin_tie_breaker(
         RecordBatch::try_from_iter(vec![("a", a), ("b", b), ("c", c)]).unwrap()
     };
 
-    let rbs = (0..batch_count).map(|_| rb.clone()).collect::<Vec<_>>();
-    let partitions = vec![rbs.clone(); partition_count];
-
     let schema = rb.schema();
+    let rbs = std::iter::repeat_n(rb, batch_count).collect::<Vec<_>>();
+    let partitions = vec![rbs.clone(); partition_count];
     let sort = [
         PhysicalSortExpr {
             expr: col("b", &schema).unwrap(),

datafusion/core/benches/spm.rs

@@ -1395,7 +1395,7 @@ impl DefaultPhysicalPlanner {
 
                 // TODO: Allow PWMJ to deal with residual equijoin conditions
                 let join: Arc<dyn ExecutionPlan> = if join_on.is_empty() {
-                    if join_filter.is_none() && matches!(join_type, JoinType::Inner) {
+                    if join_filter.is_none() && *join_type == JoinType::Inner {
                         // cross join if there is no join conditions and no join filter set
                         Arc::new(CrossJoinExec::new(physical_left, physical_right))
                     } else if num_range_filters == 1
@@ -1470,9 +1470,7 @@ impl DefaultPhysicalPlanner {
 
                         let left_side = side_of(lhs_logical)?;
                         let right_side = side_of(rhs_logical)?;
-                        if matches!(left_side, Side::Both)
-                            || matches!(right_side, Side::Both)
-                        {
+                        if left_side == Side::Both || right_side == Side::Both {
                             return Ok(Arc::new(NestedLoopJoinExec::try_new(
                                 physical_left,
                                 physical_right,
@@ -3553,12 +3551,12 @@ mod tests {
             assert!(
                 stringified_plans
                     .iter()
-                    .any(|p| matches!(p.plan_type, PlanType::FinalLogicalPlan))
+                    .any(|p| p.plan_type == PlanType::FinalLogicalPlan)
             );
             assert!(
                 stringified_plans
                     .iter()
-                    .any(|p| matches!(p.plan_type, PlanType::InitialPhysicalPlan))
+                    .any(|p| p.plan_type == PlanType::InitialPhysicalPlan)
             );
             assert!(
                 stringified_plans.iter().any(|p| matches!(
@@ -3569,7 +3567,7 @@ mod tests {
             assert!(
                 stringified_plans
                     .iter()
-                    .any(|p| matches!(p.plan_type, PlanType::FinalPhysicalPlan))
+                    .any(|p| p.plan_type == PlanType::FinalPhysicalPlan)
             );
         } else {
             panic!(

datafusion/core/src/physical_planner.rs

@@ -554,7 +554,7 @@ async fn verify_ordered_aggregate(frame: &DataFrame, expected_sort: bool) {
                         InputOrderMode::PartiallySorted(_) | InputOrderMode::Sorted
                     ));
                 } else {
-                    assert!(matches!(exec.input_order_mode(), InputOrderMode::Linear));
+                    assert_eq!(*exec.input_order_mode(), InputOrderMode::Linear);
                 }
             }
             Ok(TreeNodeRecursion::Continue)

datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs

@@ -278,9 +278,11 @@ async fn run_sort_test_with_limited_memory(
 
                     let string_item_size =
                         record_batch_memory_size / record_batch_size as usize;
-                    let string_array = Arc::new(StringArray::from_iter_values(
-                        (0..record_batch_size).map(|_| "a".repeat(string_item_size)),
-                    ));
+                    let string_array =
+                        Arc::new(StringArray::from_iter_values(std::iter::repeat_n(
+                            "a".repeat(string_item_size),
+                            record_batch_size as usize,
+                        )));
 
                     RecordBatch::try_new(
                         Arc::clone(&schema),
@@ -536,9 +538,11 @@ async fn run_test_aggregate_with_high_cardinality(
 
                     let string_item_size =
                         record_batch_memory_size / record_batch_size as usize;
-                    let string_array = Arc::new(StringArray::from_iter_values(
-                        (0..record_batch_size).map(|_| "a".repeat(string_item_size)),
-                    ));
+                    let string_array =
+                        Arc::new(StringArray::from_iter_values(std::iter::repeat_n(
+                            "a".repeat(string_item_size),
+                            record_batch_size as usize,
+                        )));
 
                     RecordBatch::try_new(
                         Arc::clone(&schema),

datafusion/core/tests/fuzz_cases/spilling_fuzz_in_memory_constrained_env.rs

@@ -589,7 +589,7 @@ async fn run_window_test(
     orderby_columns: Vec<&str>,
     search_mode: InputOrderMode,
 ) -> Result<()> {
-    let is_linear = !matches!(search_mode, Sorted);
+    let is_linear = search_mode != Sorted;
     let mut rng = StdRng::seed_from_u64(random_seed);
     let schema = input1[0].schema();
     let session_config = SessionConfig::new().with_batch_size(50);