feat: carry VariantType identity through schema serialization

[peterxcli]

Which issue does this PR close?

Closes #5548.

This is the second atomic change in #5546 and builds on the Field-aware Arrow C Data export merged in #5552. It intentionally does not enable native Variant scans.

Rationale for this change

Spark 4 maps VariantType to non-null value and metadata binary fields, while arrow-rs identifies that physical Struct as the arrow.parquet.variant extension type. Comet needs both representations: the physical Struct for Arrow transport and the Field-level extension marker to retain its logical Variant identity.

Without an explicit protobuf type and Field marker, Comet cannot carry Spark VariantType through schema serialization or recover it when importing an Arrow Field on the JVM. This change establishes that type contract while keeping native scan, expression, and operator admission closed. A later scan PR can therefore transport Variant values without treating an ordinary two-binary-field Struct as Variant.

What changes are included in this PR?

• Append VARIANT = 21 to protobuf DataTypeId without renumbering existing values.
• Serialize Spark 4 VariantType through version-specific shims while keeping Spark 3.x inert.
• Map Variant natively to Struct<value: Binary, metadata: Binary> and attach the canonical Arrow Variant extension marker to its Field.
• Preserve generated extension metadata alongside Spark and Parquet field metadata.
• Recover VariantType in Utils.fromArrowField only for the exact arrow.parquet.variant marker; identical unmarked or differently marked Structs remain StructType.
• Keep direct and nested Variant scans, expressions, and operators on explicit Spark fallback despite making schema serialization available.
• Add focused native, JVM, and SQL regression coverage for identity preservation and fallback boundaries.

This PR does not add Parquet scan admission, Variant unshredding or normalization, existence defaults, Variant functions or casts, native C2R, shuffle/spill, Python transport, writes, Iceberg support, or nested Variant execution. Those remain in later #5546 subtasks.

How are these changes tested?

cargo test -p datafusion-comet variant
cargo fmt --all -- --check
cd ..

make core
mvn -o -ntp -Pspark-4.0 -Dtest=none \
  '-Dsuites=org.apache.comet.CometVariantTypeSuite' test
mvn -o -ntp -Pspark-4.0 -Dtest=none \
  '-Dsuites=org.apache.comet.CometSqlFileTestSuite variant' test
mvn -o -ntp -Pspark-3.5 -Dtest=none \
  '-Dsuites=org.apache.comet.CometVariantTypeSuite' test

[sunchao]

Reviewed bb5ed016e86875372d7c5de76dc629a9662039bd against e98a2fbbc7dc383c9f9bdd529bff15e2d2beaf2a. No verified P1/P2 findings.

The explicit protobuf type and canonical Arrow Field marker keep Variant distinct from an ordinary two-binary-field struct. I traced nested metadata preservation and the scan, expression, operator, shuffle, and row-conversion fallback gates. Existing protobuf values remain unchanged, and the new tests check identity and explicit fallback reasons.

Validation was source review plus static compatibility checks. I did not run the Comet/Spark suites or a transport roundtrip. CodeQL and the Delta build gate passed. CI is still in progress.

[andygrove]

Nice work keeping this atomic. The explicit protobuf type plus the canonical Arrow marker does cleanly separate a logical Variant from an ordinary two-binary-field struct, and the SQL fixture already covers pruning, nesting, partition offset rebasing, and the lookalike struct, which is good to see.

I have a few things I would like to sort out before this goes in. Two of them sit on lines this PR does not touch, so I am putting them here rather than inline.

The first is CometNativeScan.scala:188. That comment justifies the nativeDataSchema pruning with "serializing it would throw", which was true when serializeDataType returned None for Variant and schema2Proto called .get on it. After this PR it no longer throws. Could you update the comment? The pruning below it is now the only thing keeping a VARIANT out of the native reader's dataSchema, and it would be good for the comment to say that rather than describe behavior that has changed.

The second is NativeColumnarToRowConverter.scala:54. That used to throw UnsupportedOperationException for anything serializeDataType could not handle, which included Variant. Now it will happily initialize native C2R with a Struct<value, metadata>. I think the operator gate keeps this unreachable today and the description lists native C2R as out of scope, so I am not worried about a live bug, but it would be good to either keep an explicit Variant rejection there or add a comment saying the operator gate is what guarantees it.

That second one generalizes, and it is really my main concern with the shape of the change. This PR turns serializeDataType from deny-by-default into allow-for-Variant across roughly forty call sites, then re-adds denial in two places. I walked the rest and I believe they are covered, casts by CometCast.isSupported, literals by supportedDataType in CometLiteral.getSupportLevel, and contraintExpressions.scala:34 transitively through the new attribute check. But would you mind adding a note on serializeDataType recording the new invariant, something to the effect that a successful serialization no longer implies native support and that containsVariantType is the gate? Right now a future contributor reading a serializeDataType(...).isDefined check has no way to know that.

[andygrove]

Could you say more about why CometScanExec is exempted here? As far as I can tell CometScanRule only produces a CometScanExec after CometScanTypeChecker has approved every output type, and this PR's own test asserts that checker rejects VariantType and ArrayType(VariantType), so I could not find a plan where the exemption changes anything.

What makes me want to pin this down is that the same change removes the other guard on that path. CometNativeScan.convert gates on scanTypes.length == scan.output.length, and before this PR a Variant output made that length check fail and the scan decline with "unsupported Comet operator ... due to unsupported data types above". Now that serializeDataType succeeds for Variant, that check can never fire for Variant either. So CometScanExec ends up being the one operator where both the old implicit guard and the new explicit guard disappear in the same commit.

If there is a real case behind the exemption, could we get a comment and a test that fails without it? If there is not, I would rather drop it. An unexplained hole in a deny gate tends to become load-bearing later. It also reads oddly next to CometBatchScanExec on the Iceberg path, which is not exempted.

[andygrove]

Arrow Java already defines this key as ArrowType.ExtensionType.EXTENSION_METADATA_KEY_NAME. Could we use that instead of a local literal? One less string that has to stay in sync by hand.

[andygrove]

The marker name is now a literal in two places that have to agree, here and Utils.VariantExtensionName, and nothing in the PR ties either of them to arrow-rs. has_valid_extension_type::<VariantType>() will not catch drift either, because arrow-rs's supports_data_type accepts any Struct. So if VariantType::NAME ever changes, all four tests in this PR still pass while the real path quietly degrades to StructType. That silent degrade is the exact failure this PR exists to prevent, so it would be good to have something that fails loudly instead.

Could the Rust test assert VariantType::NAME == "arrow.parquet.variant" directly? That is cheap and pins the upstream side. Beyond that, is a test that actually crosses the boundary feasible here, serializing a Spark VariantType schema through QueryPlanSerde.serializeDataType, running it through native, and asserting Utils.fromArrowField gives back VariantType? I realize no operator can carry Variant yet so this may need a narrow test hook, but the JVM to native contract is the whole point of the PR, and right now each half is only tested against a hand-written copy of the other half's assumptions.

For what it is worth I did check that the pieces line up today. arrow-rs 58.4.0 has NAME = "arrow.parquet.variant" in parquet-variant-compute/src/variant_array.rs, and Arrow Java 18.3.0's SchemaImporter.importField keeps unregistered extension metadata on the imported field rather than stripping it, so the marker should survive C Data import.