Merge remote-tracking branch 'upstream/main' into varchar_default_ut8view

Author: zhuqi-lucas

Cargo.lock

@@ -168,7 +168,7 @@ rstest = "0.25.0"
 serde_json = "1"
 sqlparser = { version = "0.55.0", features = ["visitor"] }
 tempfile = "3"
-tokio = { version = "1.44", features = ["macros", "rt", "sync"] }
+tokio = { version = "1.45", features = ["macros", "rt", "sync"] }
 url = "2.5.4"
 
 [profile.release]

Cargo.toml

@@ -109,29 +109,32 @@ impl RunOpt {
         };
 
         let mut benchmark_run = BenchmarkRun::new();
-        for query_id in query_range {
-            benchmark_run.start_new_case(&format!("Query {query_id}"));
-            let query_run = self.benchmark_query(query_id).await?;
-            for iter in query_run {
-                benchmark_run.write_iter(iter.elapsed, iter.row_count);
-            }
-        }
-        benchmark_run.maybe_write_json(self.output_path.as_ref())?;
-        Ok(())
-    }
-
-    async fn benchmark_query(&self, query_id: usize) -> Result<Vec<QueryResult>> {
         let mut config = self
             .common
             .config()?
             .with_collect_statistics(!self.disable_statistics);
         config.options_mut().optimizer.prefer_hash_join = self.prefer_hash_join;
         let rt_builder = self.common.runtime_env_builder()?;
         let ctx = SessionContext::new_with_config_rt(config, rt_builder.build_arc()?);
-
         // register tables
         self.register_tables(&ctx).await?;
 
+        for query_id in query_range {
+            benchmark_run.start_new_case(&format!("Query {query_id}"));
+            let query_run = self.benchmark_query(query_id, &ctx).await?;
+            for iter in query_run {
+                benchmark_run.write_iter(iter.elapsed, iter.row_count);
+            }
+        }
+        benchmark_run.maybe_write_json(self.output_path.as_ref())?;
+        Ok(())
+    }
+
+    async fn benchmark_query(
+        &self,
+        query_id: usize,
+        ctx: &SessionContext,
+    ) -> Result<Vec<QueryResult>> {
         let mut millis = vec![];
         // run benchmark
         let mut query_results = vec![];
@@ -146,14 +149,14 @@ impl RunOpt {
             if query_id == 15 {
                 for (n, query) in sql.iter().enumerate() {
                     if n == 1 {
-                        result = self.execute_query(&ctx, query).await?;
+                        result = self.execute_query(ctx, query).await?;
                     } else {
-                        self.execute_query(&ctx, query).await?;
+                        self.execute_query(ctx, query).await?;
                     }
                 }
             } else {
                 for query in sql {
-                    result = self.execute_query(&ctx, query).await?;
+                    result = self.execute_query(ctx, query).await?;
                 }
             }
 

benchmarks/src/tpch/run.rs

@@ -43,4 +43,4 @@ log = { workspace = true }
 tokio = { workspace = true }
 
 [dev-dependencies]
-tokio = { version = "1.44", features = ["rt", "rt-multi-thread", "time"] }
+tokio = { version = "1.45", features = ["rt", "rt-multi-thread", "time"] }

datafusion/common-runtime/Cargo.toml

@@ -784,7 +784,7 @@ async fn explain_logical_plan_only() {
         vec!["logical_plan", "Projection: count(Int64(1)) AS count(*)\
         \n  Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]]\
         \n    SubqueryAlias: t\
-        \n      Projection: \
+        \n      Projection:\
         \n        Values: (Utf8(\"a\"), Int64(1), Int64(100)), (Utf8(\"a\"), Int64(2), Int64(150))"]];
     assert_eq!(expected, actual);
 }

datafusion/core/tests/sql/explain_analyze.rs

@@ -1815,12 +1815,12 @@ impl LogicalPlan {
                         Ok(())
                     }
                     LogicalPlan::Projection(Projection { ref expr, .. }) => {
-                        write!(f, "Projection: ")?;
+                        write!(f, "Projection:")?;
                         for (i, expr_item) in expr.iter().enumerate() {
                             if i > 0 {
-                                write!(f, ", ")?;
+                                write!(f, ",")?;
                             }
-                            write!(f, "{expr_item}")?;
+                            write!(f, " {expr_item}")?;
                         }
                         Ok(())
                     }

datafusion/expr/src/logical_plan/plan.rs

@@ -19,13 +19,16 @@
 
 use crate::utils::make_scalar_function;
 use arrow::array::{
-    Array, ArrayRef, Int64Array, LargeListArray, ListArray, OffsetSizeTrait, UInt64Array,
+    Array, ArrayRef, FixedSizeListArray, Int64Array, LargeListArray, ListArray,
+    OffsetSizeTrait, UInt64Array,
 };
 use arrow::datatypes::{
     DataType,
     DataType::{FixedSizeList, LargeList, List, UInt64},
 };
-use datafusion_common::cast::{as_generic_list_array, as_int64_array};
+use datafusion_common::cast::{
+    as_fixed_size_list_array, as_generic_list_array, as_int64_array,
+};
 use datafusion_common::{exec_err, internal_datafusion_err, plan_err, Result};
 use datafusion_expr::{
     ColumnarValue, Documentation, ScalarUDFImpl, Signature, Volatility,
@@ -119,6 +122,23 @@ impl ScalarUDFImpl for ArrayLength {
     }
 }
 
+macro_rules! array_length_impl {
+    ($array:expr, $dimension:expr) => {{
+        let array = $array;
+        let dimension = match $dimension {
+            Some(d) => as_int64_array(d)?.clone(),
+            None => Int64Array::from_value(1, array.len()),
+        };
+        let result = array
+            .iter()
+            .zip(dimension.iter())
+            .map(|(arr, dim)| compute_array_length(arr, dim))
+            .collect::<Result<UInt64Array>>()?;
+
+        Ok(Arc::new(result) as ArrayRef)
+    }};
+}
+
 /// Array_length SQL function
 pub fn array_length_inner(args: &[ArrayRef]) -> Result<ArrayRef> {
     if args.len() != 1 && args.len() != 2 {
@@ -128,26 +148,18 @@ pub fn array_length_inner(args: &[ArrayRef]) -> Result<ArrayRef> {
     match &args[0].data_type() {
         List(_) => general_array_length::<i32>(args),
         LargeList(_) => general_array_length::<i64>(args),
+        FixedSizeList(_, _) => fixed_size_array_length(args),
         array_type => exec_err!("array_length does not support type '{array_type:?}'"),
     }
 }
 
+fn fixed_size_array_length(array: &[ArrayRef]) -> Result<ArrayRef> {
+    array_length_impl!(as_fixed_size_list_array(&array[0])?, array.get(1))
+}
+
 /// Dispatch array length computation based on the offset type.
 fn general_array_length<O: OffsetSizeTrait>(array: &[ArrayRef]) -> Result<ArrayRef> {
-    let list_array = as_generic_list_array::<O>(&array[0])?;
-    let dimension = if array.len() == 2 {
-        as_int64_array(&array[1])?.clone()
-    } else {
-        Int64Array::from_value(1, list_array.len())
-    };
-
-    let result = list_array
-        .iter()
-        .zip(dimension.iter())
-        .map(|(arr, dim)| compute_array_length(arr, dim))
-        .collect::<Result<UInt64Array>>()?;
-
-    Ok(Arc::new(result) as ArrayRef)
+    array_length_impl!(as_generic_list_array::<O>(&array[0])?, array.get(1))
 }
 
 /// Returns the length of a concrete array dimension
@@ -185,6 +197,10 @@ fn compute_array_length(
                 value = downcast_arg!(value, LargeListArray).value(0);
                 current_dimension += 1;
             }
+            FixedSizeList(_, _) => {
+                value = downcast_arg!(value, FixedSizeListArray).value(0);
+                current_dimension += 1;
+            }
             _ => return Ok(None),
         }
     }

datafusion/functions-nested/src/length.rs

@@ -1124,7 +1124,7 @@ mod tests {
             plan,
             @r"
         Aggregate: groupBy=[[]], aggr=[[count(Int32(1))]]
-          Projection: 
+          Projection:
             Aggregate: groupBy=[[]], aggr=[[count(Int32(1))]]
               TableScan: ?table? projection=[]
         "

datafusion/optimizer/src/optimize_projections/mod.rs

@@ -250,7 +250,7 @@ fn between_date32_plus_interval() -> Result<()> {
     format!("{plan}"),
     @r#"
 Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]]
-  Projection: 
+  Projection:
     Filter: test.col_date32 >= Date32("1998-03-18") AND test.col_date32 <= Date32("1998-06-16")
       TableScan: test projection=[col_date32]
 "#
@@ -268,7 +268,7 @@ fn between_date64_plus_interval() -> Result<()> {
     format!("{plan}"),
     @r#"
         Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]]
