fix: preserve map field metadata and honor target sorted flag in cast_map_to_map

[Smallfu666]

Which issue does this PR close?

Related to #5097. Intentionally not Closes, see the scope note below.

Rationale for this change

cast_map_to_map rebuilt the entries, key and value fields from scratch with Field::new, which
dropped field metadata and target nullability, so data_type() did not equal the requested
target. Comet's serde.rs runs map key and value fields through with_parquet_field_id, so a
target really can carry PARQUET:field_id, and that is the loss with a real producer.

It also used the source sorted flag for the result rather than the target's.

What changes are included in this PR?

• Rename-only fast path: when the key and value types and the sort order are unchanged, which
  is the common Parquet key_value to Spark entries relabel, delegate to arrow's cast. Arrow
  clones the target entries field rather than rebuilding it, so metadata survives.
• Hand-built path: when a child type changes, recurse with cast_array and build the result
  with the target sorted flag and the target fields. try_new replaces new, so a malformed
  target returns Err instead of panicking.
• Field-count guard before indexing entries [0] and [1], so a target with 0 or 1 fields
  returns Err rather than panicking. Without it that try_new promise is not actually true,
  because the indexing happens first.

Scope notes

Two things found while working on this are deliberately not here.

The sorted flag half of #5097 is defensive rather than a bug users can reach. serde.rs builds
every map type with sorted = false, the planner propagates it unchanged, and map_sort reuses
the input flag, so the flags always agree in a Comet plan and the hand-built branch is only
reached when a child type changes. The target flag is now simply copied, and the unreachable
rejection that was here before is gone.

The entries null buffer half of #5097 does not appear to be a live bug. MapArray::try_new
rejects entries carrying any null, StructArray::try_new discards an all-valid null buffer, and
the ArrayData route normalizes one to None in ArrayDataBuilder::build, so
entries().nulls() is None however the array was built. The hand-built path now passes None
with that reasoning recorded, rather than reading a buffer back that cannot exist. This is why
the PR says Related to rather than Closes.

A reachable TRY_CAST divergence for narrowing integral map keys is deliberately out of scope
here and is filed as #5995. It is a TRY_CAST semantics question rather than a metadata
preservation one.

How are these changes tested?

9 map-cast unit tests in cast.rs: the rename-only fast path with values, offsets and nulls
preserved, the hand-built path with metadata and child type casts, a key and value cast, the
target sorted flag, a cross-path comparison over sliced input carrying a null row, a malformed
target, and the 0, 1 and 3 entry-field cases.

All nine fail on unpatched main.

cargo test -p datafusion-comet-spark-expr passes 857 tests.
cargo clippy -p datafusion-comet-spark-expr --all-targets -- -D warnings and
cargo fmt --all -- --check are clean.

No SQL fixture was added. CometNativeCastSuite's "cast MapType to MapType" already compares
map to map casts against Spark, and a SQL answer comparison cannot observe field metadata or a
sort flag, so a fixture would not be a regression test for this change.

🤖 Generated with Claude Code

[andygrove]

Review assisted by an LLM (Claude Code). I checked the branch out, ran the tests, and verified the claims below myself.

Thanks for taking this on. The two core fixes are right. Using to_fields.clone() instead of rebuilding fields with Field::new does preserve metadata, and switching *from_sorted to *to_sorted is clearly correct. Moving to try_new is a real improvement over new as well.

What I ran locally on the branch:

• The 16 new Rust tests pass.
• cargo clippy --all-targets and cargo fmt --check are clean.
• The three existing map cast tests in CometCastSuite pass.
• I added a scratch test for a sliced input through the rename-only fast path and it passes too.

A few things I would like to resolve before merge.

The sorted rejection

Issue #5097 asked for the target sorted flag to be honored, and this does that. The rejection branch at cast_map_to_map is separate policy on top of that, and it turns what used to be wrong metadata into a runtime Internal error.

