fix: preserve duplicate named_struct fields in codegen dispatch

[RRXXZZYY]

Which issue does this PR close?

Closes #5586.

Rationale for this change

Spark permits duplicate field names in named_struct, and generated code writes those fields positionally. Arrow Java's default struct-vector factory indexes children by name, so duplicate children can collapse or reuse the wrong concrete vector type. The same behavior affects Arrow IPC and C Data stream boundaries. Arrow's struct writer also lower-cases writer-cache keys, which can drop case-distinct children such as a and A.

What changes are included in this PR?

• Keep supported CreateNamedStruct expressions on the native path; unsupported children remain subject to the existing Spark conversion boundary.
• Allocate affected struct subtrees with private, unique runtime child names while preserving Spark-visible field metadata.
• Use duplicate-safe allocation at shared IPC, C Data import, broadcast coalescing, in-memory cache, and C Data stream export boundaries.
• Materialize dispatcher outputs positionally so duplicate and case-distinct direct children remain separate.
• Publish the original struct field only after positional child initialization is complete.
• Add focused SQL, native, and cached-filter regressions for duplicate-name structs and restore dictionary-enabled/disabled coverage.
• Update the expression-support documentation to describe the native path.

How are these changes tested?

Validation on the isolated NAS Linux builder (Spark 4.1 / Scala 2.13.17 / JDK 17):

• Spark/JVM test-compile for spark -am: passed. Spotless and Scalastyle were skipped for this compile-only check.
• CometExpressionSuite native duplicate CreateNamedStruct regression: 1/1 passed with the existing native Comet library loaded and Scala-UDF codegen dispatch disabled.
• CometInMemoryCacheSuite duplicate-name cached native filter regression: 1/1 passed with the existing native Comet library loaded and Scala-UDF codegen dispatch disabled.
• git diff --check: passed.

A fresh Rust native rebuild was not available locally because dependency retrieval failed with an external TLS error; the focused JVM tests reused the unchanged native library from the isolated builder. The repository's prior CI run for the preceding commit had expression-matrix failures; a new CI run is expected for this follow-up commit. Full cross-platform and clean-checkout coverage remain unverified here. This PR makes no performance claim.

AI-assisted development disclosure: I used AI tooling to assist investigation and drafting. I reproduced the relevant failures, reviewed and refined the implementation, inspected the final diff and ownership paths, and ran the checks listed above.

[andygrove]

Thanks for taking this on. The diagnosis is right and the two unit tests are aimed at the actual mechanism, so this is close.

I checked the positional assumptions end to end and they hold: ArrayImporter.doImport walks getChildrenFromFields() positionally, CometBatchKernelCodegenOutput emits getChildByOrdinal($fi), CometScalaUDFCodegen.specFor pairs getField.getChildren.get(fi) with getChildByOrdinal(fi), and StructFieldSpec.name is only cache-key identity so duplicate names never become Java identifiers. The ordinal rename also cannot collide with a user field name, because when a struct has duplicates every child gets renamed.

The thing I would most like to see before merge is the import path staying on its old behavior when there are no duplicate names. importVector runs for every column of every batch coming back from native, and right now this rebuilds the Field tree and rewraps every complex vector regardless of whether any duplicates exist. Details inline.

One more thing worth noting: CI has not run on this branch yet. Given the local validation was Spark 4.0 only, it would be good to confirm 3.4 and 3.5 are green before merging.

[RRXXZZYY]

Thanks for the thorough review. I addressed the feedback in 65581cc2 and rebased the branch onto current main (2949fd0d).

• Ordinary imports now retain Arrow's original field.createVector(allocator) path. fieldForAllocation returns the original Field by identity when unchanged, and the import factory is hoisted per NativeUtil instance. A regression checks the existing $data$ list-child behavior.
• The duplicate-safe wrappers are now limited to the affected complex subtree. RenamedStructVector uses Arrow 18.3's Field constructor; the separate construction flag is gone, with the original exported metadata exposed only after the direct children exist.
• The internal fallback reason and doc-facing dispatcher explanation are separate, and struct is documented as Hybrid.
• Duplicate-name integration coverage moved to create_named_struct.sql and now includes struct(a, a), nested array/map/struct values, three duplicates, all-null rows, and construction after a supported primitive-key shuffle boundary.
• NativeUtilSuite now uses scoped resources and explicitly handles the Arrow array/schema handoff on failure paths.

Fresh validation on this head:

• full NativeUtilSuite on Spark 4.0: 11/11 passed;
• focused SQL-file regression on Spark 4.0, 3.5, and 3.4: passed on all three profiles;
• focused duplicate-struct codegen test on the default Spark 4.1 profile: passed;
• Scalastyle: 147 files, 0 errors/warnings;
• Spotless on all four changed Scala files and git diff --check: passed.

The PR description now records the exact checks and remaining local-validation boundaries.

[sunchao]

Summary

The positional allocation and ordinary-import fast path address the duplicate-name case while keeping the usual import path unchanged. I found one additional P2 case in the shared codegen output allocator, attached inline.

Validation

Reviewed 65581cc2 against 2949fd0d by source, including the Arrow constructor and writer contracts. I did not execute a reproduction or rerun the author-reported tests. The four current-head workflows show action_required, so they do not establish passing CI.

[sunchao]

Re-reviewed c458c31b8462fdb80a43cdff24573454ab662a2d. The previously reported import-boundary issues are addressed. No new actionable P1/P2 findings.

This was a source review. I did not run tests. Current CI still requires contributor workflow approval, so there are no passing CI results to report.

[sunchao]

Re-reviewed 91bba712752eb0191efab376805a22a52a1ed785 after the base sync. No new actionable P1/P2 findings.

