Nested (struct/list/map) Parquet schema-evolution conversions bypass Spark's type-conversion rules: silent NULLs, string parsing, and a native panic

[peterxcli]

Describe the bug

SparkPhysicalExprAdapter::replace_with_spark_cast (native/core/src/parquet/schema_adapter.rs) applies Spark's Parquet type-conversion rejection rules (the SchemaColumnConvertNotSupportedException matrix completed in #4229) only to the top-level physical/logical pair. Same-shape complex pairs (struct→struct, list→list, map→map; schema_adapter.rs:950-968) are wrapped in CometCastColumnExpr, whose parquet_convert_array (native/core/src/parquet/parquet_support.rs:234-237) recurses into every nested leaf with a plain Arrow cast_with_options(.., safe: true) whenever can_cast_types is true — and returns the unconverted array otherwise. Spark's vectorized reader runs getUpdater per leaf regardless of nesting, so every conversion Comet rejects at top level is silently performed inside a struct/array/map.

Observed with a test through the real DataSourceExec + adapter (file schema → read schema):

file	read schema	Spark	Comet
struct<x:int64> (5000000000, 1)	struct<x:int>	error	{null}, {1} (overflow silently NULLed)
struct<x:int>	struct<x:string>	error	"1", "2"
struct<d:decimal(10,2)> 123456.78	struct<d:decimal(5,2)>	error	NULL
struct<x:string> ("12","abc")	struct<x:int>	error	12, NULL (string parsed)
struct<x:int>	struct<x:array<int>>	error	[1], [2]
struct<x:array<int>>	struct<x:int>	error	native panic in StructArray::new ("Incorrect datatype for StructArray field "x", expected Int32 got List(Int32)")

The identical top-level pair (x: int64 read as int) is correctly rejected with "Parquet column cannot be converted … Expected: int, Found: INT64".

Steps to reproduce

val p = "/tmp/nested_evo"
Seq((5000000000L, 1)).toDF("x", "y").select(struct($"x").as("s")).write.mode("overwrite").parquet(p)
spark.read.schema("s struct<x:int>").parquet(p).show()

Spark: SchemaColumnConvertNotSupportedException (Column: [s, x], Expected: int, Found: INT64). Comet: prints {null} and {1}.

For the panic: write s struct<x: array<int>> and read it with spark.read.schema("s struct<x:int>").

Expected behavior

Nested leaves follow the same conversion rules as top-level columns: the same rejections as Spark, never a silent NULL/parse, and never a native panic.

Proposed solution

Apply the rejection matrix recursively when building a CometCastColumnExpr for a nested pair: walk the physical vs logical nested types (matching struct fields with the same field-id / case-fold rules as parquet_convert_struct_to_struct) and run the scalar-pair checks (string/binary ↔ non-string, primitive → string, decimal/int → decimal narrowing, long → int, float → int, date → LTZ, complex ↔ scalar, allow_type_promotion gating) on every leaf, returning parquet_schema_convert_err / RejectOnNonEmpty with the dotted column path. Make the _ => Ok(array) fallthrough in parquet_convert_array an error instead of returning a mismatched array, and use StructArray::try_new so a mismatch surfaces as a DataFusion error rather than a panic.

Additional context

Default (and only) V1 native Parquet path; requires a schema mismatch (explicit read schema, or an evolved catalog schema over older files). Related: #4297 / #4229 (top-level rules only), #5553 (nested TIMESTAMP_MILLIS overflow is unchecked for the same reason), #1633 (earlier StructArray::new panic from the nested convert).