The code comment points to the PR description for planner reachability, but I could not find that discussion there. When I traced it, native/core/src/execution/serde.rs:151 hardcodes ArrowDataType::Map(Arc::new(struct_field), false) for every map type built from protobuf. native/core/src/execution/planner.rs:200 propagates the flag unchanged. The Parquet reader derives sorted from the requested arrow type, which comes from that same false-producing path. And spark_map_sort deliberately preserves the input flag rather than setting it true.

That suggests no Comet plan ever asks for a sorted = true map target. If that is right, would it be simpler to drop the rejection and keep just the *to_sorted fix? Five of the sixteen tests exercise a branch that cannot fire. If there is a path I missed, could you add it to the PR description? That would also make the case that a hard error is the behavior we want here.

Cast options in the fast path

The fast path passes the static CAST_OPTIONS, which hardcodes safe: true. A little above, cast_array builds native_cast_options specifically so that ANSI mode gets safe: false. The fast path does not transform any values today, so this is inert. But if the condition ever widens it would silently swallow ANSI errors. Could you use native_cast_options here, or add a comment explaining why the static is fine?

Test coverage

test_cast_map_to_map_sliced casts Int32 to Int64, so it goes through the hand-built path rather than the fast path. Since there are now two independent implementations selected by a condition, it might be worth adding a sliced test where the value type is unchanged so the arrow delegation is covered too. I tried it locally and it passes, so this is about pinning the behavior rather than a suspected bug.

Along the same lines, a test that casts the same input through both paths and compares results would guard against the two drifting.

One thing I checked that is fine

Casting Map<Utf8, Int32> to Map<Int32, Int32> with a key like "abc" now returns Found unmasked nulls for non-nullable StructArray field "key" from StructArray::try_new. Before this PR the same input would have panicked in StructArray::new, so this is strictly better. It is also unreachable from a Spark plan, since Spark's canCast guards map casts with (!forceNullable(fromKey, toKey)) and forceNullable is true for any string source, so the analyzer rejects it first. I confirmed that.

Separately, Comet's CometCast.isSupported map arm only recurses into key and value support and does not replicate that forceNullable guard. Harmless today because the analyzer runs first, but worth a tracking issue so the two do not drift. Happy to file that if you would rather keep it out of this PR.

CI

There are no CI results on this yet and the branch is 35 commits behind main. Could you rebase so a full run can be triggered? My local runs only covered the Rust tests and the three map cast suites.

[Smallfu666]

Thanks for the review, and for checking out the branch and running it yourself.

Your read is right, so I dropped the rejection branch and kept the *to_sorted fix, the
to_fields.clone() metadata preservation and try_new. Agreed that turning wrong metadata into a
runtime error is a separate call from what the issue asked, and not one worth making on a branch
nothing reaches.

I confirmed your four sites and then looked for a producer you had not covered. I could not find
another production producer. Iceberg goes through the same serializeDataType and convert_spark_types_to_arrow_schema. Shuffle
builds maps only through arrow's MapBuilder, which hardcodes false. The FFI import honors
MAP_KEYS_SORTED, but Comet's only JVM side Arrow map field constructor is Utils.scala:195, which
passes false. make_all_fields_nullable and the nested comparison coercion derive a map target from
an input type rather than from protobuf, but both route to DataFusion's CastExpr rather than here
and both copy the source flag anyway. The only Map(_, true) types left in the tree are the ones my
tests hand build.

One correction on the count. 2 of the 16 tests exercised the rejection, not 5, and those are the 2 I
removed. The other 3 sorted tests pin the fix itself, and
test_cast_map_to_map_sorted_true_to_false_allowed fails on the pre-fix code with sorted=true
where false is expected.

Cast options are now built by an arrow_cast_options(eval_mode) helper shared with cast_array, so
the two call sites cannot drift. It is inert today and I checked why rather than assuming. Arrow
58.4's map cast delegates to cast_with_options for the key and value arrays, and the fast path
requires both child data types to be unchanged, so those child casts take Arrow's same-type early
return before any safe-dependent conversion. The sliced fast path test now asserts ANSI and legacy
give byte identical output.

Both coverage tests are in, and I verified with a temporary probe which branch each test actually
takes rather than inferring it from the types. The new sliced test hits the fast path, both_paths_agree
hits each branch exactly once, and the pre-existing test_cast_map_to_map_sliced hits the hand built
path, which matches your observation. Test count stays at 16.