This was a source review. I did not run tests. Current workflows still require contributor approval, and there are no current-head or merge check results.

[sunchao]

One new P2 in the C Data stream increment, attached inline. A focused Arrow 58.4.0 probe confirmed value duplication for colliding field names. The full Spark query was source-traced, not executed. Current CI still requires workflow approval.

[sunchao]

[P2] Keep stream runtime names disjoint from logical field names

Could this preserve ordinal mapping when a user names both children __comet_runtime_field_0? This root advertises __comet_runtime_field_0 and __comet_runtime_field_1. ScanStream::build_record_batch then casts back to the declared struct type, but Arrow 58.4.0 matches both target names to source child 0 and returns that child's values twice. In the existing broadcast-join regression, replacing both 'x' labels with '__comet_runtime_field_0' has this path from (1, 10) to (1, 1). A focused probe using the pinned Arrow cast reproduced the value and null duplication. I did not execute the Spark query. Please make exported runtime names disjoint from every original child name, or restore the logical schema strictly by ordinal, and cover this case in the broadcast regression.

[RRXXZZYY]

Fixed in ba8ca2c. Generated runtime names now skip every original sibling name before allocation, and the broadcast regression uses duplicate __comet_runtime_field_0 fields. The focused Spark 4.1 test passed 1/1 with the native library loaded.

[sunchao]

The runtime-name collision is addressed in ba8ca2cc. No remaining actionable P1/P2 findings.

Checked the source increment and 32 focused Arrow 58.4.0 cast controls, with the prior-head corruption reproduced as a negative control. I did not run the Scala allocator or Spark query. CI still requires workflow approval.

[sunchao]

Re-reviewed e1ecb7fb54dec61fc1cd59068d9c0437e7a8e872. The only change since approved ba8ca2cc is line wrapping in runtime field allocation. No new actionable P1/P2 findings.

This was a source review. No tests were rerun. CI still requires workflow approval.

[andygrove]

I checked this branch out and ran it rather than reading it, since the earlier rounds here were all source reviews. Build is clean on Spark 4.1 / JDK 17, and everything the PR adds passes locally: NativeUtilSuite, UtilsSuite, CometCodegenSuite and CometJsonJvmSuite at 109 of 109, the create_named_struct fixture, and the new broadcast join test. I also swept the paths this rewires for everyone, since StreamReader and the per-batch import factory are not duplicate-name-specific: CometShuffleSuite, DisableAQECometShuffleSuite, CometNativeShuffleSuite and CometInMemoryCacheSuite at 160 of 160, then CometExecSuite, CometJoinSuite and CometExpressionSuite at 316 of 316.

Then I wrote about fifty probes of my own looking for a divergence from Spark and did not find one. Every safe primitive type in a duplicate-name struct with dictionary encoding on and off at ten thousand rows, three levels of nesting, duplicate structs as map keys and as map values, array and map children inside a duplicate struct, case-distinct names alone and mixed with an exact duplicate in the same struct, a user field literally named __comet_runtime_field_0 and also $data$, entries, key and value, all-null rows, zero rows, empty field names, a struct on either side of spark.sql.codegen.maxFields, then explode, inline, to_json, from_json, union, group-by-struct, window, sort-merge join on both sides, broadcast join with AQE on, and a df.cache() round trip. All of them matched. The positional allocation holds up.

I also confirmed the piece I most expected to be wrong. ArrowReader.root, loader and initialized are all private in Arrow 18.3, so the shadow state in the two new readers is forced rather than stylistic, and every base method that reads that private state really is overridden in both classes. Nothing silently sees null or false.

The main thing I would like resolved

It is reachability, not correctness. CometShuffleExchangeExec.columnarShuffleFailureReasons still carries this guard, which is not in the diff:

case StructType(fields) =>
  fields.nonEmpty && fields.forall(f => supportedSerializableDataType(f.dataType)) &&
  // Java Arrow stream reader cannot work on duplicate field name
  fields.map(f => f.name).distinct.length == fields.length

That comment describes exactly the limitation this PR removes, and StreamReader (the reader it refers to, via ArrowReaderIterator and Utils.decodeBatches) is the columnar-shuffle read path you just moved onto CometArrowStreamReader. While the guard stays, that half of the change has no reachable effect. The only live consumer of duplicate-safe IPC reading is Utils.coalesceBroadcastBatches on the broadcast side.

It is visible to users too. With default config, SELECT /*+ REPARTITION(3) */ s FROM (SELECT named_struct('x', a, 'y', b) AS s FROM t) q plans as CometColumnarExchange, while the same query with 'x', a, 'x', b falls back to a Spark Exchange and takes a columnar-to-row transition. Only the field names differ. Meanwhile nativeShuffleFailureReasons has no equivalent guard, so with spark.comet.shuffle.mode=native a DISTRIBUTE BY over the same duplicate-name payload already goes through CometExchange and gives the right answer on this branch. The machinery works. Whether a user gets it depends only on which shuffle implementation the planner happens to pick.

I removed those two lines locally and ran nine duplicate-struct shapes (flat two-way, flat three-way with mixed types, nested, array of struct, map value, a struct holding an array and a map, case-distinct, a user field named __comet_runtime_field_0, and decimal plus timestamp) across round-robin and hash partitioning with dictionary on and off, ten thousand rows each. All thirty-six combinations pass checkSparkAnswerAndOperator with CometColumnarExchange in the plan, and CometShuffleSuite, DisableAQECometShuffleSuite and CometNativeShuffleSuite stay green at 127 of 127 with it gone.

Would you drop the guard here and add one query-level regression that shuffles a duplicate-name struct with spark.comet.shuffle.mode=jvm? If you would rather hold this PR's scope, updating the now-false comment and filing a follow-up would work too, but then the StreamReader change has nothing exercising it.

