You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/20 10:38:05 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-7085-39380cca813bc0b861353de2cb53af960dd2a87f created (now 8fc44b5274)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-7085-39380cca813bc0b861353de2cb53af960dd2a87f
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at 8fc44b5274 fix: match Spark nullness for missing nested Parquet fields (#7085)
This branch includes the following new commits:
new 8fc44b5274 fix: match Spark nullness for missing nested Parquet fields (#7085)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: fix: match Spark nullness for missing nested Parquet fields (#7085)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7085-39380cca813bc0b861353de2cb53af960dd2a87f
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 8fc44b527415957eb7b85fc95dbaefe09829a62c
Author: KUAN-HAO HUANG <10...@users.noreply.github.com>
AuthorDate: Tue Oct 20 10:37:45 2026 +0000
fix: match Spark nullness for missing nested Parquet fields (#7085)
* fix: match Spark nullness for missing nested Parquet fields
* fix: fall back for name-mapped nested Iceberg scans
* fix: decline Iceberg scans when table properties cannot be read
* docs: migrate nested-field notes to consolidated scan guides
---
docs/source/user-guide/latest/iceberg.md | 14 +-
docs/source/user-guide/latest/parquet.md | 13 +-
native/core/src/parquet/parquet_support.rs | 271 ++++++++++++++++++---
.../apache/comet/iceberg/IcebergReflection.scala | 14 +-
.../org/apache/comet/rules/CometScanRule.scala | 16 +-
.../comet/serde/operator/CometNativeScan.scala | 6 +-
.../org/apache/comet/CometIcebergNativeSuite.scala | 105 ++++++++
.../apache/comet/exec/CometNativeReaderSuite.scala | 68 ++++++
.../comet/iceberg/IcebergReflectionSuite.scala | 69 ++++++
9 files changed, 515 insertions(+), 61 deletions(-)
diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md
index 7f251a2c2d..7f9c61e18f 100644
--- a/docs/source/user-guide/latest/iceberg.md
+++ b/docs/source/user-guide/latest/iceberg.md
@@ -212,6 +212,14 @@ The following scenarios will fall back to the JVM Iceberg reader:
or renamed. The native reader cannot yet match such a field to data files written before the
change. The check uses the table's schema history, so the fallback stays after those files are
rewritten
+- Scans that read a struct, array, or map column from a table with
+ `schema.name-mapping.default`, including tables imported with `snapshot`, `migrate`, or
+ `add_files` ([#6790](https://github.com/apache/datafusion-comet/issues/6790)). The native reader
+ does not apply name mappings to nested fields. Queries that read only primitive columns can
+ still run natively. The fallback remains while the property is set, even if the data files
+ have since been rewritten with field IDs. Iceberg's nullness rules are separate from the
+ Parquet legacy setting; [#6934](https://github.com/apache/datafusion-comet/issues/6934) tracks
+ the reader update and review of this fallback
- Scans where no schema in the table's history gives a partition source or equality-delete key a
name that the scan's other columns don't use (every task of a native scan must give such a
column the same name)
@@ -783,12 +791,6 @@ Which entries apply depends on which native paths are enabled:
### Reads that differ from Spark
-- **Migrated tables that use a name mapping.** The `migrate`, `snapshot` and `add_files`
- procedures set `schema.name-mapping.default` for files without Iceberg field ids. When a query
- also reads a top-level primitive column, the native scan returns NULL for nested fields (struct
- fields, list elements and map values). Filters on those fields then select the wrong rows
- ([#6790](https://github.com/apache/datafusion-comet/issues/6790)). The fix needs an iceberg-rust
- upgrade ([#6934](https://github.com/apache/datafusion-comet/issues/6934)).
- **Data files with neither field ids nor a name mapping.** Spark reads every nested field of such
a file as NULL, and the native scan returns the values stored in the file
([#6794](https://github.com/apache/datafusion-comet/issues/6794)).
diff --git a/docs/source/user-guide/latest/parquet.md b/docs/source/user-guide/latest/parquet.md
index 4d6a8dab0f..7d69c85b23 100644
--- a/docs/source/user-guide/latest/parquet.md
+++ b/docs/source/user-guide/latest/parquet.md
@@ -111,6 +111,14 @@ reads that fail where Spark succeeds are listed in
([SPARK-47447](https://issues.apache.org/jira/browse/SPARK-47447)) and Comet matches Spark's
behavior.
+### Missing Nested Fields
+
+When all requested leaf fields under a nested column are missing from a Parquet file, Spark 3.4
+through 4.0 return NULL for the entire missing subtree, including nested arrays and maps. Comet
+follows this behavior. On Spark 4.1+, the default (`false`) preserves the file's parent nullness.
+Set `spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing=true` for legacy behavior.
+A field that exists in the file but contains only NULL values is not a missing field.
+
### Schema Mismatch Handling
The issues in this subsection apply only when the requested read schema differs from the schema written
@@ -247,11 +255,6 @@ for production, so do not turn it off to avoid an entry on this page.
read back wrong inside lists written in a legacy layout, as Hive and
`spark.sql.parquet.writeLegacyFormat=true` write them
([#7029](https://github.com/apache/datafusion-comet/issues/7029)).
-- **A struct or array with none of its requested fields in the file.** Spark returns NULL for the
- column. Comet returns a non-null struct or array that holds NULLs. This applies to Spark 3.4, 3.5
- and 4.0, and to Spark 4.1 and later with
- `spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing=true`
- ([#7030](https://github.com/apache/datafusion-comet/issues/7030)).
- **Nested BINARY values that are not valid UTF-8, read as STRING.** This applies to a BINARY
column without the UTF8 annotation that is read as `string`, through the read schema or
`spark.sql.parquet.binaryAsString=true`. Spark returns the bytes. Comet returns NULL for each
diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs
index 997bd72bbb..b357a18245 100644
--- a/native/core/src/parquet/parquet_support.rs
+++ b/native/core/src/parquet/parquet_support.rs
@@ -205,15 +205,17 @@ fn parquet_convert_array(
to_type: &DataType,
parquet_options: &SparkParquetOptions,
) -> DataFusionResult<ArrayRef> {
- parquet_convert_array_impl(array, to_type, parquet_options, None)
+ parquet_convert_array_impl(array, to_type, parquet_options, None).map(|(array, _)| array)
}
+// The boolean records whether any requested leaf exists in the file schema, independent
+// of the values' nullness. Spark propagates missing children through nested containers.
fn parquet_convert_array_impl(
array: ArrayRef,
to_type: &DataType,
parquet_options: &SparkParquetOptions,
parent_nulls: Option<&NullBuffer>,
-) -> DataFusionResult<ArrayRef> {
+) -> DataFusionResult<(ArrayRef, bool)> {
use DataType::*;
let from_type = array.data_type();
// Only checked millis-to-micros casts consume ancestor visibility. In particular,
@@ -228,14 +230,14 @@ fn parquet_convert_array_impl(
};
// Try Comet specific handlers first, then arrow-rs cast if supported, and fail otherwise.
- match (from_type, to_type) {
- (Struct(_), Struct(_)) => Ok(parquet_convert_struct_to_struct(
+ let converted: DataFusionResult<ArrayRef> = match (from_type, to_type) {
+ (Struct(_), Struct(_)) => return parquet_convert_struct_to_struct(
array.as_struct(),
from_type,
to_type,
parquet_options,
visible.as_ref(),
- )?),
+ ),
(
List(_) | LargeList(_) | FixedSizeList(_, _) | ListView(_) | LargeListView(_),
List(to_inner_type) | LargeList(to_inner_type) | FixedSizeList(to_inner_type, _)
@@ -247,12 +249,15 @@ fn parquet_convert_array_impl(
} else {
None
};
- let cast_field = parquet_convert_array_impl(
+ let (cast_field, field_overlap) = parquet_convert_array_impl(
make_array(data.child_data()[0].clone()),
to_inner_type.data_type(),
parquet_options,
child_visibility.as_ref(),
)?;
+ if !field_overlap && parquet_options.return_null_struct_if_all_fields_missing {
+ return Ok((new_null_array(to_type, array.len()), false));
+ }
// Resolve element fields with Spark's rules before Arrow changes list layout.
// Casting the original list directly can match missing struct fields by position.
let resolved_type = match from_type {
@@ -268,11 +273,12 @@ fn parquet_convert_array_impl(
.data_type(resolved_type)
.child_data(vec![cast_field.to_data()])
.build()?);
- if resolved.data_type() == to_type {
- Ok(resolved)
+ let output = if resolved.data_type() == to_type {
+ resolved
} else {
- Ok(cast_with_options(&resolved, to_type, &PARQUET_OPTIONS)?)
- }
+ cast_with_options(&resolved, to_type, &PARQUET_OPTIONS)?
+ };
+ return Ok((output, field_overlap));
}
(
Timestamp(TimeUnit::Millisecond, _),
@@ -306,7 +312,7 @@ fn parquet_convert_array_impl(
))
}
(Map(_, ordered_from), Map(_, ordered_to)) if ordered_from == ordered_to =>
- parquet_convert_map_to_map(array.as_map(), to_type, parquet_options, *ordered_to, visible.as_ref(), checked_timestamp_overflow)
+ return parquet_convert_map_to_map(array.as_map(), to_type, parquet_options, *ordered_to, visible.as_ref(), checked_timestamp_overflow)
,
// Iceberg stores UUIDs as 16-byte fixed binary but Spark expects string representation.
// Arrow doesn't support casting FixedSizeBinary to Utf8, so we handle it manually.
@@ -341,7 +347,8 @@ fn parquet_convert_array_impl(
_ => Err(DataFusionError::Execution(format!(
"Unsupported Parquet type conversion from {from_type} to {to_type}"
))),
- }
+ };
+ converted.map(|array| (array, true))
}
// Struct fields are matched by name/field ID later. This type-only check is conservative
@@ -541,7 +548,7 @@ fn parquet_convert_struct_to_struct(
to_type: &DataType,
parquet_options: &SparkParquetOptions,
parent_nulls: Option<&NullBuffer>,
-) -> DataFusionResult<ArrayRef> {
+) -> DataFusionResult<(ArrayRef, bool)> {
match (from_type, to_type) {
(DataType::Struct(from_fields), DataType::Struct(to_fields)) => {
let from_indices = match_struct_fields(from_fields, to_fields, parquet_options)?;
@@ -550,13 +557,14 @@ fn parquet_convert_struct_to_struct(
let mut cast_fields: Vec<ArrayRef> = Vec::with_capacity(to_fields.len());
for (to_field, from_index) in to_fields.iter().zip(from_indices) {
if let Some(from_index) = from_index {
- cast_fields.push(parquet_convert_array_impl(
+ let (converted, child_overlap) = parquet_convert_array_impl(
Arc::clone(array.column(from_index)),
to_field.data_type(),
parquet_options,
parent_nulls,
- )?);
- field_overlap = true;
+ )?;
+ cast_fields.push(converted);
+ field_overlap |= child_overlap;
} else {
cast_fields.push(new_null_array(to_field.data_type(), array.len()));
}
@@ -575,11 +583,10 @@ fn parquet_convert_struct_to_struct(
array.nulls().cloned()
};
- Ok(Arc::new(StructArray::try_new(
- to_fields.clone(),
- cast_fields,
- nulls,
- )?))
+ Ok((
+ Arc::new(StructArray::try_new(to_fields.clone(), cast_fields, nulls)?),
+ field_overlap,
+ ))
}
_ => unreachable!(),
}
@@ -594,7 +601,7 @@ fn parquet_convert_map_to_map(
to_ordered: bool,
parent_nulls: Option<&NullBuffer>,
checked_timestamp_overflow: bool,
-) -> Result<ArrayRef, DataFusionError> {
+) -> Result<(ArrayRef, bool), DataFusionError> {
match to_data_type {
DataType::Map(entries_field, _) => {
let key_field = key_field(entries_field).ok_or(DataFusionError::Internal(
@@ -609,30 +616,37 @@ fn parquet_convert_map_to_map(
} else {
None
};
- let key_array = parquet_convert_array_impl(
+ let (key_array, key_overlap) = parquet_convert_array_impl(
Arc::clone(from.keys()),
key_field.data_type(),
parquet_options,
child_visibility.as_ref(),
)?;
- let value_array = parquet_convert_array_impl(
+ let (value_array, value_overlap) = parquet_convert_array_impl(
Arc::clone(from.values()),
value_field.data_type(),
parquet_options,
child_visibility.as_ref(),
)?;
+ let field_overlap = key_overlap || value_overlap;
+ if !field_overlap && parquet_options.return_null_struct_if_all_fields_missing {
+ return Ok((new_null_array(to_data_type, from.len()), false));
+ }
- Ok(Arc::new(MapArray::try_new(
- Arc::<arrow::datatypes::Field>::clone(entries_field),
- from.offsets().clone(),
- StructArray::try_new(
- Fields::from(vec![key_field, value_field]),
- vec![key_array, value_array],
- from.entries().nulls().cloned(),
- )?,
- from.nulls().cloned(),
- to_ordered,
- )?))
+ Ok((
+ Arc::new(MapArray::try_new(
+ Arc::<arrow::datatypes::Field>::clone(entries_field),
+ from.offsets().clone(),
+ StructArray::try_new(
+ Fields::from(vec![key_field, value_field]),
+ vec![key_array, value_array],
+ from.entries().nulls().cloned(),
+ )?,
+ from.nulls().cloned(),
+ to_ordered,
+ )?),
+ field_overlap,
+ ))
}
dt => Err(DataFusionError::Internal(format!(
"Expected MapType. Got: {dt}"
@@ -1922,6 +1936,193 @@ mod tests {
}
}
+ #[test]
+ fn test_missing_nested_fields_nullness() {
+ use super::{parquet_convert_array, SparkParquetOptions};
+ use arrow::array::{
+ new_null_array, Array, ArrayRef, ListArray, MapArray, StringArray, StructArray,
+ };
+ use arrow::buffer::{NullBuffer, OffsetBuffer};
+ use arrow::datatypes::{DataType, Field};
+ use datafusion_comet_spark_expr::EvalMode;
+ use std::sync::Arc;
+
+ // A present field remains present even when every value happens to be NULL.
+ let values: ArrayRef = Arc::new(StructArray::from(vec![(
+ Arc::new(Field::new("x", DataType::Int32, true)),
+ new_null_array(&DataType::Int32, 3),
+ )]));
+ let missing = DataType::Struct(vec![Field::new("missing", DataType::Int32, true)].into());
+ let element = Arc::new(Field::new("element", values.data_type().clone(), true));
+ let missing_element = Arc::new(Field::new("element", missing.clone(), true));
+ let list = ListArray::new(
+ Arc::clone(&element),
+ OffsetBuffer::new(vec![0, 1, 2, 3].into()),
+ Arc::clone(&values),
+ Some(NullBuffer::from(vec![true, true, false])),
+ );
+ let representations = |field: &Arc<Field>| {
+ [
+ DataType::List(Arc::clone(field)),
+ DataType::LargeList(Arc::clone(field)),
+ DataType::FixedSizeList(Arc::clone(field), 1),
+ DataType::ListView(Arc::clone(field)),
+ DataType::LargeListView(Arc::clone(field)),
+ ]
+ };
+ let mut cases = Vec::new();
+ for (source, target) in representations(&element)
+ .into_iter()
+ .zip(representations(&missing_element))
+ {
+ cases.push((arrow::compute::cast(&list, &source).unwrap(), target, true));
+ }
+
+ let list: ArrayRef = Arc::new(list);
+ for (child, target_child) in [
+ (Arc::clone(&values), missing.clone()),
+ (
+ Arc::clone(&list),
+ DataType::List(Arc::clone(&missing_element)),
+ ),
+ ] {
+ let parent: ArrayRef = Arc::new(StructArray::from(vec![(
+ Arc::new(Field::new("child", child.data_type().clone(), true)),
+ child,
+ )]));
+ cases.push((
+ parent,
+ DataType::Struct(vec![Field::new("child", target_child, true)].into()),
+ true,
+ ));
+ }
+
+ let entries = StructArray::from(vec![
+ (
+ Arc::new(Field::new("key", DataType::Utf8, false)),
+ Arc::new(StringArray::from(vec!["a", "b", "c"])) as ArrayRef,
+ ),
+ (
+ Arc::new(Field::new("value", values.data_type().clone(), true)),
+ Arc::clone(&values),
+ ),
+ ]);
+ let map: ArrayRef = Arc::new(MapArray::new(
+ Arc::new(Field::new("entries", entries.data_type().clone(), false)),
+ OffsetBuffer::new(vec![0, 1, 2, 3].into()),
+ entries,
+ None,
+ false,
+ ));
+ let target_map = DataType::Map(
+ Arc::new(Field::new(
+ "entries",
+ DataType::Struct(
+ vec![
+ Field::new("key", DataType::Utf8, false),
+ Field::new("value", missing.clone(), true),
+ ]
+ .into(),
+ ),
+ false,
+ )),
+ false,
+ );
+ // A map retains a matching leaf through its keys, even if its value fields are absent.
+ cases.push((Arc::clone(&map), target_map.clone(), false));
+ let missing_entries = StructArray::from(vec![
+ (
+ Arc::new(Field::new("key", values.data_type().clone(), false)),
+ Arc::clone(&values),
+ ),
+ (
+ Arc::new(Field::new("value", values.data_type().clone(), true)),
+ Arc::clone(&values),
+ ),
+ ]);
+ let missing_map: ArrayRef = Arc::new(MapArray::new(
+ Arc::new(Field::new(
+ "entries",
+ missing_entries.data_type().clone(),
+ false,
+ )),
+ OffsetBuffer::new(vec![0, 1, 2, 3].into()),
+ missing_entries,
+ None,
+ false,
+ ));
+ cases.push((
+ missing_map,
+ DataType::Map(
+ Arc::new(Field::new(
+ "entries",
+ DataType::Struct(
+ vec![
+ Field::new("key", missing.clone(), false),
+ Field::new("value", missing, true),
+ ]
+ .into(),
+ ),
+ false,
+ )),
+ false,
+ ),
+ true,
+ ));
+ let partial: ArrayRef = Arc::new(StructArray::from(vec![
+ (
+ Arc::new(Field::new("missing", list.data_type().clone(), true)),
+ list,
+ ),
+ (
+ Arc::new(Field::new("kept", map.data_type().clone(), true)),
+ map,
+ ),
+ ]));
+ cases.push((
+ partial,
+ DataType::Struct(
+ vec![
+ Field::new("missing", DataType::List(missing_element), true),
+ Field::new("kept", target_map, true),
+ ]
+ .into(),
+ ),
+ false,
+ ));
+ cases.push((Arc::clone(&values), values.data_type().clone(), false));
+
+ for legacy in [true, false] {
+ let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC");
+ options.case_sensitive = true;
+ options.return_null_struct_if_all_fields_missing = legacy;
+ for (source, target, all_missing) in &cases {
+ for input in [Arc::clone(source), source.slice(1, 2), source.slice(0, 0)] {
+ let result =
+ parquet_convert_array(Arc::clone(&input), target, &options).unwrap();
+ assert_eq!(result.data_type(), target);
+ assert_eq!(result.len(), input.len());
+ let expected_nulls = if legacy && *all_missing {
+ input.len()
+ } else {
+ input.null_count()
+ };
+ assert_eq!(
+ result.null_count(),
+ expected_nulls,
+ "{target:?}, legacy={legacy}"
+ );
+ if !legacy || !all_missing {
+ for row in 0..input.len() {
+ assert_eq!(result.is_null(row), input.is_null(row));
+ }
+ assert_eq!(result.to_data().buffers(), input.to_data().buffers());
+ }
+ }
+ }
+ }
+ }
+
#[test]
fn test_millis_to_micros_overflow_checked_in_nested_fields() {
use crate::parquet::parquet_support::{parquet_convert_array, SparkParquetOptions};
diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
index 33a0593568..bd30f12496 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
@@ -2425,7 +2425,7 @@ object IcebergReflection extends Logging {
* @param metadataLocation
* Path to the table metadata file
* @param nameMapping
- * Optional name mapping from table properties (for schema evolution)
+ * Name mapping from successfully read table properties, or None when not configured
* @param tasks
* List of FileScanTask objects from Iceberg planning
* @param scanSchema
@@ -2509,16 +2509,10 @@ object CometIcebergNativeScanMetadata extends Logging {
table <- getTable(scan)
scanSchema <- getExpectedSchema(scan)
tableSchema <- getSchema(table)
+ // Reading properties must succeed; only the mapping key is optional.
+ properties <- getTableProperties(table).filter(_ != null)
} yield {
- // nameMapping is optional - if it fails we just use None
- val nameMapping = getTableProperties(table).flatMap { properties =>
- val nameMappingKey = "schema.name-mapping.default"
- if (properties.containsKey(nameMappingKey)) {
- Some(properties.get(nameMappingKey))
- } else {
- None
- }
- }
+ val nameMapping = Option(properties.get("schema.name-mapping.default"))
val globalFieldIdMapping = buildFieldIdMapping(scanSchema)
val (prunedTaskSchema, fullTaskSchema) =
diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
index 3c511b4f37..349a9dae38 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -781,6 +781,20 @@ case class CometScanRule(session: SparkSession)
val resolvedProjectedFieldIds = projectedFieldIds.flatten.toSet
val hasUnresolvedProjectedFieldIds = projectedFieldIds.exists(_.isEmpty)
+ // iceberg-rust applies name mappings only at the root (#6790). Nested columns can
+ // disappear during projection, or reach the adapter with physical field names that
+ // cannot distinguish a mapped alias from a genuinely missing field. Schema history
+ // does not detect names that already differed when the files were imported.
+ val nestedNameMappingSupported = !(metadata.nameMapping.isDefined &&
+ projectedDataColumns.exists(_.dataType match {
+ case _: StructType | _: ArrayType | _: MapType => true
+ case _ => false
+ }))
+ if (!nestedNameMappingSupported) {
+ fallbackReasons += "Nested columns with Iceberg name mapping are not yet supported " +
+ "by Comet's native reader (#6790)"
+ }
+
// The full task schema can be serialized to native, but iceberg-rust can represent
// Variant in that schema as long as no projected field contains one. Check projected
// roots strictly, and allow Variant only under entirely unprojected roots. Other
@@ -1111,7 +1125,7 @@ case class CometScanRule(session: SparkSession)
taskValidation.allParquet && allSupportedFilesystems && allLocationsOpenable &&
metadataSchemeSupported && partitionTypesSupported && unifiedPartitionTypeSupported &&
transformFunctionsSupported && deleteFileTypesSupported && dppSubqueriesSupported &&
- nestedFieldsSupported && taskSchemaNamesSupported) {
+ nestedFieldsSupported && taskSchemaNamesSupported && nestedNameMappingSupported) {
CometBatchScanExec(
scanExec.clone().asInstanceOf[BatchScanExec],
runtimeFilters = scanExec.runtimeFilters,
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala
index caab7a72be..7f66776487 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala
@@ -448,13 +448,11 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with CometTypeS
// SPARK-53535 (Spark 4.1+): when reading a struct whose requested fields are all
// missing in the Parquet file, the new default preserves the parent struct's
// nullness from the file (so non-null parents materialize as a struct of all-null
- // fields). Pre-4.1 Spark hardcodes the legacy behavior (whole struct null), which
- // matches the Comet default we use as fallback.
+ // fields). Older Spark versions always use legacy behavior and ignore this key.
val returnNullStructConfKey =
"spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing"
- val returnNullStructDefault = if (isSpark41Plus) "false" else "true"
commonBuilder.setReturnNullStructIfAllFieldsMissing(
- conf.getConfString(returnNullStructConfKey, returnNullStructDefault).toBoolean)
+ !isSpark41Plus || conf.getConfString(returnNullStructConfKey, "false").toBoolean)
// Field-ID matching: only ask the native side to do extra work when the conf is on AND
// the requested schema actually carries IDs. Spark's ParquetReadSupport applies the same
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
index e859b23be0..5dfbd19fcd 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
@@ -2165,6 +2165,111 @@ class CometIcebergNativeSuite
}
}
+ test("migration - nested name mapping falls back to Spark") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ for ((physicalName, alias) <- Seq(("x", false), ("old_x", false), ("old_x", true))) {
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ "spark.sql.iceberg.aggregate-push-down.enabled" -> "false",
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+ val sourceName = "name_mapping_source"
+ val destName = "test_cat.db.name_mapping_dest"
+ val dataPath = new File(warehouseDir, "source_data").getAbsolutePath
+ val physicalSchema = StructType.fromDDL(
+ s"id INT, s STRUCT<$physicalName:INT>, items ARRAY<STRUCT<$physicalName:INT>>, " +
+ s"m MAP<STRING,STRUCT<anchor:INT,items:ARRAY<STRUCT<$physicalName:INT,z:INT>>," +
+ s"nested:STRUCT<$physicalName:INT>>>")
+ val logicalSchema = StructType.fromDDL(
+ "id INT, s STRUCT<x:INT>, items ARRAY<STRUCT<x:INT>>, " +
+ "m MAP<STRING,STRUCT<anchor:INT,items:ARRAY<STRUCT<x:INT,z:INT>>," +
+ "nested:STRUCT<x:INT>>>")
+ val rows = Seq(
+ Row(1, Row(11), Seq(Row(22)), Map("k" -> Row(44, Seq(Row(33, 99)), Row(33)))),
+ Row(2, null, null, null),
+ Row(3, Row(null), Seq(Row(null)), Map("k" -> Row(55, Seq(Row(null, 99)), Row(null)))),
+ Row(4, Row(null), Seq.empty[Row], Map.empty[String, Row]),
+ Row(5, Row(66), Seq(null), Map("k" -> Row(77, null, null))),
+ Row(6, null, null, Map("k" -> null)))
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
+ "spark.sql.parquet.fieldId.write.enabled" -> "false") {
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(rows, 1), physicalSchema)
+ .write
+ .parquet(dataPath)
+ }
+ withTable(sourceName) {
+ spark.sql(
+ s"CREATE TABLE $sourceName (${logicalSchema.toDDL}) " +
+ s"USING parquet LOCATION '$dataPath'")
+ try {
+ val actions = Class
+ .forName("org.apache.iceberg.spark.actions.SparkActions")
+ .getMethod("get")
+ .invoke(null)
+ val snapshot = actions.getClass
+ .getMethod("snapshotTable", classOf[String])
+ .invoke(actions, sourceName)
+ val destination = snapshot.getClass
+ .getMethod("as", classOf[String])
+ .invoke(snapshot, destName)
+ destination.getClass.getMethod("execute").invoke(destination)
+ val table = Spark3Util.loadIcebergTable(spark, destName)
+ val mapping = table.properties().get("schema.name-mapping.default")
+ assert(mapping != null)
+ assert(table.schemas().size() == 1)
+ if (alias) {
+ table
+ .updateProperties()
+ .set(
+ "schema.name-mapping.default",
+ mapping.replace("\"x\"", "\"x\", \"old_x\""))
+ .commit()
+ spark.catalog.refreshTable(destName)
+ }
+ // An import-time mismatch or mapping alias need not appear in schema history.
+ assert(table.schemas().size() == 1)
+ for (pruning <- Seq(false, true)) {
+ withSQLConf(
+ "spark.sql.optimizer.nestedSchemaPruning.enabled" -> pruning.toString) {
+ // Also read a mapped sibling to cover partial map-value projection.
+ val projections = Seq(
+ "s IS NULL, items IS NULL, m IS NULL",
+ "count(s), count(items), count(m)",
+ "m['k'].anchor, m['k'].items.x IS NULL, m['k'].nested IS NULL",
+ "count(m['k'].items.x), count(m['k'].nested), count(m['k'].anchor)",
+ "m['k'].nested",
+ "id, s, items, m")
+ val queries = projections.map(p => s"SELECT $p FROM $destName") ++ Seq(
+ s"SELECT id FROM $destName WHERE s.x = 11",
+ s"SELECT id FROM $destName WHERE m['k'].anchor = 44")
+ withClue(s"physicalName=$physicalName alias=$alias pruning=$pruning: ") {
+ for (query <- queries) {
+ val (_, plan) = checkSparkAnswerAndFallbackReason(
+ query,
+ "Nested columns with Iceberg name mapping")
+ assert(collectIcebergNativeScans(plan).isEmpty)
+ }
+ }
+ }
+ }
+ // Unprojected nested fields do not prevent primitive or row-count scans.
+ checkIcebergNativeScan(s"SELECT id FROM $destName")
+ checkIcebergNativeScan(s"SELECT count(*) FROM $destName")
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $destName")
+ }
+ }
+ }
+ }
+ }
+ }
+
test("migration - basic read after migration (fallback for no field ID)") {
assume(icebergAvailable, "Iceberg not available in classpath")
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala
index d5dff80cb2..7e35d22002 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala
@@ -1161,6 +1161,74 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper
}
}
+ Seq(None, Some("true"), Some("false")).foreach { legacy =>
+ test(s"issue #7030: missing nested fields - legacy ${legacy.getOrElse("default")}") {
+ val fileSchema = StructType.fromDDL(
+ "id int, arrs array<struct<x:int>>, s struct<c:struct<x:int>>, " +
+ "mixed struct<mm:map<string,struct<x:int>>, sa:array<struct<x:int>>>, " +
+ "present struct<x:int>")
+ val readSchema = StructType.fromDDL(
+ "id int, arrs array<struct<missing:int>>, s struct<c:struct<missing:int>>, " +
+ "mixed struct<mm:map<string,struct<missing:int>>, " +
+ "sa:array<struct<missing:int>>>, present struct<x:int>")
+ val rows = java.util.Arrays.asList(
+ Row(1, Seq(Row(1)), Row(Row(1)), Row(Map("k" -> Row(1)), Seq(Row(1))), Row(null)),
+ Row(2, Seq.empty, Row(Row(null)), Row(Map.empty, Seq.empty), Row(null)),
+ Row(3, null, null, null, null),
+ Row(4, Seq(null), Row(null), Row(Map("k" -> null), Seq(null)), Row(null)))
+
+ withTempPath { path =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ // Keep a file without null parents so its statistics cannot hide incorrect pruning.
+ spark
+ .createDataFrame(rows.subList(0, 1), fileSchema)
+ .coalesce(1)
+ .write
+ .parquet(path.toString)
+ spark
+ .createDataFrame(rows.subList(1, rows.size()), fileSchema)
+ .coalesce(1)
+ .write
+ .mode("append")
+ .parquet(path.toString)
+ }
+ val legacyConf =
+ legacy.toSeq.map("spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing" -> _)
+ for (pruning <- Seq("true", "false")) {
+ withSQLConf(
+ (legacyConf ++ Seq(
+ "spark.sql.parquet.enableNestedColumnVectorizedReader" -> "true",
+ "spark.sql.optimizer.nestedSchemaPruning.enabled" -> pruning,
+ CometConf.COMET_BATCH_SIZE.key -> "1")): _*) {
+ def read(): DataFrame = spark.read.schema(readSchema).parquet(path.toString)
+ val nativeScan = Seq(classOf[CometNativeScanExec])
+ checkSparkAnswerAndOperator(read(), nativeScan)
+ for (predicate <- Seq("arrs IS NULL", "s IS NULL", "mixed.sa IS NULL")) {
+ checkSparkAnswerAndOperator(read().where(predicate), nativeScan)
+ }
+ checkSparkAnswerAndOperator(
+ read().selectExpr(
+ "id",
+ "arrs IS NULL",
+ "s IS NULL",
+ "mixed IS NULL",
+ "mixed.sa IS NULL",
+ "present IS NULL"),
+ nativeScan)
+ checkSparkAnswerAndOperator(
+ read().selectExpr(
+ "count(arrs)",
+ "count(s)",
+ "count(mixed)",
+ "count(mixed.sa)",
+ "count(present)"),
+ nativeScan)
+ }
+ }
+ }
+ }
+ }
+
test("issue #4136: struct with all requested fields missing in file") {
// SPARK-53535 (Spark 4.1) added LEGACY_PARQUET_RETURN_NULL_STRUCT_IF_ALL_FIELDS_MISSING.
// With the new default (false), Spark's vectorized reader appends a "marker" leaf field to
diff --git a/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala b/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala
index 7f1cfa35e8..02ffb737a7 100644
--- a/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala
@@ -66,6 +66,75 @@ class IcebergReflectionSuite extends AnyFunSuite {
val metadata = IcebergReflection.getTableMetadata(table)
assert(metadata.isDefined)
assert(metadata.get.isInstanceOf[TableMetadata])
+ val extracted = extractScanMetadata(new MetadataScan(table, schema)).get
+ assert(extracted.nameMapping.isEmpty)
+ assert(extracted.fullTaskSchema.isDefined)
+ assert(extracted.prunedTaskSchema.isDefined)
+ }
+
+ private val scanSchema =
+ new Schema(Types.NestedField.optional(1, "id", Types.IntegerType.get()))
+
+ private def extractScanMetadata(scan: MetadataScan): Option[CometIcebergNativeScanMetadata] =
+ CometIcebergNativeScanMetadata.extract(
+ scan,
+ "file:///tmp/test-table/metadata.json",
+ Map.empty,
+ Collections.emptyList[AnyRef](),
+ Set.empty)
+
+ test("scan metadata preserves name mapping when present and None when absent") {
+ val mapping = """[{"field-id":1,"names":["id","old_id"]}]"""
+ for (expected <- Seq(None, Some(mapping))) {
+ val properties = new java.util.HashMap[String, String]()
+ expected.foreach(properties.put("schema.name-mapping.default", _))
+ val table = new MetadataTable(scanSchema, new PropertiesMetadata(() => properties))
+ val metadata = extractScanMetadata(new MetadataScan(table, scanSchema)).get
+ assert(metadata.nameMapping == expected)
+ assert(metadata.fullTaskSchema.isDefined)
+ assert(metadata.prunedTaskSchema.isDefined)
+ assert(metadata.globalFieldIdMapping == Map("id" -> 1))
+ }
+ }
+
+ test("scan metadata declines when the properties accessor throws") {
+ val metadata =
+ new PropertiesMetadata(() => throw new IllegalStateException("properties failed"))
+ val table = new MetadataTable(scanSchema, metadata)
+ assert(extractScanMetadata(new MetadataScan(table, scanSchema)).isEmpty)
+ }
+
+ test("scan metadata declines when the properties accessor is missing") {
+ val table = new MetadataTable(scanSchema, new Object)
+ assert(extractScanMetadata(new MetadataScan(table, scanSchema)).isEmpty)
+ }
+
+ test("scan metadata declines when the properties accessor returns null") {
+ val table = new MetadataTable(scanSchema, new PropertiesMetadata(() => null))
+ assert(extractScanMetadata(new MetadataScan(table, scanSchema)).isEmpty)
+ }
+
+ // Only the metadata's properties accessor varies. Every case has a valid scan/table schema,
+ // schema history and operations.current(), so other reflection failures cannot mask the bug.
+ class MetadataScan(scanTable: AnyRef, scanSchema: Schema) {
+ def table(): AnyRef = scanTable
+ def projection(): Schema = scanSchema
+ }
+
+ class MetadataTable(tableSchema: Schema, metadata: AnyRef) {
+ def schema(): Schema = tableSchema
+ def schemas(): java.util.Map[Integer, Schema] =
+ Collections.singletonMap(Integer.valueOf(0), tableSchema)
+ def operations(): MetadataOperations = new MetadataOperations(metadata)
+ def name(): String = "test-table"
+ }
+
+ class MetadataOperations(metadata: AnyRef) {
+ def current(): AnyRef = metadata
+ }
+
+ class PropertiesMetadata(read: () => java.util.Map[String, String]) {
+ def properties(): java.util.Map[String, String] = read()
}
test("findMethod resolves a method once and returns the cached instance") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org