peterxcli opened a new pull request, #5407: URL: https://github.com/apache/datafusion-comet/pull/5407
## Which issue does this PR close? This is Phase A of #4295. It enables whole-value projection of top-level Spark 4 `VariantType` columns in ordinary native Parquet scans. It uses Arrow/Parquet's existing whole-value unshredding, but does not take on #3983's Parquet writer, subfield-pruning, or predicate-pushdown scope. Iceberg projection remains a later phase. ## Rationale for this change Spark 4 exposes semi-structured values as the atomic [`VariantType`](https://github.com/apache/spark/blob/c7d67e3f5d4c9d88a480367b44fc54d26adf99ab/sql/api/src/main/scala/org/apache/spark/sql/types/VariantType.scala#L22-L43): ```scala class VariantType private () extends AtomicType case object VariantType extends VariantType ``` The logical type is backed by structured physical storage at the Arrow boundary. Spark's Arrow conversion produces a Struct with non-null Binary children in [`[value, metadata]` order](https://github.com/apache/spark/blob/c7d67e3f5d4c9d88a480367b44fc54d26adf99ab/sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala#L143-L157), and [`ColumnVector.getVariant`](https://github.com/apache/spark/blob/c7d67e3f5d4c9d88a480367b44fc54d26adf99ab/sql/catalyst/src/main/java/org/apache/spark/sql/vectorized/ColumnVector.java#L315-L323) hard-codes that child order: ```java return new VariantVal(getChild(0).getBinary(rowId), getChild(1).getBinary(rowId)); ``` Arrow/Parquet identifies the same logical value with the Field-level extension name [`arrow.parquet.variant`](https://github.com/apache/arrow-rs/blob/0ff81c1215cc026a1de93ce3d2078df1ecba6f09/parquet-variant-compute/src/variant_array.rs#L77-L109), whose storage type is a Struct. A Parquet reader can expose `metadata`, `value`, and optional shredded `typed_value` children in arbitrary order; Arrow-rs resolves [`metadata`](https://github.com/apache/arrow-rs/blob/0ff81c1215cc026a1de93ce3d2078df1ecba6f09/parquet-variant-compute/src/variant_array.rs#L276-L317) and [`value` / `typed_value`](https://github.com/apache/arrow-rs/blob/0ff81c1215cc026a1de93ce3d2078df1ecba6f09/parquet-variant-compute/src/variant_array.rs#L901-L914) by name, while [`unshred_variant`](https://github.com/apache/arrow-rs/blob/0ff81c1215cc026a1de93ce3d2078df1ecba6f09/parquet-variant-compute/src/unshred_variant.rs#L46-L97) reconstructs whole values and removes `typed_value`. Before this PR, that logical identity could not survive Comet's native path: * The protobuf enum ended at `CALENDAR_INTERVAL = 20` and had [no Variant type ID](https://github.com/apache/datafusion-comet/blob/bdd2e13f57c052b2047c2e630bc0a3b1dcd7fe6d/native/proto/src/proto/types.proto#L43-L66), and [`QueryPlanSerde.serializeDataType`](https://github.com/apache/datafusion-comet/blob/bdd2e13f57c052b2047c2e630bc0a3b1dcd7fe6d/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala#L556-L594) could not encode `VariantType`. * The native FFI exporter constructed its schema from [`ArrayData.data_type()`](https://github.com/apache/datafusion-comet/blob/bdd2e13f57c052b2047c2e630bc0a3b1dcd7fe6d/native/core/src/execution/utils.rs#L25-L64): ```rust FFI_ArrowSchema::try_from(self.data_type()) ``` That discarded parent Field metadata. Arrow's [`TryFrom<&DataType>`](https://github.com/apache/arrow-rs/blob/0ff81c1215cc026a1de93ce3d2078df1ecba6f09/arrow-schema/src/ffi.rs#L667-L708) has no Field metadata to export, whereas [`TryFrom<&Field>`](https://github.com/apache/arrow-rs/blob/0ff81c1215cc026a1de93ce3d2078df1ecba6f09/arrow-schema/src/ffi.rs#L798-L816) explicitly forwards `field.metadata()`. * On the JVM side, [`Utils.fromArrowField`](https://github.com/apache/datafusion-comet/blob/bdd2e13f57c052b2047c2e630bc0a3b1dcd7fe6d/spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala#L76-L88) mapped every Arrow Struct to Spark `StructType`, so even a marked Variant Field could not become Spark `VariantType`. Consequently, a query as small as `SELECT v FROM parquet_table` fell back rather than remaining an ordinary Comet native Parquet scan. ## What changes are included in this PR? The resulting path is: ```text Spark VariantType -> Comet protobuf VARIANT -> Arrow Field<Struct[value, metadata], arrow.parquet.variant> -> ordinary Parquet schema adapter -> VariantArray::try_new + unshred_variant -> Struct[value: Binary, metadata: Binary] -> Arrow C Data Interface Field -> CometStructVector (logical Spark VariantType) -> ColumnVector.getVariant ``` ### Preserve logical type identity * Appends protobuf `VARIANT = 21` without renumbering existing values and serializes Spark 4 `VariantType` through the version shim ([proto](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/native/proto/src/proto/types.proto#L43-L67), [Spark serialization](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala#L580-L594)). * Adds a protobuf-to-Arrow Field path that builds physical `Struct<value: Binary, metadata: Binary>` storage and attaches the canonical parent extension marker ([native serde](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/native/core/src/execution/serde.rs#L109-L112), [Field construction](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/native/core/src/execution/serde.rs#L203-L219)). * Spark 4 shims recognize `VariantType` and recursively detect nested uses; Spark 3.x shims remain `false`/`None`, so Spark 3 behavior and compilation remain unchanged ([Spark 4 shim](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala#L50-L70), [Spark 3 shim](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala#L36-L46)). ### Normalize once at the ordinary Parquet boundary The normalization is installed through the existing Parquet schema-adaptation path. The schema adapter pairs the logical target Field with the physical Parquet Field and installs `CometCastColumnExpr`; that expression normalizes the reader StructArray at evaluation. Marked Variant targets take this path even for otherwise-identity casts ([schema adapter](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/native/core/src/parquet/schema_adapter.rs#L581-L617), [identity-cast handling](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/native/core/src/parquet/schema_adapter.rs#L628-L662)). [`normalize_variant_array`](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/native/core/src/parquet/cast_column.rs#L183-L217): 1. constructs Arrow-rs `VariantArray` from the reader Struct; 2. delegates whole-value reconstruction to Arrow-rs `unshred_variant`; 3. converts BinaryView/LargeBinary output to ordinary Binary; 4. rebuilds exactly two children in Spark's required `[value, metadata]` order; and 5. preserves the parent null bitmap while the target Field retains its name, nullability, and extension metadata. This handles both already-unshredded `value + metadata` input and shredded input containing `typed_value`, without adding another Variant dependency or a Comet-owned value decoder. ### Preserve the Field through FFI and bridge it back to Spark The shared FFI exporter now accepts the corresponding RecordBatch Field and builds `FFI_ArrowSchema` from that Field, leaving Arrow array export unchanged ([export helper](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/native/core/src/execution/utils.rs#L25-L61), [batch export](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/native/core/src/execution/jni_api.rs#L689-L738)). Offset-normalized arrays still use the original output Field. On import, [`Utils.fromArrowField`](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala#L72-L96) maps only the explicit `ARROW:extension:name = arrow.parquet.variant` marker to the version-shim Variant type. Plain Structs keep their existing `StructType` behavior. No new vector is needed: the existing [`CometStructVector`](https://github.com/apache/datafusion-comet/blob/bdd2e13f57c052b2047c2e630bc0a3b1dcd7fe6d/spark/src/main/java/org/apache/comet/vector/CometStructVector.java#L33-L59) preserves child ordinals, so Spark's inherited `getVariant` consumes child 0 as value and child 1 as metadata. ### Keep Phase A's fallback boundary explicit Protobuf serialization here is schema transport, not general native Variant-expression support. The ordinary FileSource Parquet scan admits only a direct top-level `VariantType` field ([scan gate](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala#L963-L969)). Direct Variant attributes are rejected by expression serde, and non-scan operators recursively reject Variant-bearing schemas ([expression gate](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/spark/src/main/scala/org/apache/comet/serde/namedExpressions.scala#L37-L44), [operator gate](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala#L732-L742)). The following remain deliberate Spark fallbacks: * `PushVariantIntoScan` / annotated `VariantStruct` output; * `variant_get`, predicates, casts, Variant functions, and Parquet writing; * native columnar-to-row, sort/limit and other native operators, shuffle, and spill; * nested Variant inside ARRAY/MAP/STRUCT; and * Iceberg Variant projection and equality deletes. This builds on #5377: an unread Variant root is still pruned and its scan ordinals are still rebased instead of decoding it. ## How are these changes tested? The focused Spark test covers both unshredded and forced-shredded Parquet input, `SELECT v`, `SELECT id, v, tail`, object/array/scalar values, JSON null, SQL null, nullable parents, native scan retention, fallback from `CometNativeColumnarToRowExec` to JVM `CometColumnarToRowExec`, the logical Spark type, parent Field name/nullability/extension, exact `[value, metadata]` child order, Binary child types, and `getVariant` consumption ([test](https://github.com/peterxcli/datafusion-comet/blob/45a0ed44ed9c58ede31410de077a0882e72fd4f8/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala#L91-L204)). The SQL regression file also checks full-value projection plus the fallback matrix and #5377 pruning behavior. Unshredded input uses Comet's byte-exact Spark answer checker. The shredded comparison is semantic because Spark's builder [range-compresses integral values](https://github.com/apache/spark/blob/c7d67e3f5d4c9d88a480367b44fc54d26adf99ab/common/variant/src/main/java/org/apache/spark/types/variant/VariantBuilder.java#L145-L165), while Arrow-rs selects its unshred builder from [`typed_value`'s Arrow primitive type](https://github.com/apache/arrow-rs/blob/0ff81c1215cc026a1de93ce3d2078df1ecba6f09/parquet-variant-compute/src/unshred_variant.rs#L178-L204) and [appends that typed value](https://github.com/apache/arrow-rs/blob/0ff81c1215cc026a1de93ce3d2078df1ecba6f09/parquet-variant-compute/src/unshred_variant.rs#L420-L451); Spark's [`VariantVal.equals`](https://github.com/apache/spark/blob/c7d67e3f5d4c9d88a480367b44fc54d26adf99ab/common/unsafe/src/main/java/org/apache/spark/unsafe/types/VariantVal.java#L117-L128) compares the raw value and metadata byte arrays. The test separately verifies that Spark consumes the returned bytes through `getVariant` and that the reconstructed JSON values match. Commands run: ```shell make core cd native cargo fmt --all -- --check env DYLD_LIBRARY_PATH="$JAVA_HOME/lib/server" \ cargo test -p datafusion-comet test_normalize_shredded_variant_for_spark -- --nocapture cargo clippy -p datafusion-comet --lib --tests -- -D warnings cd .. mvn -o -ntp -Pspark-4.0 -Dtest=none \ '-Dsuites=org.apache.comet.CometSqlFileTestSuite variant' test mvn -o -ntp -Pspark-4.0 -Dtest=none \ '-Dsuites=org.apache.comet.parquet.ParquetReadV1Suite native scan projects Variant' test mvn -o -ntp -Pspark-4.1 -Dtest=none \ '-Dsuites=org.apache.comet.parquet.ParquetReadV1Suite native scan projects Variant' test mvn -o -ntp -Pspark-4.1 -Dtest=none \ '-Dsuites=org.apache.comet.CometIcebergNativeSuite variant' test mvn -o -ntp -Pspark-4.0 -Dtest=none \ '-Dsuites=org.apache.comet.rules.CometScanRuleSuite,org.apache.comet.rules.CometScanContribSuite' test mvn -o -ntp -Pspark-3.5 -DskipTests test-compile git diff --check upstream/main...HEAD ``` All commands passed. The focused results were 1/1 Rust normalization test, 1/1 Spark 4.0 SQL suite, 1/1 Spark 4.0 vector test, 1/1 Spark 4.1 vector test, 4/4 Spark 4.1 Iceberg fallback tests, and 16/16 Spark 4.0 scan-rule/contrib tests. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
