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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
