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

[Smallfu666]

Which issue does this PR close?

Related to #5097. Intentionally not Closes, see the scope note below: the "entries null buffer"
part of the issue is not representable through safe Arrow APIs.

Rationale for this change

cast_map_to_map had two bugs:

• It used the source sorted flag for the result, so a cast to a map type with a different
  sorted flag returned the wrong type.
• It rebuilt the entries, key and value fields from scratch, dropping field metadata and target
  nullability, so data_type() did not equal the requested target.

What changes are included in this PR?

• Rename-only fast path: when 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. It relabels to
  the target fields and preserves their metadata with no value transformation.
• Hand-built path: when a child type changes, recurse with Comet's cast_array, and build the
  result with the target sorted flag rather than the source one. try_new replaces new so a
  malformed target returns Err instead of panicking, and the result data_type() equals the
  requested target.
• Field-count guard before indexing entries [0] and [1], so a malformed entries struct with
  0, 1 or 3 or more fields returns Err rather than panicking.
• Cast options: both call sites now build their CastOptions through a shared
  arrow_cast_options(eval_mode) helper. This is behaviorally inert today. Arrow 58.4's map cast
  delegates to cast_with_options for the key and value arrays, and this 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 shared helper still prevents this call site from drifting into
  a hardcoded safe: true if the fast-path condition is ever widened.

Scope note. Within the inspected safe Arrow constructors, an all-valid entries validity buffer is
normalized to None, while actual null entries are invalid for a map. Entries-level null-buffer
preservation is therefore not observable through those construction paths and is not claimed here.
ArrayData and FFI construction were not audited. Map-level null preservation is covered.

How are these changes tested?

16 Rust unit tests in cast.rs, covering:

• the rename-only fast path, asserting the full target schema and unchanged values, offsets and map
  nulls
• sorted flags equal, true to false, and a value-only cast
• metadata preservation together with child type casts
• empty maps, mixed null and empty rows, and sliced maps through both paths
• key and value casts
• malformed targets with 0, 1 or 3 entry fields, and a non-nullable field carrying a null, all
  returning Err without panicking

Two are new from review. test_cast_map_to_map_sliced_rename_only_fast_path covers a sliced input
whose value type is unchanged, so it exercises the arrow delegation rather than the hand-built path.
test_cast_map_to_map_both_paths_agree sends equivalent input through both implementations, compares
the results, and asserts the shared result is correct so the two cannot agree on a wrong answer and
still pass. Which branch each test takes was verified with a temporary probe rather than inferred
from the types.

Spark differential coverage in cast_complex.sql, since Rust assertions only prove agreement
with my own expectations. The new cases cover map value casts, a map key cast, null maps, empty maps
and null values, plus the DATATYPE_MISMATCH that a key cast which could introduce nulls should
raise.

Results on cb36aa5fe, Spark 3.5 and Scala 2.12:

• cargo test --workspace: 890 passed, 0 failed.
• CometSqlFileTestSuite: 444 run, 444 succeeded, 0 failed.
• CometNativeCastSuite: 167 run, 167 succeeded, 0 failed, 1 canceled.
• cargo fmt --check and cargo clippy --all-targets -- -D warnings clean.

Production negative control: reverting only the cast_map_to_map implementation to main while
retaining the 16 tests gives 8 passed and 8 failed. The failures cover the target sorted flag, field
metadata and schema preservation, and the malformed-target panic cases.

SQL harness negative control: adding cases to an existing SQL file does not move the suite count,
so the new map-cast expectation was broken on its own. CometSqlFileTestSuite then reports 443
succeeded and 1 failed, confirming the added cast_complex.sql case is active and independently
checked.

[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.