Pin nested-shredded variant base files through the Spark-native internal read path; extend the full-variant rewrite to struct paths if red #19775

Context

Part of #18937. Follow-up to #19689 (nested shredding parity on the Avro write path), which lets both record types write a variant shredded below the top level (a struct member at any depth, under the forced shredding DDL). Split out of #19689 by design: the write parity and the footer-strip recursion land there; the Spark-native INTERNAL read path is pinned and, if needed, extended here.

What is deferred today

Four sites rewrite only TOP-LEVEL variant columns into the full-variant projection shape (struct<"0": variant>, VariantMetadata("$")) for internal reads of parquet base files (compaction, clustering, small-file merge on the SPARK record type, CDC BASE_FILE_INSERT, legacy RDD paths with the file-group reader disabled), and say the nested leg is deferred because no production path wrote a nested-shredded file:

• SparkFileFormatInternalRowReaderContext (fullVariantReadSchemaWithOrdinals + the restore projection, and the comment block above readStructTypeForScan)
• BaseSpark4Adapter.rewriteTopLevelVariantsForFullRead (behind SparkAdapter.buildFullVariantReadSchema, Spark 4.1 / 4.2 adapters)
• HoodieMergeOnReadRDDV2.shouldRerouteVariantSplit
• CDCFileGroupIterator (BASE_FILE_INSERT)

Finding to verify first

By code, Spark 4.1's row reader does not clip a VariantType leaf at a struct depth: ParquetReadSupport.clipParquetType returns the file's shredded group unchanged for VariantType (case _ => parquetType, at any struct depth, and via the list/map converters for array elements and map values), and ParquetRowConverter reconstructs it with ParquetVariantConverter (spark.sql.variant.allowReadingShredded, default true). On the Hudi side SparkSchemaTransformUtils.isDataTypeEqual and HoodieParquetReadSupport.trimParquetSchema keep the group. The existing section F leg of TestVariantShreddingMixedLayouts already reads a row-writer nested-shredded COW base correctly with the vectorized reader on and off. So a MOR compaction / clustering / CDC read over a struct-nested shredded base is most likely correct today, and the deferral comments describe the wrong mechanism: the projection shape is an explicit, Spark-4.0-safe contract, not what prevents clipping. None of that is pinned: there is no MOR or CDC leg over a nested-shredded base anywhere.

What needs to be done

1. Add the legs to TestVariantShreddingMixedLayouts (section F scaffolds: withVariantTable(..., extraCols = "s struct<inner: variant>"), withWriteLayout(Forced("k string")), runCompaction, runClustering, assertVariantLayout(..., column = "s.inner")), Spark 4.1 gate, both record types where the scaffold allows:
   • MOR: nested-shredded base + log update -> snapshot read (vectorized reader swept true/false) -> read-optimized read -> compaction under Forced and under Unshredded (layout re-derived per run, values intact) -> clustering for both row.writer.enable values.
   • CDC (SPARK record type, both hoodie.table.cdc.supplemental.logging.mode values) over a nested-shredded COW base: before/after images of s.inner non-null through hudi_table_changes.
   • variant_get(s.inner, '$.k') projection and filter over the nested-shredded table (nested PushVariantIntoScan shape; the overlay in the reader context is top-level only).
   • One array<variant> column shredded through a declared write schema: snapshot read and one compaction (arrays/maps are read natively; Spark's VariantInRelation rewrites struct paths only).
2. Outcome A (all green): rewrite the four comment blocks to state the real mechanism and keep the rewrite top-level.
3. Outcome B (a leg red): extend rewriteTopLevelVariantsForFullRead to recurse into StructType only (matching Spark's VariantInRelation.StructPathToVariant; arrays/maps stay native), make fullVariantReadSchemaWithOrdinals return ordinal paths (Set[Seq[Int]]), build the restore projection recursively (GetStructField(_, 0, "0") at a variant path, a null-preserving CreateNamedStruct for each enclosing struct), and keep HoodieMergeOnReadRDDV2 and CDCFileGroupIterator on the shared helpers.

Out of scope

Inference below the top level; Hive / Trino / Flink reconstruction (they keep failing fast); Spark 4.0 (its reader cannot rebuild shredded values, adapter returns None).