I have left the smaller points inline. None of them are blocking.

[andygrove]

This case does what the comment above it says, which is the problem: the REPARTITION(2, a) hint is inside the subquery, so a and b cross the exchange as plain int and string and the struct is built afterwards. Nothing in the PR carries a duplicate-name struct through a shuffle, which is the boundary the StreamReader change is for.

Moving the hint outside would cover it:

SELECT /*+ REPARTITION(3) */ s
FROM (SELECT named_struct('x', a, 'x', b) AS s FROM test_named_struct) shuffled

Worth knowing before you try it that this version currently fails the fixture's coverage assertion, for the shuffle-guard reason in my summary. It passes once that guard is gone.

[andygrove]

Removing this makes sense, since it asserted ProjectExec and that is precisely the behaviour the PR changes. The input diversity went with it though: ten thousand rows, dictionaryEnabled both ways, and a literal child in named_struct('a', _1, 'a', 2). The new fixture has three rows, no dictionary variation, and no duplicate-plus-literal case.

I reran these queries against the branch and they all match, so nothing is broken. But a dictionary-encoded string child inside a duplicate-name struct is cheap to keep and awkward to notice losing. Would you add a dictionaryEnabled loop over a duplicate-name struct somewhere, or a --CONFIG matrix line on the fixture?

[andygrove]

The identity fast path works as intended. A scalar column now costs one getChildren().isEmpty() and goes straight to field.createVector, and the new $data$ regression pins that.

One allocation is left on the per-batch path. names is a fresh HashSet[String] for every struct node with more than one child, created on every importVector call for every complex column, and in the overwhelmingly common no-duplicate case it gets filled and discarded having proved nothing. For the child counts structs usually have, a linear scan over the already-materialised children list would avoid it, or the set could be created only on the first repeat.

This is second-order next to the Field tree that importField already rebuilds on the same call, so only worth doing if it stays a one-liner.

[andygrove]

This override drops three things ArrowReader.initialize did: the per-field DictionaryUtility.toMemoryFormat call, originalSchema.getCustomMetadata(), and populating dictionaries.

I traced all three and they look safe. CometArrowStream.reconcileStreamSchema already decodes CometDictionaryVector columns down to the dictionary's value type, so the schema reaching here never carries a DictionaryEncoding, and neither Utils.toArrowSchema nor reconcileStreamSchema ever sets schema-level metadata, so there is none to lose.

That took a while to establish though, and the commit directly under this branch (5552) was specifically about Arrow metadata surviving C Data exports. Could a sentence here record why skipping both is correct, so the next person does not have to re-derive it?

[andygrove]

This class and ColumnarBatchArrowReader now carry the same forty-odd lines of shadow state: cometInitialized, cometRoot and cometLoader, plus overrides of ensureInitialized, getVectorSchemaRoot, prepareLoadNextBatch, loadRecordBatch, lookup, getDictionaryIds, getDictionaryVectors and close.

The reason is good and not at all obvious from reading either file. ArrowReader.root, loader and initialized are all private in Arrow 18.3, so there is no way to reuse them, and every base method that reads them has to be overridden or it silently sees null or false. I enumerated those methods and the override set is complete in both classes, which is the part I most expected to be wrong. The class doc here explains the allocation motive but not the shadowing one.

Two smaller things suggest the second copy did not get the same pass as the first. getDictionaryIds throws IllegalStateException when uninitialized here but calls ensureInitialized() in ColumnarBatchArrowReader, and Arrow's base does neither. And ColumnarBatchArrowReader.close(closeReadSource: Boolean) ignores its parameter, which is harmless because closeReadSource() is () in that class but reads as an oversight.

Would a shared private[comet] base in org.apache.comet.vector holding the shadow root and loader and the mechanical overrides be worth it? Mostly so the private-field constraint has one place to be written down, since that is the thing most likely to get lost when someone next touches either class.

[sunchao]

Correctness

Follow-up on e1ecb7fb after the new review: I found no additional P1/P2 issue. Maintained Spark 3.5 and 4.0 preserve duplicate labels, field order, and each child's nullability in CreateNamedStruct; ordinal access remains consistent with those semantics. The constructor itself has no ANSI-dependent arithmetic or overflow behavior. I did not independently qualify maintained Spark 3.4 or 4.1.

I confirmed the existing JVM columnar-shuffle guard and the SQL fixture limitation: the duplicate struct is constructed above the exchange. The guard is unchanged by this PR, and the existing discussion already covers that admission/coverage gap.

One clarification: the shared StreamReader is exercised with duplicate fields. The new UtilsSuite test calls decodeBatches, which reaches ArrowReaderIterator → StreamReader → CometArrowStreamReader, then checks both ordinal values and both original labels. Production CometBatchRDD broadcast consumers use this same decoding path. This coverage does not establish duplicate-payload JVM shuffle support.

The Spark 3.5 execution job passed that test, the duplicate-field broadcast join, and the C Data import regression (743 tests passed overall). The expression job passed the SQL, codegen, and JSON regressions (1,292 passed overall). Both loaded the native artifact built at merge 48892f46, whose HEAD parent matches this PR but whose base differs from the assigned review base. All changed production files and the duplicate-field test bodies match HEAD; the broader trees differ. These are CI merge results, not a local rerun or current-pair merge qualification.

Performance

The per-import HashSet allocation for each struct with multiple children is present. Leaves return immediately, unchanged schemas retain their original Field objects, and imports without duplicate names still use Arrow's default vector allocation. I have no measurement establishing a material regression. The removed test's explicit dictionary on/off, 10,000-row, and duplicate column-plus-literal coverage is reduced; the new direct codegen test still covers duplicate literal values. The existing feedback already identifies the coverage reduction.

Design

