feat: add Spark-compatible arrays_zip function

[CuteChuanChuan]

Which issue does this PR close?

Closes #20888.

Rationale for this change

• Spark's arrays_zip returns a list of structs whose field names follow the original SQL arguments — column refs use the column name, aliases use the alias, anything else falls back to 0-based ordinals ("0", "1", ...).
• DataFusion's native arrays_zip uses 1-based ordinals. To support Spark compatibility without altering DataFusion's native semantics, this PR adds a SparkArraysZip wrapper plus an analyzer-time rewrite in the datafusion-spark crate.

What changes are included in this PR?

1. SparkArraysZip — a wrapper that delegates to the native ArraysZip and renames the inner List<Struct<..>> fields. Both the planning-time DataType and the execution-time ArrayRef are renamed so the declared schema matches the produced data.

2. SparkArraysZipRewrite — an analyzer-time FunctionRewrite that captures each argument's natural name from the original Expr tree and embeds it as a marked List<Utf8> literal first argument. Running before optimizer passes that could rename arg_fields is what keeps the names stable through planning.

Are these changes tested?

Yes:

• SLT (datafusion/sqllogictest/test_files/spark/array/arrays_zip.slt)
• Unit tests in arrays_zip_rewrite.rs
• Unit tests in arrays_zip.rs

Are there any user-facing changes?

Yes — in a session built with SessionStateBuilderSpark::with_spark_features, arrays_zip now produces struct field names following Spark semantics (column name / alias / 0-based ordinal) instead of DataFusion's 1-based ordinals. DataFusion's native arrays_zip is unaffected and still reachable under its list_zip alias.

[CuteChuanChuan]

Working on renaming part.

[CuteChuanChuan]

Hi @comphead ,
could you PTAL when you have chances? Thanks! 🙏

[comphead]

thats a nice test, just checked Spark does the same

scala> spark.sql("select arrays_zip(a, array(3, 4, 5)) from (select array(1, 2, 3) a, array(3, 4, 5))").printSchema
root
 |-- arrays_zip(a, array(3, 4, 5)): array (nullable = false)
 |    |-- element: struct (containsNull = false)
 |    |    |-- a: integer (nullable = true)
 |    |    |-- 1: integer (nullable = true)

[comphead]

this is actually not how Spark works, checking 4.1.1

scala> spark.sql("select arrays_zip(a, a, a) from (select array(1, 2, 3) a, array(3, 4, 5))").printSchema
warning: 1 deprecation (since 2.13.3); for details, enable `:setting -deprecation` or `:replay -deprecation`
root
 |-- arrays_zip(a, a, a): array (nullable = false)
 |    |-- element: struct (containsNull = false)
 |    |    |-- a: integer (nullable = true)
 |    |    |-- a: integer (nullable = true)
 |    |    |-- a: integer (nullable = true)

It reminds me a spark issue, because this construction returns data, but you cannot really access it by the field name. We prob can leave it as is for now

[CuteChuanChuan]

@comphead ,
Thanks for pointing this out. I've added a note in the SLT next to that test clarifying the divergence from Spark.

[comphead]

@CuteChuanChuan speaking to downstream projects that uses array_zip directly from datafusion-spark, how naming would work in this case?

Can we try to reuse function_nested::arrayzip as much as possible and pass the column names in there?

[CuteChuanChuan]

@CuteChuanChuan speaking to downstream projects that uses array_zip directly from datafusion-spark, how naming would work in this case?

Can we try to reuse function_nested::arrayzip as much as possible and pass the column names in there?

No problem. I will give it a try. Thanks you for pointing this out. 🙏

[CuteChuanChuan]

Hi @comphead ,
could you PTAL when you have a chance?

Following up on the downstream-naming question: I added an ArraysZip::with_field_names constructor to datafusion-functions-nested so the Spark wrapper injects the resolved field names instead of rebuilding the List to rename it. This lets the spark crate keep only the Spark-specific logic (name policy + the analyzer rewrite) and reuse the native implementation for everything else — the zip itself, and applying the names.

For downstream projects using arrays_zip directly: the native functions-nested function defaults to ordinal field names and now also accepts custom names via with_field_names; the Spark-faithful naming (column names 0-based ordinals) lives in the spark crate and relies on the registered rewrite rule.

