You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "0lai0 (via GitHub)" <gi...@apache.org> on 2026/09/14 19:06:50 UTC
[PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
0lai0 opened a new pull request, #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932
## Which issue does this PR close?
Closes #5741.
## Rationale for this change
Reading a VARIANT-annotated Parquet field as an ordinary `struct<value binary, metadata binary>` silently returned the storage bytes. Spark rejects that read in `ParquetToSparkSchemaConverter.convertGroupField` with `_LEGACY_ERROR_TEMP_3071` unless `spark.sql.parquet.ignoreVariantAnnotation` is set.
`CometScanRule` cannot catch it: it only sees the requested schema, and a hand-written struct carries none of the `VariantMetadata` that `isVariantStruct` looks for. The annotation lives in the file, so the check has to happen in the native reader.
## What changes are included in this PR?
arrow-rs surfaces the annotation as the `arrow.parquet.variant` Arrow extension type, and Comet's serde marks a requested `VariantType` the same way. `check_variant_annotation` compares the two sides symmetrically in `SparkPhysicalExprAdapterFactory::create`, so a marked request stays a legitimate Variant read and only an unmarked request against an annotated file is rejected.
Matching on `value`/`metadata` child names instead would misclassify ordinary structs, which #5741 rules out.
- `spark.sql.parquet.ignoreVariantAnnotation` plumbed through `NativeScanCommon`, read by key since the conf is 4.1-only while this file compiles against 3.4 through 4.1.
- Gated to Spark 4.1+. The converter branch and the conf both arrived in 4.1, so on 3.4, 3.5 and 4.0 the check is off rather than inventing a failure Spark does not have.
- Recurses through struct fields, list elements and map values, pairing nested fields by field id the way `spark_parquet_convert` does.
- Runs at file open, before any row group, matching Spark rejecting during schema conversion. An empty file fails too.
- `SparkError::ParquetVariantAnnotationMismatch` converted by the 4.x shim into Spark's `AnalysisException` wrapped in `FAILED_READ_FILE`.
- `dev/diffs/4.1.3.diff` regenerated per `spark-sql-tests.md` to drop the `IgnoreComet` exclusion.
Known gap, documented on `is_variant_marked`: the extension type carries no spec version, so with `ignoreVariantAnnotation=true` on a non-v1 annotation Comet reads a file Spark refuses. Both engines reject it when the conf is off, with different error classes.
## How are these changes tested?
`ParquetVariantShreddingSuite / variant logical type annotation - ignore variant annotation` is no longer excluded. I ran the suite locally against Spark 4.1.3 with `ENABLE_COMET=true`: 7 tests pass, none ignored. With the check disabled it fails with `Expected exception org.apache.spark.SparkException to be thrown, but no exception was thrown`, the failure #5741
reports, so the suite is exercising this change.
15 native tests in `schema_adapter.rs` cover rejection at each nesting shape, a real `VARIANT(1)` file written through the low-level writer, `ignoreVariantAnnotation=true`, a marked Variant request not being rejected, an empty file, nested field-id resolution, and probes asserting the annotation still reaches `create` as an extension type.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on code in PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#discussion_r4283982416
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -4643,6 +4884,930 @@ pub(crate) mod test {
assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap());
}
+ /// Verification probe for apache/datafusion-comet#5741.
+ ///
+ /// Captures the `physical_file_schema` DataFusion hands to
+ /// `PhysicalExprAdapterFactory::create`, so the tests below can assert both that `create`
+ /// was reached at all and what the file schema looked like when it was.
+ #[derive(Debug)]
+ struct ProbeFactory {
+ inner: SparkPhysicalExprAdapterFactory,
+ seen: Arc<Mutex<Option<SchemaRef>>>,
+ }
+
+ impl PhysicalExprAdapterFactory for ProbeFactory {
+ fn create(
+ &self,
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ ) -> datafusion::common::Result<
+ Arc<dyn datafusion_physical_expr_adapter::PhysicalExprAdapter>,
+ > {
+ *self.seen.lock().unwrap() = Some(Arc::clone(&physical_file_schema));
+ self.inner.create(logical_file_schema, physical_file_schema)
+ }
+ }
+
+ /// Recursively drop every Arrow extension marker from `field`, producing the schema a Spark
+ /// user's hand-written DDL yields: the same structure, no variant identity. Mirrors the
+ /// `.schema("v struct<value binary, metadata binary>")` in Spark's
+ /// `ParquetVariantShreddingSuite`, test
+ /// `variant logical type annotation - ignore variant annotation`.
+ fn strip_extension_markers(field: &FieldRef) -> FieldRef {
+ let data_type = match field.data_type() {
+ DataType::Struct(fields) => {
+ DataType::Struct(fields.iter().map(strip_extension_markers).collect())
+ }
+ DataType::List(item) => DataType::List(strip_extension_markers(item)),
+ DataType::Map(entries, sorted) => {
+ DataType::Map(strip_extension_markers(entries), *sorted)
+ }
+ other => other.clone(),
+ };
+ let mut metadata = field.metadata().clone();
+ metadata.remove("ARROW:extension:name");
+ metadata.remove("ARROW:extension:metadata");
+ Arc::new(Field::new(field.name(), data_type, field.is_nullable()).with_metadata(metadata))
+ }
+
+ /// Write `batch` to a temp Parquet file, then scan it back through `DataSourceExec` asking for
+ /// the same schema with every extension marker stripped. Reports the Parquet schema the writer
+ /// actually produced alongside the physical file schema `create` observed (`None` when `create`
+ /// never ran).
+ ///
+ /// Returning the Parquet schema separates the two ways a shape can fail: the *writer* never
+ /// emitting the VARIANT annotation, versus the *reader* not surfacing it as an Arrow extension
+ /// type. Only the second one is a problem for #5741; the first is a limitation of the test.
+ ///
+ /// `skip_arrow_metadata` controls whether the file embeds an `ARROW:schema` key-value entry.
+ /// Spark's writer embeds none, so the Spark-shaped cases pass `true`: a marker on the way back
+ /// can then only have come from the Parquet annotation, and a file carrying the hint would let
+ /// every assertion below pass vacuously.
+ struct VariantProbe {
+ /// The Parquet schema the writer actually produced.
+ parquet_schema: String,
+ /// The physical file schema `create` observed, or `None` when it never ran.
+ observed_physical: Option<SchemaRef>,
+ /// Total rows read, or the error the scan raised.
+ scan: Result<usize, DataFusionError>,
+ }
+
+ struct ProbeOptions {
+ /// Whether the file omits the embedded `ARROW:schema` hint, as Spark's writer does.
+ skip_arrow_metadata: bool,
+ /// `spark.sql.parquet.ignoreVariantAnnotation`.
+ ignore_variant_annotation: bool,
+ /// Keep the Variant markers on the requested schema, modelling a genuine native Variant
+ /// projection (`projects_variant` in `parquet_exec.rs`) instead of a hand-written struct.
+ keep_request_markers: bool,
+ /// Request this schema instead of one derived from the file schema, for cases where
+ /// Spark's requested names differ from the names the file was written with.
+ requested_schema: Option<SchemaRef>,
+ }
+
+ impl Default for ProbeOptions {
+ fn default() -> Self {
+ Self {
+ skip_arrow_metadata: true,
+ ignore_variant_annotation: false,
+ keep_request_markers: false,
+ requested_schema: None,
+ }
+ }
+ }
+
+ async fn probe_variant_annotation(
+ batch: RecordBatch,
+ options: ProbeOptions,
+ ) -> Result<VariantProbe, DataFusionError> {
+ let file_schema = batch.schema();
+ let filename = get_temp_filename();
+ let filename = filename.as_path().as_os_str().to_str().unwrap().to_string();
+ let file = File::create(&filename)?;
+ let mut writer = ArrowWriter::try_new_with_options(
+ file,
+ Arc::clone(&file_schema),
+ ArrowWriterOptions::new().with_skip_arrow_metadata(options.skip_arrow_metadata),
+ )?;
+ writer.write(&batch)?;
+ writer.close()?;
+
+ probe_variant_annotation_file(&filename, &file_schema, options).await
+ }
+
+ /// The reading half of `probe_variant_annotation`: reads back the Parquet schema `filename`
+ /// actually holds, derives the requested schema from `file_schema` per `options`, and scans
+ /// the file through a `ProbeFactory`. Split out so a test can hand it a file the `ArrowWriter`
+ /// alone cannot produce, such as one whose `ARROW:schema` hint disagrees with its annotations.
+ async fn probe_variant_annotation_file(
+ filename: &str,
+ file_schema: &SchemaRef,
+ options: ProbeOptions,
+ ) -> Result<VariantProbe, DataFusionError> {
+ let filename = filename.to_string();
+ let file_schema = Arc::clone(file_schema);
+
+ let mut printed = Vec::new();
+ let reader = ParquetRecordBatchReaderBuilder::try_new(File::open(&filename)?)?;
+ print_schema(&mut printed, reader.parquet_schema().root_schema());
+ let printed = String::from_utf8(printed).unwrap();
+
+ let requested = if let Some(requested) = options.requested_schema {
+ requested
+ } else if options.keep_request_markers {
+ Arc::clone(&file_schema)
+ } else {
+ Arc::new(Schema::new(
+ file_schema
+ .fields()
+ .iter()
+ .map(strip_extension_markers)
+ .collect::<Fields>(),
+ ))
+ };
+
+ let mut spark_parquet_options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ spark_parquet_options.ignore_variant_annotation = options.ignore_variant_annotation;
+ let seen = Arc::new(Mutex::new(None));
+ let expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory> = Arc::new(ProbeFactory {
+ inner: SparkPhysicalExprAdapterFactory::new(spark_parquet_options, None),
+ seen: Arc::clone(&seen),
+ });
+
+ // Install the reader factory `init_datasource_exec` installs, gated the way it gates it:
+ // the footer rewrites that decide which fields the reader marks as Variant live there, so
+ // a probe without it would read a hint the real scan never sees.
+ let projects_variant = requested
+ .fields()
+ .iter()
+ .any(|field| field.has_valid_extension_type::<VariantType>());
+ let runtime = Arc::new(RuntimeEnv::default());
+ let store = runtime.object_store(ObjectStoreUrl::local_filesystem())?;
+ let reader_factory = Arc::new(
+ EagerPageIndexReaderFactory::new(
+ store,
+ runtime.cache_manager.get_file_metadata_cache(),
+ ScanIoSource::Local,
+ &ExecutionPlanMetricsSet::new(),
+ )
+ .with_spark_variant_schema(projects_variant),
+ );
+ let parquet_source =
+ ParquetSource::new(requested).with_parquet_file_reader_factory(reader_factory);
+ let files = FileGroup::new(vec![PartitionedFile::from_path(filename)?]);
+ let file_scan_config = FileScanConfigBuilder::new(
+ ObjectStoreUrl::local_filesystem(),
+ Arc::new(parquet_source),
+ )
+ .with_file_groups(vec![files])
+ .with_expr_adapter(Some(expr_adapter_factory))
+ .build();
+ let parquet_exec = DataSourceExec::new(Arc::new(file_scan_config));
+ let scan = async {
+ let mut stream = parquet_exec.execute(0, Arc::new(TaskContext::default()))?;
+ let mut rows = 0;
+ while let Some(batch) = stream.next().await {
+ rows += batch?.num_rows();
+ }
+ Ok(rows)
+ }
+ .await;
+
+ let observed_physical = seen.lock().unwrap().clone();
+ Ok(VariantProbe {
+ parquet_schema: printed,
+ observed_physical,
+ scan,
+ })
+ }
+
+ /// The two-field Variant storage group Spark writes when shredding is off.
+ fn variant_storage_fields() -> Fields {
+ Fields::from(vec![
+ Field::new("value", DataType::Binary, false),
+ Field::new("metadata", DataType::Binary, false),
+ ])
+ }
+
+ /// A `StructArray` holding one Variant value: the encoded integer `1`, matching the
+ /// `Row(Array[Byte](12, 1), Array[Byte](1, 0, 0))` the Spark suite expects.
+ fn variant_storage_array() -> StructArray {
+ let value = Arc::new(BinaryArray::from(vec![Some(&[12u8, 1u8][..])])) as ArrayRef;
+ let meta = Arc::new(BinaryArray::from(vec![Some(&[1u8, 0u8, 0u8][..])])) as ArrayRef;
+ StructArray::new(variant_storage_fields(), vec![value, meta], None)
+ }
+
+ /// An Arrow field named `name` carrying the Variant storage type and the
+ /// `arrow.parquet.variant` extension marker, which parquet-rs's writer turns into the Parquet
+ /// VARIANT logical type annotation.
+ fn variant_field(name: &str) -> FieldRef {
+ Arc::new(
+ Field::new(name, DataType::Struct(variant_storage_fields()), true)
+ .with_extension_type(VariantType),
+ )
+ }
+
+ /// Assert that `probe_variant_annotation` saw a live `create` call on a file the writer really
+ /// annotated, then hand the observed physical schema to `locate` to check the marker survived
+ /// at the shape-specific position.
+ fn assert_annotation_survived(
+ shape: &str,
+ probe: &VariantProbe,
+ locate: impl Fn(&SchemaRef) -> Option<FieldRef>,
+ ) {
+ let printed = &probe.parquet_schema;
+ assert!(
+ printed.contains("VARIANT"),
+ "{shape}: parquet-rs did not write the VARIANT annotation, so this file cannot test \
+ the reader at all. Parquet schema:\n{printed}"
+ );
+ let observed = probe.observed_physical.clone().unwrap_or_else(|| {
+ panic!(
+ "{shape}: PhysicalExprAdapterFactory::create was never called: the logical and \
+ physical file schemas compared equal, so the variant annotation did not survive \
+ into the physical schema"
+ )
+ });
+ let field = locate(&observed).unwrap_or_else(|| {
+ panic!("{shape}: could not locate the variant field in {observed:?}")
+ });
+ assert!(
+ field.has_valid_extension_type::<VariantType>(),
+ "{shape}: physical file schema lost the variant annotation: {field:?}"
+ );
+ }
+
+ /// One row of `map<string, variant>`, the `mv` column of the Spark suite, with the map value
+ /// child named `value_name` in the file.
+ fn map_of_variant_batch(value_name: &str) -> Result<RecordBatch, DataFusionError> {
+ let value_field = variant_field(value_name);
+ let entry_fields = Fields::from(vec![
+ Arc::new(Field::new("key", DataType::Utf8, false)),
+ Arc::clone(&value_field),
+ ]);
+ let entries = StructArray::new(
+ entry_fields.clone(),
+ vec![
+ Arc::new(StringArray::from(vec!["v2"])) as ArrayRef,
+ Arc::new(variant_storage_array()) as ArrayRef,
+ ],
+ None,
+ );
+ let entries_field = Arc::new(Field::new("entries", DataType::Struct(entry_fields), false));
+ let map = MapArray::new(
+ Arc::clone(&entries_field),
+ OffsetBuffer::new(vec![0, 1].into()),
+ entries,
+ None,
+ false,
+ );
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "mv",
+ DataType::Map(entries_field, false),
+ true,
+ )]));
+ Ok(RecordBatch::try_new(schema, vec![Arc::new(map)])?)
+ }
+
+ /// Assert the scan was rejected with Spark's `_LEGACY_ERROR_TEMP_3071` shape, naming `column`
+ /// (a dotted path) and the requested type.
+ fn assert_variant_annotation_rejected(probe: &VariantProbe, column: &str, spark_type: &str) {
+ let err = probe
+ .scan
+ .as_ref()
+ .expect_err("reading a VARIANT-annotated field as a plain struct must be rejected");
+ let msg = err.to_string();
+ assert!(
+ msg.contains("_LEGACY_ERROR_TEMP_3071")
+ && msg.contains("Invalid Spark read type")
+ && msg.contains(column)
+ && msg.contains(spark_type),
+ "unexpected error for {column}: {msg}"
+ );
+ }
+
+ /// The plain struct a hand-written Spark read schema produces for a Variant column.
+ fn plain_variant_storage_sql() -> &'static str {
+ "STRUCT<value: BINARY, metadata: BINARY>"
+ }
+
+ /// #5741: reading a VARIANT-annotated top-level field as
+ /// `struct<value binary, metadata binary>` must raise Spark's `_LEGACY_ERROR_TEMP_3071`
+ /// instead of silently returning the storage
+ /// struct. This is the `v` column of the Spark suite's `false` arm.
+ #[tokio::test]
+ async fn variant_annotation_read_as_plain_struct_is_rejected() -> Result<(), DataFusionError> {
+ let schema = Arc::new(Schema::new(vec![variant_field("v")]));
+ let batch = RecordBatch::try_new(schema, vec![Arc::new(variant_storage_array())])?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ assert_variant_annotation_rejected(&probe, "v", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// The `ns struct<nv variant>` case: the annotation is on a nested field, and the rejection
+ /// must name the dotted path rather than just the root column.
+ #[tokio::test]
+ async fn variant_annotation_inside_struct_is_rejected() -> Result<(), DataFusionError> {
+ let outer_fields = Fields::from(vec![variant_field("nv")]);
+ let outer = StructArray::new(
+ outer_fields.clone(),
+ vec![Arc::new(variant_storage_array()) as ArrayRef],
+ None,
+ );
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "ns",
+ DataType::Struct(outer_fields),
+ true,
+ )]));
+ let batch = RecordBatch::try_new(schema, vec![Arc::new(outer)])?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ assert_variant_annotation_rejected(&probe, "ns.nv", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// The `av array<variant>` case.
+ #[tokio::test]
+ async fn variant_annotation_as_list_element_is_rejected() -> Result<(), DataFusionError> {
+ let element = variant_field("element");
+ let list = ListArray::new(
+ Arc::clone(&element),
+ OffsetBuffer::new(vec![0, 1].into()),
+ Arc::new(variant_storage_array()),
+ None,
+ );
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "av",
+ DataType::List(element),
+ true,
+ )]));
+ let batch = RecordBatch::try_new(schema, vec![Arc::new(list)])?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ assert_variant_annotation_rejected(&probe, "av.element", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// The `mv map<string, variant>` case.
+ #[tokio::test]
+ async fn variant_annotation_as_map_value_is_rejected() -> Result<(), DataFusionError> {
+ let batch = map_of_variant_batch("value")?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ // The path walks through the Parquet map's `entries` group, as Spark's own error does.
+ assert_variant_annotation_rejected(&probe, "mv.entries.value", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// In field-id read mode a nested logical field resolves to the file field with the matching
+ /// ID, not the matching name -- that is what `spark_parquet_convert` does when it builds the
+ /// read. Pairing by name here instead would inspect the wrong file field (or none) and let the
+ /// annotation through unchecked, so the rejection must follow the ID.
+ #[test]
+ fn variant_annotation_is_found_through_a_nested_field_id_match() {
+ let storage = DataType::Struct(variant_storage_fields());
+ let logical = Arc::new(Schema::new(vec![Field::new(
+ "ns",
+ DataType::Struct(Fields::from(vec![Field::new(
+ "renamed_since_write",
+ storage.clone(),
+ true,
+ )
+ .with_metadata(id_meta("7"))])),
+ true,
+ )]));
+ // The file holds the Variant under a different name, plus a decoy that would win a
+ // name-only match and carries no annotation.
+ let physical = Arc::new(Schema::new(vec![Field::new(
+ "ns",
+ DataType::Struct(Fields::from(vec![
+ Field::new("renamed_since_write", DataType::Int32, true)
+ .with_metadata(id_meta("9")),
+ Field::new("nv", storage, true)
+ .with_metadata(id_meta("7"))
+ .with_extension_type(VariantType),
+ ])),
+ true,
+ )]));
+
+ let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ options.use_field_id = true;
+ let err = SparkPhysicalExprAdapterFactory::new(options, None)
+ .create(logical, physical)
+ .expect_err("an ID-resolved Variant field must still be rejected");
+ let msg = err.to_string();
+ assert!(
+ msg.contains("_LEGACY_ERROR_TEMP_3071") && msg.contains("ns.renamed_since_write"),
+ "unexpected error: {msg}"
+ );
+ }
+
+ /// Write a one-row Parquet file holding a single top-level group annotated as VARIANT with
+ /// `spec_version`, using the low-level writer so the annotation is exactly what Spark's writer
+ /// emits. `ArrowWriter` cannot do this: `logical_type_for_struct` hardcodes
+ /// `LogicalType::variant(None)`, which is a different annotation from Spark's `VARIANT(1)`.
+ fn write_annotated_variant_file(spec_version: Option<i8>) -> String {
+ let value = Arc::new(
+ ParquetType::primitive_type_builder("value", PhysicalType::BYTE_ARRAY)
+ .with_repetition(Repetition::REQUIRED)
+ .build()
+ .unwrap(),
+ );
+ let metadata = Arc::new(
+ ParquetType::primitive_type_builder("metadata", PhysicalType::BYTE_ARRAY)
+ .with_repetition(Repetition::REQUIRED)
+ .build()
+ .unwrap(),
+ );
+ let variant = Arc::new(
+ ParquetType::group_type_builder("v")
+ .with_repetition(Repetition::REQUIRED)
+ .with_logical_type(Some(LogicalType::variant(spec_version)))
+ .with_fields(vec![value, metadata])
+ .build()
+ .unwrap(),
+ );
+ let schema = Arc::new(
+ ParquetType::group_type_builder("schema")
+ .with_fields(vec![variant])
+ .build()
+ .unwrap(),
+ );
+
+ let filename = get_temp_filename();
+ let filename = filename.as_path().as_os_str().to_str().unwrap().to_string();
+ let file = File::create(&filename).unwrap();
+ let mut writer = SerializedFileWriter::new(file, schema, Default::default()).unwrap();
+ let mut row_group = writer.next_row_group().unwrap();
+ for bytes in [vec![12u8, 1u8], vec![1u8, 0u8, 0u8]] {
+ let mut column = row_group.next_column().unwrap().unwrap();
+ column
+ .typed::<ByteArrayType>()
+ .write_batch(&[ByteArray::from(bytes)], None, None)
+ .unwrap();
+ column.close().unwrap();
+ }
+ row_group.close().unwrap();
+ writer.close().unwrap();
+ filename
+ }
+
+ /// Scan `filename` asking for the plain `struct<value binary, metadata binary>` a hand-written
+ /// Spark read schema produces, and report the scan outcome.
+ async fn scan_as_plain_variant_storage(filename: String) -> Result<usize, DataFusionError> {
+ let requested = Arc::new(Schema::new(vec![Field::new(
+ "v",
+ DataType::Struct(variant_storage_fields()),
+ false,
+ )]));
+ let expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory> =
+ Arc::new(SparkPhysicalExprAdapterFactory::new(
+ SparkParquetOptions::new(EvalMode::Legacy, "UTC", false),
+ None,
+ ));
+ let file_scan_config = FileScanConfigBuilder::new(
+ ObjectStoreUrl::local_filesystem(),
+ Arc::new(ParquetSource::new(requested)),
+ )
+ .with_file_groups(vec![FileGroup::new(vec![PartitionedFile::from_path(
+ filename,
+ )?])])
+ .with_expr_adapter(Some(expr_adapter_factory))
+ .build();
+ let exec = DataSourceExec::new(Arc::new(file_scan_config));
+ let mut stream = exec.execute(0, Arc::new(TaskContext::default()))?;
+ let mut rows = 0;
+ while let Some(batch) = stream.next().await {
+ rows += batch?.num_rows();
+ }
+ Ok(rows)
+ }
+
+ /// The annotation Spark itself writes is `VARIANT(1)`, and every other rejection test in this
+ /// module goes through `ArrowWriter`, which emits `VARIANT(None)` instead. This one writes the
+ /// real Spark shape so the rejection is pinned against the annotation users actually have on
+ /// disk, not only against the one the test writer happens to produce.
+ #[tokio::test]
+ async fn variant_annotation_with_spec_version_1_is_rejected() -> Result<(), DataFusionError> {
+ let err = scan_as_plain_variant_storage(write_annotated_variant_file(Some(1)))
+ .await
+ .expect_err("a VARIANT(1) field read as a plain struct must be rejected");
+ let msg = err.to_string();
+ assert!(
+ msg.contains("_LEGACY_ERROR_TEMP_3071") && msg.contains("Invalid Spark read type"),
+ "unexpected error: {msg}"
+ );
+ Ok(())
+ }
+
+ /// An annotation with no spec version, which is what an Arrow-based writer emits and
+ /// parquet-java reads back as 0. Spark rejects this too, but through its
+ /// `unrecognizedParquetTypeError` catch-all rather than the variant branch, so the error class
+ /// differs: `PARQUET_TYPE_NOT_RECOGNIZED` there against `_LEGACY_ERROR_TEMP_3071` here. The
+ /// `arrow.parquet.variant` extension type carries no version, so this check cannot tell the
+ /// file apart from `VARIANT(1)`. See `check_variant_annotation` for the full comparison. If a
+ /// future arrow-rs exposes the version, this test is the one to revisit.
+ #[tokio::test]
+ async fn variant_annotation_without_a_spec_version_is_also_rejected(
+ ) -> Result<(), DataFusionError> {
+ let err = scan_as_plain_variant_storage(write_annotated_variant_file(None))
+ .await
+ .expect_err("Comet rejects an unversioned VARIANT annotation that Spark would accept");
Review Comment:
Both fixed. The message now reads "as Spark does", and the doc keeps only the sentence about reading the spec version.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on code in PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#discussion_r4283998992
##########
native/core/src/parquet/eager_page_index_reader_factory.rs:
##########
@@ -646,7 +802,7 @@ impl AsyncFileReader for EagerPageIndexReader {
let metadata = if spark_variant_schema {
with_spark_arrow_schema(metadata)?
} else {
- metadata
+ with_reconciled_variant_markers(metadata)?
Review Comment:
Ohh good catch, and it is worse than you put it: `ignoreVariantAnnotation` is `!isSpark41Plus || conf`, so every scan below Spark 4.1 paid for a rewrite whose only consumer never runs. Gated behind `with_reconcile_variant_markers`, set from `!ignore_variant_annotation`. Nothing else reads the physical side's marker, and `the_footer_is_left_alone_when_the_annotation_check_is_off` pins the skip.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#issuecomment-5848629083
No problem, @sunchao. Merged main
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#discussion_r4277235651
##########
native/core/src/parquet/eager_page_index_reader_factory.rs:
##########
@@ -646,7 +802,7 @@ impl AsyncFileReader for EagerPageIndexReader {
let metadata = if spark_variant_schema {
with_spark_arrow_schema(metadata)?
} else {
- metadata
+ with_reconciled_variant_markers(metadata)?
Review Comment:
This runs on every open of a file that carries an `ARROW:schema` hint, whether or not the variant check is enabled. Each run does two `parquet_to_arrow_schema` conversions and decodes the hint, and the metadata cache keeps the original footer, so a cached file repeats the work on every open. The reconciled marker is only read by `check_variant_annotation`, and `CometParquetUtils.ignoreVariantAnnotation` is `true` below Spark 4.1. Comet's own writer, pyarrow and other arrow-rs writers emit the hint. Could this be skipped when `ignore_variant_annotation` is set? The options are already on the factory through `with_conversion_check`. I have not measured the cost.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -4643,6 +4884,930 @@ pub(crate) mod test {
assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap());
}
+ /// Verification probe for apache/datafusion-comet#5741.
+ ///
+ /// Captures the `physical_file_schema` DataFusion hands to
+ /// `PhysicalExprAdapterFactory::create`, so the tests below can assert both that `create`
+ /// was reached at all and what the file schema looked like when it was.
+ #[derive(Debug)]
+ struct ProbeFactory {
+ inner: SparkPhysicalExprAdapterFactory,
+ seen: Arc<Mutex<Option<SchemaRef>>>,
+ }
+
+ impl PhysicalExprAdapterFactory for ProbeFactory {
+ fn create(
+ &self,
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ ) -> datafusion::common::Result<
+ Arc<dyn datafusion_physical_expr_adapter::PhysicalExprAdapter>,
+ > {
+ *self.seen.lock().unwrap() = Some(Arc::clone(&physical_file_schema));
+ self.inner.create(logical_file_schema, physical_file_schema)
+ }
+ }
+
+ /// Recursively drop every Arrow extension marker from `field`, producing the schema a Spark
+ /// user's hand-written DDL yields: the same structure, no variant identity. Mirrors the
+ /// `.schema("v struct<value binary, metadata binary>")` in Spark's
+ /// `ParquetVariantShreddingSuite`, test
+ /// `variant logical type annotation - ignore variant annotation`.
+ fn strip_extension_markers(field: &FieldRef) -> FieldRef {
+ let data_type = match field.data_type() {
+ DataType::Struct(fields) => {
+ DataType::Struct(fields.iter().map(strip_extension_markers).collect())
+ }
+ DataType::List(item) => DataType::List(strip_extension_markers(item)),
+ DataType::Map(entries, sorted) => {
+ DataType::Map(strip_extension_markers(entries), *sorted)
+ }
+ other => other.clone(),
+ };
+ let mut metadata = field.metadata().clone();
+ metadata.remove("ARROW:extension:name");
+ metadata.remove("ARROW:extension:metadata");
+ Arc::new(Field::new(field.name(), data_type, field.is_nullable()).with_metadata(metadata))
+ }
+
+ /// Write `batch` to a temp Parquet file, then scan it back through `DataSourceExec` asking for
+ /// the same schema with every extension marker stripped. Reports the Parquet schema the writer
+ /// actually produced alongside the physical file schema `create` observed (`None` when `create`
+ /// never ran).
+ ///
+ /// Returning the Parquet schema separates the two ways a shape can fail: the *writer* never
+ /// emitting the VARIANT annotation, versus the *reader* not surfacing it as an Arrow extension
+ /// type. Only the second one is a problem for #5741; the first is a limitation of the test.
+ ///
+ /// `skip_arrow_metadata` controls whether the file embeds an `ARROW:schema` key-value entry.
+ /// Spark's writer embeds none, so the Spark-shaped cases pass `true`: a marker on the way back
+ /// can then only have come from the Parquet annotation, and a file carrying the hint would let
+ /// every assertion below pass vacuously.
+ struct VariantProbe {
+ /// The Parquet schema the writer actually produced.
+ parquet_schema: String,
+ /// The physical file schema `create` observed, or `None` when it never ran.
+ observed_physical: Option<SchemaRef>,
+ /// Total rows read, or the error the scan raised.
+ scan: Result<usize, DataFusionError>,
+ }
+
+ struct ProbeOptions {
+ /// Whether the file omits the embedded `ARROW:schema` hint, as Spark's writer does.
+ skip_arrow_metadata: bool,
+ /// `spark.sql.parquet.ignoreVariantAnnotation`.
+ ignore_variant_annotation: bool,
+ /// Keep the Variant markers on the requested schema, modelling a genuine native Variant
+ /// projection (`projects_variant` in `parquet_exec.rs`) instead of a hand-written struct.
+ keep_request_markers: bool,
+ /// Request this schema instead of one derived from the file schema, for cases where
+ /// Spark's requested names differ from the names the file was written with.
+ requested_schema: Option<SchemaRef>,
+ }
+
+ impl Default for ProbeOptions {
+ fn default() -> Self {
+ Self {
+ skip_arrow_metadata: true,
+ ignore_variant_annotation: false,
+ keep_request_markers: false,
+ requested_schema: None,
+ }
+ }
+ }
+
+ async fn probe_variant_annotation(
+ batch: RecordBatch,
+ options: ProbeOptions,
+ ) -> Result<VariantProbe, DataFusionError> {
+ let file_schema = batch.schema();
+ let filename = get_temp_filename();
+ let filename = filename.as_path().as_os_str().to_str().unwrap().to_string();
+ let file = File::create(&filename)?;
+ let mut writer = ArrowWriter::try_new_with_options(
+ file,
+ Arc::clone(&file_schema),
+ ArrowWriterOptions::new().with_skip_arrow_metadata(options.skip_arrow_metadata),
+ )?;
+ writer.write(&batch)?;
+ writer.close()?;
+
+ probe_variant_annotation_file(&filename, &file_schema, options).await
+ }
+
+ /// The reading half of `probe_variant_annotation`: reads back the Parquet schema `filename`
+ /// actually holds, derives the requested schema from `file_schema` per `options`, and scans
+ /// the file through a `ProbeFactory`. Split out so a test can hand it a file the `ArrowWriter`
+ /// alone cannot produce, such as one whose `ARROW:schema` hint disagrees with its annotations.
+ async fn probe_variant_annotation_file(
+ filename: &str,
+ file_schema: &SchemaRef,
+ options: ProbeOptions,
+ ) -> Result<VariantProbe, DataFusionError> {
+ let filename = filename.to_string();
+ let file_schema = Arc::clone(file_schema);
+
+ let mut printed = Vec::new();
+ let reader = ParquetRecordBatchReaderBuilder::try_new(File::open(&filename)?)?;
+ print_schema(&mut printed, reader.parquet_schema().root_schema());
+ let printed = String::from_utf8(printed).unwrap();
+
+ let requested = if let Some(requested) = options.requested_schema {
+ requested
+ } else if options.keep_request_markers {
+ Arc::clone(&file_schema)
+ } else {
+ Arc::new(Schema::new(
+ file_schema
+ .fields()
+ .iter()
+ .map(strip_extension_markers)
+ .collect::<Fields>(),
+ ))
+ };
+
+ let mut spark_parquet_options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ spark_parquet_options.ignore_variant_annotation = options.ignore_variant_annotation;
+ let seen = Arc::new(Mutex::new(None));
+ let expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory> = Arc::new(ProbeFactory {
+ inner: SparkPhysicalExprAdapterFactory::new(spark_parquet_options, None),
+ seen: Arc::clone(&seen),
+ });
+
+ // Install the reader factory `init_datasource_exec` installs, gated the way it gates it:
+ // the footer rewrites that decide which fields the reader marks as Variant live there, so
+ // a probe without it would read a hint the real scan never sees.
+ let projects_variant = requested
+ .fields()
+ .iter()
+ .any(|field| field.has_valid_extension_type::<VariantType>());
+ let runtime = Arc::new(RuntimeEnv::default());
+ let store = runtime.object_store(ObjectStoreUrl::local_filesystem())?;
+ let reader_factory = Arc::new(
+ EagerPageIndexReaderFactory::new(
+ store,
+ runtime.cache_manager.get_file_metadata_cache(),
+ ScanIoSource::Local,
+ &ExecutionPlanMetricsSet::new(),
+ )
+ .with_spark_variant_schema(projects_variant),
+ );
+ let parquet_source =
+ ParquetSource::new(requested).with_parquet_file_reader_factory(reader_factory);
+ let files = FileGroup::new(vec![PartitionedFile::from_path(filename)?]);
+ let file_scan_config = FileScanConfigBuilder::new(
+ ObjectStoreUrl::local_filesystem(),
+ Arc::new(parquet_source),
+ )
+ .with_file_groups(vec![files])
+ .with_expr_adapter(Some(expr_adapter_factory))
+ .build();
+ let parquet_exec = DataSourceExec::new(Arc::new(file_scan_config));
+ let scan = async {
+ let mut stream = parquet_exec.execute(0, Arc::new(TaskContext::default()))?;
+ let mut rows = 0;
+ while let Some(batch) = stream.next().await {
+ rows += batch?.num_rows();
+ }
+ Ok(rows)
+ }
+ .await;
+
+ let observed_physical = seen.lock().unwrap().clone();
+ Ok(VariantProbe {
+ parquet_schema: printed,
+ observed_physical,
+ scan,
+ })
+ }
+
+ /// The two-field Variant storage group Spark writes when shredding is off.
+ fn variant_storage_fields() -> Fields {
+ Fields::from(vec![
+ Field::new("value", DataType::Binary, false),
+ Field::new("metadata", DataType::Binary, false),
+ ])
+ }
+
+ /// A `StructArray` holding one Variant value: the encoded integer `1`, matching the
+ /// `Row(Array[Byte](12, 1), Array[Byte](1, 0, 0))` the Spark suite expects.
+ fn variant_storage_array() -> StructArray {
+ let value = Arc::new(BinaryArray::from(vec![Some(&[12u8, 1u8][..])])) as ArrayRef;
+ let meta = Arc::new(BinaryArray::from(vec![Some(&[1u8, 0u8, 0u8][..])])) as ArrayRef;
+ StructArray::new(variant_storage_fields(), vec![value, meta], None)
+ }
+
+ /// An Arrow field named `name` carrying the Variant storage type and the
+ /// `arrow.parquet.variant` extension marker, which parquet-rs's writer turns into the Parquet
+ /// VARIANT logical type annotation.
+ fn variant_field(name: &str) -> FieldRef {
+ Arc::new(
+ Field::new(name, DataType::Struct(variant_storage_fields()), true)
+ .with_extension_type(VariantType),
+ )
+ }
+
+ /// Assert that `probe_variant_annotation` saw a live `create` call on a file the writer really
+ /// annotated, then hand the observed physical schema to `locate` to check the marker survived
+ /// at the shape-specific position.
+ fn assert_annotation_survived(
+ shape: &str,
+ probe: &VariantProbe,
+ locate: impl Fn(&SchemaRef) -> Option<FieldRef>,
+ ) {
+ let printed = &probe.parquet_schema;
+ assert!(
+ printed.contains("VARIANT"),
+ "{shape}: parquet-rs did not write the VARIANT annotation, so this file cannot test \
+ the reader at all. Parquet schema:\n{printed}"
+ );
+ let observed = probe.observed_physical.clone().unwrap_or_else(|| {
+ panic!(
+ "{shape}: PhysicalExprAdapterFactory::create was never called: the logical and \
+ physical file schemas compared equal, so the variant annotation did not survive \
+ into the physical schema"
+ )
+ });
+ let field = locate(&observed).unwrap_or_else(|| {
+ panic!("{shape}: could not locate the variant field in {observed:?}")
+ });
+ assert!(
+ field.has_valid_extension_type::<VariantType>(),
+ "{shape}: physical file schema lost the variant annotation: {field:?}"
+ );
+ }
+
+ /// One row of `map<string, variant>`, the `mv` column of the Spark suite, with the map value
+ /// child named `value_name` in the file.
+ fn map_of_variant_batch(value_name: &str) -> Result<RecordBatch, DataFusionError> {
+ let value_field = variant_field(value_name);
+ let entry_fields = Fields::from(vec![
+ Arc::new(Field::new("key", DataType::Utf8, false)),
+ Arc::clone(&value_field),
+ ]);
+ let entries = StructArray::new(
+ entry_fields.clone(),
+ vec![
+ Arc::new(StringArray::from(vec!["v2"])) as ArrayRef,
+ Arc::new(variant_storage_array()) as ArrayRef,
+ ],
+ None,
+ );
+ let entries_field = Arc::new(Field::new("entries", DataType::Struct(entry_fields), false));
+ let map = MapArray::new(
+ Arc::clone(&entries_field),
+ OffsetBuffer::new(vec![0, 1].into()),
+ entries,
+ None,
+ false,
+ );
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "mv",
+ DataType::Map(entries_field, false),
+ true,
+ )]));
+ Ok(RecordBatch::try_new(schema, vec![Arc::new(map)])?)
+ }
+
+ /// Assert the scan was rejected with Spark's `_LEGACY_ERROR_TEMP_3071` shape, naming `column`
+ /// (a dotted path) and the requested type.
+ fn assert_variant_annotation_rejected(probe: &VariantProbe, column: &str, spark_type: &str) {
+ let err = probe
+ .scan
+ .as_ref()
+ .expect_err("reading a VARIANT-annotated field as a plain struct must be rejected");
+ let msg = err.to_string();
+ assert!(
+ msg.contains("_LEGACY_ERROR_TEMP_3071")
+ && msg.contains("Invalid Spark read type")
+ && msg.contains(column)
+ && msg.contains(spark_type),
+ "unexpected error for {column}: {msg}"
+ );
+ }
+
+ /// The plain struct a hand-written Spark read schema produces for a Variant column.
+ fn plain_variant_storage_sql() -> &'static str {
+ "STRUCT<value: BINARY, metadata: BINARY>"
+ }
+
+ /// #5741: reading a VARIANT-annotated top-level field as
+ /// `struct<value binary, metadata binary>` must raise Spark's `_LEGACY_ERROR_TEMP_3071`
+ /// instead of silently returning the storage
+ /// struct. This is the `v` column of the Spark suite's `false` arm.
+ #[tokio::test]
+ async fn variant_annotation_read_as_plain_struct_is_rejected() -> Result<(), DataFusionError> {
+ let schema = Arc::new(Schema::new(vec![variant_field("v")]));
+ let batch = RecordBatch::try_new(schema, vec![Arc::new(variant_storage_array())])?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ assert_variant_annotation_rejected(&probe, "v", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// The `ns struct<nv variant>` case: the annotation is on a nested field, and the rejection
+ /// must name the dotted path rather than just the root column.
+ #[tokio::test]
+ async fn variant_annotation_inside_struct_is_rejected() -> Result<(), DataFusionError> {
+ let outer_fields = Fields::from(vec![variant_field("nv")]);
+ let outer = StructArray::new(
+ outer_fields.clone(),
+ vec![Arc::new(variant_storage_array()) as ArrayRef],
+ None,
+ );
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "ns",
+ DataType::Struct(outer_fields),
+ true,
+ )]));
+ let batch = RecordBatch::try_new(schema, vec![Arc::new(outer)])?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ assert_variant_annotation_rejected(&probe, "ns.nv", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// The `av array<variant>` case.
+ #[tokio::test]
+ async fn variant_annotation_as_list_element_is_rejected() -> Result<(), DataFusionError> {
+ let element = variant_field("element");
+ let list = ListArray::new(
+ Arc::clone(&element),
+ OffsetBuffer::new(vec![0, 1].into()),
+ Arc::new(variant_storage_array()),
+ None,
+ );
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "av",
+ DataType::List(element),
+ true,
+ )]));
+ let batch = RecordBatch::try_new(schema, vec![Arc::new(list)])?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ assert_variant_annotation_rejected(&probe, "av.element", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// The `mv map<string, variant>` case.
+ #[tokio::test]
+ async fn variant_annotation_as_map_value_is_rejected() -> Result<(), DataFusionError> {
+ let batch = map_of_variant_batch("value")?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ // The path walks through the Parquet map's `entries` group, as Spark's own error does.
+ assert_variant_annotation_rejected(&probe, "mv.entries.value", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// In field-id read mode a nested logical field resolves to the file field with the matching
+ /// ID, not the matching name -- that is what `spark_parquet_convert` does when it builds the
+ /// read. Pairing by name here instead would inspect the wrong file field (or none) and let the
+ /// annotation through unchecked, so the rejection must follow the ID.
+ #[test]
+ fn variant_annotation_is_found_through_a_nested_field_id_match() {
+ let storage = DataType::Struct(variant_storage_fields());
+ let logical = Arc::new(Schema::new(vec![Field::new(
+ "ns",
+ DataType::Struct(Fields::from(vec![Field::new(
+ "renamed_since_write",
+ storage.clone(),
+ true,
+ )
+ .with_metadata(id_meta("7"))])),
+ true,
+ )]));
+ // The file holds the Variant under a different name, plus a decoy that would win a
+ // name-only match and carries no annotation.
+ let physical = Arc::new(Schema::new(vec![Field::new(
+ "ns",
+ DataType::Struct(Fields::from(vec![
+ Field::new("renamed_since_write", DataType::Int32, true)
+ .with_metadata(id_meta("9")),
+ Field::new("nv", storage, true)
+ .with_metadata(id_meta("7"))
+ .with_extension_type(VariantType),
+ ])),
+ true,
+ )]));
+
+ let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ options.use_field_id = true;
+ let err = SparkPhysicalExprAdapterFactory::new(options, None)
+ .create(logical, physical)
+ .expect_err("an ID-resolved Variant field must still be rejected");
+ let msg = err.to_string();
+ assert!(
+ msg.contains("_LEGACY_ERROR_TEMP_3071") && msg.contains("ns.renamed_since_write"),
+ "unexpected error: {msg}"
+ );
+ }
+
+ /// Write a one-row Parquet file holding a single top-level group annotated as VARIANT with
+ /// `spec_version`, using the low-level writer so the annotation is exactly what Spark's writer
+ /// emits. `ArrowWriter` cannot do this: `logical_type_for_struct` hardcodes
+ /// `LogicalType::variant(None)`, which is a different annotation from Spark's `VARIANT(1)`.
+ fn write_annotated_variant_file(spec_version: Option<i8>) -> String {
+ let value = Arc::new(
+ ParquetType::primitive_type_builder("value", PhysicalType::BYTE_ARRAY)
+ .with_repetition(Repetition::REQUIRED)
+ .build()
+ .unwrap(),
+ );
+ let metadata = Arc::new(
+ ParquetType::primitive_type_builder("metadata", PhysicalType::BYTE_ARRAY)
+ .with_repetition(Repetition::REQUIRED)
+ .build()
+ .unwrap(),
+ );
+ let variant = Arc::new(
+ ParquetType::group_type_builder("v")
+ .with_repetition(Repetition::REQUIRED)
+ .with_logical_type(Some(LogicalType::variant(spec_version)))
+ .with_fields(vec![value, metadata])
+ .build()
+ .unwrap(),
+ );
+ let schema = Arc::new(
+ ParquetType::group_type_builder("schema")
+ .with_fields(vec![variant])
+ .build()
+ .unwrap(),
+ );
+
+ let filename = get_temp_filename();
+ let filename = filename.as_path().as_os_str().to_str().unwrap().to_string();
+ let file = File::create(&filename).unwrap();
+ let mut writer = SerializedFileWriter::new(file, schema, Default::default()).unwrap();
+ let mut row_group = writer.next_row_group().unwrap();
+ for bytes in [vec![12u8, 1u8], vec![1u8, 0u8, 0u8]] {
+ let mut column = row_group.next_column().unwrap().unwrap();
+ column
+ .typed::<ByteArrayType>()
+ .write_batch(&[ByteArray::from(bytes)], None, None)
+ .unwrap();
+ column.close().unwrap();
+ }
+ row_group.close().unwrap();
+ writer.close().unwrap();
+ filename
+ }
+
+ /// Scan `filename` asking for the plain `struct<value binary, metadata binary>` a hand-written
+ /// Spark read schema produces, and report the scan outcome.
+ async fn scan_as_plain_variant_storage(filename: String) -> Result<usize, DataFusionError> {
+ let requested = Arc::new(Schema::new(vec![Field::new(
+ "v",
+ DataType::Struct(variant_storage_fields()),
+ false,
+ )]));
+ let expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory> =
+ Arc::new(SparkPhysicalExprAdapterFactory::new(
+ SparkParquetOptions::new(EvalMode::Legacy, "UTC", false),
+ None,
+ ));
+ let file_scan_config = FileScanConfigBuilder::new(
+ ObjectStoreUrl::local_filesystem(),
+ Arc::new(ParquetSource::new(requested)),
+ )
+ .with_file_groups(vec![FileGroup::new(vec![PartitionedFile::from_path(
+ filename,
+ )?])])
+ .with_expr_adapter(Some(expr_adapter_factory))
+ .build();
+ let exec = DataSourceExec::new(Arc::new(file_scan_config));
+ let mut stream = exec.execute(0, Arc::new(TaskContext::default()))?;
+ let mut rows = 0;
+ while let Some(batch) = stream.next().await {
+ rows += batch?.num_rows();
+ }
+ Ok(rows)
+ }
+
+ /// The annotation Spark itself writes is `VARIANT(1)`, and every other rejection test in this
+ /// module goes through `ArrowWriter`, which emits `VARIANT(None)` instead. This one writes the
+ /// real Spark shape so the rejection is pinned against the annotation users actually have on
+ /// disk, not only against the one the test writer happens to produce.
+ #[tokio::test]
+ async fn variant_annotation_with_spec_version_1_is_rejected() -> Result<(), DataFusionError> {
+ let err = scan_as_plain_variant_storage(write_annotated_variant_file(Some(1)))
+ .await
+ .expect_err("a VARIANT(1) field read as a plain struct must be rejected");
+ let msg = err.to_string();
+ assert!(
+ msg.contains("_LEGACY_ERROR_TEMP_3071") && msg.contains("Invalid Spark read type"),
+ "unexpected error: {msg}"
+ );
+ Ok(())
+ }
+
+ /// An annotation with no spec version, which is what an Arrow-based writer emits and
+ /// parquet-java reads back as 0. Spark rejects this too, but through its
+ /// `unrecognizedParquetTypeError` catch-all rather than the variant branch, so the error class
+ /// differs: `PARQUET_TYPE_NOT_RECOGNIZED` there against `_LEGACY_ERROR_TEMP_3071` here. The
+ /// `arrow.parquet.variant` extension type carries no version, so this check cannot tell the
+ /// file apart from `VARIANT(1)`. See `check_variant_annotation` for the full comparison. If a
+ /// future arrow-rs exposes the version, this test is the one to revisit.
+ #[tokio::test]
+ async fn variant_annotation_without_a_spec_version_is_also_rejected(
+ ) -> Result<(), DataFusionError> {
+ let err = scan_as_plain_variant_storage(write_annotated_variant_file(None))
+ .await
+ .expect_err("Comet rejects an unversioned VARIANT annotation that Spark would accept");
Review Comment:
The message says Comet rejects an unversioned annotation "that Spark would accept", but the doc on this test and the doc on `is_variant_marked` both say Spark rejects it with `PARQUET_TYPE_NOT_RECOGNIZED`. Could the message match them? The `is_variant_marked` doc around line 518 also ends with two similar "Closing it ..." sentences, so one can go.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -4643,6 +4884,930 @@ pub(crate) mod test {
assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap());
}
+ /// Verification probe for apache/datafusion-comet#5741.
+ ///
+ /// Captures the `physical_file_schema` DataFusion hands to
+ /// `PhysicalExprAdapterFactory::create`, so the tests below can assert both that `create`
+ /// was reached at all and what the file schema looked like when it was.
+ #[derive(Debug)]
+ struct ProbeFactory {
+ inner: SparkPhysicalExprAdapterFactory,
+ seen: Arc<Mutex<Option<SchemaRef>>>,
+ }
+
+ impl PhysicalExprAdapterFactory for ProbeFactory {
+ fn create(
+ &self,
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ ) -> datafusion::common::Result<
+ Arc<dyn datafusion_physical_expr_adapter::PhysicalExprAdapter>,
+ > {
+ *self.seen.lock().unwrap() = Some(Arc::clone(&physical_file_schema));
+ self.inner.create(logical_file_schema, physical_file_schema)
+ }
+ }
+
+ /// Recursively drop every Arrow extension marker from `field`, producing the schema a Spark
+ /// user's hand-written DDL yields: the same structure, no variant identity. Mirrors the
+ /// `.schema("v struct<value binary, metadata binary>")` in Spark's
+ /// `ParquetVariantShreddingSuite`, test
+ /// `variant logical type annotation - ignore variant annotation`.
+ fn strip_extension_markers(field: &FieldRef) -> FieldRef {
+ let data_type = match field.data_type() {
+ DataType::Struct(fields) => {
+ DataType::Struct(fields.iter().map(strip_extension_markers).collect())
+ }
+ DataType::List(item) => DataType::List(strip_extension_markers(item)),
+ DataType::Map(entries, sorted) => {
+ DataType::Map(strip_extension_markers(entries), *sorted)
+ }
+ other => other.clone(),
+ };
+ let mut metadata = field.metadata().clone();
+ metadata.remove("ARROW:extension:name");
+ metadata.remove("ARROW:extension:metadata");
+ Arc::new(Field::new(field.name(), data_type, field.is_nullable()).with_metadata(metadata))
+ }
+
+ /// Write `batch` to a temp Parquet file, then scan it back through `DataSourceExec` asking for
+ /// the same schema with every extension marker stripped. Reports the Parquet schema the writer
+ /// actually produced alongside the physical file schema `create` observed (`None` when `create`
+ /// never ran).
+ ///
+ /// Returning the Parquet schema separates the two ways a shape can fail: the *writer* never
+ /// emitting the VARIANT annotation, versus the *reader* not surfacing it as an Arrow extension
+ /// type. Only the second one is a problem for #5741; the first is a limitation of the test.
+ ///
+ /// `skip_arrow_metadata` controls whether the file embeds an `ARROW:schema` key-value entry.
+ /// Spark's writer embeds none, so the Spark-shaped cases pass `true`: a marker on the way back
+ /// can then only have come from the Parquet annotation, and a file carrying the hint would let
+ /// every assertion below pass vacuously.
+ struct VariantProbe {
+ /// The Parquet schema the writer actually produced.
+ parquet_schema: String,
+ /// The physical file schema `create` observed, or `None` when it never ran.
+ observed_physical: Option<SchemaRef>,
+ /// Total rows read, or the error the scan raised.
+ scan: Result<usize, DataFusionError>,
+ }
+
+ struct ProbeOptions {
+ /// Whether the file omits the embedded `ARROW:schema` hint, as Spark's writer does.
+ skip_arrow_metadata: bool,
+ /// `spark.sql.parquet.ignoreVariantAnnotation`.
+ ignore_variant_annotation: bool,
+ /// Keep the Variant markers on the requested schema, modelling a genuine native Variant
+ /// projection (`projects_variant` in `parquet_exec.rs`) instead of a hand-written struct.
+ keep_request_markers: bool,
+ /// Request this schema instead of one derived from the file schema, for cases where
+ /// Spark's requested names differ from the names the file was written with.
+ requested_schema: Option<SchemaRef>,
+ }
+
+ impl Default for ProbeOptions {
+ fn default() -> Self {
+ Self {
+ skip_arrow_metadata: true,
+ ignore_variant_annotation: false,
+ keep_request_markers: false,
+ requested_schema: None,
+ }
+ }
+ }
+
+ async fn probe_variant_annotation(
+ batch: RecordBatch,
+ options: ProbeOptions,
+ ) -> Result<VariantProbe, DataFusionError> {
+ let file_schema = batch.schema();
+ let filename = get_temp_filename();
+ let filename = filename.as_path().as_os_str().to_str().unwrap().to_string();
+ let file = File::create(&filename)?;
+ let mut writer = ArrowWriter::try_new_with_options(
+ file,
+ Arc::clone(&file_schema),
+ ArrowWriterOptions::new().with_skip_arrow_metadata(options.skip_arrow_metadata),
+ )?;
+ writer.write(&batch)?;
+ writer.close()?;
+
+ probe_variant_annotation_file(&filename, &file_schema, options).await
+ }
+
+ /// The reading half of `probe_variant_annotation`: reads back the Parquet schema `filename`
+ /// actually holds, derives the requested schema from `file_schema` per `options`, and scans
+ /// the file through a `ProbeFactory`. Split out so a test can hand it a file the `ArrowWriter`
+ /// alone cannot produce, such as one whose `ARROW:schema` hint disagrees with its annotations.
+ async fn probe_variant_annotation_file(
+ filename: &str,
+ file_schema: &SchemaRef,
+ options: ProbeOptions,
+ ) -> Result<VariantProbe, DataFusionError> {
+ let filename = filename.to_string();
+ let file_schema = Arc::clone(file_schema);
+
+ let mut printed = Vec::new();
+ let reader = ParquetRecordBatchReaderBuilder::try_new(File::open(&filename)?)?;
+ print_schema(&mut printed, reader.parquet_schema().root_schema());
+ let printed = String::from_utf8(printed).unwrap();
+
+ let requested = if let Some(requested) = options.requested_schema {
+ requested
+ } else if options.keep_request_markers {
+ Arc::clone(&file_schema)
+ } else {
+ Arc::new(Schema::new(
+ file_schema
+ .fields()
+ .iter()
+ .map(strip_extension_markers)
+ .collect::<Fields>(),
+ ))
+ };
+
+ let mut spark_parquet_options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ spark_parquet_options.ignore_variant_annotation = options.ignore_variant_annotation;
+ let seen = Arc::new(Mutex::new(None));
+ let expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory> = Arc::new(ProbeFactory {
+ inner: SparkPhysicalExprAdapterFactory::new(spark_parquet_options, None),
+ seen: Arc::clone(&seen),
+ });
+
+ // Install the reader factory `init_datasource_exec` installs, gated the way it gates it:
+ // the footer rewrites that decide which fields the reader marks as Variant live there, so
+ // a probe without it would read a hint the real scan never sees.
+ let projects_variant = requested
+ .fields()
+ .iter()
+ .any(|field| field.has_valid_extension_type::<VariantType>());
+ let runtime = Arc::new(RuntimeEnv::default());
+ let store = runtime.object_store(ObjectStoreUrl::local_filesystem())?;
+ let reader_factory = Arc::new(
+ EagerPageIndexReaderFactory::new(
+ store,
+ runtime.cache_manager.get_file_metadata_cache(),
+ ScanIoSource::Local,
+ &ExecutionPlanMetricsSet::new(),
+ )
+ .with_spark_variant_schema(projects_variant),
+ );
+ let parquet_source =
+ ParquetSource::new(requested).with_parquet_file_reader_factory(reader_factory);
+ let files = FileGroup::new(vec![PartitionedFile::from_path(filename)?]);
+ let file_scan_config = FileScanConfigBuilder::new(
+ ObjectStoreUrl::local_filesystem(),
+ Arc::new(parquet_source),
+ )
+ .with_file_groups(vec![files])
+ .with_expr_adapter(Some(expr_adapter_factory))
+ .build();
+ let parquet_exec = DataSourceExec::new(Arc::new(file_scan_config));
+ let scan = async {
+ let mut stream = parquet_exec.execute(0, Arc::new(TaskContext::default()))?;
+ let mut rows = 0;
+ while let Some(batch) = stream.next().await {
+ rows += batch?.num_rows();
+ }
+ Ok(rows)
+ }
+ .await;
+
+ let observed_physical = seen.lock().unwrap().clone();
+ Ok(VariantProbe {
+ parquet_schema: printed,
+ observed_physical,
+ scan,
+ })
+ }
+
+ /// The two-field Variant storage group Spark writes when shredding is off.
+ fn variant_storage_fields() -> Fields {
+ Fields::from(vec![
+ Field::new("value", DataType::Binary, false),
+ Field::new("metadata", DataType::Binary, false),
+ ])
+ }
+
+ /// A `StructArray` holding one Variant value: the encoded integer `1`, matching the
+ /// `Row(Array[Byte](12, 1), Array[Byte](1, 0, 0))` the Spark suite expects.
+ fn variant_storage_array() -> StructArray {
+ let value = Arc::new(BinaryArray::from(vec![Some(&[12u8, 1u8][..])])) as ArrayRef;
+ let meta = Arc::new(BinaryArray::from(vec![Some(&[1u8, 0u8, 0u8][..])])) as ArrayRef;
+ StructArray::new(variant_storage_fields(), vec![value, meta], None)
+ }
+
+ /// An Arrow field named `name` carrying the Variant storage type and the
+ /// `arrow.parquet.variant` extension marker, which parquet-rs's writer turns into the Parquet
+ /// VARIANT logical type annotation.
+ fn variant_field(name: &str) -> FieldRef {
+ Arc::new(
+ Field::new(name, DataType::Struct(variant_storage_fields()), true)
+ .with_extension_type(VariantType),
+ )
+ }
+
+ /// Assert that `probe_variant_annotation` saw a live `create` call on a file the writer really
+ /// annotated, then hand the observed physical schema to `locate` to check the marker survived
+ /// at the shape-specific position.
+ fn assert_annotation_survived(
+ shape: &str,
+ probe: &VariantProbe,
+ locate: impl Fn(&SchemaRef) -> Option<FieldRef>,
+ ) {
+ let printed = &probe.parquet_schema;
+ assert!(
+ printed.contains("VARIANT"),
+ "{shape}: parquet-rs did not write the VARIANT annotation, so this file cannot test \
+ the reader at all. Parquet schema:\n{printed}"
+ );
+ let observed = probe.observed_physical.clone().unwrap_or_else(|| {
+ panic!(
+ "{shape}: PhysicalExprAdapterFactory::create was never called: the logical and \
+ physical file schemas compared equal, so the variant annotation did not survive \
+ into the physical schema"
+ )
+ });
+ let field = locate(&observed).unwrap_or_else(|| {
+ panic!("{shape}: could not locate the variant field in {observed:?}")
+ });
+ assert!(
+ field.has_valid_extension_type::<VariantType>(),
+ "{shape}: physical file schema lost the variant annotation: {field:?}"
+ );
+ }
+
+ /// One row of `map<string, variant>`, the `mv` column of the Spark suite, with the map value
+ /// child named `value_name` in the file.
+ fn map_of_variant_batch(value_name: &str) -> Result<RecordBatch, DataFusionError> {
+ let value_field = variant_field(value_name);
+ let entry_fields = Fields::from(vec![
+ Arc::new(Field::new("key", DataType::Utf8, false)),
+ Arc::clone(&value_field),
+ ]);
+ let entries = StructArray::new(
+ entry_fields.clone(),
+ vec![
+ Arc::new(StringArray::from(vec!["v2"])) as ArrayRef,
+ Arc::new(variant_storage_array()) as ArrayRef,
+ ],
+ None,
+ );
+ let entries_field = Arc::new(Field::new("entries", DataType::Struct(entry_fields), false));
+ let map = MapArray::new(
+ Arc::clone(&entries_field),
+ OffsetBuffer::new(vec![0, 1].into()),
+ entries,
+ None,
+ false,
+ );
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "mv",
+ DataType::Map(entries_field, false),
+ true,
+ )]));
+ Ok(RecordBatch::try_new(schema, vec![Arc::new(map)])?)
+ }
+
+ /// Assert the scan was rejected with Spark's `_LEGACY_ERROR_TEMP_3071` shape, naming `column`
+ /// (a dotted path) and the requested type.
+ fn assert_variant_annotation_rejected(probe: &VariantProbe, column: &str, spark_type: &str) {
+ let err = probe
+ .scan
+ .as_ref()
+ .expect_err("reading a VARIANT-annotated field as a plain struct must be rejected");
+ let msg = err.to_string();
+ assert!(
+ msg.contains("_LEGACY_ERROR_TEMP_3071")
+ && msg.contains("Invalid Spark read type")
+ && msg.contains(column)
+ && msg.contains(spark_type),
+ "unexpected error for {column}: {msg}"
+ );
+ }
+
+ /// The plain struct a hand-written Spark read schema produces for a Variant column.
+ fn plain_variant_storage_sql() -> &'static str {
+ "STRUCT<value: BINARY, metadata: BINARY>"
+ }
+
+ /// #5741: reading a VARIANT-annotated top-level field as
+ /// `struct<value binary, metadata binary>` must raise Spark's `_LEGACY_ERROR_TEMP_3071`
+ /// instead of silently returning the storage
+ /// struct. This is the `v` column of the Spark suite's `false` arm.
+ #[tokio::test]
+ async fn variant_annotation_read_as_plain_struct_is_rejected() -> Result<(), DataFusionError> {
+ let schema = Arc::new(Schema::new(vec![variant_field("v")]));
+ let batch = RecordBatch::try_new(schema, vec![Arc::new(variant_storage_array())])?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ assert_variant_annotation_rejected(&probe, "v", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// The `ns struct<nv variant>` case: the annotation is on a nested field, and the rejection
+ /// must name the dotted path rather than just the root column.
+ #[tokio::test]
+ async fn variant_annotation_inside_struct_is_rejected() -> Result<(), DataFusionError> {
+ let outer_fields = Fields::from(vec![variant_field("nv")]);
+ let outer = StructArray::new(
+ outer_fields.clone(),
+ vec![Arc::new(variant_storage_array()) as ArrayRef],
+ None,
+ );
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "ns",
+ DataType::Struct(outer_fields),
+ true,
+ )]));
+ let batch = RecordBatch::try_new(schema, vec![Arc::new(outer)])?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ assert_variant_annotation_rejected(&probe, "ns.nv", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// The `av array<variant>` case.
+ #[tokio::test]
+ async fn variant_annotation_as_list_element_is_rejected() -> Result<(), DataFusionError> {
+ let element = variant_field("element");
+ let list = ListArray::new(
+ Arc::clone(&element),
+ OffsetBuffer::new(vec![0, 1].into()),
+ Arc::new(variant_storage_array()),
+ None,
+ );
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "av",
+ DataType::List(element),
+ true,
+ )]));
+ let batch = RecordBatch::try_new(schema, vec![Arc::new(list)])?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ assert_variant_annotation_rejected(&probe, "av.element", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// The `mv map<string, variant>` case.
+ #[tokio::test]
+ async fn variant_annotation_as_map_value_is_rejected() -> Result<(), DataFusionError> {
+ let batch = map_of_variant_batch("value")?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ // The path walks through the Parquet map's `entries` group, as Spark's own error does.
+ assert_variant_annotation_rejected(&probe, "mv.entries.value", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// In field-id read mode a nested logical field resolves to the file field with the matching
+ /// ID, not the matching name -- that is what `spark_parquet_convert` does when it builds the
+ /// read. Pairing by name here instead would inspect the wrong file field (or none) and let the
+ /// annotation through unchecked, so the rejection must follow the ID.
+ #[test]
+ fn variant_annotation_is_found_through_a_nested_field_id_match() {
+ let storage = DataType::Struct(variant_storage_fields());
+ let logical = Arc::new(Schema::new(vec![Field::new(
+ "ns",
+ DataType::Struct(Fields::from(vec![Field::new(
+ "renamed_since_write",
+ storage.clone(),
+ true,
+ )
+ .with_metadata(id_meta("7"))])),
+ true,
+ )]));
+ // The file holds the Variant under a different name, plus a decoy that would win a
+ // name-only match and carries no annotation.
+ let physical = Arc::new(Schema::new(vec![Field::new(
+ "ns",
+ DataType::Struct(Fields::from(vec![
+ Field::new("renamed_since_write", DataType::Int32, true)
+ .with_metadata(id_meta("9")),
+ Field::new("nv", storage, true)
+ .with_metadata(id_meta("7"))
+ .with_extension_type(VariantType),
+ ])),
+ true,
+ )]));
+
+ let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ options.use_field_id = true;
+ let err = SparkPhysicalExprAdapterFactory::new(options, None)
+ .create(logical, physical)
+ .expect_err("an ID-resolved Variant field must still be rejected");
+ let msg = err.to_string();
+ assert!(
+ msg.contains("_LEGACY_ERROR_TEMP_3071") && msg.contains("ns.renamed_since_write"),
+ "unexpected error: {msg}"
+ );
+ }
+
+ /// Write a one-row Parquet file holding a single top-level group annotated as VARIANT with
+ /// `spec_version`, using the low-level writer so the annotation is exactly what Spark's writer
+ /// emits. `ArrowWriter` cannot do this: `logical_type_for_struct` hardcodes
+ /// `LogicalType::variant(None)`, which is a different annotation from Spark's `VARIANT(1)`.
+ fn write_annotated_variant_file(spec_version: Option<i8>) -> String {
+ let value = Arc::new(
+ ParquetType::primitive_type_builder("value", PhysicalType::BYTE_ARRAY)
+ .with_repetition(Repetition::REQUIRED)
+ .build()
+ .unwrap(),
+ );
+ let metadata = Arc::new(
+ ParquetType::primitive_type_builder("metadata", PhysicalType::BYTE_ARRAY)
+ .with_repetition(Repetition::REQUIRED)
+ .build()
+ .unwrap(),
+ );
+ let variant = Arc::new(
+ ParquetType::group_type_builder("v")
+ .with_repetition(Repetition::REQUIRED)
+ .with_logical_type(Some(LogicalType::variant(spec_version)))
+ .with_fields(vec![value, metadata])
+ .build()
+ .unwrap(),
+ );
+ let schema = Arc::new(
+ ParquetType::group_type_builder("schema")
+ .with_fields(vec![variant])
+ .build()
+ .unwrap(),
+ );
+
+ let filename = get_temp_filename();
+ let filename = filename.as_path().as_os_str().to_str().unwrap().to_string();
+ let file = File::create(&filename).unwrap();
+ let mut writer = SerializedFileWriter::new(file, schema, Default::default()).unwrap();
+ let mut row_group = writer.next_row_group().unwrap();
+ for bytes in [vec![12u8, 1u8], vec![1u8, 0u8, 0u8]] {
+ let mut column = row_group.next_column().unwrap().unwrap();
+ column
+ .typed::<ByteArrayType>()
+ .write_batch(&[ByteArray::from(bytes)], None, None)
+ .unwrap();
+ column.close().unwrap();
+ }
+ row_group.close().unwrap();
+ writer.close().unwrap();
+ filename
+ }
+
+ /// Scan `filename` asking for the plain `struct<value binary, metadata binary>` a hand-written
+ /// Spark read schema produces, and report the scan outcome.
+ async fn scan_as_plain_variant_storage(filename: String) -> Result<usize, DataFusionError> {
+ let requested = Arc::new(Schema::new(vec![Field::new(
+ "v",
+ DataType::Struct(variant_storage_fields()),
+ false,
+ )]));
+ let expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory> =
+ Arc::new(SparkPhysicalExprAdapterFactory::new(
+ SparkParquetOptions::new(EvalMode::Legacy, "UTC", false),
+ None,
+ ));
+ let file_scan_config = FileScanConfigBuilder::new(
+ ObjectStoreUrl::local_filesystem(),
+ Arc::new(ParquetSource::new(requested)),
+ )
+ .with_file_groups(vec![FileGroup::new(vec![PartitionedFile::from_path(
+ filename,
+ )?])])
+ .with_expr_adapter(Some(expr_adapter_factory))
+ .build();
+ let exec = DataSourceExec::new(Arc::new(file_scan_config));
+ let mut stream = exec.execute(0, Arc::new(TaskContext::default()))?;
+ let mut rows = 0;
+ while let Some(batch) = stream.next().await {
+ rows += batch?.num_rows();
+ }
+ Ok(rows)
+ }
+
+ /// The annotation Spark itself writes is `VARIANT(1)`, and every other rejection test in this
+ /// module goes through `ArrowWriter`, which emits `VARIANT(None)` instead. This one writes the
+ /// real Spark shape so the rejection is pinned against the annotation users actually have on
+ /// disk, not only against the one the test writer happens to produce.
+ #[tokio::test]
+ async fn variant_annotation_with_spec_version_1_is_rejected() -> Result<(), DataFusionError> {
+ let err = scan_as_plain_variant_storage(write_annotated_variant_file(Some(1)))
+ .await
+ .expect_err("a VARIANT(1) field read as a plain struct must be rejected");
+ let msg = err.to_string();
+ assert!(
+ msg.contains("_LEGACY_ERROR_TEMP_3071") && msg.contains("Invalid Spark read type"),
+ "unexpected error: {msg}"
+ );
+ Ok(())
+ }
+
+ /// An annotation with no spec version, which is what an Arrow-based writer emits and
+ /// parquet-java reads back as 0. Spark rejects this too, but through its
+ /// `unrecognizedParquetTypeError` catch-all rather than the variant branch, so the error class
+ /// differs: `PARQUET_TYPE_NOT_RECOGNIZED` there against `_LEGACY_ERROR_TEMP_3071` here. The
+ /// `arrow.parquet.variant` extension type carries no version, so this check cannot tell the
+ /// file apart from `VARIANT(1)`. See `check_variant_annotation` for the full comparison. If a
+ /// future arrow-rs exposes the version, this test is the one to revisit.
+ #[tokio::test]
+ async fn variant_annotation_without_a_spec_version_is_also_rejected(
+ ) -> Result<(), DataFusionError> {
+ let err = scan_as_plain_variant_storage(write_annotated_variant_file(None))
+ .await
+ .expect_err("Comet rejects an unversioned VARIANT annotation that Spark would accept");
+ assert!(
+ err.to_string().contains("_LEGACY_ERROR_TEMP_3071"),
+ "unexpected error: {err}"
+ );
+ Ok(())
+ }
+
+ /// `spark.sql.parquet.ignoreVariantAnnotation=true` is the explicit opt-in to read the
+ /// annotated group as its plain underlying struct. The `true` arm of the Spark suite: the read
+ /// must succeed and return the Variant storage bytes unchanged.
+ #[tokio::test]
+ async fn variant_annotation_is_ignored_when_conf_is_set() -> Result<(), DataFusionError> {
+ let top_level = Arc::new(Schema::new(vec![variant_field("v")]));
+ let cases = vec![
+ (
+ "top level",
+ RecordBatch::try_new(top_level, vec![Arc::new(variant_storage_array())])?,
+ ),
+ // The conf short-circuits the whole walk, so a nested annotation must be ignored too.
+ ("map value", map_of_variant_batch("value")?),
+ ];
+ for (shape, batch) in cases {
+ let probe = probe_variant_annotation(
+ batch,
+ ProbeOptions {
+ ignore_variant_annotation: true,
+ ..Default::default()
+ },
+ )
+ .await?;
+ let rows = probe.scan.unwrap_or_else(|err| {
+ panic!(
+ "{shape}: ignoreVariantAnnotation=true must allow the plain struct read: {err}"
+ )
+ });
+ assert_eq!(rows, 1, "{shape}: unexpected row count");
+ }
+ Ok(())
+ }
+
+ /// The other half of the symmetric check: when the *requested* schema is itself marked as a
+ /// Variant -- a genuine native Variant projection, which #5794 made reachable -- the annotation
+ /// agrees with the request and the read must not be rejected. Without this test, "simplifying"
+ /// the check to fire on the physical marker alone would break Variant projection with nothing
+ /// turning red.
+ #[tokio::test]
+ async fn marked_variant_request_is_not_rejected() -> Result<(), DataFusionError> {
+ let schema = Arc::new(Schema::new(vec![variant_field("v")]));
+ let batch = RecordBatch::try_new(schema, vec![Arc::new(variant_storage_array())])?;
+ let probe = probe_variant_annotation(
+ batch,
+ ProbeOptions {
+ keep_request_markers: true,
+ ..Default::default()
+ },
+ )
+ .await?;
+ assert_eq!(
+ probe
+ .scan
+ .expect("a marked Variant request must not trip the annotation check"),
+ 1
+ );
+ Ok(())
+ }
+
+ /// Spark rejects this read inside `ParquetToSparkSchemaConverter`, while converting the schema,
+ /// so a file with no rows fails just the same. That is the opposite of the type-promotion
+ /// checks in this module, which `RejectOnNonEmpty` defers to match Spark's per-row-group
+ /// behavior (see `parquet_empty_file_disallowed_widening`, SPARK-26709). Pinning it here keeps
+ /// a later reader from "fixing" this check to match its neighbours.
+ #[tokio::test]
+ async fn variant_annotation_is_rejected_on_an_empty_file() -> Result<(), DataFusionError> {
Review Comment:
`variant_annotation_is_rejected_on_an_empty_file` looks covered by `requested_annotated_root_is_rejected_on_an_empty_file` in `variant_tests.rs`, which goes through `init_datasource_exec`. The four `variant_annotation_survives_*` tests also each have a `*_is_rejected` twin for the same shape. A rejection can only happen when the marker reached `create`, so the twins may already prove what the survives tests check. Could those be merged or dropped? `ProbeFactory`, `observed_physical` and `assert_annotation_survived` would then be unused. `variant_annotation_survives_with_embedded_arrow_schema` looks like the one distinct shape.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#issuecomment-5876680489
This is a light fully automated review since there are so many PRs open.
The physical side of the check at `native/core/src/parquet/schema_adapter.rs:549` is the reader's Arrow field, and the doc on `is_variant_marked` (line 497) treats that marker as the file's VARIANT annotation. As far as I can tell that only holds when the file has no embedded `ARROW:schema`. When it has one, `convert_field` in parquet 59.3's `arrow/schema/complex.rs` copies each field's metadata from the hint and never derives the extension from the Parquet logical type. For a plain-struct request `projects_variant` is false, so `parquet_exec.rs:198` keeps the hint. A concrete case is a file written by arrow-rs `ArrowWriter` without the non-default `variant_experimental` feature, from a column built with `parquet-variant-compute`'s `VariantArray::field`. `logical_type_for_struct` returns `None` without that feature, so the group has no annotation but the hint carries `arrow.parquet.variant`. Spark 4.1 infers and reads that column as `struct<metadata binary, value binary>`, and mai
n reads it the same way, but with this change the read fails with `_LEGACY_ERROR_TEMP_3071`. The reverse holds too: an annotated group whose hint lacks the marker still returns the storage bytes. Could the physical side come from the Parquet schema instead, for example by reconciling the hint's Variant markers with the file's annotations in `EagerPageIndexReaderFactory`, which already rewrites the hint for Variant projections? A test that writes an unannotated group with a hand-built `ARROW:schema` carrying the marker would pin this down.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on code in PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#discussion_r4034249279
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -911,6 +1083,51 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory {
None
};
+ // Compare the file's VARIANT annotations against the requested types before handing the
+ // schemas to the default adapter. `adapted_physical_schema` is used so that field-id and
+ // case-insensitive resolution has already aligned the two sides' top-level names.
+ if !self.parquet_options.ignore_variant_annotation {
+ let mut physical_by_folded: HashMap<&str, usize> = HashMap::new();
+ for (i, name) in physical_folded.iter().enumerate() {
+ physical_by_folded.entry(name.as_str()).or_insert(i);
+ }
+ // When the read schema is known, check each requested root against its requested
+ // type and skip roots Spark would not read. Requested roots share their Spark names
+ // with the logical file schema, so the same fold pairs them.
+ let required_by_folded: Option<HashMap<String, &FieldRef>> =
+ self.required_schema.as_ref().map(|required| {
+ let mut map = HashMap::new();
+ for (field, folded) in required
+ .fields()
+ .iter()
+ .zip(fold_schema_names(required, case_sensitive))
Review Comment:
Done. Merged current main, and the requested-schema fold now propagates its error through a fallible closure and transpose().
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#issuecomment-5771105066
@0lai0 sorry for the late reply - could you resolve the conflict? I'll merge this afterwards
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#issuecomment-5895475227
Thanks @andygrove. You are right, and both directions were broken. With a hint present convert_field copies its metadata and never derives the extension from the logical type, so the marker the check read was the hint's, not the file's.
Fixed in EagerPageIndexReaderFactory, as you suggested: `with_reconciled_variant_markers `takes the hinted schema and the annotation-only schema (`parquet_to_arrow_schema` with and without the key-value metadata) and moves the markers to match the annotations, leaving everything else the hint decides alone. A file with no hint returns after one scan of the key-value metadata.
Two tests, one per direction, each writing the data with one schema and the hint arrow-rs writes for another. Both fail with the reconciliation disabled.
Also worth flagging: `probe_variant_annotation` built its own `ParquetSource` and never installed the reader factory, so no Variant test exercised the footer rewrites and none of them could have caught this. It installs the factory now.
`cargo test --lib`: 530 passed. `dev/local-ci.sh spark sql_core-1` (4.1.3): 12835 passed, 0 failed.
On the 4.2 diff: will do. It is not on this branch yet — #4950 landed it after my last merge — so I will merge main and regenerate it without the `IgnoreComet`. Waiting for #6398 first, since until it lands the 4.2 run stops in "Pre-compile Spark Test classes" and tells us nothing about the fix.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#issuecomment-5892104227
`dev/diffs/4.2.0.diff` landed on main in #4950. It has the same `IgnoreComet("https://github.com/apache/datafusion-comet/issues/5741")` on "variant logical type annotation - ignore variant annotation" that this PR removes from the 4.1 diff. Since this PR closes #5741, could you remove it from the 4.2 diff too? Otherwise the Spark 4.2 suite will keep skipping that test after the fix lands. The [Spark SQL Tests guide](https://datafusion.apache.org/comet/contributor-guide/spark-sql-tests.html) covers regenerating a diff.
I've added the `run-spark-4.2-tests` label so CI runs the Spark 4.2 SQL suite on this PR. The first run will fail in "Pre-compile Spark Test classes", because main's 4.2 diff still asks for Comet `1.1.0-SNAPSHOT` (#6380). #6398 fixes that. Once it lands, merging main into this branch will rerun the suite and give a real result.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#discussion_r4029599526
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -911,6 +1083,51 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory {
None
};
+ // Compare the file's VARIANT annotations against the requested types before handing the
+ // schemas to the default adapter. `adapted_physical_schema` is used so that field-id and
+ // case-insensitive resolution has already aligned the two sides' top-level names.
+ if !self.parquet_options.ignore_variant_annotation {
+ let mut physical_by_folded: HashMap<&str, usize> = HashMap::new();
+ for (i, name) in physical_folded.iter().enumerate() {
+ physical_by_folded.entry(name.as_str()).or_insert(i);
+ }
+ // When the read schema is known, check each requested root against its requested
+ // type and skip roots Spark would not read. Requested roots share their Spark names
+ // with the logical file schema, so the same fold pairs them.
+ let required_by_folded: Option<HashMap<String, &FieldRef>> =
+ self.required_schema.as_ref().map(|required| {
+ let mut map = HashMap::new();
+ for (field, folded) in required
+ .fields()
+ .iter()
+ .zip(fold_schema_names(required, case_sensitive))
Review Comment:
### Correctness
[P1] Propagate the fallible fold before building this map
Current `main` changed `fold_schema_names` to return `DataFusionResult<Vec<String>>` in [#5845](https://github.com/apache/datafusion-comet/pull/5845). This new call remains unchanged in the PR's [merge commit](https://github.com/apache/datafusion-comet/blob/3aa01e9b3cd7408e2d6cf6af3f604e6ef64ac2dc/native/core/src/parquet/schema_adapter.rs#L1109). Iterating that `Result` yields a `Vec<String>`, so the closure produces `Option<HashMap<Vec<String>, &FieldRef>>` instead of the declared `Option<HashMap<String, &FieldRef>>`, and the current merge result cannot compile. A bounded `rustc` reproduction using the exact added block and folding function confirms E0308. the same block compiles with the head's older folding API. Please update this call to propagate the fold error before zipping, and make the optional-schema closure fallible (for example, return a `Result` and use `transpose()`), when updating to current `main`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#discussion_r4012091659
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -911,6 +1088,29 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory {
None
};
+ // Compare the file's VARIANT annotations against the requested types before handing the
+ // schemas to the default adapter. `adapted_physical_schema` is used so that field-id and
+ // case-insensitive resolution has already aligned the two sides' top-level names.
+ if !self.parquet_options.ignore_variant_annotation {
+ let mut physical_by_folded: HashMap<&str, usize> = HashMap::new();
+ for (i, name) in physical_folded.iter().enumerate() {
+ physical_by_folded.entry(name.as_str()).or_insert(i);
+ }
+ let mut path = Vec::new();
+ for (logical_field, folded) in logical_file_schema.fields().iter().zip(&logical_folded)
Review Comment:
### Correctness
[P2] Could this validation use the actual requested read schema rather than every field of `logical_file_schema`? `CometNativeScan` keeps an ordinary struct in `nativeDataSchema` even when it is unprojected, and `init_datasource_exec` passes that full schema to `ParquetSource` with a separate projection. DataFusion 55.1.0 consequently calls this factory with the full schema before rewriting the projection. For a file with `id INT, v VARIANT(1)`, a read declared as `id INT, v STRUCT<value BINARY, metadata BINARY>` followed by `.select("id")` now fails on `v`, although no read of `v` was requested. Spark clips its Parquet schema to the requested columns before conversion. Please keep the eager check for requested fields, including empty files, and add a regression covering an omitted annotated root.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -434,6 +434,183 @@ fn is_string_or_binary(dt: &DataType) -> bool {
)
}
+/// Approximate Spark's `DataType.sql` for the variant rejection message. `spark_catalog_name`
+/// bottoms out at "unknown" for nested types, which is exactly the shape a hand-written Variant
+/// read schema has, so the nested cases are spelled out here.
+///
+/// This is close to `DataType.sql` but not identical. Spark's `StructField.sql` wraps names in
+/// `QuotingUtils.quoteIfNeeded` and appends the nullability and comment DDL, none of which is
+/// reproduced here, so a field named `has space` renders bare. The whole message already differs
+/// from Spark's by design (see `variant_annotation_err`), and the rendering exists to tell a user
+/// which type they asked for, so it is kept simple rather than made byte-identical.
+fn spark_read_type_name(dt: &DataType) -> String {
+ match dt {
+ DataType::Struct(fields) => {
+ let rendered = fields
+ .iter()
+ .map(|field| {
+ format!(
+ "{}: {}",
+ field.name(),
+ spark_read_type_name(field.data_type())
+ )
+ })
+ .collect::<Vec<_>>()
+ .join(", ");
+ format!("STRUCT<{rendered}>")
+ }
+ DataType::List(item) | DataType::LargeList(item) => {
+ format!("ARRAY<{}>", spark_read_type_name(item.data_type()))
+ }
+ DataType::Map(entries, _) => match entries.data_type() {
+ DataType::Struct(fields) if fields.len() == 2 => format!(
+ "MAP<{}, {}>",
+ spark_read_type_name(fields[0].data_type()),
+ spark_read_type_name(fields[1].data_type())
+ ),
+ other => format!("MAP<{}>", spark_read_type_name(other)),
+ },
+ other => spark_catalog_name(other).to_uppercase(),
+ }
+}
+
+/// Build the carrier for a Parquet field whose VARIANT logical type annotation is incompatible
+/// with the requested Spark read type. The JVM shim turns these into Spark's
+/// `_LEGACY_ERROR_TEMP_3071` `AnalysisException`.
+///
+/// `column` is the dotted path to the offending field. Spark's own message interpolates the
+/// Parquet `Type` itself, roughly `optional group v (VARIANT(1)) { ... }`, so the two messages
+/// deliberately differ. `ParquetVariantShreddingSuite` matches on `Invalid Spark read type` and
+/// accepts either, and a path is the more useful of the two once the field is nested.
+fn variant_annotation_err(column: &str, target_type: &DataType) -> DataFusionError {
+ DataFusionError::External(Box::new(SparkError::ParquetVariantAnnotationMismatch {
+ file_path: String::new(),
+ column: column.to_string(),
+ spark_type: spark_read_type_name(target_type),
+ }))
+}
+
+/// Whether `field` is marked as a Variant. On a physical field the marker is the Parquet VARIANT
+/// logical type annotation, which arrow-rs surfaces as the `arrow.parquet.variant` Arrow
+/// extension type for struct fields, list elements and map values alike. On a requested field it
+/// is the same marker Comet's serde applies for `VariantType`. `check_variant_annotation` relies
+/// on both sides using it, so it compares them through this one predicate.
+///
+/// The marker does not carry the annotation's spec version, and Spark keys its handling off that
+/// version. `ParquetToSparkSchemaConverter.convertGroupField` matches
+/// `case v: VariantLogicalTypeAnnotation if v.getSpecVersion == 1`, and an annotation failing
+/// that guard falls through to `case _ => throw unrecognizedParquetTypeError(...)`, which is
+/// `PARQUET_TYPE_NOT_RECOGNIZED`. arrow-rs matches `LogicalType::Variant(_)` for any version, so
+/// this predicate cannot tell the two apart. Spark's writer emits version 1. An Arrow-based
+/// writer emits no version at all, which parquet-java reads back as 0.
+///
+/// Both engines therefore reject a non-v1 annotation, with different error classes: Spark raises
+/// `PARQUET_TYPE_NOT_RECOGNIZED` and Comet raises `_LEGACY_ERROR_TEMP_3071`.
+/// `variant_annotation_without_a_spec_version_is_also_rejected` pins that.
+///
+/// The one behavioral gap is `ignore_variant_annotation` on a non-v1 annotation. Spark's guard
+/// fails before it reaches its own ignore branch, so Spark still raises
+/// `PARQUET_TYPE_NOT_RECOGNIZED`, while Comet skips the check entirely and reads the plain
+/// struct. Comet is the more permissive of the two there. Closing it would mean pairing the
+/// requested schema against the Parquet `SchemaDescriptor` rather than the Arrow schema, which
+/// duplicates the field-id and case-folding rules `check_variant_annotation` reuses. Closing it
+/// means either reading the version from the Parquet schema or arrow-rs carrying it on the
+/// extension type.
+fn is_variant_marked(field: &Field) -> bool {
+ field.has_valid_extension_type::<VariantType>()
+}
+
+/// Reject reading a Parquet field that carries the VARIANT logical type annotation as anything
+/// other than Spark's `VariantType`, mirroring the `checkConversionRequirement` in Spark's
+/// `ParquetToSparkSchemaConverter.convertGroupField`.
+///
+/// Comet's scan never runs Spark's schema converter, so this is the only place the file's
+/// annotation is ever compared against the requested type. The two sides are compared
+/// symmetrically through `is_variant_marked`: a marked request is a legitimate Variant read (see
+/// `parquet_exec::init_datasource_exec`'s `projects_variant`) and must not be rejected.
+/// Identifying a Variant by its `value`/`metadata` child names instead would misclassify ordinary
+/// structs, which is what apache/datafusion-comet#5741 rules out.
+///
+/// Runs once per file when the reader opens it, before any row group is inspected, so a file with
+/// no row groups is rejected too. That matches Spark, which rejects while converting the schema,
+/// and is the opposite of the type-promotion checks in this module, which `RejectOnNonEmpty`
+/// defers to execution to match Spark's per-row-group behavior.
+fn check_variant_annotation(
+ logical: &FieldRef,
+ physical: &FieldRef,
+ parquet_options: &SparkParquetOptions,
+ path: &mut Vec<String>,
+) -> DataFusionResult<()> {
+ path.push(logical.name().clone());
+ if is_variant_marked(physical) && !is_variant_marked(logical) {
+ return Err(variant_annotation_err(&path.join("."), logical.data_type()));
+ }
+ match (logical.data_type(), physical.data_type()) {
+ (DataType::Struct(logical_fields), DataType::Struct(physical_fields)) => {
+ // Pair nested fields the same way `spark_parquet_convert` does, so the field this
+ // check inspects is the one the read will actually pull from the file: when the
+ // logical struct carries Parquet field IDs anywhere, ID-bearing logical fields match
+ // ONLY by ID and the rest fall back to the folded name. A logical field with no
+ // counterpart in the file is null-filled and carries no annotation to check.
+ let should_match_by_id = parquet_options.use_field_id
+ && logical_fields.iter().any(|f| parse_field_id(f).is_some());
+ let physical_id_to_index: HashMap<i32, usize> = if should_match_by_id {
+ let mut map = HashMap::new();
+ for (i, field) in physical_fields.iter().enumerate() {
+ if let Some(id) = parse_field_id(field) {
+ map.entry(id).or_insert(i);
+ }
+ }
+ map
+ } else {
+ HashMap::new()
+ };
+ let physical_names = physical_fields
+ .iter()
+ .map(|field| field.name().as_str())
+ .collect::<Vec<_>>();
+ let physical_folded = fold_names(&physical_names, parquet_options.case_sensitive);
+ // First match wins on a folded-name collision, as `spark_parquet_convert` does when it
+ // falls through its ambiguity check; `collect` would arbitrarily keep the last.
+ let mut folded_to_index: HashMap<&str, usize> = HashMap::new();
+ for (i, folded) in physical_folded.iter().enumerate() {
+ folded_to_index.entry(folded.as_str()).or_insert(i);
+ }
+
+ for logical_child in logical_fields {
+ let physical_index = match (should_match_by_id, parse_field_id(logical_child)) {
+ (true, Some(id)) => physical_id_to_index.get(&id).copied(),
+ _ => {
+ let folded =
+ fold_name(logical_child.name(), parquet_options.case_sensitive);
+ folded_to_index.get(folded.as_str()).copied()
+ }
+ };
+ if let Some(i) = physical_index {
+ check_variant_annotation(
+ logical_child,
+ &physical_fields[i],
+ parquet_options,
+ path,
+ )?;
+ }
+ }
+ }
+ (DataType::List(logical_item), DataType::List(physical_item))
+ | (DataType::LargeList(logical_item), DataType::LargeList(physical_item))
+ | (DataType::List(logical_item), DataType::LargeList(physical_item))
+ | (DataType::LargeList(logical_item), DataType::List(physical_item)) => {
+ check_variant_annotation(logical_item, physical_item, parquet_options, path)?;
+ }
+ (DataType::Map(logical_entries, _), DataType::Map(physical_entries, _)) => {
Review Comment:
### Correctness
[P2] Could the map branch validate the key and value by position instead of recursing through the entries struct's name/field-ID matcher? The requested Arrow map uses `key`/`value`, but parquet-rs retains the file's child names. If the second child is named `payload` and carries `VARIANT(1)`, the struct matcher finds no `value` field and skips its annotation. Both `check_conversion` and `parquet_convert_map_to_map` still read that second child positionally, so the incompatible plain-struct map value gets past this new guard. Spark's map reader also selects children 0 and 1 rather than requiring those names. Please use the same positional pairing here and cover an annotated map value whose Parquet field name differs from `value`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#issuecomment-5847887609
Oops I'm sorry @0lai0 ! could you resolve the conflict again?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#issuecomment-5928989794
## Status summary
Comet's scan never runs Spark's schema converter, so nothing compared a Parquet file's VARIANT annotation against the type the read asked for: a VARIANT column read as `struct<value binary, metadata binary>` silently returned the storage bytes where Spark raises `_LEGACY_ERROR_TEMP_3071`. This adds that check. Closes #5741.
### The five commits, in reading order
| Commit | What it does |
|---|---|
| `5508366` | The check itself: `check_variant_annotation` in `schema_adapter.rs`, the error carrier, the serde and proto plumbing for `spark.sql.parquet.ignoreVariantAnnotation` |
| `03f9da0` | Re-enables the Spark test the 4.1 diff was skipping for #5741 |
| `8c1cb78` | Scopes the check to the requested read schema, and pairs map children by position the way `MapArray` reads them |
| `96087ba` | Review fix: takes the marker from the file's annotation rather than its `ARROW:schema` hint (below) |
| `a29268d` | Stops skipping the same Spark test on 4.2, whose diff landed on main in #4950 |
### Both review points are addressed
@andygrove found that the physical side of the check was the reader's Arrow field, which parquet-rs builds from the `ARROW:schema` hint when a file carries one, never from the Parquet logical type.
That made the marker the hint's rather than the file's, so Comet rejected a column Spark reads (hint marked, group unannotated) and read one Spark rejects (the reverse). `96087ba` reconciles the hint's Variant markers against the annotations in `EagerPageIndexReaderFactory`, before the adapter sees either, and adds a test per direction; both fail with the reconciliation disabled. It also installs the reader factory in `probe_variant_annotation`, which had been building its own `ParquetSource` without it — the reason no Variant test could have caught this.
`a29268d` regenerates `dev/diffs/4.2.0.diff` from a clean `v4.2.0` clone per the Spark SQL Tests guide. The only change from the previous diff is that one hunk's removal.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on code in PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#discussion_r4029410259
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -434,6 +434,183 @@ fn is_string_or_binary(dt: &DataType) -> bool {
)
}
+/// Approximate Spark's `DataType.sql` for the variant rejection message. `spark_catalog_name`
+/// bottoms out at "unknown" for nested types, which is exactly the shape a hand-written Variant
+/// read schema has, so the nested cases are spelled out here.
+///
+/// This is close to `DataType.sql` but not identical. Spark's `StructField.sql` wraps names in
+/// `QuotingUtils.quoteIfNeeded` and appends the nullability and comment DDL, none of which is
+/// reproduced here, so a field named `has space` renders bare. The whole message already differs
+/// from Spark's by design (see `variant_annotation_err`), and the rendering exists to tell a user
+/// which type they asked for, so it is kept simple rather than made byte-identical.
+fn spark_read_type_name(dt: &DataType) -> String {
+ match dt {
+ DataType::Struct(fields) => {
+ let rendered = fields
+ .iter()
+ .map(|field| {
+ format!(
+ "{}: {}",
+ field.name(),
+ spark_read_type_name(field.data_type())
+ )
+ })
+ .collect::<Vec<_>>()
+ .join(", ");
+ format!("STRUCT<{rendered}>")
+ }
+ DataType::List(item) | DataType::LargeList(item) => {
+ format!("ARRAY<{}>", spark_read_type_name(item.data_type()))
+ }
+ DataType::Map(entries, _) => match entries.data_type() {
+ DataType::Struct(fields) if fields.len() == 2 => format!(
+ "MAP<{}, {}>",
+ spark_read_type_name(fields[0].data_type()),
+ spark_read_type_name(fields[1].data_type())
+ ),
+ other => format!("MAP<{}>", spark_read_type_name(other)),
+ },
+ other => spark_catalog_name(other).to_uppercase(),
+ }
+}
+
+/// Build the carrier for a Parquet field whose VARIANT logical type annotation is incompatible
+/// with the requested Spark read type. The JVM shim turns these into Spark's
+/// `_LEGACY_ERROR_TEMP_3071` `AnalysisException`.
+///
+/// `column` is the dotted path to the offending field. Spark's own message interpolates the
+/// Parquet `Type` itself, roughly `optional group v (VARIANT(1)) { ... }`, so the two messages
+/// deliberately differ. `ParquetVariantShreddingSuite` matches on `Invalid Spark read type` and
+/// accepts either, and a path is the more useful of the two once the field is nested.
+fn variant_annotation_err(column: &str, target_type: &DataType) -> DataFusionError {
+ DataFusionError::External(Box::new(SparkError::ParquetVariantAnnotationMismatch {
+ file_path: String::new(),
+ column: column.to_string(),
+ spark_type: spark_read_type_name(target_type),
+ }))
+}
+
+/// Whether `field` is marked as a Variant. On a physical field the marker is the Parquet VARIANT
+/// logical type annotation, which arrow-rs surfaces as the `arrow.parquet.variant` Arrow
+/// extension type for struct fields, list elements and map values alike. On a requested field it
+/// is the same marker Comet's serde applies for `VariantType`. `check_variant_annotation` relies
+/// on both sides using it, so it compares them through this one predicate.
+///
+/// The marker does not carry the annotation's spec version, and Spark keys its handling off that
+/// version. `ParquetToSparkSchemaConverter.convertGroupField` matches
+/// `case v: VariantLogicalTypeAnnotation if v.getSpecVersion == 1`, and an annotation failing
+/// that guard falls through to `case _ => throw unrecognizedParquetTypeError(...)`, which is
+/// `PARQUET_TYPE_NOT_RECOGNIZED`. arrow-rs matches `LogicalType::Variant(_)` for any version, so
+/// this predicate cannot tell the two apart. Spark's writer emits version 1. An Arrow-based
+/// writer emits no version at all, which parquet-java reads back as 0.
+///
+/// Both engines therefore reject a non-v1 annotation, with different error classes: Spark raises
+/// `PARQUET_TYPE_NOT_RECOGNIZED` and Comet raises `_LEGACY_ERROR_TEMP_3071`.
+/// `variant_annotation_without_a_spec_version_is_also_rejected` pins that.
+///
+/// The one behavioral gap is `ignore_variant_annotation` on a non-v1 annotation. Spark's guard
+/// fails before it reaches its own ignore branch, so Spark still raises
+/// `PARQUET_TYPE_NOT_RECOGNIZED`, while Comet skips the check entirely and reads the plain
+/// struct. Comet is the more permissive of the two there. Closing it would mean pairing the
+/// requested schema against the Parquet `SchemaDescriptor` rather than the Arrow schema, which
+/// duplicates the field-id and case-folding rules `check_variant_annotation` reuses. Closing it
+/// means either reading the version from the Parquet schema or arrow-rs carrying it on the
+/// extension type.
+fn is_variant_marked(field: &Field) -> bool {
+ field.has_valid_extension_type::<VariantType>()
+}
+
+/// Reject reading a Parquet field that carries the VARIANT logical type annotation as anything
+/// other than Spark's `VariantType`, mirroring the `checkConversionRequirement` in Spark's
+/// `ParquetToSparkSchemaConverter.convertGroupField`.
+///
+/// Comet's scan never runs Spark's schema converter, so this is the only place the file's
+/// annotation is ever compared against the requested type. The two sides are compared
+/// symmetrically through `is_variant_marked`: a marked request is a legitimate Variant read (see
+/// `parquet_exec::init_datasource_exec`'s `projects_variant`) and must not be rejected.
+/// Identifying a Variant by its `value`/`metadata` child names instead would misclassify ordinary
+/// structs, which is what apache/datafusion-comet#5741 rules out.
+///
+/// Runs once per file when the reader opens it, before any row group is inspected, so a file with
+/// no row groups is rejected too. That matches Spark, which rejects while converting the schema,
+/// and is the opposite of the type-promotion checks in this module, which `RejectOnNonEmpty`
+/// defers to execution to match Spark's per-row-group behavior.
+fn check_variant_annotation(
+ logical: &FieldRef,
+ physical: &FieldRef,
+ parquet_options: &SparkParquetOptions,
+ path: &mut Vec<String>,
+) -> DataFusionResult<()> {
+ path.push(logical.name().clone());
+ if is_variant_marked(physical) && !is_variant_marked(logical) {
+ return Err(variant_annotation_err(&path.join("."), logical.data_type()));
+ }
+ match (logical.data_type(), physical.data_type()) {
+ (DataType::Struct(logical_fields), DataType::Struct(physical_fields)) => {
+ // Pair nested fields the same way `spark_parquet_convert` does, so the field this
+ // check inspects is the one the read will actually pull from the file: when the
+ // logical struct carries Parquet field IDs anywhere, ID-bearing logical fields match
+ // ONLY by ID and the rest fall back to the folded name. A logical field with no
+ // counterpart in the file is null-filled and carries no annotation to check.
+ let should_match_by_id = parquet_options.use_field_id
+ && logical_fields.iter().any(|f| parse_field_id(f).is_some());
+ let physical_id_to_index: HashMap<i32, usize> = if should_match_by_id {
+ let mut map = HashMap::new();
+ for (i, field) in physical_fields.iter().enumerate() {
+ if let Some(id) = parse_field_id(field) {
+ map.entry(id).or_insert(i);
+ }
+ }
+ map
+ } else {
+ HashMap::new()
+ };
+ let physical_names = physical_fields
+ .iter()
+ .map(|field| field.name().as_str())
+ .collect::<Vec<_>>();
+ let physical_folded = fold_names(&physical_names, parquet_options.case_sensitive);
+ // First match wins on a folded-name collision, as `spark_parquet_convert` does when it
+ // falls through its ambiguity check; `collect` would arbitrarily keep the last.
+ let mut folded_to_index: HashMap<&str, usize> = HashMap::new();
+ for (i, folded) in physical_folded.iter().enumerate() {
+ folded_to_index.entry(folded.as_str()).or_insert(i);
+ }
+
+ for logical_child in logical_fields {
+ let physical_index = match (should_match_by_id, parse_field_id(logical_child)) {
+ (true, Some(id)) => physical_id_to_index.get(&id).copied(),
+ _ => {
+ let folded =
+ fold_name(logical_child.name(), parquet_options.case_sensitive);
+ folded_to_index.get(folded.as_str()).copied()
+ }
+ };
+ if let Some(i) = physical_index {
+ check_variant_annotation(
+ logical_child,
+ &physical_fields[i],
+ parquet_options,
+ path,
+ )?;
+ }
+ }
+ }
+ (DataType::List(logical_item), DataType::List(physical_item))
+ | (DataType::LargeList(logical_item), DataType::LargeList(physical_item))
+ | (DataType::List(logical_item), DataType::LargeList(physical_item))
+ | (DataType::LargeList(logical_item), DataType::List(physical_item)) => {
+ check_variant_annotation(logical_item, physical_item, parquet_options, path)?;
+ }
+ (DataType::Map(logical_entries, _), DataType::Map(physical_entries, _)) => {
Review Comment:
Sure, fixed. The map branch now validates the key and value by position. variant_annotation_on_a_renamed_map_value_is_rejected covers an annotated map value named payload.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on code in PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#discussion_r4029395323
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -911,6 +1088,29 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory {
None
};
+ // Compare the file's VARIANT annotations against the requested types before handing the
+ // schemas to the default adapter. `adapted_physical_schema` is used so that field-id and
+ // case-insensitive resolution has already aligned the two sides' top-level names.
+ if !self.parquet_options.ignore_variant_annotation {
+ let mut physical_by_folded: HashMap<&str, usize> = HashMap::new();
+ for (i, name) in physical_folded.iter().enumerate() {
+ physical_by_folded.entry(name.as_str()).or_insert(i);
+ }
+ let mut path = Vec::new();
+ for (logical_field, folded) in logical_file_schema.fields().iter().zip(&logical_folded)
Review Comment:
Thanks @sunchao. Sure, fixed. The validation now uses the requested read schema. `init_datasource_exec `passes required_schema to the factory via `with_required_schema`, so unrequested roots are skipped, and requested fields are still checked eagerly in `create()`, including empty files. Regressions in `parquet_exec/variant_tests.rs` cover an omitted annotated root (reads fine) and a requested one (rejected).
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#issuecomment-5772062914
Resolved, Thanks @sunchao
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on code in PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#discussion_r4284000199
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -4643,6 +4884,930 @@ pub(crate) mod test {
assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap());
}
+ /// Verification probe for apache/datafusion-comet#5741.
+ ///
+ /// Captures the `physical_file_schema` DataFusion hands to
+ /// `PhysicalExprAdapterFactory::create`, so the tests below can assert both that `create`
+ /// was reached at all and what the file schema looked like when it was.
+ #[derive(Debug)]
+ struct ProbeFactory {
+ inner: SparkPhysicalExprAdapterFactory,
+ seen: Arc<Mutex<Option<SchemaRef>>>,
+ }
+
+ impl PhysicalExprAdapterFactory for ProbeFactory {
+ fn create(
+ &self,
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ ) -> datafusion::common::Result<
+ Arc<dyn datafusion_physical_expr_adapter::PhysicalExprAdapter>,
+ > {
+ *self.seen.lock().unwrap() = Some(Arc::clone(&physical_file_schema));
+ self.inner.create(logical_file_schema, physical_file_schema)
+ }
+ }
+
+ /// Recursively drop every Arrow extension marker from `field`, producing the schema a Spark
+ /// user's hand-written DDL yields: the same structure, no variant identity. Mirrors the
+ /// `.schema("v struct<value binary, metadata binary>")` in Spark's
+ /// `ParquetVariantShreddingSuite`, test
+ /// `variant logical type annotation - ignore variant annotation`.
+ fn strip_extension_markers(field: &FieldRef) -> FieldRef {
+ let data_type = match field.data_type() {
+ DataType::Struct(fields) => {
+ DataType::Struct(fields.iter().map(strip_extension_markers).collect())
+ }
+ DataType::List(item) => DataType::List(strip_extension_markers(item)),
+ DataType::Map(entries, sorted) => {
+ DataType::Map(strip_extension_markers(entries), *sorted)
+ }
+ other => other.clone(),
+ };
+ let mut metadata = field.metadata().clone();
+ metadata.remove("ARROW:extension:name");
+ metadata.remove("ARROW:extension:metadata");
+ Arc::new(Field::new(field.name(), data_type, field.is_nullable()).with_metadata(metadata))
+ }
+
+ /// Write `batch` to a temp Parquet file, then scan it back through `DataSourceExec` asking for
+ /// the same schema with every extension marker stripped. Reports the Parquet schema the writer
+ /// actually produced alongside the physical file schema `create` observed (`None` when `create`
+ /// never ran).
+ ///
+ /// Returning the Parquet schema separates the two ways a shape can fail: the *writer* never
+ /// emitting the VARIANT annotation, versus the *reader* not surfacing it as an Arrow extension
+ /// type. Only the second one is a problem for #5741; the first is a limitation of the test.
+ ///
+ /// `skip_arrow_metadata` controls whether the file embeds an `ARROW:schema` key-value entry.
+ /// Spark's writer embeds none, so the Spark-shaped cases pass `true`: a marker on the way back
+ /// can then only have come from the Parquet annotation, and a file carrying the hint would let
+ /// every assertion below pass vacuously.
+ struct VariantProbe {
+ /// The Parquet schema the writer actually produced.
+ parquet_schema: String,
+ /// The physical file schema `create` observed, or `None` when it never ran.
+ observed_physical: Option<SchemaRef>,
+ /// Total rows read, or the error the scan raised.
+ scan: Result<usize, DataFusionError>,
+ }
+
+ struct ProbeOptions {
+ /// Whether the file omits the embedded `ARROW:schema` hint, as Spark's writer does.
+ skip_arrow_metadata: bool,
+ /// `spark.sql.parquet.ignoreVariantAnnotation`.
+ ignore_variant_annotation: bool,
+ /// Keep the Variant markers on the requested schema, modelling a genuine native Variant
+ /// projection (`projects_variant` in `parquet_exec.rs`) instead of a hand-written struct.
+ keep_request_markers: bool,
+ /// Request this schema instead of one derived from the file schema, for cases where
+ /// Spark's requested names differ from the names the file was written with.
+ requested_schema: Option<SchemaRef>,
+ }
+
+ impl Default for ProbeOptions {
+ fn default() -> Self {
+ Self {
+ skip_arrow_metadata: true,
+ ignore_variant_annotation: false,
+ keep_request_markers: false,
+ requested_schema: None,
+ }
+ }
+ }
+
+ async fn probe_variant_annotation(
+ batch: RecordBatch,
+ options: ProbeOptions,
+ ) -> Result<VariantProbe, DataFusionError> {
+ let file_schema = batch.schema();
+ let filename = get_temp_filename();
+ let filename = filename.as_path().as_os_str().to_str().unwrap().to_string();
+ let file = File::create(&filename)?;
+ let mut writer = ArrowWriter::try_new_with_options(
+ file,
+ Arc::clone(&file_schema),
+ ArrowWriterOptions::new().with_skip_arrow_metadata(options.skip_arrow_metadata),
+ )?;
+ writer.write(&batch)?;
+ writer.close()?;
+
+ probe_variant_annotation_file(&filename, &file_schema, options).await
+ }
+
+ /// The reading half of `probe_variant_annotation`: reads back the Parquet schema `filename`
+ /// actually holds, derives the requested schema from `file_schema` per `options`, and scans
+ /// the file through a `ProbeFactory`. Split out so a test can hand it a file the `ArrowWriter`
+ /// alone cannot produce, such as one whose `ARROW:schema` hint disagrees with its annotations.
+ async fn probe_variant_annotation_file(
+ filename: &str,
+ file_schema: &SchemaRef,
+ options: ProbeOptions,
+ ) -> Result<VariantProbe, DataFusionError> {
+ let filename = filename.to_string();
+ let file_schema = Arc::clone(file_schema);
+
+ let mut printed = Vec::new();
+ let reader = ParquetRecordBatchReaderBuilder::try_new(File::open(&filename)?)?;
+ print_schema(&mut printed, reader.parquet_schema().root_schema());
+ let printed = String::from_utf8(printed).unwrap();
+
+ let requested = if let Some(requested) = options.requested_schema {
+ requested
+ } else if options.keep_request_markers {
+ Arc::clone(&file_schema)
+ } else {
+ Arc::new(Schema::new(
+ file_schema
+ .fields()
+ .iter()
+ .map(strip_extension_markers)
+ .collect::<Fields>(),
+ ))
+ };
+
+ let mut spark_parquet_options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ spark_parquet_options.ignore_variant_annotation = options.ignore_variant_annotation;
+ let seen = Arc::new(Mutex::new(None));
+ let expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory> = Arc::new(ProbeFactory {
+ inner: SparkPhysicalExprAdapterFactory::new(spark_parquet_options, None),
+ seen: Arc::clone(&seen),
+ });
+
+ // Install the reader factory `init_datasource_exec` installs, gated the way it gates it:
+ // the footer rewrites that decide which fields the reader marks as Variant live there, so
+ // a probe without it would read a hint the real scan never sees.
+ let projects_variant = requested
+ .fields()
+ .iter()
+ .any(|field| field.has_valid_extension_type::<VariantType>());
+ let runtime = Arc::new(RuntimeEnv::default());
+ let store = runtime.object_store(ObjectStoreUrl::local_filesystem())?;
+ let reader_factory = Arc::new(
+ EagerPageIndexReaderFactory::new(
+ store,
+ runtime.cache_manager.get_file_metadata_cache(),
+ ScanIoSource::Local,
+ &ExecutionPlanMetricsSet::new(),
+ )
+ .with_spark_variant_schema(projects_variant),
+ );
+ let parquet_source =
+ ParquetSource::new(requested).with_parquet_file_reader_factory(reader_factory);
+ let files = FileGroup::new(vec![PartitionedFile::from_path(filename)?]);
+ let file_scan_config = FileScanConfigBuilder::new(
+ ObjectStoreUrl::local_filesystem(),
+ Arc::new(parquet_source),
+ )
+ .with_file_groups(vec![files])
+ .with_expr_adapter(Some(expr_adapter_factory))
+ .build();
+ let parquet_exec = DataSourceExec::new(Arc::new(file_scan_config));
+ let scan = async {
+ let mut stream = parquet_exec.execute(0, Arc::new(TaskContext::default()))?;
+ let mut rows = 0;
+ while let Some(batch) = stream.next().await {
+ rows += batch?.num_rows();
+ }
+ Ok(rows)
+ }
+ .await;
+
+ let observed_physical = seen.lock().unwrap().clone();
+ Ok(VariantProbe {
+ parquet_schema: printed,
+ observed_physical,
+ scan,
+ })
+ }
+
+ /// The two-field Variant storage group Spark writes when shredding is off.
+ fn variant_storage_fields() -> Fields {
+ Fields::from(vec![
+ Field::new("value", DataType::Binary, false),
+ Field::new("metadata", DataType::Binary, false),
+ ])
+ }
+
+ /// A `StructArray` holding one Variant value: the encoded integer `1`, matching the
+ /// `Row(Array[Byte](12, 1), Array[Byte](1, 0, 0))` the Spark suite expects.
+ fn variant_storage_array() -> StructArray {
+ let value = Arc::new(BinaryArray::from(vec![Some(&[12u8, 1u8][..])])) as ArrayRef;
+ let meta = Arc::new(BinaryArray::from(vec![Some(&[1u8, 0u8, 0u8][..])])) as ArrayRef;
+ StructArray::new(variant_storage_fields(), vec![value, meta], None)
+ }
+
+ /// An Arrow field named `name` carrying the Variant storage type and the
+ /// `arrow.parquet.variant` extension marker, which parquet-rs's writer turns into the Parquet
+ /// VARIANT logical type annotation.
+ fn variant_field(name: &str) -> FieldRef {
+ Arc::new(
+ Field::new(name, DataType::Struct(variant_storage_fields()), true)
+ .with_extension_type(VariantType),
+ )
+ }
+
+ /// Assert that `probe_variant_annotation` saw a live `create` call on a file the writer really
+ /// annotated, then hand the observed physical schema to `locate` to check the marker survived
+ /// at the shape-specific position.
+ fn assert_annotation_survived(
+ shape: &str,
+ probe: &VariantProbe,
+ locate: impl Fn(&SchemaRef) -> Option<FieldRef>,
+ ) {
+ let printed = &probe.parquet_schema;
+ assert!(
+ printed.contains("VARIANT"),
+ "{shape}: parquet-rs did not write the VARIANT annotation, so this file cannot test \
+ the reader at all. Parquet schema:\n{printed}"
+ );
+ let observed = probe.observed_physical.clone().unwrap_or_else(|| {
+ panic!(
+ "{shape}: PhysicalExprAdapterFactory::create was never called: the logical and \
+ physical file schemas compared equal, so the variant annotation did not survive \
+ into the physical schema"
+ )
+ });
+ let field = locate(&observed).unwrap_or_else(|| {
+ panic!("{shape}: could not locate the variant field in {observed:?}")
+ });
+ assert!(
+ field.has_valid_extension_type::<VariantType>(),
+ "{shape}: physical file schema lost the variant annotation: {field:?}"
+ );
+ }
+
+ /// One row of `map<string, variant>`, the `mv` column of the Spark suite, with the map value
+ /// child named `value_name` in the file.
+ fn map_of_variant_batch(value_name: &str) -> Result<RecordBatch, DataFusionError> {
+ let value_field = variant_field(value_name);
+ let entry_fields = Fields::from(vec![
+ Arc::new(Field::new("key", DataType::Utf8, false)),
+ Arc::clone(&value_field),
+ ]);
+ let entries = StructArray::new(
+ entry_fields.clone(),
+ vec![
+ Arc::new(StringArray::from(vec!["v2"])) as ArrayRef,
+ Arc::new(variant_storage_array()) as ArrayRef,
+ ],
+ None,
+ );
+ let entries_field = Arc::new(Field::new("entries", DataType::Struct(entry_fields), false));
+ let map = MapArray::new(
+ Arc::clone(&entries_field),
+ OffsetBuffer::new(vec![0, 1].into()),
+ entries,
+ None,
+ false,
+ );
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "mv",
+ DataType::Map(entries_field, false),
+ true,
+ )]));
+ Ok(RecordBatch::try_new(schema, vec![Arc::new(map)])?)
+ }
+
+ /// Assert the scan was rejected with Spark's `_LEGACY_ERROR_TEMP_3071` shape, naming `column`
+ /// (a dotted path) and the requested type.
+ fn assert_variant_annotation_rejected(probe: &VariantProbe, column: &str, spark_type: &str) {
+ let err = probe
+ .scan
+ .as_ref()
+ .expect_err("reading a VARIANT-annotated field as a plain struct must be rejected");
+ let msg = err.to_string();
+ assert!(
+ msg.contains("_LEGACY_ERROR_TEMP_3071")
+ && msg.contains("Invalid Spark read type")
+ && msg.contains(column)
+ && msg.contains(spark_type),
+ "unexpected error for {column}: {msg}"
+ );
+ }
+
+ /// The plain struct a hand-written Spark read schema produces for a Variant column.
+ fn plain_variant_storage_sql() -> &'static str {
+ "STRUCT<value: BINARY, metadata: BINARY>"
+ }
+
+ /// #5741: reading a VARIANT-annotated top-level field as
+ /// `struct<value binary, metadata binary>` must raise Spark's `_LEGACY_ERROR_TEMP_3071`
+ /// instead of silently returning the storage
+ /// struct. This is the `v` column of the Spark suite's `false` arm.
+ #[tokio::test]
+ async fn variant_annotation_read_as_plain_struct_is_rejected() -> Result<(), DataFusionError> {
+ let schema = Arc::new(Schema::new(vec![variant_field("v")]));
+ let batch = RecordBatch::try_new(schema, vec![Arc::new(variant_storage_array())])?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ assert_variant_annotation_rejected(&probe, "v", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// The `ns struct<nv variant>` case: the annotation is on a nested field, and the rejection
+ /// must name the dotted path rather than just the root column.
+ #[tokio::test]
+ async fn variant_annotation_inside_struct_is_rejected() -> Result<(), DataFusionError> {
+ let outer_fields = Fields::from(vec![variant_field("nv")]);
+ let outer = StructArray::new(
+ outer_fields.clone(),
+ vec![Arc::new(variant_storage_array()) as ArrayRef],
+ None,
+ );
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "ns",
+ DataType::Struct(outer_fields),
+ true,
+ )]));
+ let batch = RecordBatch::try_new(schema, vec![Arc::new(outer)])?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ assert_variant_annotation_rejected(&probe, "ns.nv", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// The `av array<variant>` case.
+ #[tokio::test]
+ async fn variant_annotation_as_list_element_is_rejected() -> Result<(), DataFusionError> {
+ let element = variant_field("element");
+ let list = ListArray::new(
+ Arc::clone(&element),
+ OffsetBuffer::new(vec![0, 1].into()),
+ Arc::new(variant_storage_array()),
+ None,
+ );
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "av",
+ DataType::List(element),
+ true,
+ )]));
+ let batch = RecordBatch::try_new(schema, vec![Arc::new(list)])?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ assert_variant_annotation_rejected(&probe, "av.element", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// The `mv map<string, variant>` case.
+ #[tokio::test]
+ async fn variant_annotation_as_map_value_is_rejected() -> Result<(), DataFusionError> {
+ let batch = map_of_variant_batch("value")?;
+ let probe = probe_variant_annotation(batch, ProbeOptions::default()).await?;
+ // The path walks through the Parquet map's `entries` group, as Spark's own error does.
+ assert_variant_annotation_rejected(&probe, "mv.entries.value", plain_variant_storage_sql());
+ Ok(())
+ }
+
+ /// In field-id read mode a nested logical field resolves to the file field with the matching
+ /// ID, not the matching name -- that is what `spark_parquet_convert` does when it builds the
+ /// read. Pairing by name here instead would inspect the wrong file field (or none) and let the
+ /// annotation through unchecked, so the rejection must follow the ID.
+ #[test]
+ fn variant_annotation_is_found_through_a_nested_field_id_match() {
+ let storage = DataType::Struct(variant_storage_fields());
+ let logical = Arc::new(Schema::new(vec![Field::new(
+ "ns",
+ DataType::Struct(Fields::from(vec![Field::new(
+ "renamed_since_write",
+ storage.clone(),
+ true,
+ )
+ .with_metadata(id_meta("7"))])),
+ true,
+ )]));
+ // The file holds the Variant under a different name, plus a decoy that would win a
+ // name-only match and carries no annotation.
+ let physical = Arc::new(Schema::new(vec![Field::new(
+ "ns",
+ DataType::Struct(Fields::from(vec![
+ Field::new("renamed_since_write", DataType::Int32, true)
+ .with_metadata(id_meta("9")),
+ Field::new("nv", storage, true)
+ .with_metadata(id_meta("7"))
+ .with_extension_type(VariantType),
+ ])),
+ true,
+ )]));
+
+ let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ options.use_field_id = true;
+ let err = SparkPhysicalExprAdapterFactory::new(options, None)
+ .create(logical, physical)
+ .expect_err("an ID-resolved Variant field must still be rejected");
+ let msg = err.to_string();
+ assert!(
+ msg.contains("_LEGACY_ERROR_TEMP_3071") && msg.contains("ns.renamed_since_write"),
+ "unexpected error: {msg}"
+ );
+ }
+
+ /// Write a one-row Parquet file holding a single top-level group annotated as VARIANT with
+ /// `spec_version`, using the low-level writer so the annotation is exactly what Spark's writer
+ /// emits. `ArrowWriter` cannot do this: `logical_type_for_struct` hardcodes
+ /// `LogicalType::variant(None)`, which is a different annotation from Spark's `VARIANT(1)`.
+ fn write_annotated_variant_file(spec_version: Option<i8>) -> String {
+ let value = Arc::new(
+ ParquetType::primitive_type_builder("value", PhysicalType::BYTE_ARRAY)
+ .with_repetition(Repetition::REQUIRED)
+ .build()
+ .unwrap(),
+ );
+ let metadata = Arc::new(
+ ParquetType::primitive_type_builder("metadata", PhysicalType::BYTE_ARRAY)
+ .with_repetition(Repetition::REQUIRED)
+ .build()
+ .unwrap(),
+ );
+ let variant = Arc::new(
+ ParquetType::group_type_builder("v")
+ .with_repetition(Repetition::REQUIRED)
+ .with_logical_type(Some(LogicalType::variant(spec_version)))
+ .with_fields(vec![value, metadata])
+ .build()
+ .unwrap(),
+ );
+ let schema = Arc::new(
+ ParquetType::group_type_builder("schema")
+ .with_fields(vec![variant])
+ .build()
+ .unwrap(),
+ );
+
+ let filename = get_temp_filename();
+ let filename = filename.as_path().as_os_str().to_str().unwrap().to_string();
+ let file = File::create(&filename).unwrap();
+ let mut writer = SerializedFileWriter::new(file, schema, Default::default()).unwrap();
+ let mut row_group = writer.next_row_group().unwrap();
+ for bytes in [vec![12u8, 1u8], vec![1u8, 0u8, 0u8]] {
+ let mut column = row_group.next_column().unwrap().unwrap();
+ column
+ .typed::<ByteArrayType>()
+ .write_batch(&[ByteArray::from(bytes)], None, None)
+ .unwrap();
+ column.close().unwrap();
+ }
+ row_group.close().unwrap();
+ writer.close().unwrap();
+ filename
+ }
+
+ /// Scan `filename` asking for the plain `struct<value binary, metadata binary>` a hand-written
+ /// Spark read schema produces, and report the scan outcome.
+ async fn scan_as_plain_variant_storage(filename: String) -> Result<usize, DataFusionError> {
+ let requested = Arc::new(Schema::new(vec![Field::new(
+ "v",
+ DataType::Struct(variant_storage_fields()),
+ false,
+ )]));
+ let expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory> =
+ Arc::new(SparkPhysicalExprAdapterFactory::new(
+ SparkParquetOptions::new(EvalMode::Legacy, "UTC", false),
+ None,
+ ));
+ let file_scan_config = FileScanConfigBuilder::new(
+ ObjectStoreUrl::local_filesystem(),
+ Arc::new(ParquetSource::new(requested)),
+ )
+ .with_file_groups(vec![FileGroup::new(vec![PartitionedFile::from_path(
+ filename,
+ )?])])
+ .with_expr_adapter(Some(expr_adapter_factory))
+ .build();
+ let exec = DataSourceExec::new(Arc::new(file_scan_config));
+ let mut stream = exec.execute(0, Arc::new(TaskContext::default()))?;
+ let mut rows = 0;
+ while let Some(batch) = stream.next().await {
+ rows += batch?.num_rows();
+ }
+ Ok(rows)
+ }
+
+ /// The annotation Spark itself writes is `VARIANT(1)`, and every other rejection test in this
+ /// module goes through `ArrowWriter`, which emits `VARIANT(None)` instead. This one writes the
+ /// real Spark shape so the rejection is pinned against the annotation users actually have on
+ /// disk, not only against the one the test writer happens to produce.
+ #[tokio::test]
+ async fn variant_annotation_with_spec_version_1_is_rejected() -> Result<(), DataFusionError> {
+ let err = scan_as_plain_variant_storage(write_annotated_variant_file(Some(1)))
+ .await
+ .expect_err("a VARIANT(1) field read as a plain struct must be rejected");
+ let msg = err.to_string();
+ assert!(
+ msg.contains("_LEGACY_ERROR_TEMP_3071") && msg.contains("Invalid Spark read type"),
+ "unexpected error: {msg}"
+ );
+ Ok(())
+ }
+
+ /// An annotation with no spec version, which is what an Arrow-based writer emits and
+ /// parquet-java reads back as 0. Spark rejects this too, but through its
+ /// `unrecognizedParquetTypeError` catch-all rather than the variant branch, so the error class
+ /// differs: `PARQUET_TYPE_NOT_RECOGNIZED` there against `_LEGACY_ERROR_TEMP_3071` here. The
+ /// `arrow.parquet.variant` extension type carries no version, so this check cannot tell the
+ /// file apart from `VARIANT(1)`. See `check_variant_annotation` for the full comparison. If a
+ /// future arrow-rs exposes the version, this test is the one to revisit.
+ #[tokio::test]
+ async fn variant_annotation_without_a_spec_version_is_also_rejected(
+ ) -> Result<(), DataFusionError> {
+ let err = scan_as_plain_variant_storage(write_annotated_variant_file(None))
+ .await
+ .expect_err("Comet rejects an unversioned VARIANT annotation that Spark would accept");
+ assert!(
+ err.to_string().contains("_LEGACY_ERROR_TEMP_3071"),
+ "unexpected error: {err}"
+ );
+ Ok(())
+ }
+
+ /// `spark.sql.parquet.ignoreVariantAnnotation=true` is the explicit opt-in to read the
+ /// annotated group as its plain underlying struct. The `true` arm of the Spark suite: the read
+ /// must succeed and return the Variant storage bytes unchanged.
+ #[tokio::test]
+ async fn variant_annotation_is_ignored_when_conf_is_set() -> Result<(), DataFusionError> {
+ let top_level = Arc::new(Schema::new(vec![variant_field("v")]));
+ let cases = vec![
+ (
+ "top level",
+ RecordBatch::try_new(top_level, vec![Arc::new(variant_storage_array())])?,
+ ),
+ // The conf short-circuits the whole walk, so a nested annotation must be ignored too.
+ ("map value", map_of_variant_batch("value")?),
+ ];
+ for (shape, batch) in cases {
+ let probe = probe_variant_annotation(
+ batch,
+ ProbeOptions {
+ ignore_variant_annotation: true,
+ ..Default::default()
+ },
+ )
+ .await?;
+ let rows = probe.scan.unwrap_or_else(|err| {
+ panic!(
+ "{shape}: ignoreVariantAnnotation=true must allow the plain struct read: {err}"
+ )
+ });
+ assert_eq!(rows, 1, "{shape}: unexpected row count");
+ }
+ Ok(())
+ }
+
+ /// The other half of the symmetric check: when the *requested* schema is itself marked as a
+ /// Variant -- a genuine native Variant projection, which #5794 made reachable -- the annotation
+ /// agrees with the request and the read must not be rejected. Without this test, "simplifying"
+ /// the check to fire on the physical marker alone would break Variant projection with nothing
+ /// turning red.
+ #[tokio::test]
+ async fn marked_variant_request_is_not_rejected() -> Result<(), DataFusionError> {
+ let schema = Arc::new(Schema::new(vec![variant_field("v")]));
+ let batch = RecordBatch::try_new(schema, vec![Arc::new(variant_storage_array())])?;
+ let probe = probe_variant_annotation(
+ batch,
+ ProbeOptions {
+ keep_request_markers: true,
+ ..Default::default()
+ },
+ )
+ .await?;
+ assert_eq!(
+ probe
+ .scan
+ .expect("a marked Variant request must not trip the annotation check"),
+ 1
+ );
+ Ok(())
+ }
+
+ /// Spark rejects this read inside `ParquetToSparkSchemaConverter`, while converting the schema,
+ /// so a file with no rows fails just the same. That is the opposite of the type-promotion
+ /// checks in this module, which `RejectOnNonEmpty` defers to match Spark's per-row-group
+ /// behavior (see `parquet_empty_file_disallowed_widening`, SPARK-26709). Pinning it here keeps
+ /// a later reader from "fixing" this check to match its neighbours.
+ #[tokio::test]
+ async fn variant_annotation_is_rejected_on_an_empty_file() -> Result<(), DataFusionError> {
Review Comment:
Dropped, 102 lines. You are right that the twins carry it: each asserts the full dotted path (v, ns.nv, av.element, mv.entries.value), which the error can only produce if the marker was found at that position. The empty-file reasoning moved onto `requested_annotated_root_is_rejected_on_an_empty_file`.
One correction: the helpers stay in use, since `survives_with_embedded_arrow_schema` still calls them. Clippy is clean.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: validate Parquet VARIANT annotation against the requested read type [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#issuecomment-6194156426
Merging main also extracted the serde body into `buildNativeScanCommon`, which contrib/delta-spark's CometDeltaNativeScan also calls, so setIgnoreVariantAnnotation now covers the Delta arm.
That is the behavior it should have had: before, Delta took the proto default and would raise `_LEGACY_ERROR_TEMP_3071` on Spark versions whose converter has no variant branch at all.
893 tests pass, clippy and fmt clean, `./mvnw -DskipTests` compile green.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org