Against Arrow 18.3.0, both readers override the entry points that otherwise depend on its private initialization, root, and loader state. I found no missing override. ColumnarBatchArrowReader receives schemas without schema-level metadata from its current production constructors, and dictionary columns are decoded before loading its stable root. Field metadata remains preserved. The reported omission therefore does not demonstrate a current caller failure.

Abstraction & complexity

The repeated reader lifecycle code follows Arrow's private-state constraint. I found no additional correctness or maintenance issue requiring another abstraction before merge. This follow-up adds clarification to the existing discussion and leaves the same-head approval unchanged.

[RRXXZZYY]

I removed the JVM columnar-shuffle duplicate-name guard now that the stream reader preserves children positionally. The former fallback test is now a query-level regression that carries a duplicate-name struct through a JVM repartition and asserts one Comet exchange plus Spark-equivalent output.\n\nI ran the targeted suite in the NAS builder, but its complete result was not recoverable from the builder's streamed output; the earlier full run is independently blocked by an existing Scalastyle violation in Utils.scala outside this PR. I have therefore left CI as the authoritative full validation for this small follow-up.

[sunchao]

Correctness

Re-reviewed bd79bd3a6ef72c476137386222053c0779c10a5d against bb9e74020adc228e486f6f4d0fa68292b30bff31, after e1ecb7fb.

The shuffle follow-up addresses the remaining admission gap: a nonempty struct with duplicate field names can now use JVM columnar shuffle, and the replacement regression carries the struct itself through REPARTITION(3). Its helper requires exactly one CometColumnarShuffle exchange and compares the result with Spark. The suite runs with AQE both enabled and disabled.

This matches the maintained Spark 3.5 and 4.0 constructor semantics: CreateNamedStruct preserves names, field order, child types and nullability, and generated evaluation writes values by ordinal. Duplicate names do not make the constructor invalid. The surrounding constructor remains non-null even when its values are null. Invalid names and child-expression errors retain Spark's analysis/evaluation behavior. No new ANSI or overflow behavior is introduced by removing the shuffle eligibility check. The writer uses ordinal struct fields, and the shuffle decoder reaches the updated C Data import factory. Spark 3.4 and 4.1 maintained sources were unavailable, so this review does not qualify those versions.

One new P1 blocks this head: the merge left an extra closing brace in NativeUtilSuite.scala at line 518. Both Scala 2.12.18 and 2.13.17 reject it before test compilation can proceed. The inline comment has the exact diagnostic.

Validation

Parser-only compiler checks cover all 15 changed Scala files. Both compiler versions fail at the same brace. The exact base and prior approved NativeUtilSuite.scala pass under both as controls. These checks do not establish type checking, generated-kernel execution, JNI/Arrow round trips or Spark query correctness. At the September 8, 14:38 UTC refresh, all three current-head Actions runs require maintainer action, and focused job reads return zero jobs. The PR body's validation is explicitly for e1ecb7fb. The new author reply reports an attempted shuffle run without a complete result. Current-head Spark/Arrow runtime qualification therefore remains pending after the syntax fix.

Performance

Removing the distinct-name check removes a small planning-time scan/allocation and permits the existing shuffle implementation for these schemas. It adds no new per-row conversion or copy in this follow-up. The vector factory's schema traversal and allocation strategy are unchanged from the prior review. No measured shuffle speedup or Arrow 59.3 runtime benefit is established here.

Design

Removing the eligibility restriction is consistent with the existing positional writer and repaired import boundary. Replacing a fallback assertion with an actual duplicate payload crossing the exchange is the appropriate regression scope. The rebase preserves the earlier factory and dispatch design, but the merged test file must compile before that integration can be validated.

Abstraction & complexity

The authored follow-up adds no abstraction: it simplifies the recursive type predicate and reuses the existing exchange/result assertion helper. No new actionable complexity issue was found in the changed scope.

[sunchao]

Correctness

[P1] Remove the extra closing brace introduced by the merge

This closing brace leaves NativeUtilSuite.scala unbalanced, so the shared test sources cannot compile. Running scala.tools.nsc.Main -Ystop-after:parser on the changed Scala files fails under both Scala 2.12.18 and 2.13.17 with NativeUtilSuite.scala:518: error: Unmatched closing brace '}' ignored here. The exact base and prior approved versions of this suite parse under both compilers. Please remove the extra brace and rerun test compilation and the targeted shuffle/import regressions. The new tests cannot execute on this head as written.

[RRXXZZYY]

Thanks for catching this. I removed the extra closing brace in NativeUtilSuite.scala and pushed the one-line syntax fix in 071ba483. I did not claim a new full runtime result locally; the current-head workflows still require maintainer approval, so CI remains the authoritative validation for this follow-up.

[sunchao]

Correctness

Re-reviewed 071ba483b4c54385bdcb89c8e06de2933052d180 against bb9e74020adc228e486f6f4d0fa68292b30bff31, after bd79bd3a.

The prior P1 is fixed in 071ba483. Relative to bd79bd3a, the normalized change removes exactly the extra closing brace from NativeUtilSuite.scala. The other differences are nearby line endings. All 15 Scala files in the full PR contribution now pass parser checks with Scala 2.12.18 and 2.13.17. The prior-head suite still fails at line 518 under both compilers, while the exact base passes.

I found no new or remaining actionable P1/P2. The production implementation and all other test files are byte-identical to the previous review, and the base is unchanged. The maintained Spark 3.5 and 4.0 comparison still holds: duplicate labels preserve field order, types and child nullability, and values are accessed by ordinal. This syntax fix changes no constructor error, fallback, ANSI or overflow behavior. Maintained Spark 3.4 and 4.1 remain unqualified.

Validation