Separately, all of this was pinned only by Rust assertions, which proves agreement with my own
expectations rather than with Spark. I added map cases to cast_complex.sql: value casts, key casts, and the null map, empty map and null value rows, plus the
DATATYPE_MISMATCH a key cast that could introduce nulls should raise. CometSqlFileTestSuite is 444 of 444 on Spark 3.5.
Adding cases to an existing file does not move the suite count, so I broke the new expectation on its
own and confirmed the suite goes red on that case specifically rather than assuming it ran.

Nothing under docs/source/user-guide/latest/ claims a support level for map casts, so there is no
audit page to update.

Rebased onto current main. CI has not started, since fork runs still need a maintainer to approve
each one.

If you still think the CometCast.isSupported divergence is worth tracking separately, please do
file it. I am happy to pick it up.

[andygrove]

Note on this review: this was generated by an LLM (Claude Code) at my request while I worked through a review backlog. I have not verified the individual findings myself. Please treat everything below as suggestions to evaluate rather than as authoritative review feedback, and push back on anything that is wrong or already handled.

This is careful work. The comments citing specific arrow-array and arrow-data line numbers for why a guard is or is not reachable are unusually good, and the "intentionally not Closes" scope note with the reason is the right way to handle a partial fix. test_cast_map_to_map_both_paths_agree pinning the equivalence between the delegated and hand-built paths is exactly the test I would have asked for.

Three questions.

Target field nullability can now make a TRY_CAST fail

StructArray::try_new(to_fields.clone(), vec![cast_keys, cast_values], ...) builds the entries with the target's declared nullability. If the target key field is nullable = false and cast_array produced nulls in the keys, try_new returns Err and the whole query fails.

Under EvalMode::Try that is reachable: TRY_CAST(m AS MAP<INT, INT>) where m is MAP<STRING, INT> and some key is not parseable as an int would produce a null key. What does Spark do there? If Spark returns a null map for the row, or throws a specific error, the current behavior of surfacing an Arrow try_new failure is probably not right. If Spark also fails, it would be worth a test pinning that the error is comparable.

Is a differing sorted flag reachable from Spark?

Spark's MapType has no sort-order concept, so I would expect from_sorted and to_sorted to always agree in a Comet-produced plan, which would make the hand-built branch reachable only when a child type changes. Is that right?

If it is, the sort-flag half of the fix is defensive rather than a bug users can hit, and saying so would help set expectations. If it is not, an example of where the flags diverge would be valuable in the description, since "cast to a map type with a different sorted flag returned the wrong type" is currently the headline bug and there is no repro for it.

The delegated path and eval mode

The rename-only path calls cast_with_options(array, to_type, &arrow_cast_options(cast_options.eval_mode)). Since neither child type changes, no value conversion happens and the eval mode should be irrelevant. Is it worth passing DEFAULT_CAST_OPTIONS there instead, with a comment saying the mode cannot matter for a pure relabel? Threading the mode through suggests it does something, and the next person will wonder what.

[Smallfu666]

Thanks again for the review, and sorry for the churn on this branch. I went back through your
comments and reduced the PR to the parts that are directly relevant to the map cast fix.

The unreachable sorted rejection is gone. The target sorted flag is simply preserved, and the
fast path is limited to the rename-only case where neither child type changes. The tests now
cover the sliced fast path, the hand-built path, and a direct comparison between the two.

I also kept the field-count guard because the indexing happens before try_new, so without it
malformed targets can still panic.

While rechecking the surrounding behavior I found that the entries null buffer mentioned in #5097
does not appear to be representable through arrow's valid MapArray construction paths, so the
PR body now documents that rather than claiming to preserve it.

I also found a separate reachable TRY_CAST issue for narrowing integral map keys, filed as
#5995. I am keeping that out of this PR rather than widening the scope again.

The branch is now rebased on current main and the Rust tests, clippy, and fmt are clean. CI is
still action_required, so if you have a chance to approve the workflow run, that should be the
remaining step.