fix: apply Spark's Parquet conversion rules to nested struct/list/map fields

[peterxcli]

Which issue does this PR close?

Closes #5671.

Rationale for this change

SparkPhysicalExprAdapter applied Spark's SchemaColumnConvertNotSupportedException matrix only to the top-level physical/logical pair. Same-shape struct/list/map pairs were wrapped in CometCastColumnExpr, whose parquet_convert_array cast every nested leaf with a plain Arrow cast_with_options(safe: true) — silently NULLing overflowing INT64 -> int, stringifying ints, parsing strings, wrapping scalars into arrays — and returned the unconverted array for pairs Arrow cannot cast, which then panicked in StructArray::new. Spark's vectorized reader runs ParquetVectorUpdaterFactory.getUpdater on every leaf regardless of nesting, so nested leaves must follow the same rules.

What changes are included in this PR?

• Factor the scalar rule set out of replace_with_spark_cast into check_leaf_conversion (identical conditions and order, so top-level behaviour is unchanged) returning Accept / Reject / RejectOnNonEmpty.
• Add check_conversion, which walks same-shape struct/list/map pairs at any depth, resolves struct fields with the runtime convert's field-id / case-fold rules (new shared match_struct_fields, extracted from parquet_convert_struct_to_struct), applies the rules to every leaf, reports the Spark-style column path (Column: [s, x]) and honours RejectOnNonEmpty (SPARK-26709) like the top level. Missing nested fields still read as null/default.
• Apply the check on both the CastExpr path and the default-adapter fallback path (wrap_all_type_mismatches), which previously enforced only the string/binary rule.
• parquet_convert_array: turn the _ => Ok(array) fallthrough into an error and use try_new for struct/list/map so a mismatch is an error, never a panic.
• spark_catalog_name: render array<..> / struct<..> / map<..> per Spark's catalogString (was unknown).

Behaviour notes for reviewers:

• The fallback path now applies the full rule set to its top-level pair (previously only the string/binary rule).
• A complex-vs-complex shape mismatch (e.g. STRUCT read as ARRAY) and a dictionary-encoded physical column (files carrying an ARROW:schema) now get the Spark-shaped rejection at plan time instead of falling into a Spark Cast.
• The nested case-insensitive duplicate-field error now fires at plan time (previously on the first batch).

How are these changes tested?

• New Rust tests in schema_adapter.rs, run through a real DataSourceExec + SparkPhysicalExprAdapterFactory: the six cases from the issue (INT64→int, int→string, decimal narrowing, string→int, int→array, array→int — the former panic), list-of-struct and map-value leaves, case-insensitive and field-id nested matching, RejectOnNonEmpty for non-empty vs. empty files, and positives (int32→int64 with type promotion, TIMESTAMP_MILLIS→micros inside a list, missing nested field → null); plus a parquet_support.rs test that array<int> -> int inside a struct is an error rather than a panic. cargo test -p datafusion-comet parquet: 112 passed, 0 failed.
• New Scala tests in ParquetReadSuite: native scan rejects nested Parquet conversions Spark rejects (asserts Spark and Comet both throw and Comet's cause chain contains SchemaColumnConvertNotSupportedException for struct/array/map cases), nested schema evolution follows Spark's per-version widening rules, nested TIMESTAMP_MILLIS columns read as timestamp. ParquetReadV1Suite + ParquetTimestampLtzAsNtzSuite on Spark 4.1.3: 68 passed, 0 failed (3 canceled are pre-existing pre-Spark-4 assume gates).
• cargo fmt / cargo clippy --all-targets --workspace -- -D warnings clean; spotless:apply applied.

[sunchao]

Summary

The reader previously applied these Spark conversion restrictions at the top level while nested struct, list and map fields could reach a more permissive conversion path. This patch applies the leaf policy recursively, shares field matching between validation and conversion, and replaces unchecked nested construction with recoverable errors.

Correctness and compatibility

I checked the maintained Spark 3.5 and 4.0 reader sources, including leaf conversion, field-ID precedence, missing fields, case matching and nested shape handling. I also checked the distinction between an empty file and a nonempty file containing empty or null collections. The recursive walk and fallback validation are consistent with those paths. This remains a partial conversion-rejection policy, not a claim that every Arrow conversion matches every Spark version.

Validation and CI

The review's component validation passed 32 base tests, 47 head tests and eight additional real-Parquet probes. These use the production adapter and conversion code with matching dependency versions. They are not a full native-core build, local planner-test run or Spark/JVM execution. The fixture reproduction uses ArrowWriter and DataSourceExec, while the existing planner helper writes through write_parquet.

At 14:36 UTC, CI had 59 successful checks, eight skipped, five running and one failure. The Rust job failed on test_nested_types_list_of_struct_by_index. Its actual checkout is synthetic merge 5338cef07e79101552ae84ca3b0ab8caf4f492d0, with the exact reviewed base/head parents and matching relevant source. The inline P2 describes the fixture change needed to keep that test valid under the stricter rule.

Performance

Validation walks the schema rather than individual rows. Shared struct matching adds an index vector proportional to the requested fields for each converted struct batch, but does not copy child data buffers solely to validate their types. I found no demonstrated P1/P2 performance regression. No throughput or memory improvement was measured, and no benchmark was run.

Design

Using one field resolver for validation and conversion avoids validating one field while reading another after field-ID matching, renaming or reordering. Applying the policy to both normal rewrites and the fallback closes the bypass without introducing a separate conversion framework. The deferred rejection preserves the relevant empty-file behavior.

Abstraction & complexity

One recursive walker and the Accept/Reject/RejectOnNonEmpty distinction fit the policy being enforced. Shared matching removes duplicated selection logic, while checked constructors keep invalid nested arrays on the error path. No additional actionable concern was verified in these areas.

[sunchao]

[P2] Update the nested-list fixture for the stricter leaf check

The new recursion correctly rejects the narrowing in execution::planner::tests::test_nested_types_list_of_struct_by_index, but the fixture still creates nested a with DataFusion SQL's untyped 1 (Int64) and requests Int32. The Rust CI job now fails with [c0, item, a], INT64 -> int.

Using the same SQL and a real Parquet scan, the base adapter succeeds and the head adapter rejects it. An explicitly typed CAST(1 AS INT) succeeds with the head adapter and preserves the intended a/c projection. Could you type the fixture as INT, or change its requested and expected type to Int64? This fixes the deterministic CI failure without weakening the new rejection rule, which agrees with both maintained Spark branches.