These are parser-only checks, not full test compilation or Spark/JNI/shuffle execution. At the September 8, 15:45 UTC refresh, all three current-head workflows require maintainer action, and their job lists are empty. The author also explicitly reports no new full runtime result. Current-head integration validation remains pending, and the older e1ecb7fb results do not establish it.

Performance

The allocation, copying and shuffle paths are byte-identical to the prior review. This test-syntax repair introduces no runtime work or performance claim, so it does not call for a new microbenchmark.

Design

Deleting the unmatched brace is the direct fix. It retains the import/default-allocation assertions and the existing duplicate-payload shuffle regression without changing their scope. No new design issue was found.

Abstraction & complexity

The fix restores the existing test block structure and adds no helper or abstraction. The earlier factory and reader design is unchanged, with no new actionable complexity concern in this update.

[RRXXZZYY]

I also synced the branch with the current main and corrected the CRLF-only upload noise from the earlier API update. The final PR diff is back to the intended 19 files, with the brace removal retained; 95fad218 is the current merge commit and GitHub now reports the PR as mergeable. I have not claimed a new full runtime run locally, so CI remains the authoritative validation once its workflow is approved.

[sunchao]

Correctness

Re-reviewed 95fad218e12c6fda4500d860e41512e76390319a against 8e6846850c525506dd2b9194f2014e8acd2ab60a, the actual merge base. All 19 authored patches match approved 071ba483 after line-ending and diff-metadata normalization. All 15 authored Scala files retain identical normalized text, including the corrected NativeUtilSuite brace. The remaining incremental changes match the inherited base changes.

No new or remaining P1/P2 was found. Maintained Spark 3.5/4.0 still constructs a non-null struct with ordered fields, child types and child nullability, and evaluates values by ordinal. Duplicate-name dispatch, fallback reporting, same-arity duplicate children, and disjoint runtime names are unchanged. The literal regression invokes the kernel directly, so its source coverage does not depend on SQL constant folding. The inherited remote-shuffle schema cache continues through the same duplicate-safe NativeUtil import. Decoder creation, use and release remain under the iterator monitor.

At September 8, 17:31 UTC, all three current-head workflows were action_required, with no head or merge check results. The synthetic merge has the expected parents and equals the head tree. The prior Scala 2.12/2.13 parser results support preservation of the syntax fix through source identity. They are not new test runs or full compilation. No current-head build, Spark/JNI/shuffle test or benchmark ran here. The author likewise reports no new runtime run. Maintained Spark 3.4/4.1 remain unqualified.

Performance

Line-ending normalization adds no runtime work. The struct allocation and copying paths retain the approved implementation. The inherited decoder caches expected types once per iterator and leaves per-batch import behavior intact. No measured speedup or new benchmark result is claimed for this pair.

Design

The merge preserves the existing separation between logical duplicate names and private allocation names. Schema caching does not replace the positional import boundary. No new design correction is needed in this update.

Abstraction & complexity

The authored factory and reader abstractions are unchanged after normalization. The base integration adds no duplicate-name-specific layer, and no new actionable complexity concern emerged.

[sunchao]

Thanks for the PR @RRXXZZYY . This overall looks good to me (and Codex). However, I wonder if we can just implement the support in native instead of codegen dispatch.

[sunchao]

(Summary from discussion with Codex)

Yes. I’d prefer enabling the existing native constructor while retaining the Arrow boundary fixes.

Much of the native support already exists

The Rust CreateNamedStruct already pairs field names with child arrays by position and constructs a StructArray. Neither that constructor nor its native planner rejects duplicate names. The explicit rejection is in the Scala serializer. Native constructor, Scala guard.

So this looks primarily like enabling and validating an existing capability, rather than writing a new native implementation.

Why I prefer that

For:

named_struct('x', a + 1, 'x', b * 2)

The PR’s dispatcher executes the whole expression on the JVM. A native route would allow both arithmetic children and struct construction to stay native.

That should eliminate concrete work:

• The JVM callback for this constructor.
• Spark’s per-row struct-object allocation.
• Copying those row fields back into Arrow output vectors.

For existing column arrays, native struct construction can reuse the child arrays; scalar children may still need expansion. These are source-based advantages, not a measured query speedup.

The Arrow repairs still matter

Native-produced duplicate structs still reach Java when returning results, shuffling, broadcasting, or invoking JVM expressions. Arrow Java would encounter the same name-collision problem there.

Therefore, native support still needs the duplicate-safe import, IPC, broadcast, and stream-export changes. The shared codegen allocation repair also remains useful for other JVM expressions that return structs. Shared allocation changes.

My recommendation

Make duplicate-name construction native-first. Use JVM dispatch where child expressions actually require it, rather than making duplicate labels themselves trigger dispatch.

Before accepting that alternative, I’d run the boundary regressions with struct dispatch disabled and assert native execution, including mixed types, nulls, scalar/column mixtures, nested structs, shuffle, and broadcast.

The source makes this approach look preferable and feasible. I haven’t tested that planner change end to end; the current PR’s passing CI validates its dispatcher-based implementation.

[sunchao]

Correctness

Discussion follow-up on 95fad218e12c6fda4500d860e41512e76390319a against 8e6846850c525506dd2b9194f2014e8acd2ab60a. The pair and full 19-file contribution are unchanged since review 5145002442. I rechecked constructor/serde routing, ownership and the boundary tests. No new or remaining verified P1/P2 was found. My existing approval stands.

Maintained Spark 3.5/4.0 still require an ordered, non-null parent struct whose children retain their types and nullability, including all-null child values. The inspected native constructor and positional Arrow construction support that contract at source level. This does not qualify the proposed routing change end to end. Maintained Spark 3.4/4.1 source gaps remain.

