sunchao commented on code in PR #5407:
URL: https://github.com/apache/datafusion-comet/pull/5407#discussion_r3865538958
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala:
##########
@@ -102,6 +118,21 @@ object CometNativeScan extends
CometOperatorSerde[CometScanExec] with CometTypeS
withFallbackReason(scanExec, "Full native scan disabled because
ignoreMissingFiles enabled")
}
+ if (serializeExistenceDefaultValues(scanExec.requiredSchema,
scanExec.output).isEmpty) {
+ withFallbackReason(scanExec, unsupportedDefaultReason)
+ }
+
+ // Spark's strict mode validates the legacy two-field Variant layout and
reports SPARK-47546
+ // errors itself: https://issues.apache.org/jira/browse/SPARK-47546
+ if (scanExec.requiredSchema.fields.exists(field =>
isVariantType(field.dataType)) &&
+ !SQLConf.get
+ .getConfString("spark.sql.variant.allowReadingShredded", "true")
Review Comment:
[P2] Honor Spark 4.0's registered strict-reader default
The two-argument `getConfString` returns its supplied default when the
setting is unset, but Spark 4.0 registers
`spark.sql.variant.allowReadingShredded` as **false**. Consequently, the
ordinary unset case bypasses this fallback even though Spark's strict reader is
active. I reproduced this on Spark 4.0.4 with a Spark-written shredded Variant
file: the effective setting is false and Spark rejects it with
`INVALID_VARIANT_FROM_PARQUET.WRONG_NUM_FIELDS`, while this head selects
`CometNativeScanExec` and returns the value. Explicit false correctly falls
back, so the new test misses the default case. Please read the registered
default through the version shim (or the no-default lookup for versions with
Variant) and cover unset as well as explicit false/true.
##########
native/core/src/parquet/cast_column/variant.rs:
##########
@@ -0,0 +1,1122 @@
+// 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::{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 = decode_variant_metadata_dictionary(array)?;
+ let array = normalize_variant_typed_value(&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 =
+ prepare_variant_for_unshredding(variant).and_then(|array|
Ok(unshred_variant(&array)?));
+ let first_error = match first {
+ Ok(array) => return Ok(array),
+ Err(error) => error,
+ };
+ let Some(variant) = canonicalize_spark_empty_key_metadata(variant)? else {
+ return Err(first_error);
+ };
+ let variant = prepare_variant_for_unshredding(&variant)?;
+ Ok(unshred_variant(&variant)?)
+}
+
+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::UInt8 => Some(DataType::Int16),
+ DataType::UInt16 => Some(DataType::Int32),
+ DataType::UInt32 => Some(DataType::Int64),
+ DataType::Timestamp(TimeUnit::Millisecond, timezone) => {
Review Comment:
[P2] Preserve integer semantics for nanosAsLong Variant children
This normalizer leaves nanosecond timestamps unchanged, but with
`spark.sql.legacy.parquet.nanosAsLong=true`, Spark's vectorized Parquet reader
interprets those children as integers. On Spark 4.0.4, with
`spark.sql.variant.allowReadingShredded=true`, a whole-Variant read of
`typed_value: INT64 TIMESTAMP(NANOS,true)` containing `1704067200123000000`
returns that integer without Comet; this head, with `CometNativeScanExec`
asserted in the executed plan, instead returns a timestamp (`"2024-01-01
00:00:00.123+00:00"` when cast to STRING). The fixture has no embedded
`ARROW:schema`. A non-microsecond-aligned value, `1704067200123456789`, instead
fails with `UNKNOWN_PRIMITIVE_TYPE_IN_VARIANT` (type 18; the NTZ case produces
19). Please preserve the configured raw-Int64 interpretation before Variant
construction, or fall back for these inputs. Merely converting nanos to micros
would still change the Variant value type. The passing baseline here is Spark's
vectorized reader; its row reade
r rejects the fixture.
--
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]
