Merge branch 'main' into alamb/enable_pushdown

Author: Dandandan

.github/workflows/audit.yml

@@ -42,7 +42,7 @@ jobs:
     steps:
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
       - name: Install cargo-audit
-        uses: taiki-e/install-action@710817a1645ef40daad5bcde7431ceccf6cc3528  # v2.67.13
+        uses: taiki-e/install-action@650c5ca14212efbbf3e580844b04bdccf68dac31  # v2.67.18
         with:
           tool: cargo-audit
       - name: Run audit check

.github/workflows/rust.yml

@@ -45,10 +45,11 @@ jobs:
   # Check crate compiles and base cargo check passes
   linux-build-lib:
     name: linux build test
-    runs-on: ubuntu-latest
+    runs-on: ${{ github.repository_owner == 'apache' && format('runs-on={0},family=m7a,cpu=16,image=ubuntu24-full-x64,extras=s3-cache,disk=large,tag=datafusion', github.run_id) || 'ubuntu-latest' }}
     container:
       image: amd64/rust
     steps:
+      - uses: runs-on/action@cd2b598b0515d39d78c38a02d529db87d2196d1e  # v2.0.3
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
       - name: Setup Rust toolchain
         uses: ./.github/actions/setup-builder
@@ -266,12 +267,13 @@ jobs:
   linux-test:
     name: cargo test (amd64)
     needs: linux-build-lib
-    runs-on: ubuntu-latest
+    runs-on: ${{ github.repository_owner == 'apache' && format('runs-on={0},family=m7a,cpu=16,image=ubuntu24-full-x64,extras=s3-cache,disk=large,tag=datafusion', github.run_id) || 'ubuntu-latest' }}
     container:
       image: amd64/rust
       volumes:
         - /usr/local:/host/usr/local
     steps:
+      - uses: runs-on/action@cd2b598b0515d39d78c38a02d529db87d2196d1e  # v2.0.3
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
         with:
           submodules: true
@@ -421,7 +423,7 @@ jobs:
           sudo apt-get update -qq
           sudo apt-get install -y -qq clang
       - name: Setup wasm-pack
-        uses: taiki-e/install-action@710817a1645ef40daad5bcde7431ceccf6cc3528  # v2.67.13
+        uses: taiki-e/install-action@650c5ca14212efbbf3e580844b04bdccf68dac31  # v2.67.18
         with:
           tool: wasm-pack
       - name: Run tests with headless mode
@@ -733,7 +735,7 @@ jobs:
 
       - name: Set up Node.js (required for prettier)
         # doc_prettier_check.sh uses npx to run prettier for Markdown formatting
-        uses: actions/setup-node@v4
+        uses: actions/setup-node@v6
         with:
           node-version: '18'
 
@@ -756,7 +758,7 @@ jobs:
       - name: Setup Rust toolchain
         uses: ./.github/actions/setup-builder
       - name: Install cargo-msrv
-        uses: taiki-e/install-action@710817a1645ef40daad5bcde7431ceccf6cc3528  # v2.67.13
+        uses: taiki-e/install-action@650c5ca14212efbbf3e580844b04bdccf68dac31  # v2.67.18
         with:
           tool: cargo-msrv
 

Cargo.lock

@@ -152,14 +152,14 @@ datafusion-substrait = { path = "datafusion/substrait", version = "52.1.0" }
 
 doc-comment = "0.3"
 env_logger = "0.11"
-flate2 = "1.1.8"
+flate2 = "1.1.9"
 futures = "0.3"
 glob = "0.3.0"
 half = { version = "2.7.0", default-features = false }
 hashbrown = { version = "0.16.1" }
 hex = { version = "0.4.3" }
 indexmap = "2.13.0"
-insta = { version = "1.46.1", features = ["glob", "filters"] }
+insta = { version = "1.46.3", features = ["glob", "filters"] }
 itertools = "0.14"
 liblzma = { version = "0.4.4", features = ["static"] }
 log = "^0.4"

Cargo.toml

@@ -40,7 +40,7 @@ mimalloc_extended = ["libmimalloc-sys/extended"]
 
 [dependencies]
 arrow = { workspace = true }
-clap = { version = "4.5.53", features = ["derive"] }
+clap = { version = "4.5.56", features = ["derive"] }
 datafusion = { workspace = true, default-features = true }
 datafusion-common = { workspace = true, default-features = true }
 env_logger = { workspace = true }