At 2026-09-09T19:16:28Z, checks are now 64 successful and nine skipped. The duplicate-struct SQL, direct kernel, case-distinct JSON and retained-producer JSON regressions passed in the inspected Spark 3.5 and 4.0 expression jobs. The 3.5 exec job passed the import, coalescing and broadcast collision regressions, and the shuffle job passed the duplicate-struct repartition test. Builder and consumers checked out 10bbc8b81f96ca35f0616125f0333453769f558d, whose parents are the assigned pair and whose tree equals HEAD. These passes validate the current dispatcher implementation. No local product test or benchmark was run.

Performance

The ownership check confirms that existing array children share their backing buffers through reference-counted arrays. The constructor still builds field/vector metadata and expands scalar children. Child expressions retain their own computation and allocation costs. No new timing result or measured benefit for the proposed route is available.

Design

The native-first request remains pending: duplicate names still return Unsupported before child conversion. One validation detail is that spark.comet.exec.scalaUDF.codegen.enabled=false disables all dispatch, including the JVM JSON consumer. Use that setting for pure native constructor/boundary controls. For the retained to_json test, leave JSON dispatch enabled and separately assert that its producer serializes as native CreateNamedStruct. A Comet project alone does not establish that distinction. The duplicate-safe Arrow boundary repairs remain necessary with either producer route.

Abstraction & complexity

The existing factory and reader changes continue to serve imports, IPC, broadcast and stream export independently of constructor routing. Enabling the native constructor should not introduce another allocation wrapper or duplicate those boundary repairs. I found no additional abstraction issue in the unchanged implementation.

[andygrove]

@sunchao I think you are right, and the evidence is stronger than the summary states, so I prototyped it rather than argue from the source.

Nothing in the native path resolves a struct child by name

That is the property the native route needs, and it holds:

• CometGetStructField serializes setOrdinal(expr.ordinal), and the planner builds GetStructField::new(child, expr.ordinal as usize). Purely positional on both sides.
• There is no column_by_name anywhere in native/spark-expr/src or native/core/src outside tests.
• Native to_json iterates fields() positionally and emits each field's own name, so a duplicate-name struct produces both keys, which is what Spark does.

Field access by name is not even reachable. named_struct('x', a, 'x', b).x is rejected by Spark's own analyzer before Comet sees it:

[AMBIGUOUS_REFERENCE_TO_FIELDS] Ambiguous reference to the field `x`. It appears 2 times in the schema.
  at ExtractValue$.findField(complexTypeExtractors.scala:161)

So Spark itself guarantees the only way into a duplicate-name struct is by ordinal.

The prototype works

I removed the four lines of getSupportLevel on this branch, leaving everything else in the PR in place, and ran ten shapes under both shuffle modes with checkSparkAnswer, asserting no dispatcher note in the extended explain:

PROBE[jvm]    OK  flat / computed children / case-distinct + dup / nested dup /
                  to_json / array of dup / map value dup / shuffled / sorted / grouped
PROBE[native] OK  (same ten)

Twenty for twenty, all native, all matching Spark. CometCodegenSuite still passes 94 of 94 and the new create_named_struct.sql fixture still passes with the guard gone, so the PR's own routing tests do not depend on the rejection either.

The two positions are not in conflict

The native route depends on this PR's Arrow-boundary work, and that is the part to keep. A native named_struct produces a duplicate-name StructArray that then has to survive shuffle IPC, broadcast coalescing, the per-batch import factory and a cache round trip. CometArrowStreamReader, ColumnarBatchArrowReader and the runtime field allocation are what make that hold, and the dispatcher route needs them just as much, since the JVM result crosses the same boundaries. Nothing in the PR is wasted by switching the primary path.

@RRXXZZYY the shape I would suggest, if you agree: keep the boundary fixes exactly as they are, replace the four-line getSupportLevel rejection with Compatible(), and let CodegenDispatchFallback stay as the fallback for whatever native declines for other reasons rather than as the primary path for duplicates. The getUnsupportedReasons text and the generated expressions.md entry then need updating, since they currently promise the dispatcher.

One caveat on my own earlier review: my roughly fifty divergence probes were run against the dispatcher route, so they do not carry over unchanged. The ten above are a start, not a replacement. The shapes I would want re-run natively are dictionary-encoded inputs, an all-null and a zero-row batch, empty field names, a struct straddling spark.sql.codegen.maxFields, and a df.cache() round trip, since those are the ones where the two routes could differ.

Separately, my main request from last time is addressed: the JVM columnar-shuffle duplicate-name guard is gone and the former fallback test is now a query-level regression carrying a duplicate-name struct through a JVM repartition. Thanks for that.

[RRXXZZYY]

Implemented the native-first route for duplicate-name CreateNamedStruct: supported children now stay native, while unsupported children retain the existing dispatcher fallback.

I added checks with Scala-UDF codegen dispatch disabled for the constructor, the columnar shuffle boundary, and the broadcast-hash-join boundary. The focused Spark 4.1 tests passed (1/1 each), along with the SQL fixture. I kept the SQL fixture's default setting because it also covers map(...), which has no native path in this environment. The PR description has the full validation scope and boundaries.

[andygrove]

I checked out db15072b and ran it, since the native-first switch is new since my last round and has no review or CI on it yet. Build is clean on Spark 4.1 / JDK 17 and everything the PR touches passes locally: CometExpressionSuite, CometCodegenSuite, CometJsonJvmSuite, NativeUtilSuite, UtilsSuite and CometJoinSuite at 291 of 291, then CometShuffleSuite, DisableAQECometShuffleSuite, CometNativeShuffleSuite and the create_named_struct fixture at 141 of 141.