[comphead]

Thanks @CuteChuanChuan I would expect something like below. Just extend existing arrays_zip_inner to accept fieldnames and in datafusion-spark, create another constructor that accepts field names.

Spark caller will create a function with necessary fields.

It can be constructor or some .with_field_names to avoid recreate function

 flowchart TB
     subgraph spark["datafusion-spark"]
         SparkUDF["SparkArraysZip (UDF)<br/>function/array/arrays_zip.rs"]
         SparkNew["new()<br/>names from arg_fields"]
         SparkCtor["with_field_names(Vec&lt;String&gt;)<br/>caller-supplied names"]
         SparkMod["function/array/mod.rs<br/>+ pub mod arrays_zip<br/>+ make_udf_function!<br/>+ export_functions!<br/>+ functions() entry"]

         SparkUDF --> SparkNew
         SparkUDF --> SparkCtor
         SparkMod -.registers.-> SparkUDF
     end

     subgraph nested["datafusion-functions-nested"]
         ArraysZipUDF["ArraysZip (UDF)<br/>names: '1','2','3'..."]
         Wrapper["arrays_zip_inner(args)<br/>(private wrapper)<br/>generates default names"]
         Pub["pub fn arrays_zip_inner_with_names<br/>(args, field_names)"]
         Perfect["try_perfect_list_zip<br/>(args, field_names)"]

         ArraysZipUDF --> Wrapper
         Wrapper --> Pub
         Pub --> Perfect
     end

     SparkNew -->|resolve names<br/>then call| Pub
     SparkCtor -->|resolve names<br/>then call| Pub

Loading

Name resolution inside SparkArraysZip

flowchart LR
    Start["SparkArraysZip.field_names"]
    Some["Some(names)"]
    None["None"]
    Validate["validate length<br/>matches args"]
    FromArgs["arg_fields[i].name()<br/>for each i"]
    Use["use names"]

    Start --> Some
    Start --> None
    Some --> Validate --> Use
    None -->|Spark default:<br/>column/alias names| FromArgs --> Use

Loading

[CuteChuanChuan]

Thanks @CuteChuanChuan I would expect something like below. Just extend existing arrays_zip_inner to accept fieldnames and in datafusion-spark, create another constructor that accepts field names.

Got it. Thanks for providing these details.❤️ I will give it a try.

[CuteChuanChuan]

Thanks @comphead for the detailed and well-guided review — I've reworked to match your diagram. Two deviations below.

Matches the diagram

functions-nested: pub arrays_zip_inner_with_names(args, field_names) + pub arrays_zip_return_type(arg_types, field_names), with a private arrays_zip_inner(args) wrapper supplying the default 1-based ordinals (native UDF unchanged).

spark: SparkArraysZip holds the names — with_field_names(Vec) for callers, new() for none — and resolve_names() passes them into both native fns (single naming source, no post-hoc rename). A programmatic caller uses with_field_names directly and never hits the rewrite.

Deviation 1 — None → 0-based ordinals, not arg_fields names

In the previous design, new() derives from arg_fields[i].name(); I tried that and it panics, because those names aren't stable across optimizer passes (a literal arg gets renamed to lit):

SELECT arrays_zip(a, a, a) FROM (VALUES ([1,2],[1,2],[1,2])) AS t(a, b, c);
-- optimize_projections: schema mismatch
-- declared: List<Struct<"0","1","2">>
-- recomputed: List<Struct<"a","a","a">>

So None produces positional ordinals (depend only on arg count). Spark-faithful names for SQL come from the rewrite below instead.

Deviation 2 — the SQL rewrite is now an AnalyzerRule, not a FunctionRewrite

ApplyFunctionRewrites rewrites expressions but never recomputes the plan node's schema, so the cached schema goes stale when a rewrite changes the output type (as pinning names does) — the same panic as above. An AnalyzerRule can recompute_schema() after pinning. This fixes every SLT case including a,a,a.

Duplicate names are kept as-is per Spark 4.1.1 check (arrays_zip(a,a) → struct<a,a>); I dropped the earlier de-dup. IMO mimicking Spark here brings no surprise to users.

PTAL — happy to adjust the feeback.