sunchao commented on code in PR #5407:
URL: https://github.com/apache/datafusion-comet/pull/5407#discussion_r3830165353
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -963,8 +963,10 @@ case class CometScanRule(session: SparkSession)
private def isSchemaSupported(scanExec: FileSourceScanExec, r:
HadoopFsRelation): Boolean = {
val fallbackReasons = new ListBuffer[String]()
val typeChecker = CometScanTypeChecker()
- val schemaSupported =
- typeChecker.isSchemaSupported(scanExec.requiredSchema, fallbackReasons)
+ val schemaSupported = scanExec.requiredSchema.fields.forall { field =>
+ isVariantType(field.dataType) ||
+ typeChecker.isTypeSupported(field.dataType, field.name, fallbackReasons)
Review Comment:
[P2] Reject unsupported Variant defaults before admitting the scan
Could we keep this scan on Spark when a required Variant field has a
non-null existence default? `CometNativeScan` drops failed default
serializations with `flatMap`, but retains every default index, and
`CometLiteral` still rejects Variant. I reproduced this on Spark 4.0.4:
```sql
CREATE TABLE t(v VARIANT DEFAULT parse_json('1')) USING parquet;
INSERT INTO t VALUES (parse_json('42'));
ALTER TABLE t ADD COLUMNS(n INT DEFAULT 7);
SELECT v, n FROM t;
```
Spark returns `(42, 7)`, while this head's native scan returns `(42, NULL)`.
The remaining default `7` is zipped to index 0 (`v`), where it is ignored
because that column exists physically, leaving `n` without its default. Please
reject an unserializable default or preserve and validate each value/index pair
before enabling the scan.
##########
native/core/src/parquet/cast_column.rs:
##########
@@ -176,6 +180,42 @@ fn cast_timestamp_micros_to_millis_scalar(
ScalarValue::TimestampMillisecond(new_val, target_tz)
}
+fn normalize_variant_array(
+ array: &ArrayRef,
+ target_field: &FieldRef,
+) -> DataFusionResult<ArrayRef> {
+ let DataType::Struct(fields) = target_field.data_type() else {
+ return Err(DataFusionError::Execution(
+ "Variant extension field must use Struct storage".to_string(),
+ ));
+ };
+ if fields.len() != 2
+ || fields[0].name() != "value"
+ || fields[1].name() != "metadata"
+ || fields
+ .iter()
+ .any(|field| field.data_type() != &DataType::Binary)
+ {
+ return Err(DataFusionError::Execution(
+ "Variant output must contain Binary children [value,
metadata]".to_string(),
+ ));
+ }
+
+ let variant = VariantArray::try_new(array.as_ref())?;
+ let unshredded = unshred_variant(&variant)?;
Review Comment:
[P2] Preserve Spark lookup compatibility when rebuilding Unicode objects
Could we account for Spark's object-key ordering before returning these
reconstructed bytes? Arrow sorts object keys in UTF-8 order, but the supported
Spark versions use Java `String.compareTo` and switch to binary search at 32
fields. I wrote a shredded Parquet object with Spark containing `k00` through
`k29`, `U+E000`, and `😀`. With `pushVariantIntoScan=false` and
`allowReadingShredded=true`, `variant_get(v, '$.😀', 'int')` returns `531` on
Spark but `NULL` with this native scan. The expression itself correctly falls
back to Spark, but it consumes the incompatible reconstructed ordering. The
byte-level mismatch also reproduces on Spark 4.1.3. Please normalize for the
Spark consumer or retain fallback for affected values, with a 32-key Unicode
regression.
##########
native/core/src/execution/utils.rs:
##########
@@ -56,7 +57,7 @@ impl SparkArrowConvert for ArrayData {
);
unsafe {
std::ptr::write(array_ptr, FFI_ArrowArray::new(self));
- std::ptr::write(schema_ptr,
FFI_ArrowSchema::try_from(self.data_type())?);
+ std::ptr::write(schema_ptr, FFI_ArrowSchema::try_from(field)?);
Review Comment:
[P2] Handle NUL-containing field names before C schema export
Could we preserve the Field metadata without passing an embedded-NUL name to
Arrow's C-string exporter? Spark accepts a top-level Parquet column named
`v\u0000suffix`. I wrote and read that ordinary BIGINT column successfully with
Spark 4.0.4, but the native scan at this head fails with `NulError`. Arrow
58.4.0's `FFI_ArrowSchema::try_from(field)` calls
`CString::new(field.name()).unwrap()`, whereas the previous datatype-only
export did not serialize the parent name. This affects non-Variant columns too,
and the unaligned branch has the same issue. Please use a safe exported name or
an explicit pre-execution fallback while retaining the logical metadata.
##########
native/core/src/parquet/cast_column.rs:
##########
@@ -176,6 +180,42 @@ fn cast_timestamp_micros_to_millis_scalar(
ScalarValue::TimestampMillisecond(new_val, target_tz)
}
+fn normalize_variant_array(
+ array: &ArrayRef,
+ target_field: &FieldRef,
+) -> DataFusionResult<ArrayRef> {
+ let DataType::Struct(fields) = target_field.data_type() else {
+ return Err(DataFusionError::Execution(
+ "Variant extension field must use Struct storage".to_string(),
+ ));
+ };
+ if fields.len() != 2
+ || fields[0].name() != "value"
+ || fields[1].name() != "metadata"
+ || fields
+ .iter()
+ .any(|field| field.data_type() != &DataType::Binary)
+ {
+ return Err(DataFusionError::Execution(
+ "Variant output must contain Binary children [value,
metadata]".to_string(),
+ ));
+ }
+
+ let variant = VariantArray::try_new(array.as_ref())?;
Review Comment:
[P2] Decode dictionary metadata before constructing VariantArray
Could we decode dictionary-encoded metadata before this call? The [canonical
Arrow Variant
representation](https://arrow.apache.org/docs/format/CanonicalExtensions.html#parquet-variant)
permits it, and an Arrow-written Parquet file can retain `metadata:
Dictionary(Int32, Binary)` in its embedded `ARROW:schema` while storing
ordinary required BINARY children physically. I reproduced a file containing
`42, 43, 44`: Spark 4.0.4 reads it successfully, but this head's native scan
throws `Illegal shredded value type: Dictionary(Int32, Binary)`. Arrow/Parquet
58.4.0 restores the nested dictionary, which `VariantArray::try_new` rejects,
so the Binary cast below is never reached. Decoding the metadata child first
makes the same values readable.
##########
spark/src/test/resources/sql-tests/expressions/misc/variant.sql:
##########
@@ -43,8 +45,21 @@ SELECT tail FROM test_variant ORDER BY id
query
SELECT id, tail FROM test_variant WHERE tail IS NOT NULL ORDER BY id
+-- Full-value projection is scan-only: no native expression or pass-through
operator carries v.
+query
+SELECT v FROM test_variant
Review Comment:
[P2] Pin Variant pushdown for the native projection SQL assertions
Could we set `spark.sql.variant.pushVariantIntoScan=false` for these
native-projection cases, as the new Scala vector test does? Spark 4.1 and 4.2
enable that optimizer rule by default, so even `SELECT v` becomes the annotated
VariantStruct representation that this PR deliberately keeps on Spark. The
plain `query` assertion then requires a native plan that cannot be produced.
This is the actual failure in both [Spark 4.1
CI](https://github.com/apache/datafusion-comet/actions/runs/32476998515/job/96761947281)
and [Spark 4.2
CI](https://github.com/apache/datafusion-comet/actions/runs/32476998515/job/96761947274):
`variant.sql:50` fails with `Expected only Comet native operators, but found
Project`. Please configure the whole-value test path explicitly and keep a
separate fallback assertion for pushed VariantStruct.
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -731,6 +732,15 @@ case class CometExecRule(session: SparkSession)
private def tryConvertToComet(
op: SparkPlan,
handler: CometOperatorSerde[_]): Option[SparkPlan] = {
+ if (!op.isInstanceOf[CometScanExec] &&
+ (op.output ++ op.children.flatMap(_.output)).exists(attr =>
+ containsVariantType(attr.dataType))) {
Review Comment:
[P2] Check the write input beneath WriteFilesExec
Could this guard inspect the same unwrapped data-producing children used by
`requiresNativeChildren` below? For
`DataWritingCommandExec(WriteFilesExec(CometNativeScan[Variant]))`, both the
command output and `WriteFilesExec.output` are empty, so the Variant check
misses the input. With `spark.comet.parquet.write.enabled=true` and
`spark.comet.operator.DataWritingCommandExec.allowIncompatible=true`, copying a
Spark-written Variant Parquet column now selects `CometNativeWriteExec` and
fails in `CometArrowStream.inputObjects -> Utils.toArrowSchema` with
`Unsupported data type: ... VariantType ... variant`. The intended Spark write
fallback succeeds. Please apply the Variant check after unwrapping
`WriteFilesExec` so this scan-only change does not enable the unsupported
writer.
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -731,6 +732,15 @@ case class CometExecRule(session: SparkSession)
private def tryConvertToComet(
op: SparkPlan,
handler: CometOperatorSerde[_]): Option[SparkPlan] = {
+ if (!op.isInstanceOf[CometScanExec] &&
+ (op.output ++ op.children.flatMap(_.output)).exists(attr =>
+ containsVariantType(attr.dataType))) {
+ withFallbackReason(
+ op,
+ "Native operators do not support schemas containing type VariantType")
+ return None
Review Comment:
[P2] Apply the Variant fallback to the later Python rewrite too
Could we apply this boundary to
`EliminateRedundantTransitions.EligibleMapInBatch` as well? That later rule
creates `CometMapInBatchExec` without passing through this guard. I reproduced
a native Variant scan followed by `df.mapInPandas(lambda batches: batches,
df.schema)`: it succeeds with `spark.comet.exec.pyarrowUDF.enabled=false`, but
fails with the flag enabled. The accelerated runner forwards the new Arrow
schema, which lacks Spark's `variant=true` metadata on the `metadata` child.
Spark's Pandas serializer therefore supplies a `dict` rather than `VariantVal`,
and the identity result fails `assert isinstance(variant, VariantVal)` during
output conversion. Keeping Variant-bearing inputs on the ordinary Spark Python
path would preserve the intended fallback.
--
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]