benchmarks/Cargo.toml

@@ -40,7 +40,7 @@ async-trait = { workspace = true }
 aws-config = "1.8.12"
 aws-credential-types = "1.2.7"
 chrono = { workspace = true }
-clap = { version = "4.5.53", features = ["cargo", "derive"] }
+clap = { version = "4.5.56", features = ["cargo", "derive"] }
 datafusion = { workspace = true, features = [
     "avro",
     "compression",

datafusion-cli/Cargo.toml

@@ -25,8 +25,9 @@ use arrow::array::{
     BinaryViewArray, Decimal32Array, Decimal64Array, DurationMicrosecondArray,
     DurationMillisecondArray, DurationNanosecondArray, DurationSecondArray, Float16Array,
     Int8Array, Int16Array, LargeBinaryArray, LargeListViewArray, LargeStringArray,
-    ListViewArray, StringViewArray, UInt16Array,
+    ListViewArray, RunArray, StringViewArray, UInt16Array,
 };
+use arrow::datatypes::RunEndIndexType;
 use arrow::{
     array::{
         Array, BinaryArray, BooleanArray, Date32Array, Date64Array, Decimal128Array,
@@ -334,3 +335,8 @@ pub fn as_list_view_array(array: &dyn Array) -> Result<&ListViewArray> {
 pub fn as_large_list_view_array(array: &dyn Array) -> Result<&LargeListViewArray> {
     Ok(downcast_value!(array, LargeListViewArray))
 }
+
+// Downcast Array to RunArray
+pub fn as_run_array<T: RunEndIndexType>(array: &dyn Array) -> Result<&RunArray<T>> {
+    Ok(downcast_value!(array, RunArray, T))
+}

datafusion/common/src/cast.rs

@@ -31,7 +31,7 @@ use std::{collections::HashSet, sync::Arc};
 ///
 /// ## Field Matching Strategy
 /// - **By Name**: Source struct fields are matched to target fields by name (case-sensitive)
-/// - **By Position**: When there is no name overlap and the field counts match, fields are cast by index
+/// - **No Positional Mapping**: Structs with no overlapping field names are rejected
 /// - **Type Adaptation**: When a matching field is found, it is recursively cast to the target field's type
 /// - **Missing Fields**: Target fields not present in the source are filled with null values
 /// - **Extra Fields**: Source fields not present in the target are ignored
@@ -67,24 +67,16 @@ fn cast_struct_column(
     if let Some(source_struct) = source_col.as_any().downcast_ref::<StructArray>() {
         let source_fields = source_struct.fields();
         validate_struct_compatibility(source_fields, target_fields)?;
-        let has_overlap = has_one_of_more_common_fields(source_fields, target_fields);
-
         let mut fields: Vec<Arc<Field>> = Vec::with_capacity(target_fields.len());
         let mut arrays: Vec<ArrayRef> = Vec::with_capacity(target_fields.len());
         let num_rows = source_col.len();
 
-        // Iterate target fields and pick source child either by name (when fields overlap)
-        // or by position (when there is no name overlap).
-        for (index, target_child_field) in target_fields.iter().enumerate() {
+        // Iterate target fields and pick source child by name when present.
+        for target_child_field in target_fields.iter() {
             fields.push(Arc::clone(target_child_field));
 
-            // Determine the source child column: by name when overlapping names exist,
-            // otherwise by position.
-            let source_child_opt: Option<&ArrayRef> = if has_overlap {
-                source_struct.column_by_name(target_child_field.name())
-            } else {
-                Some(source_struct.column(index))
-            };
+            let source_child_opt =
+                source_struct.column_by_name(target_child_field.name());
 
             match source_child_opt {
                 Some(source_child_col) => {
@@ -230,20 +222,11 @@ pub fn validate_struct_compatibility(
 ) -> Result<()> {
     let has_overlap = has_one_of_more_common_fields(source_fields, target_fields);
     if !has_overlap {
-        if source_fields.len() != target_fields.len() {
-            return _plan_err!(
-                "Cannot cast struct with {} fields to {} fields without name overlap; positional mapping is ambiguous",
-                source_fields.len(),
-                target_fields.len()
-            );
-        }
-
-        for (source_field, target_field) in source_fields.iter().zip(target_fields.iter())
-        {
-            validate_field_compatibility(source_field, target_field)?;
-        }
-
-        return Ok(());
+        return _plan_err!(
+            "Cannot cast struct with {} fields to {} fields because there is no field name overlap",
+            source_fields.len(),
+            target_fields.len()
+        );
     }
 
     // Check compatibility for each target field
@@ -323,7 +306,11 @@ fn validate_field_compatibility(
     Ok(())
 }
 
-fn has_one_of_more_common_fields(
+/// Check if two field lists have at least one common field by name.
+///
+/// This is useful for validating struct compatibility when casting between structs,
+/// ensuring that source and target fields have overlapping names.
+pub fn has_one_of_more_common_fields(
     source_fields: &[FieldRef],
     target_fields: &[FieldRef],
 ) -> bool {
@@ -546,7 +533,7 @@ mod tests {
     }
 
     #[test]
-    fn test_validate_struct_compatibility_positional_no_overlap_mismatch_len() {
+    fn test_validate_struct_compatibility_no_overlap_mismatch_len() {
         let source_fields = vec![
             arc_field("left", DataType::Int32),
             arc_field("right", DataType::Int32),
@@ -556,7 +543,7 @@ mod tests {
         let result = validate_struct_compatibility(&source_fields, &target_fields);
         assert!(result.is_err());
         let error_msg = result.unwrap_err().to_string();
-        assert!(error_msg.contains("positional mapping is ambiguous"));
+        assert_contains!(error_msg, "no field name overlap");
     }
 
     #[test]
@@ -665,21 +652,21 @@ mod tests {
     }
 
     #[test]
-    fn test_validate_struct_compatibility_positional_with_type_mismatch() {
-        // Source struct: {left: Struct} - nested struct
-        let source_fields =
-            vec![arc_struct_field("left", vec![field("x", DataType::Int32)])];
+    fn test_validate_struct_compatibility_no_overlap_equal_len() {
+        let source_fields = vec![
+            arc_field("left", DataType::Int32),
+            arc_field("right", DataType::Utf8),
+        ];
 
-        // Target struct: {alpha: Int32} (no name overlap, incompatible type at position 0)
-        let target_fields = vec![arc_field("alpha", DataType::Int32)];
+        let target_fields = vec![
+            arc_field("alpha", DataType::Int32),
+            arc_field("beta", DataType::Utf8),
+        ];
 
         let result = validate_struct_compatibility(&source_fields, &target_fields);
         assert!(result.is_err());
         let error_msg = result.unwrap_err().to_string();
-        assert_contains!(
-            error_msg,
-            "Cannot cast struct field 'alpha' from type Struct(\"x\": Int32) to type Int32"
-        );
+        assert_contains!(error_msg, "no field name overlap");
     }
 
     #[test]
@@ -948,7 +935,7 @@ mod tests {
     }
 
     #[test]
-    fn test_cast_struct_positional_when_no_overlap() {
+    fn test_cast_struct_no_overlap_rejected() {
         let first = Arc::new(Int32Array::from(vec![Some(10), Some(20)])) as ArrayRef;
         let second =
             Arc::new(StringArray::from(vec![Some("alpha"), Some("beta")])) as ArrayRef;
@@ -964,17 +951,10 @@ mod tests {
             vec![field("a", DataType::Int64), field("b", DataType::Utf8)],
         );
 
-        let result =
-            cast_column(&source_col, &target_field, &DEFAULT_CAST_OPTIONS).unwrap();
-        let struct_array = result.as_any().downcast_ref::<StructArray>().unwrap();
-
-        let a_col = get_column_as!(&struct_array, "a", Int64Array);
-        assert_eq!(a_col.value(0), 10);
-        assert_eq!(a_col.value(1), 20);
-
-        let b_col = get_column_as!(&struct_array, "b", StringArray);
-        assert_eq!(b_col.value(0), "alpha");
-        assert_eq!(b_col.value(1), "beta");
+        let result = cast_column(&source_col, &target_field, &DEFAULT_CAST_OPTIONS);
+        assert!(result.is_err());
+        let error_msg = result.unwrap_err().to_string();
+        assert_contains!(error_msg, "no field name overlap");
     }
 
     #[test]