-          Projection: 
+          Projection:
             Filter: test.col_date64 >= Date64("1998-03-18") AND test.col_date64 <= Date64("1998-06-16")
               TableScan: test projection=[col_date64]
         "#

datafusion/optimizer/tests/optimizer_integration.rs

@@ -1307,8 +1307,8 @@ fn lookup_join_hashmap(
     limit: usize,
     offset: JoinHashMapOffset,
 ) -> Result<(UInt64Array, UInt32Array, Option<JoinHashMapOffset>)> {
-    let (probe_indices, build_indices, next_offset) = build_hashmap
-        .get_matched_indices_with_limit_offset(hashes_buffer, None, limit, offset);
+    let (probe_indices, build_indices, next_offset) =
+        build_hashmap.get_matched_indices_with_limit_offset(hashes_buffer, limit, offset);
 
     let build_indices: UInt64Array = build_indices.into();
     let probe_indices: UInt32Array = probe_indices.into();
@@ -3333,7 +3333,7 @@ mod tests {
 
     #[test]
     fn join_with_hash_collision() -> Result<()> {
-        let mut hashmap_left = HashTable::with_capacity(2);
+        let mut hashmap_left = HashTable::with_capacity(4);
         let left = build_table_i32(
             ("a", &vec![10, 20]),
             ("x", &vec![100, 200]),
@@ -3348,9 +3348,15 @@ mod tests {
             hashes_buff,
         )?;
 
-        // Create hash collisions (same hashes)
+        // 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];