My last comment listed the shapes my earlier probes did not cover on the native route, so I re-ran those. Thirty probes with spark.comet.exec.scalaUDF.codegen.enabled=false, so anything still needing the dispatcher shows up as a Spark fallback and fails the operator assertion: dictionary on and off at ten thousand rows over every safe primitive type, all-null rows, zero rows, empty field names, a struct on either side of spark.sql.codegen.maxFields, df.cache(), case-distinct alone and mixed with an exact duplicate, user fields literally named __comet_runtime_field_0, __comet_runtime_field_1, $data$, key and value, an all-scalar struct with ConstantFolding excluded, array and map of duplicate struct, explode, inline, both shuffle modes and a sort-merge join payload. Then a second round using native to_json as a leak detector, over a natively built duplicate struct and above a JVM columnar shuffle, a native shuffle, a broadcast join and a cache round trip. Every one printed {"x":1,"x":10}. No private runtime name reaches user-visible output. The native route holds up, and I like it better than the dispatcher one.

I also microbenched the import factory, since createVectorForImport now runs per column per batch. Two to five nanoseconds per column against a baseline where ArrowImporter.importField already rebuilds a whole Field tree on the same call. That is inside the noise floor and I would not spend more time on it.

Two things I would like resolved, and neither is about the native route being wrong.

The dispatcher is unreachable now, but three places still promise it

Deleting getSupportLevel leaves CometCreateNamedStruct on the inherited Compatible(None). exprToProtoInternal only calls dispatchIfFallback from the Unsupported and Incompatible arms, so when convert returns None for an unsupported child the whole projection falls back to Spark and the dispatcher never runs. I confirmed it with SELECT named_struct('x', java_method('java.lang.Math', 'abs', _1), 'x', _1) FROM tbl, where the extended explain reads Project [COMET: unsupported arguments for CreateNamedStruct, java_method is not supported] and the summary line says 0 native, 0 codegen dispatch.

That is not a regression, main fell back too. What is new is the claim. with CodegenDispatchFallback is now inert for this serde, the rewritten getUnsupportedReasons text describes a path that cannot be taken, and the two expressions.md rows read Hybrid | Unsupported child expressions route through the JVM codegen dispatcher. GenerateDocs.classifySerde derives Hybrid purely from the mixin, so the page will keep saying it as long as the mixin is there.

Would you drop the mixin and the getUnsupportedReasons override and regenerate the page? That leaves named_struct and struct as plain Native, which is what they now are. If you would rather actually keep a dispatcher route for unsupported children, that is a real improvement, but it needs getSupportLevel to return Unsupported for that case, which means deciding it before convert runs.

Two sibling readers still collapse duplicate children, and native panics

RowArrowReader and SparkColumnarArrowReader still inherit ArrowReader.initialize(), which allocates through VectorSchemaRoot.create(schema, allocator). That is the same name-indexed path you replaced in ColumnarBatchArrowReader, and those two are the readers CometSparkToColumnarExec actually uses. So a duplicate-name struct arriving from a non-Arrow source collapses to one physical child while the advertised schema still declares two:

sql("SELECT named_struct('x', _1, 'x', _2) AS s, _1 AS k FROM tbl").createOrReplaceTempView("v")
spark.catalog.cacheTable("v")
sql("SELECT s FROM v WHERE k > 1").collect()

Comet native panic: panicked at arrow-array-59.3.0/src/ffi.rs:382:17:
assertion failed: fields.len() == self.array.num_children()
org.apache.comet.CometNativeException
  at org.apache.comet.Native.executePlan(Native Method)

Default config, no allowIncompatible, and the plan is just CometFilter over CometSparkRowToColumnar over Scan In-memory table. A join over the same cached view does it too, as does native to_json over it.

I ran the same probes against the merge base and they panic there identically, so this is pre-existing and not something you broke. I am raising it here because it is the same defect this PR is about, in the same package, and NativeUtil.createVectorSchemaRootForExport already fixes it. I overrode initialize() in RowArrowReader with it and all five crashing probes go green with answers matching Spark. CometArrowConverters lines 69 and 99 have the same VectorSchemaRoot.create for the Comet in-memory cache serializer, and CometLocalTableScanExec shares RowArrowReader. CometArrowPythonRunnerBase line 153 has the same shape, though I did not build a repro for it.

Would you fold that in? Otherwise #5586 closes as "duplicate names work now" while a plain df.cache() and a filter still panic. Filing a follow-up before merge would work too, but I would rather it not merge with nothing tracking it.

The dictionary coverage came back

db15072b deletes named_struct with duplicate field names, which ran three shapes over makeParquetFileAllPrimitiveTypes at ten thousand rows with dictionary encoding both on and off, and replaces it with a two-row two-column table. Dictionary encoding is the axis I would least want to drop, because the fifty probes I ran earlier in this thread were all against the dispatcher route and do not carry over.

Would you keep the old test body and wrap it in withSQLConf(CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "false") so it now asserts the native route instead of the old fallback? I ran exactly that on this head, plus decimal, timestamp, date, binary and dictionary-encoded string at ten thousand rows with dictionary both ways, and it all passes. It costs a few seconds of suite time.

Smaller points

createVectorSchemaRootForExport builds new Schema(runtimeFields) and loses schema.getCustomMetadata, while createVectorSchemaRootForImport six lines above passes the original Schema through. No caller sets schema-level metadata today so nothing breaks, but the two functions are easy to mistake for each other. Would new Schema(runtimeFields, schema.getCustomMetadata) be right?

The export root also advertises the private runtime names to native rather than pinning getField back the way the import side does. That is only correct because native ScanExec.build_record_batch sees the struct DataType differ from the declared one and casts positionally back to x, x. My probes confirm it works, but the invariant is currently only visible from scan.rs. Could the scaladoc say the consumer is expected to re-derive names from its declared schema?

