sunchao commented on code in PR #5407:
URL: https://github.com/apache/datafusion-comet/pull/5407#discussion_r3869971570
##########
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) ||
Review Comment:
[P2] Preserve Unicode column matching when admitting Variant scans
This newly routes Variant columns through the native adapter's existing
ASCII-only case-insensitive matcher. With `spark.sql.caseSensitive=false`,
`allowReadingShredded=true` and pushdown disabled, I wrote a Spark Parquet
column named `É` containing Variant `42`, then read it with the explicit schema
`é VARIANT`. Both Spark 4.0.4 readers return `42`; this head selects one
`CometNativeScanExec` but returns SQL NULL because the field is treated as
missing. `Σ`/`σ` reproduces too, while exact-name, ASCII-case and
case-sensitive controls agree. The matcher predates this PR, but these Variant
reads previously stayed on Spark. Please preserve Spark's Unicode name
resolution, or keep affected Variant scans on the fallback path.
##########
native/core/src/parquet/cast_column/variant.rs:
##########
@@ -0,0 +1,1414 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use arrow::{
+ array::{
+ make_array, Array, ArrayRef, AsArray, BinaryArray, BinaryBuilder,
ListLikeArray,
+ StructArray,
+ },
+ buffer::NullBuffer,
+ compute::{cast, cast_with_options},
+ datatypes::{DataType, FieldRef, TimeUnit},
+ error::ArrowError,
+};
+use datafusion::common::{
+ format::DEFAULT_CAST_OPTIONS, DataFusionError, Result as DataFusionResult,
+};
+use parquet::variant::{
+ unshred_variant, BorrowedShreddingState, ListBuilder, MetadataBuilder,
ObjectBuilder,
+ ParentState, ReadOnlyMetadataBuilder, ValueBuilder, Variant, VariantArray,
VariantBuilder,
+ VariantDecimal4, VariantDecimal8, VariantMetadata, WritableMetadataBuilder,
+};
+use std::{
+ collections::HashSet,
+ panic::{catch_unwind, AssertUnwindSafe},
+ sync::Arc,
+};
+
+pub(super) 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 array = normalize_variant_storage(array)?;
+ let variant = VariantArray::try_new(array.as_ref())?;
+ let was_shredded = variant.typed_value_field().is_some();
+ let unshredded = unshred_variant_for_spark(&variant)?;
+ let value = unshredded.value_field().ok_or_else(|| {
+ DataFusionError::Execution("Unshredded Variant is missing its value
field".to_string())
+ })?;
+ let value = cast(value.as_ref(), &DataType::Binary)?;
+ let metadata = cast(unshredded.metadata_field().as_ref(),
&DataType::Binary)?;
+ let (value, metadata) = if was_shredded {
+ rebuild_shredded_variant_for_spark(&variant, &value, &metadata,
unshredded.inner().nulls())?
+ } else {
+ let value = reorder_variant_values(
+ &value,
+ &metadata,
+ unshredded.inner().nulls(),
+ VariantObjectKeyOrder::SparkUtf16,
+ false,
+ )?;
+ (value, metadata)
+ };
+ let output = StructArray::try_new(
+ fields.clone(),
+ vec![value, metadata],
+ unshredded.inner().nulls().cloned(),
+ )?;
+ Ok(Arc::new(output))
+}
+
+fn unshred_variant_for_spark(variant: &VariantArray) ->
DataFusionResult<VariantArray> {
+ let first_error = match unshred_variant(variant) {
+ Ok(array) => return Ok(array),
+ Err(error) => DataFusionError::from(error),
+ };
+
+ if let Ok(prepared) = prepare_variant_for_unshredding(variant, None) {
+ if let Ok(array) = unshred_variant(&prepared) {
+ return Ok(array);
+ }
+ }
+ let Some(metadata) = canonicalize_spark_empty_key_metadata(variant)? else {
+ return Err(first_error);
+ };
+ let Ok(prepared) = prepare_variant_for_unshredding(variant,
Some(metadata.as_binary::<i32>()))
+ else {
+ return Err(first_error);
+ };
+ match unshred_variant(&prepared) {
+ Ok(array) => Ok(array),
+ Err(_) => Err(first_error),
+ }
+}
+
+fn normalize_variant_type(data_type: &DataType) -> Option<DataType> {
+ fn normalize_field(field: &FieldRef) -> Option<FieldRef> {
+ normalize_variant_type(field.data_type())
+ .map(|data_type|
Arc::new(field.as_ref().clone().with_data_type(data_type)))
+ }
+
+ match data_type {
+ DataType::Dictionary(_, value_type) => {
+ Some(normalize_variant_type(value_type).unwrap_or_else(||
value_type.as_ref().clone()))
+ }
+ DataType::UInt8 => Some(DataType::Int16),
+ DataType::UInt16 => Some(DataType::Int32),
+ DataType::UInt32 => Some(DataType::Int64),
+ // Spark reads Parquet UINT_64 as Decimal(20, 0). This is lossless for
the full range and
+ // lets the existing Spark-compatible rebuild choose the Variant
decimal width per value.
+ DataType::UInt64 => Some(DataType::Decimal128(20, 0)),
+ DataType::Timestamp(TimeUnit::Millisecond, timezone) => {
+ Some(DataType::Timestamp(TimeUnit::Microsecond, timezone.clone()))
+ }
+ DataType::FixedSizeBinary(_) => Some(DataType::Binary),
+ DataType::FixedSizeList(field, _) => Some(DataType::List(
+ normalize_field(field).unwrap_or_else(|| Arc::clone(field)),
+ )),
+ DataType::List(field) => normalize_field(field).map(DataType::List),
+ DataType::LargeList(field) =>
normalize_field(field).map(DataType::LargeList),
+ DataType::ListView(field) =>
normalize_field(field).map(DataType::ListView),
+ DataType::LargeListView(field) =>
normalize_field(field).map(DataType::LargeListView),
+ DataType::Struct(fields) => {
+ let mut changed = false;
+ let fields = fields
+ .iter()
+ .map(|field| match normalize_field(field) {
+ Some(field) => {
+ changed = true;
+ field
+ }
+ None => Arc::clone(field),
+ })
+ .collect::<Vec<_>>();
+ changed.then(|| DataType::Struct(fields.into()))
+ }
+ _ => None,
Review Comment:
[P2] Normalize retained Arrow leaf types before Variant validation
`ARROW:schema` can restore Spark-readable shredded leaves as types this
match leaves untouched. With `allowReadingShredded=true` and pushdown disabled,
an ArrowWriter 58.4 file retaining `typed_value: Decimal256(38,2)` (unscaled
value `123`) has ordinary physical `DECIMAL(38,2)`: both Spark 4.0.4 readers
return `1.23`, but the asserted native scan fails with `Illegal shredded value
type: Decimal256(38, 2)`. Nested decimal reproduces; Decimal128 and
identical-physical/no-Arrow-hint controls pass. Retained Date64 also fails, for
both its plain-INT64 and DATE physical encodings. Please preserve Spark's
physical interpretation before constructing `VariantArray`, or retain fallback.
Simply casting Date64 to Date32 would be wrong for the INT64 encoding, which
Spark interprets as BIGINT.
--
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]