RenamedStructVector.getField reading size() == exportField.getChildren.size() as "construction finished" means any future disagreement surfaces as the private names silently reaching FFI rather than as a failure. createPinnedVector is the only construction site, so an explicit publish() right after initializeChildrenFromFields would be louder. I could not get the current form to misfire, so this is about the next person to touch it.

Last, repeating my point from the previous round with a bit more weight behind it. CometArrowStreamReader and ColumnarBatchArrowReader carry the same forty lines of shadow state for ArrowReader's private root, loader and initialized, and only initialize() and loadNextBatch() actually differ. If the reader fix above lands here that becomes four copies. A private[comet] abstract class in org.apache.comet.vector holding the fields plus ensureInitialized, getVectorSchemaRoot, prepareLoadNextBatch and close would give the constraint one place to be written down.

One process note. Everything above is the default Spark 4.1 profile. The three workflows on this head are still action_required, so a green cross-version run is the remaining gate either way.

[RRXXZZYY]

Follow-up in 714aa8d9b: folded the duplicate-safe Arrow reader/cache allocation into the native-first path, restored the dictionary-enabled/disabled expression regression, preserved Arrow custom metadata, and made export-field publication explicit after child initialization. On the isolated NAS builder, Spark 4.1 test-compile and the focused native expression/cache regressions pass (1/1 each); the PR body records the SMB formatting and native-rebuild boundaries.

[sunchao]

Correctness

Reviewed head 714aa8d9bb609b34a720efd00c2a381e1c97de64 against base 8e6846850c525506dd2b9194f2014e8acd2ab60a. I found one confirmed issue to fix before merge: the serializer declaration fails the normal formatting gate, detailed inline. I did not reproduce an additional runtime correctness defect.

Local validation

On Linux with Spark 4.1.3 and JDK 17, I built the native library from this exact head, then built and tested from the root Maven reactor. The packaged native library matched the freshly built library by SHA-256.

• 116 repository tests passed: 113 across NativeUtilSuite, UtilsSuite, and CometCodegenSuite, plus the named-struct SQL fixture and two duplicate-expression regressions. The latter include dictionary encoding enabled and disabled over 10,000 rows.
• All 14 distinct independent Spark probes passed after correcting two probe setup/expectation errors. They cover empty and colliding names, nulls, nested and wide structs, constant/column mixtures, both shuffle modes, cache, and a retained producer feeding JVM JSON. Min/max retains the existing Spark SortAggregate fallback and was checked for matching results and Comet shuffle.
• 100 Arrow/C Data component rounds passed, covering duplicate structs, dictionary replacement/delta batches, exported-batch lifetime, repeated closure, and zero outstanding allocator bytes.

Runtime tests skipped Spotless and Scalastyle. Formatting was checked independently: the exact base passes and this head fails. I did not run a local cross-version or cross-platform runtime matrix.

CI

The final snapshot at 2026-09-11 02:32 UTC has 17 successful, 39 failed, 11 skipped, and none running checks. Of the failures, 37 report the same Spotless violation. Iceberg shard coverage then fails because its four failed shard builds produced no manifests. The separate Delta build gate suppresses its Maven error, so its exact cause remains unconfirmed. CI run, Delta gate.

These CI jobs use merge commit 87c52a65, which includes a newer base and differs from the direct head tested locally. The affected serializer file is identical in both.

Performance

I found no demonstrated meaningful regression. The ordinary import path retains its fast path, although complex schemas are still traversed. No query-speedup claim was benchmarked.

Design

Keeping supported struct construction native while repairing the shared Arrow boundaries is appropriate. Unsupported child expressions remain subject to the existing conversion boundary.

Abstraction & complexity

The four reader implementations repeat lifecycle bookkeeping because Arrow keeps its root/loader state private. That adds maintenance cost, but the inspected overrides and cleanup checks revealed no additional actionable defect.

[sunchao]

[P2] Fix the declaration formatting to unblock CI

Could you put this declaration on one line and rerun CI?

object CometCreateNamedStruct extends CometExpressionSerde[CreateNamedStruct] {

I independently ran the root-reactor Spotless check on the exact base and head. Base 8e684685 passes, while head 714aa8d9 fails solely on this declaration. The Spark 4.1 build and both Celeborn compatibility jobs report the same formatting error before compilation. The runtime checks I ran with style checks skipped passed, but the normal build remains blocked here.

[RRXXZZYY]

CI follow-up: the failed compile jobs pointed to a Spotless-only formatting issue in structs.scala. I pushed d0a20e46c with the project-formatted one-line declaration; no behavior changed. I did not alter the separate shard-coverage or cross-platform jobs.

[sunchao]

Re-reviewed d0a20e46c98c from the assigned baseline 95fad218e12c. The intervening changes implement the native constructor route, extend duplicate-safe allocation to the row/columnar readers and cache converters, restore dictionary-on/off coverage, and publish struct metadata after child initialization. The misleading dispatcher mixin and documentation are removed.

The P2 formatting finding is fixed. I ran the repository-pinned Scalafmt 3.6.1 with its checked-in configuration against the exact file versions: the base passes, 714aa8d9 reproduces the one-line declaration violation, and this head passes. The current file exactly equals that formatter's output. I found no new or remaining verified P1/P2 in the reviewed changes.

I also rechecked positional field construction, nullability, reader lifecycle and the native conversion boundaries against maintained Spark 3.5/4.0. The only change since the later runtime-tested revision is this formatting edit. Those runtime results remain the published reviewer's results. My local checks were source review, formatter execution and git diff --check, not a Spark/JNI suite or root-reactor Spotless run.

Current CI requires workflow approval and has zero jobs. The successful label check provides no test coverage. Maintained Spark 3.4/4.1 source gaps remain, and no query-performance benefit was measured.