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/09/11 22:35:23 UTC
(datafusion-comet) 01/01: fix: apply Spark's Parquet conversion rules to nested struct/list/map fields (#5681)
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-5681-c369d04b4694a4792a1a7cce7d1b7487e5b791ef
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 6751af0291faede7f966b1bd1f826321c8cadcfc
Author: Peter Lee <pe...@gmail.com>
AuthorDate: Fri Sep 11 22:35:00 2026 +0000
fix: apply Spark's Parquet conversion rules to nested struct/list/map fields (#5681)
* fix: apply Spark's Parquet conversion rules to nested struct/list/map fields
`SparkPhysicalExprAdapter` applied Spark's `SchemaColumnConvertNotSupportedException`
matrix only to the top-level physical/logical pair. Same-shape complex pairs were
wrapped in `CometCastColumnExpr`, whose `parquet_convert_array` cast every nested
leaf with a plain Arrow `cast_with_options(safe: true)` (silent NULLs on overflow,
parsed strings, scalars wrapped into arrays) and returned the unconverted array for
pairs Arrow cannot cast, which then panicked in `StructArray::new`.
- Factor the scalar rule set out of `replace_with_spark_cast` into
`check_leaf_conversion` (same conditions, same order) and add `check_conversion`,
which walks same-shape struct/list/map pairs at any depth, resolves struct fields
with the runtime convert's field-id / case-fold rules (now shared via
`match_struct_fields`), applies the rules to every leaf, reports the Spark-style
column path (`Column: [s, x]`) and honours `RejectOnNonEmpty` like the top level.
- Apply it on both the `CastExpr` path and the default-adapter fallback path.
- Make `parquet_convert_array` fail on unsupported pairs and use `try_new` for
struct/list/map so a mismatch is an error, never a panic.
- Render array/struct/map targets with Spark's `catalogString` in the error.
Closes #5671
Co-Authored-By: Claude Fable 5.1 <no...@anthropic.com>
* test: type nested list fixture as int
* fix: address nested Parquet conversion review feedback
* fix: validate fixed-size Parquet list elements
* fix: validate Parquet list-view elements
* fix: preserve Parquet struct field matching across list representations
---------
Co-authored-by: Claude Fable 5.1 <no...@anthropic.com>
---
.../user-guide/latest/compatibility/scans.md | 4 +
native/core/src/execution/planner.rs | 3 +-
native/core/src/parquet/parquet_support.rs | 475 ++++--
native/core/src/parquet/schema_adapter.rs | 1515 ++++++++++++++++----
.../apache/comet/parquet/ParquetReadSuite.scala | 89 ++
5 files changed, 1732 insertions(+), 354 deletions(-)
diff --git a/docs/source/user-guide/latest/compatibility/scans.md b/docs/source/user-guide/latest/compatibility/scans.md
index 071b83ba5a..d5ab79b4a9 100644
--- a/docs/source/user-guide/latest/compatibility/scans.md
+++ b/docs/source/user-guide/latest/compatibility/scans.md
@@ -102,6 +102,10 @@ and `INT32 → DOUBLE` widening that Spark 4.0+ accepts unconditionally; `Timest
is rejected by Spark 3.x but accepted by Spark 4.0+). Comet aims to follow the per-version Spark
behavior.
+- **List conversion error paths assume Spark's standard encoding**. Comet inserts `list`
+ before the element name when reporting a rejected array element conversion. Arrow's schema
+ omits the repeated group name, so paths for legacy LIST encodings or custom group names may
+ differ from Spark's Parquet column path.
- **`ParquetSchemaConvert` errors do not include the file path**. The mismatch itself is detected and
rejected correctly, but the resulting Spark error message reads
`Encountered error while reading file . Data type mismatches…` (note the empty path). Behavior is
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index d66638ba32..319ddc3a30 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -6802,7 +6802,8 @@ mod tests {
*/
#[tokio::test]
async fn test_nested_types_list_of_struct_by_index() -> Result<(), DataFusionError> {
- let test_data = "select make_array(named_struct('a', 1, 'b', 'n', 'c', 'x')) c0";
+ let test_data =
+ "select make_array(named_struct('a', cast(1 as int), 'b', 'n', 'c', 'x')) c0";
// Define schema Comet reads with
let required_schema = Schema::new(Fields::from(vec![Field::new(
diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs
index 25abaa8a4c..c815944c46 100644
--- a/native/core/src/parquet/parquet_support.rs
+++ b/native/core/src/parquet/parquet_support.rs
@@ -17,7 +17,9 @@
use crate::execution::operators::ExecutionError;
use crate::parquet::name_fold::fold_names;
-use arrow::array::{FixedSizeBinaryArray, ListArray, MapArray, StringArray};
+use arrow::array::{
+ make_array, FixedSizeBinaryArray, GenericListViewArray, MapArray, OffsetSizeTrait, StringArray,
+};
use arrow::buffer::NullBuffer;
use arrow::compute::can_cast_types;
use arrow::datatypes::{FieldRef, Fields};
@@ -203,8 +205,7 @@ fn parquet_convert_array_impl(
None
};
- // Try Comet specific handlers first, then arrow-rs cast if supported,
- // return uncasted data otherwise
+ // 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(
array.as_struct(),
@@ -213,27 +214,43 @@ fn parquet_convert_array_impl(
parquet_options,
visible.as_ref(),
)?),
- (List(_), List(to_inner_type)) => {
- let list_arr: &ListArray = array.as_list();
+ (
+ List(_) | LargeList(_) | FixedSizeList(_, _) | ListView(_) | LargeListView(_),
+ List(to_inner_type) | LargeList(to_inner_type) | FixedSizeList(to_inner_type, _)
+ | ListView(to_inner_type) | LargeListView(to_inner_type),
+ ) => {
+ let data = array.to_data();
let child_visibility = if checked_timestamp_overflow {
- repeated_visibility(
- list_arr.value_offsets(), list_arr.values().len(), visible.as_ref())
+ list_child_visibility(array.as_ref(), visible.as_ref())
} else {
None
};
let cast_field = parquet_convert_array_impl(
- Arc::clone(list_arr.values()),
+ make_array(data.child_data()[0].clone()),
to_inner_type.data_type(),
parquet_options,
child_visibility.as_ref(),
)?;
-
- Ok(Arc::new(ListArray::new(
- Arc::clone(to_inner_type),
- list_arr.offsets().clone(),
- cast_field,
- list_arr.nulls().cloned(),
- )))
+ // 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 {
+ List(_) => List(Arc::clone(to_inner_type)),
+ LargeList(_) => LargeList(Arc::clone(to_inner_type)),
+ FixedSizeList(_, size) => FixedSizeList(Arc::clone(to_inner_type), *size),
+ ListView(_) => ListView(Arc::clone(to_inner_type)),
+ LargeListView(_) => LargeListView(Arc::clone(to_inner_type)),
+ _ => unreachable!(),
+ };
+ // Retain the source offsets, sizes and null buffer while replacing its values.
+ let resolved = make_array(data.into_builder()
+ .data_type(resolved_type)
+ .child_data(vec![cast_field.to_data()])
+ .build()?);
+ if resolved.data_type() == to_type {
+ Ok(resolved)
+ } else {
+ Ok(cast_with_options(&resolved, to_type, &PARQUET_OPTIONS)?)
+ }
}
(
Timestamp(TimeUnit::Millisecond, _),
@@ -295,7 +312,13 @@ fn parquet_convert_array_impl(
_ if can_cast_types(from_type, to_type) => {
Ok(cast_with_options(&array, to_type, &PARQUET_OPTIONS)?)
}
- _ => Ok(array),
+ // Every pair reaching here should already have passed the schema adapter's
+ // `check_conversion` (Spark's `getUpdater` matrix), so this is a gap in that gate. Fail
+ // instead of handing back an array of the wrong type, which a parent `StructArray` /
+ // `ListArray` constructor would otherwise panic on (#5671).
+ _ => Err(DataFusionError::Execution(format!(
+ "Unsupported Parquet type conversion from {from_type} to {to_type}"
+ ))),
}
}
@@ -307,26 +330,71 @@ fn has_timestamp_unit(data_type: &DataType, unit: TimeUnit) -> bool {
DataType::Struct(fields) => fields
.iter()
.any(|field| has_timestamp_unit(field.data_type(), unit)),
- DataType::List(field) | DataType::Map(field, _) => {
- has_timestamp_unit(field.data_type(), unit)
- }
+ DataType::List(field)
+ | DataType::LargeList(field)
+ | DataType::FixedSizeList(field, _)
+ | DataType::ListView(field)
+ | DataType::LargeListView(field)
+ | DataType::Map(field, _) => has_timestamp_unit(field.data_type(), unit),
_ => false,
}
}
// List/map values can include entries outside a slice or beneath a null parent.
-fn repeated_visibility(
- offsets: &[i32],
+fn repeated_visibility<O: OffsetSizeTrait>(
+ offsets: &[O],
len: usize,
nulls: Option<&NullBuffer>,
) -> Option<NullBuffer> {
- if nulls.is_none() && offsets[0] == 0 && *offsets.last().unwrap() as usize == len {
+ if nulls.is_none() && offsets[0].as_usize() == 0 && offsets.last().unwrap().as_usize() == len {
return None;
}
let mut valid = vec![false; len];
for (row, range) in offsets.windows(2).enumerate() {
if nulls.is_none_or(|nulls| nulls.is_valid(row)) {
- valid[range[0] as usize..range[1] as usize].fill(true);
+ valid[range[0].as_usize()..range[1].as_usize()].fill(true);
+ }
+ }
+ Some(NullBuffer::from(valid))
+}
+
+fn list_child_visibility(array: &dyn Array, nulls: Option<&NullBuffer>) -> Option<NullBuffer> {
+ match array.data_type() {
+ DataType::List(_) => {
+ let list = array.as_list::<i32>();
+ repeated_visibility(list.value_offsets(), list.values().len(), nulls)
+ }
+ DataType::LargeList(_) => {
+ let list = array.as_list::<i64>();
+ repeated_visibility(list.value_offsets(), list.values().len(), nulls)
+ }
+ DataType::FixedSizeList(_, size) => {
+ // Arrow slices the child values along with a fixed-size list. Expand in those
+ // child coordinates; applying a parent slice offset again would shift the mask.
+ nulls.map(|nulls| nulls.expand(*size as usize))
+ }
+ DataType::ListView(_) => list_view_visibility(array.as_list_view::<i32>(), nulls),
+ DataType::LargeListView(_) => list_view_visibility(array.as_list_view::<i64>(), nulls),
+ _ => unreachable!("only called for list arrays"),
+ }
+}
+
+fn list_view_visibility<O: OffsetSizeTrait>(
+ list: &GenericListViewArray<O>,
+ nulls: Option<&NullBuffer>,
+) -> Option<NullBuffer> {
+ let mut valid = vec![false; list.values().len()];
+ for (row, (offset, size)) in list
+ .value_offsets()
+ .iter()
+ .zip(list.value_sizes())
+ .enumerate()
+ {
+ if nulls.is_none_or(|nulls| nulls.is_valid(row)) {
+ let start = offset.as_usize();
+ // Views can overlap or be out of order. Accumulate the union of visible ranges;
+ // a null parent must never clear values another parent can see.
+ valid[start..start + size.as_usize()].fill(true);
}
}
Some(NullBuffer::from(valid))
@@ -340,6 +408,89 @@ fn field_id(field: &arrow::datatypes::Field) -> Option<i32> {
.and_then(|v| v.parse::<i32>().ok())
}
+/// Resolve each requested (`to`) struct field to the index of the file (`from`) field it reads
+/// from, or `None` when the file holds no such field. Mirrors Spark's `clipParquetGroupFields`:
+/// when the requested struct carries Parquet field IDs anywhere (and `use_field_id` is set),
+/// ID-bearing requested fields match ONLY by ID (a missing ID is a missing column, never a name
+/// fallback); other fields match by name, folded with the same `toLowerCase(Locale.ROOT)` fold
+/// the top-level schema adapter uses when `case_sensitive` is false. A requested field whose
+/// folded name matches more than one file field in case-insensitive mode raises Spark's
+/// `foundDuplicateFieldInCaseInsensitiveModeError`.
+///
+/// Shared by the runtime convert (`parquet_convert_struct_to_struct`) and the plan-time
+/// conversion check in `schema_adapter`, so both resolve nested fields identically.
+pub(crate) fn match_struct_fields(
+ from_fields: &[FieldRef],
+ to_fields: &[FieldRef],
+ parquet_options: &SparkParquetOptions,
+) -> DataFusionResult<Vec<Option<usize>>> {
+ let should_match_by_id =
+ parquet_options.use_field_id && to_fields.iter().any(|f| field_id(f).is_some());
+
+ let from_id_to_index: HashMap<i32, usize> = if should_match_by_id {
+ let mut map = HashMap::new();
+ for (i, field) in from_fields.iter().enumerate() {
+ if let Some(id) = field_id(field) {
+ map.entry(id).or_insert(i);
+ }
+ }
+ map
+ } else {
+ HashMap::new()
+ };
+
+ // Fold the file (`from`) and requested (`to`) field names once via the JVM's
+ // `toLowerCase(Locale.ROOT)` (the same fold the top-level schema adapter uses), so
+ // nested case-insensitive matching is byte-for-byte consistent with the top level.
+ let mut all_names: Vec<&str> = Vec::with_capacity(from_fields.len() + to_fields.len());
+ all_names.extend(from_fields.iter().map(|f| f.name().as_str()));
+ all_names.extend(to_fields.iter().map(|f| f.name().as_str()));
+ let all_folded = fold_names(&all_names, parquet_options.case_sensitive);
+ let (from_folded, to_folded) = all_folded.split_at(from_fields.len());
+
+ // Group file field indices by folded name so a case-insensitive collision is detected
+ // (Spark's `caseInsensitiveParquetFieldMap`) rather than silently overwritten.
+ let mut folded_to_indices: HashMap<&str, Vec<usize>> = HashMap::new();
+ for (i, folded) in from_folded.iter().enumerate() {
+ folded_to_indices
+ .entry(folded.as_str())
+ .or_default()
+ .push(i);
+ }
+
+ to_fields
+ .iter()
+ .enumerate()
+ .map(
+ |(to_pos, to_field)| match (should_match_by_id, field_id(to_field)) {
+ // Spark treats a missing ID match as a missing column rather than
+ // falling back to name match.
+ (true, Some(id)) => Ok(from_id_to_index.get(&id).copied()),
+ _ => match folded_to_indices.get(to_folded[to_pos].as_str()) {
+ // Mirror Spark's `foundDuplicateFieldInCaseInsensitiveModeError`: a
+ // requested field matching more than one file field is ambiguous. Gated on
+ // case-insensitive mode to match the top-level check (which only runs when
+ // `!case_sensitive`): when case-sensitive the fold is identity, so a
+ // collision means byte-identical sibling names, and raising an error whose
+ // message says "in case-insensitive mode" would be wrong. Fall through to
+ // the first match in that case.
+ Some(indices) if indices.len() > 1 && !parquet_options.case_sensitive => {
+ let matched: Vec<&str> = indices
+ .iter()
+ .map(|&i| from_fields[i].name().as_str())
+ .collect();
+ Err(DataFusionError::External(Box::new(
+ SparkError::duplicate_field_case_insensitive(to_field.name(), &matched),
+ )))
+ }
+ Some(indices) => Ok(Some(indices[0])),
+ None => Ok(None),
+ },
+ },
+ )
+ .collect()
+}
+
/// Cast between struct types based on logic in
/// `org.apache.spark.sql.catalyst.expressions.Cast#castStruct`.
fn parquet_convert_struct_to_struct(
@@ -351,77 +502,11 @@ fn parquet_convert_struct_to_struct(
) -> DataFusionResult<ArrayRef> {
match (from_type, to_type) {
(DataType::Struct(from_fields), DataType::Struct(to_fields)) => {
- // Match `from` (file) fields to `to` (logical) fields. Mirrors Spark's
- // `clipParquetGroupFields`: when the logical struct carries Parquet field IDs
- // anywhere, ID-bearing logical fields match ONLY by ID; non-ID-bearing fields
- // fall back to name match. When no logical field carries an ID, fall back to
- // name match across the board.
- let should_match_by_id =
- parquet_options.use_field_id && to_fields.iter().any(|f| field_id(f).is_some());
-
- let from_id_to_index: HashMap<i32, usize> = if should_match_by_id {
- let mut map = HashMap::new();
- for (i, field) in from_fields.iter().enumerate() {
- if let Some(id) = field_id(field) {
- map.entry(id).or_insert(i);
- }
- }
- map
- } else {
- HashMap::new()
- };
-
- // Fold the file (`from`) and requested (`to`) field names once via the JVM's
- // `toLowerCase(Locale.ROOT)` (the same fold the top-level schema adapter uses), so
- // nested case-insensitive matching is byte-for-byte consistent with the top level.
- let mut all_names: Vec<&str> = Vec::with_capacity(from_fields.len() + to_fields.len());
- all_names.extend(from_fields.iter().map(|f| f.name().as_str()));
- all_names.extend(to_fields.iter().map(|f| f.name().as_str()));
- let all_folded = fold_names(&all_names, parquet_options.case_sensitive);
- let (from_folded, to_folded) = all_folded.split_at(from_fields.len());
-
- // Group file field indices by folded name so a case-insensitive collision is detected
- // (Spark's `caseInsensitiveParquetFieldMap`) rather than silently overwritten.
- let mut folded_to_indices: HashMap<&str, Vec<usize>> = HashMap::new();
- for (i, folded) in from_folded.iter().enumerate() {
- folded_to_indices
- .entry(folded.as_str())
- .or_default()
- .push(i);
- }
+ let from_indices = match_struct_fields(from_fields, to_fields, parquet_options)?;
let mut field_overlap = false;
let mut cast_fields: Vec<ArrayRef> = Vec::with_capacity(to_fields.len());
- for (to_pos, to_field) in to_fields.iter().enumerate() {
- let from_index = match (should_match_by_id, field_id(to_field)) {
- // Spark treats a missing ID match as a missing column rather than
- // falling back to name match.
- (true, Some(id)) => from_id_to_index.get(&id).copied(),
- _ => match folded_to_indices.get(to_folded[to_pos].as_str()) {
- // Mirror Spark's `foundDuplicateFieldInCaseInsensitiveModeError`: a
- // requested field matching more than one file field is ambiguous. Gated on
- // case-insensitive mode to match the top-level check (which only runs when
- // `!case_sensitive`): when case-sensitive the fold is identity, so a
- // collision means byte-identical sibling names, and raising an error whose
- // message says "in case-insensitive mode" would be wrong. Fall through to
- // the first match in that case.
- Some(indices) if indices.len() > 1 && !parquet_options.case_sensitive => {
- let matched: Vec<&str> = indices
- .iter()
- .map(|&i| from_fields[i].name().as_str())
- .collect();
- return Err(DataFusionError::External(Box::new(
- SparkError::duplicate_field_case_insensitive(
- to_field.name(),
- &matched,
- ),
- )));
- }
- Some(indices) => Some(indices[0]),
- None => None,
- },
- };
-
+ 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(
Arc::clone(array.column(from_index)),
@@ -448,11 +533,11 @@ fn parquet_convert_struct_to_struct(
array.nulls().cloned()
};
- Ok(Arc::new(StructArray::new(
+ Ok(Arc::new(StructArray::try_new(
to_fields.clone(),
cast_fields,
nulls,
- )))
+ )?))
}
_ => unreachable!(),
}
@@ -495,17 +580,17 @@ fn parquet_convert_map_to_map(
child_visibility.as_ref(),
)?;
- Ok(Arc::new(MapArray::new(
+ Ok(Arc::new(MapArray::try_new(
Arc::<arrow::datatypes::Field>::clone(entries_field),
from.offsets().clone(),
- StructArray::new(
+ StructArray::try_new(
Fields::from(vec![key_field, value_field]),
vec![key_array, value_array],
from.entries().nulls().cloned(),
- ),
+ )?,
from.nulls().cloned(),
to_ordered,
- )))
+ )?))
}
dt => Err(DataFusionError::Internal(format!(
"Expected MapType. Got: {dt}"
@@ -1164,6 +1249,34 @@ mod tests {
.map(|(url, path, _)| (url, path))
}
+ /// A conversion the schema adapter should have rejected must surface as an error, never
+ /// as a mismatched child array that `StructArray::new` panics on (#5671).
+ #[test]
+ fn convert_list_to_int_inside_struct_errors_instead_of_panicking() {
+ use crate::parquet::parquet_support::{spark_parquet_convert, SparkParquetOptions};
+ use arrow::array::{Array, ListArray, StructArray};
+ use arrow::datatypes::{DataType, Field, Fields, Int32Type};
+ use datafusion::physical_plan::ColumnarValue;
+ use datafusion_comet_spark_expr::EvalMode;
+ use std::sync::Arc;
+
+ let list = ListArray::from_iter_primitive::<Int32Type, _, _>(vec![Some(vec![Some(1)])]);
+ let from_fields = Fields::from(vec![Field::new("x", list.data_type().clone(), true)]);
+ let array = StructArray::new(from_fields, vec![Arc::new(list)], None);
+ let to_type = DataType::Struct(Fields::from(vec![Field::new("x", DataType::Int32, true)]));
+ let err = spark_parquet_convert(
+ ColumnarValue::Array(Arc::new(array)),
+ &to_type,
+ &SparkParquetOptions::new(EvalMode::Legacy, "UTC", false),
+ )
+ .expect_err("array<int> -> int must be an error");
+ assert!(
+ err.to_string()
+ .contains("Unsupported Parquet type conversion"),
+ "unexpected error: {err}"
+ );
+ }
+
#[cfg(not(feature = "hdfs-opendal"))]
#[test]
fn test_prepare_object_store() {
@@ -1512,6 +1625,174 @@ mod tests {
}
}
+ #[test]
+ fn test_millis_to_micros_list_representations_visibility() {
+ use super::{parquet_convert_array, SparkParquetOptions};
+ use arrow::array::{
+ cast::AsArray, make_array, Array, ArrayRef, ListArray, StructArray,
+ TimestampMillisecondArray,
+ };
+ use arrow::buffer::{NullBuffer, OffsetBuffer};
+ use arrow::datatypes::{DataType, Field, TimeUnit, TimestampMicrosecondType};
+ use datafusion_comet_spark_expr::EvalMode;
+ use std::sync::Arc;
+
+ let options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ for timezone in [None::<Arc<str>>, Some(Arc::from("UTC"))] {
+ for overflow in [i64::MIN, i64::MAX] {
+ let values: ArrayRef = Arc::new(
+ TimestampMillisecondArray::from(vec![
+ overflow, overflow, 7, 8, overflow, overflow,
+ ])
+ .with_timezone_opt(timezone.clone()),
+ );
+ let field = Arc::new(Field::new("item", values.data_type().clone(), false));
+ let target_field = Arc::new(Field::new(
+ "item",
+ DataType::Timestamp(TimeUnit::Microsecond, timezone.clone()),
+ false,
+ ));
+ let representations = |field: &Arc<Field>| {
+ [
+ DataType::List(Arc::clone(field)),
+ DataType::LargeList(Arc::clone(field)),
+ DataType::FixedSizeList(Arc::clone(field), 2),
+ DataType::ListView(Arc::clone(field)),
+ DataType::LargeListView(Arc::clone(field)),
+ ]
+ };
+ let list = ListArray::new(
+ Arc::clone(&field),
+ OffsetBuffer::new(vec![0, 2, 4, 6].into()),
+ values,
+ None,
+ );
+ for (source, target) in representations(&field)
+ .into_iter()
+ .zip(representations(&target_field))
+ {
+ let unmasked = arrow::compute::cast(&list, &source).unwrap();
+ let validity = NullBuffer::from(vec![false, true, false]);
+ let masked = make_array(
+ unmasked
+ .to_data()
+ .into_builder()
+ .nulls(Some(validity.clone()))
+ .build()
+ .unwrap(),
+ );
+ for input in [
+ Arc::clone(&masked),
+ masked.slice(1, 1),
+ unmasked.slice(1, 1),
+ ] {
+ let output = parquet_convert_array(input, &target, &options).unwrap();
+ let output = arrow::compute::cast(
+ &output,
+ &DataType::List(Arc::clone(&target_field)),
+ )
+ .unwrap();
+ let output = output.as_list::<i32>();
+ let row = if output.len() == 1 { 0 } else { 1 };
+ if output.len() == 3 {
+ assert!(output.is_null(0) && output.is_null(2));
+ }
+ let values = output.value(row);
+ let values = values.as_primitive::<TimestampMicrosecondType>();
+ assert_eq!(values.values().as_ref(), &[7000, 8000], "{source:?}");
+ assert_eq!(values.null_count(), 0);
+ }
+ assert!(
+ parquet_convert_array(Arc::clone(&unmasked), &target, &options).is_err(),
+ "{source:?}"
+ );
+ parquet_convert_array(unmasked.slice(0, 0), &target, &options).unwrap();
+ // The list itself is non-null: visibility comes from the enclosing struct.
+ let outer: ArrayRef = Arc::new(StructArray::new(
+ vec![Arc::new(Field::new("a", source, false))].into(),
+ vec![unmasked],
+ Some(validity),
+ ));
+ let outer_target =
+ DataType::Struct(vec![Arc::new(Field::new("a", target, false))].into());
+ parquet_convert_array(outer, &outer_target, &options).unwrap();
+ }
+ }
+ }
+ }
+
+ #[test]
+ fn test_millis_to_micros_overlapping_list_views() {
+ use super::{parquet_convert_array, SparkParquetOptions};
+ use arrow::array::{
+ cast::AsArray, make_array, Array, ArrayRef, LargeListViewArray, ListViewArray,
+ TimestampMillisecondArray,
+ };
+ use arrow::buffer::NullBuffer;
+ use arrow::datatypes::{DataType, Field, TimeUnit, TimestampMicrosecondType};
+ use datafusion_comet_spark_expr::EvalMode;
+ use std::sync::Arc;
+
+ let options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ let values: ArrayRef = Arc::new(TimestampMillisecondArray::from(vec![
+ i64::MAX,
+ 7,
+ 8,
+ 9,
+ i64::MIN,
+ ]));
+ let field = Arc::new(Field::new("item", values.data_type().clone(), false));
+ let target_field = Arc::new(Field::new(
+ "item",
+ DataType::Timestamp(TimeUnit::Microsecond, None),
+ false,
+ ));
+ let nulls = Some(NullBuffer::from(vec![
+ false, true, true, false, false, true,
+ ]));
+ let small: ArrayRef = Arc::new(ListViewArray::new(
+ Arc::clone(&field),
+ vec![0, 2, 1, 3, 0, 4].into(),
+ vec![4, 2, 2, 1, 1, 0].into(),
+ Arc::clone(&values),
+ nulls.clone(),
+ ));
+ let large: ArrayRef = Arc::new(LargeListViewArray::new(
+ field,
+ vec![0, 2, 1, 3, 0, 4].into(),
+ vec![4, 2, 2, 1, 1, 0].into(),
+ values,
+ nulls,
+ ));
+ for (array, target) in [
+ (small, DataType::ListView(Arc::clone(&target_field))),
+ (large, DataType::LargeListView(Arc::clone(&target_field))),
+ ] {
+ // Null views overlap live views both before and after them. Unreferenced prefix
+ // and suffix values overflow, and the final live view is empty.
+ for input in [Arc::clone(&array), array.slice(1, 2)] {
+ let output = parquet_convert_array(input, &target, &options).unwrap();
+ let output =
+ arrow::compute::cast(&output, &DataType::List(Arc::clone(&target_field)))
+ .unwrap();
+ let output = output.as_list::<i32>();
+ let first = if output.len() == 2 { 0 } else { 1 };
+ for (row, expected) in [(first, [8000, 9000]), (first + 1, [7000, 8000])] {
+ let values = output.value(row);
+ let values = values.as_primitive::<TimestampMicrosecondType>();
+ assert_eq!(values.values().as_ref(), &expected);
+ assert_eq!(values.null_count(), 0);
+ }
+ if output.len() == 6 {
+ assert!(output.is_null(0) && output.is_null(3) && output.is_null(4));
+ assert!(output.value(5).is_empty());
+ }
+ }
+ let visible = make_array(array.to_data().into_builder().nulls(None).build().unwrap());
+ assert!(parquet_convert_array(visible, &target, &options).is_err());
+ }
+ }
+
/// Constructs S3 stores using anonymous credentials, without reading remote objects, and
/// checks that both alias URL forms return the normalized bucket, key, and remote I/O label.
#[cfg(not(feature = "hdfs-opendal"))]
diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs
index 818916f126..5a4b3e1630 100644
--- a/native/core/src/parquet/schema_adapter.rs
+++ b/native/core/src/parquet/schema_adapter.rs
@@ -17,7 +17,9 @@
use crate::parquet::cast_column::CometCastColumnExpr;
use crate::parquet::name_fold::{fold_name, fold_names, fold_schema_names};
-use crate::parquet::parquet_support::{spark_parquet_convert, SparkParquetOptions};
+use crate::parquet::parquet_support::{
+ match_struct_fields, spark_parquet_convert, SparkParquetOptions,
+};
use arrow::array::new_empty_array;
use arrow::compute::can_cast_types;
use arrow::datatypes::{DataType, Field, FieldRef, Schema, SchemaRef};
@@ -364,6 +366,31 @@ fn spark_catalog_name(dt: &DataType) -> String {
DataType::Decimal128(p, s) | DataType::Decimal256(p, s) => {
format!("decimal({p},{s})")
}
+ // Spark's `catalogString` for the complex types (e.g. `array<int>`), so a
+ // scalar-vs-complex rejection reads like Spark's.
+ DataType::List(item)
+ | DataType::LargeList(item)
+ | DataType::FixedSizeList(item, _)
+ | DataType::ListView(item)
+ | DataType::LargeListView(item) => {
+ format!("array<{}>", spark_catalog_name(item.data_type()))
+ }
+ DataType::Map(entries, _) => match entries.data_type() {
+ DataType::Struct(kv) if kv.len() == 2 => format!(
+ "map<{},{}>",
+ spark_catalog_name(kv[0].data_type()),
+ spark_catalog_name(kv[1].data_type())
+ ),
+ _ => "unknown".to_string(),
+ },
+ DataType::Struct(fields) => format!(
+ "struct<{}>",
+ fields
+ .iter()
+ .map(|f| format!("{}:{}", f.name(), spark_catalog_name(f.data_type())))
+ .collect::<Vec<_>>()
+ .join(",")
+ ),
_ => "unknown".to_string(),
}
}
@@ -408,16 +435,17 @@ fn is_string_or_binary(dt: &DataType) -> bool {
}
/// Build a Spark-shaped `SchemaColumnConvertNotSupportedException` carrier for a
-/// rejected Parquet -> Spark conversion. The bracketed column wrapping mirrors
+/// rejected Parquet -> Spark conversion. `column` is the Spark-style column path (`a`, or
+/// `s, x` for a nested leaf); the bracketed wrapping mirrors
/// `Arrays.toString(descriptor.getPath())` in Spark's vectorized reader.
fn parquet_schema_convert_err(
- field_name: &str,
+ column: &str,
physical_type: &DataType,
target_type: &DataType,
) -> DataFusionError {
DataFusionError::External(Box::new(SparkError::ParquetSchemaConvert {
file_path: String::new(),
- column: format!("[{}]", field_name),
+ column: format!("[{}]", column),
physical_type: parquet_primitive_name(physical_type).to_string(),
spark_type: spark_catalog_name(target_type),
}))
@@ -425,22 +453,357 @@ fn parquet_schema_convert_err(
/// Build a `RejectOnNonEmpty` expr wrapping `child`. The rejection fires only
/// when the input batch is non-empty (mirrors Spark's per-row-group check).
+/// `column` is the Spark-style column path, as for [`parquet_schema_convert_err`].
fn reject_on_non_empty_expr(
child: Arc<dyn PhysicalExpr>,
target_field: &FieldRef,
- field_name: &str,
+ column: &str,
physical_type: &DataType,
target_type: &DataType,
) -> Arc<dyn PhysicalExpr> {
Arc::new(RejectOnNonEmpty {
child,
target_field: Arc::clone(target_field),
- column: format!("[{}]", field_name),
+ column: format!("[{}]", column),
physical_type: parquet_primitive_name(physical_type).to_string(),
spark_type: spark_catalog_name(target_type),
})
}
+/// Outcome of checking one Parquet (physical) -> Spark (logical) type pair against the
+/// conversion rules of Spark's vectorized Parquet reader.
+enum ConversionCheck {
+ /// Spark has an updater for the pair (for a same-shape complex pair: for every leaf).
+ Accept,
+ /// Spark rejects the pair; raised at plan time.
+ Reject(DataFusionError),
+ /// Spark rejects the pair, but only while decoding a row group, so the rejection is
+ /// deferred to runtime via [`RejectOnNonEmpty`] (SPARK-26709). Carries the offending
+ /// leaf's column path and physical / requested types for the error message.
+ RejectOnNonEmpty {
+ column: String,
+ physical_type: DataType,
+ target_type: DataType,
+ },
+}
+
+/// Apply the rejection matrix of Spark's `ParquetVectorUpdaterFactory.getUpdater` to a single
+/// physical/logical leaf pair. `column` is the Spark-style column path used in the error (`a`
+/// for a top-level column, `s, x` for a nested leaf, mirroring
+/// `Arrays.toString(descriptor.getPath())`). The rules and their order are exactly those the
+/// adapter applies to top-level columns; [`check_conversion`] applies them to nested leaves.
+fn check_leaf_conversion(
+ physical_type: &DataType,
+ target_type: &DataType,
+ column: &str,
+ options: &SparkParquetOptions,
+) -> ConversionCheck {
+ if physical_type == target_type {
+ return ConversionCheck::Accept;
+ }
+ let reject = || {
+ ConversionCheck::Reject(parquet_schema_convert_err(
+ column,
+ physical_type,
+ target_type,
+ ))
+ };
+ let reject_on_non_empty = || ConversionCheck::RejectOnNonEmpty {
+ column: column.to_string(),
+ physical_type: physical_type.clone(),
+ target_type: target_type.clone(),
+ };
+
+ // Reject reading a string/binary Parquet column as anything else. Spark's
+ // `ParquetVectorUpdaterFactory.getUpdater` BINARY case allows StringType /
+ // BinaryType, or DecimalType only when the column carries a
+ // `DecimalLogicalTypeAnnotation` (which arrow-rs surfaces as `Decimal128`,
+ // not `Binary`). Without this guard, runtime cast paths silently return
+ // nulls, parse strings, or surface as a generic Arrow type-mismatch error.
+ // See #4088 and #4351.
+ if is_string_or_binary(physical_type) && !is_string_or_binary(target_type) {
+ return reject();
+ }
+
+ // Reject reading a primitive numeric Parquet column as StringType /
+ // BinaryType. Spark has no `int -> string` etc. updater. Defer to
+ // runtime via `RejectOnNonEmpty` so empty Parquet files (SPARK-26709)
+ // pass and the JVM shim translates to
+ // `SchemaColumnConvertNotSupportedException`.
+ let physical_is_primitive_numeric = matches!(
+ physical_type,
+ DataType::Boolean
+ | DataType::Int8
+ | DataType::Int16
+ | DataType::Int32
+ | DataType::Int64
+ | DataType::Float32
+ | DataType::Float64
+ );
+ if physical_is_primitive_numeric && is_string_or_binary(target_type) {
+ return reject_on_non_empty();
+ }
+
+ // Decimal-to-decimal narrowing. Spark's `isDecimalTypeMatched` (the
+ // `DecimalLogicalTypeAnnotation` branch) allows the read only when
+ // `dst_scale >= src_scale` AND
+ // `dst_precision - dst_scale >= src_precision - src_scale`.
+ // Either failure means silently dropping fractional digits or losing
+ // integer-side magnitude. See #4089 and #4343.
+ if let (DataType::Decimal128(src_p, src_s), DataType::Decimal128(dst_p, dst_s)) =
+ (physical_type, target_type)
+ {
+ let src_int_precision = i32::from(*src_p) - i32::from(*src_s);
+ let dst_int_precision = i32::from(*dst_p) - i32::from(*dst_s);
+ if dst_s < src_s || dst_int_precision < src_int_precision {
+ return reject();
+ }
+ }
+
+ // Integer-to-decimal narrowing. Spark's `canReadAsDecimal` requires
+ // `precision - scale >= 10` for an INT32 source and `>= 20` for INT64.
+ // Unconditional in all Spark versions, so reject at plan time. See #4344.
+ let int_decimal_min_int_precision = match physical_type {
+ DataType::Int8 | DataType::Int16 | DataType::Int32 => Some(10i32),
+ DataType::Int64 => Some(20i32),
+ _ => None,
+ };
+ if let Some(min_int_precision) = int_decimal_min_int_precision {
+ let dst_precision_scale = match target_type {
+ DataType::Decimal128(p, s) | DataType::Decimal256(p, s) => Some((*p, *s)),
+ _ => None,
+ };
+ if let Some((dst_p, dst_s)) = dst_precision_scale {
+ let dst_int_precision = i32::from(dst_p) - i32::from(dst_s);
+ if dst_int_precision < min_int_precision {
+ return reject();
+ }
+ }
+ }
+
+ // Type promotion (widening). When `allow_type_promotion` is false,
+ // reject the three widenings (INT32→INT64, FLOAT→DOUBLE, INT32→DOUBLE)
+ // that Spark 3.x's vectorized reader rejects. The flag tracks Comet's
+ // per-Spark-version constant in ShimCometConf. Deferred to runtime so
+ // empty files (SPARK-26709) pass.
+ if !options.allow_type_promotion {
+ let is_disallowed_promotion = matches!(
+ (physical_type, target_type),
+ (DataType::Int32, DataType::Int64)
+ | (DataType::Float32, DataType::Float64)
+ | (DataType::Int32, DataType::Float64)
+ );
+ if is_disallowed_promotion {
+ return reject_on_non_empty();
+ }
+ }
+
+ // Reject primitive Parquet conversions Spark's vectorized reader rejects
+ // on every supported version (no matching branch in
+ // `ParquetVectorUpdaterFactory.getUpdater`):
+ //
+ // - `INT64 -> Int*` truncates lower bits.
+ // - `INT64 -> Float*` and `INT32 -> Float32` lose precision.
+ // - `Float* -> Int*` and `Float64 -> Float32` truncate / overflow.
+ // - `INT32 -> Timestamp` / `INT64 -> Date32` / `INT64 -> Timestamp`:
+ // date/timestamp-annotated columns surface as Date32 / Timestamp,
+ // so reaching this branch means the column was un-annotated.
+ // - `Date32 -> Timestamp(LTZ)`: Spark only allows Date -> TimestampNTZ.
+ // - `Timestamp -> Date32`: no Timestamp updater branches into Date.
+ //
+ // Deferred to runtime (SPARK-26709). See #4297.
+ let is_spark_rejected_conversion = matches!(
+ (physical_type, target_type),
+ // Long -> narrower int.
+ (
+ DataType::Int64,
+ DataType::Int8 | DataType::Int16 | DataType::Int32,
+ )
+ // Long -> floating point.
+ | (DataType::Int64, DataType::Float32 | DataType::Float64)
+ // Long -> date / timestamp (raw INT64; annotated columns surface as Date32/Timestamp).
+ | (DataType::Int64, DataType::Date32)
+ | (DataType::Int64, DataType::Timestamp(_, _))
+ // Int -> float (DoubleType is allowed via IntegerToDoubleUpdater; FloatType is not).
+ | (
+ DataType::Int8 | DataType::Int16 | DataType::Int32,
+ DataType::Float32,
+ )
+ // Int -> timestamp (raw INT32; DATE-annotated columns surface as Date32).
+ | (
+ DataType::Int8 | DataType::Int16 | DataType::Int32,
+ DataType::Timestamp(_, _),
+ )
+ // Float -> int / Double -> int (no integer branches under FLOAT/DOUBLE).
+ | (
+ DataType::Float32 | DataType::Float64,
+ DataType::Int8 | DataType::Int16 | DataType::Int32 | DataType::Int64,
+ )
+ // Double -> float (narrowing).
+ | (DataType::Float64, DataType::Float32)
+ // Date -> Timestamp(LTZ). Spark allows Date -> TimestampNTZ only.
+ | (DataType::Date32, DataType::Timestamp(_, Some(_)))
+ // Timestamp -> Date.
+ | (DataType::Timestamp(_, _), DataType::Date32)
+ );
+ if is_spark_rejected_conversion {
+ return reject_on_non_empty();
+ }
+
+ // Spark 3.x refuses to read a Parquet TimestampLTZ column as
+ // TimestampNTZ (SPARK-36182); Spark 4.0 (SPARK-47447) lifted that.
+ // The flag tracks Comet's per-Spark-version constant in
+ // ShimCometConf. Deferred to runtime so empty files (SPARK-26709)
+ // still pass. See #4219.
+ //
+ // This catches all LTZ physical encodings: TIMESTAMP_MICROS /
+ // TIMESTAMP_MILLIS arrive as `Timestamp(_, Some(_))` directly, and
+ // INT96 arrives as `Timestamp(_, Some("UTC"))` because `coerce_int96_tz`
+ // attaches the UTC timezone (see `get_options`) instead of letting
+ // `coerce_int96` strip it to a timezone-free `Timestamp(_, None)`.
+ if !options.allow_timestamp_ltz_to_ntz
+ && matches!(
+ (physical_type, target_type),
+ (
+ DataType::Timestamp(_, Some(_)),
+ DataType::Timestamp(_, None)
+ )
+ )
+ {
+ return reject_on_non_empty();
+ }
+
+ // Scalar/complex mismatch (e.g. TIMESTAMP read as ARRAY<TIMESTAMP>):
+ // Spark's vectorized reader rejects with
+ // SchemaColumnConvertNotSupportedException (SPARK-45604). Same-shape
+ // complex pairs never reach this leaf check (`check_conversion` walks their
+ // leaves instead), so two complex types here differ in shape (e.g. STRUCT
+ // read as ARRAY), which Spark rejects just the same.
+ let is_complex = |t: &DataType| {
+ matches!(
+ t,
+ DataType::Struct(_)
+ | DataType::List(_)
+ | DataType::LargeList(_)
+ | DataType::FixedSizeList(_, _)
+ | DataType::ListView(_)
+ | DataType::LargeListView(_)
+ | DataType::Map(_, _)
+ )
+ };
+ if is_complex(physical_type) || is_complex(target_type) {
+ return reject();
+ }
+
+ ConversionCheck::Accept
+}
+
+/// Check a physical/logical type pair the way Spark's vectorized reader does. Spark runs
+/// `getUpdater` on every *leaf* column regardless of nesting, so same-shape complex pairs
+/// (struct / list / map, at any depth) are walked and [`check_leaf_conversion`] is applied to
+/// each leaf, extending the column path the way `descriptor.getPath()` does (struct field
+/// names, the map entries field plus its `key` / `value`, and `list` plus the element field).
+/// Arrow omits the repeated list group name, so `list` assumes the standard three-level
+/// encoding used by Spark. Legacy or custom group names cannot be recovered from this schema.
+/// Requested struct fields resolve to file fields with the same field-id / case-fold rules the runtime
+/// convert uses ([`match_struct_fields`]); requested fields missing from the file are skipped
+/// (they read as null / default, as before). The first non-`Accept` verdict in leaf order
+/// wins, like Spark, which raises for the first offending column it initializes.
+fn check_conversion(
+ physical_type: &DataType,
+ target_type: &DataType,
+ column: &str,
+ options: &SparkParquetOptions,
+) -> DataFusionResult<ConversionCheck> {
+ // Normalize before dispatching on shape, including dictionary-wrapped containers.
+ if let DataType::Dictionary(_, value_type) = physical_type {
+ return check_conversion(value_type, target_type, column, options);
+ }
+ match (physical_type, target_type) {
+ (DataType::Struct(physical_fields), DataType::Struct(target_fields)) => {
+ let physical_indices = match_struct_fields(physical_fields, target_fields, options)?;
+ for (target_field, physical_index) in target_fields.iter().zip(physical_indices) {
+ let Some(physical_index) = physical_index else {
+ continue;
+ };
+ let physical_field = &physical_fields[physical_index];
+ let check = check_conversion(
+ physical_field.data_type(),
+ target_field.data_type(),
+ &format!("{column}, {}", physical_field.name()),
+ options,
+ )?;
+ if !matches!(check, ConversionCheck::Accept) {
+ return Ok(check);
+ }
+ }
+ Ok(ConversionCheck::Accept)
+ }
+ (
+ DataType::List(physical_item)
+ | DataType::LargeList(physical_item)
+ | DataType::FixedSizeList(physical_item, _)
+ | DataType::ListView(physical_item)
+ | DataType::LargeListView(physical_item),
+ DataType::List(target_item)
+ | DataType::LargeList(target_item)
+ | DataType::FixedSizeList(target_item, _)
+ | DataType::ListView(target_item)
+ | DataType::LargeListView(target_item),
+ ) => check_conversion(
+ physical_item.data_type(),
+ target_item.data_type(),
+ &format!("{column}, list, {}", physical_item.name()),
+ options,
+ ),
+ (
+ DataType::Map(physical_entries, physical_sorted),
+ DataType::Map(target_entries, target_sorted),
+ ) if physical_sorted == target_sorted => {
+ // Map entries are `key` / `value` structs that the runtime convert pairs
+ // positionally (`parquet_convert_map_to_map`), so do the same here.
+ if let (DataType::Struct(physical_kv), DataType::Struct(target_kv)) =
+ (physical_entries.data_type(), target_entries.data_type())
+ {
+ if physical_kv.len() != 2 || target_kv.len() != 2 {
+ return Ok(ConversionCheck::Reject(parquet_schema_convert_err(
+ column,
+ physical_type,
+ target_type,
+ )));
+ }
+ for (physical_field, target_field) in physical_kv.iter().zip(target_kv.iter()) {
+ let check = check_conversion(
+ physical_field.data_type(),
+ target_field.data_type(),
+ &format!(
+ "{column}, {}, {}",
+ physical_entries.name(),
+ physical_field.name()
+ ),
+ options,
+ )?;
+ if !matches!(check, ConversionCheck::Accept) {
+ return Ok(check);
+ }
+ }
+ return Ok(ConversionCheck::Accept);
+ }
+ Ok(ConversionCheck::Reject(parquet_schema_convert_err(
+ column,
+ physical_type,
+ target_type,
+ )))
+ }
+ _ => Ok(check_leaf_conversion(
+ physical_type,
+ target_type,
+ column,
+ options,
+ )),
+ }
+}
+
/// Whether `col_name` (with folded form `col_folded`) is case-insensitively ambiguous in the
/// file. `folded_to_indices` maps each folded physical name to the indices of the original
/// physical fields that fold to it (built once in `create`), so more than one index under the
@@ -781,19 +1144,32 @@ impl SparkPhysicalExprAdapter {
if logical_field.has_valid_extension_type::<VariantType>()
|| logical_field.data_type() != physical_field.data_type()
{
- // Mirror the same string/binary -> non-string/binary rejection in
- // `replace_with_spark_cast`; this branch is reached when the default
- // adapter rejected the cast and we'd otherwise build a CometCastColumnExpr
- // that can't actually convert (e.g. BINARY -> DECIMAL with no
- // `DecimalLogicalTypeAnnotation`). See #4088 and #4351.
- let physical_type = physical_field.data_type();
- let target_type = logical_field.data_type();
- if is_string_or_binary(physical_type) && !is_string_or_binary(target_type) {
- return Err(parquet_schema_convert_err(
- physical_field.name(),
- physical_type,
- target_type,
- ));
+ // Apply the same Spark conversion rules as `replace_with_spark_cast`;
+ // this branch is reached when the default adapter rejected the cast and
+ // we'd otherwise build a CometCastColumnExpr that silently converts, or
+ // can't actually convert (e.g. BINARY -> DECIMAL with no
+ // `DecimalLogicalTypeAnnotation`). See #4088, #4351 and #5671.
+ match check_conversion(
+ physical_field.data_type(),
+ logical_field.data_type(),
+ physical_field.name(),
+ &self.parquet_options,
+ )? {
+ ConversionCheck::Accept => {}
+ ConversionCheck::Reject(err) => return Err(err),
+ ConversionCheck::RejectOnNonEmpty {
+ column,
+ physical_type: leaf_physical_type,
+ target_type: leaf_target_type,
+ } => {
+ return Ok(Transformed::yes(reject_on_non_empty_expr(
+ remapped,
+ logical_field,
+ &column,
+ &leaf_physical_type,
+ &leaf_target_type,
+ )));
+ }
}
let cast_expr: Arc<dyn PhysicalExpr> = Arc::new(
@@ -870,224 +1246,33 @@ impl SparkPhysicalExprAdapter {
return Ok(Transformed::yes(child));
}
- // Reject reading a string/binary Parquet column as anything else. Spark's
- // `ParquetVectorUpdaterFactory.getUpdater` BINARY case allows StringType /
- // BinaryType, or DecimalType only when the column carries a
- // `DecimalLogicalTypeAnnotation` (which arrow-rs surfaces as `Decimal128`,
- // not `Binary`). Without this guard, runtime cast paths silently return
- // nulls, parse strings, or surface as a generic Arrow type-mismatch error.
- // See #4088 and #4351.
- if is_string_or_binary(physical_type) && !is_string_or_binary(target_type) {
- return Err(parquet_schema_convert_err(
- input_field.name(),
- physical_type,
- target_type,
- ));
- }
-
- // Reject reading a primitive numeric Parquet column as StringType /
- // BinaryType. Spark has no `int -> string` etc. updater. Defer to
- // runtime via `RejectOnNonEmpty` so empty Parquet files (SPARK-26709)
- // pass and the JVM shim translates to
- // `SchemaColumnConvertNotSupportedException`.
- let physical_is_primitive_numeric = matches!(
+ // Spark's vectorized reader validates every (file type, requested type) leaf pair
+ // in `ParquetVectorUpdaterFactory.getUpdater`, nested or not. Apply the same rules
+ // here (see `check_leaf_conversion`) so a pair Spark rejects never reaches a
+ // runtime cast that would silently null, parse, or reinterpret values (#5671).
+ match check_conversion(
physical_type,
- DataType::Boolean
- | DataType::Int8
- | DataType::Int16
- | DataType::Int32
- | DataType::Int64
- | DataType::Float32
- | DataType::Float64
- );
- if physical_is_primitive_numeric && is_string_or_binary(target_type) {
- let rejection = reject_on_non_empty_expr(
- child,
- cast.target_field(),
- input_field.name(),
- physical_type,
- target_type,
- );
- return Ok(Transformed::yes(rejection));
- }
-
- // Decimal-to-decimal narrowing. Spark's `isDecimalTypeMatched` (the
- // `DecimalLogicalTypeAnnotation` branch) allows the read only when
- // `dst_scale >= src_scale` AND
- // `dst_precision - dst_scale >= src_precision - src_scale`.
- // Either failure means silently dropping fractional digits or losing
- // integer-side magnitude. See #4089 and #4343.
- if let (DataType::Decimal128(src_p, src_s), DataType::Decimal128(dst_p, dst_s)) =
- (physical_type, target_type)
- {
- let src_int_precision = i32::from(*src_p) - i32::from(*src_s);
- let dst_int_precision = i32::from(*dst_p) - i32::from(*dst_s);
- if dst_s < src_s || dst_int_precision < src_int_precision {
- return Err(parquet_schema_convert_err(
- input_field.name(),
- physical_type,
- target_type,
- ));
- }
- }
-
- // Integer-to-decimal narrowing. Spark's `canReadAsDecimal` requires
- // `precision - scale >= 10` for an INT32 source and `>= 20` for INT64.
- // Unconditional in all Spark versions, so reject at plan time. See #4344.
- let int_decimal_min_int_precision = match physical_type {
- DataType::Int8 | DataType::Int16 | DataType::Int32 => Some(10i32),
- DataType::Int64 => Some(20i32),
- _ => None,
- };
- if let Some(min_int_precision) = int_decimal_min_int_precision {
- let dst_precision_scale = match target_type {
- DataType::Decimal128(p, s) | DataType::Decimal256(p, s) => Some((*p, *s)),
- _ => None,
- };
- if let Some((dst_p, dst_s)) = dst_precision_scale {
- let dst_int_precision = i32::from(dst_p) - i32::from(dst_s);
- if dst_int_precision < min_int_precision {
- return Err(parquet_schema_convert_err(
- input_field.name(),
- physical_type,
- target_type,
- ));
- }
- }
- }
-
- // Type promotion (widening). When `allow_type_promotion` is false,
- // reject the three widenings (INT32→INT64, FLOAT→DOUBLE, INT32→DOUBLE)
- // that Spark 3.x's vectorized reader rejects. The flag tracks Comet's
- // per-Spark-version constant in ShimCometConf. Deferred to runtime so
- // empty files (SPARK-26709) pass.
- if !self.parquet_options.allow_type_promotion {
- let is_disallowed_promotion = matches!(
- (physical_type, target_type),
- (DataType::Int32, DataType::Int64)
- | (DataType::Float32, DataType::Float64)
- | (DataType::Int32, DataType::Float64)
- );
- if is_disallowed_promotion {
- let rejection = reject_on_non_empty_expr(
- Arc::clone(&child),
+ target_type,
+ input_field.name(),
+ &self.parquet_options,
+ )? {
+ ConversionCheck::Accept => {}
+ ConversionCheck::Reject(err) => return Err(err),
+ ConversionCheck::RejectOnNonEmpty {
+ column,
+ physical_type: leaf_physical_type,
+ target_type: leaf_target_type,
+ } => {
+ return Ok(Transformed::yes(reject_on_non_empty_expr(
+ child,
cast.target_field(),
- input_field.name(),
- physical_type,
- target_type,
- );
- return Ok(Transformed::yes(rejection));
+ &column,
+ &leaf_physical_type,
+ &leaf_target_type,
+ )));
}
}
- // Reject primitive Parquet conversions Spark's vectorized reader rejects
- // on every supported version (no matching branch in
- // `ParquetVectorUpdaterFactory.getUpdater`):
- //
- // - `INT64 -> Int*` truncates lower bits.
- // - `INT64 -> Float*` and `INT32 -> Float32` lose precision.
- // - `Float* -> Int*` and `Float64 -> Float32` truncate / overflow.
- // - `INT32 -> Timestamp` / `INT64 -> Date32` / `INT64 -> Timestamp`:
- // date/timestamp-annotated columns surface as Date32 / Timestamp,
- // so reaching this branch means the column was un-annotated.
- // - `Date32 -> Timestamp(LTZ)`: Spark only allows Date -> TimestampNTZ.
- // - `Timestamp -> Date32`: no Timestamp updater branches into Date.
- //
- // Deferred to runtime (SPARK-26709). See #4297.
- let is_spark_rejected_conversion = matches!(
- (physical_type, target_type),
- // Long -> narrower int.
- (
- DataType::Int64,
- DataType::Int8 | DataType::Int16 | DataType::Int32,
- )
- // Long -> floating point.
- | (DataType::Int64, DataType::Float32 | DataType::Float64)
- // Long -> date / timestamp (raw INT64; annotated columns surface as Date32/Timestamp).
- | (DataType::Int64, DataType::Date32)
- | (DataType::Int64, DataType::Timestamp(_, _))
- // Int -> float (DoubleType is allowed via IntegerToDoubleUpdater; FloatType is not).
- | (
- DataType::Int8 | DataType::Int16 | DataType::Int32,
- DataType::Float32,
- )
- // Int -> timestamp (raw INT32; DATE-annotated columns surface as Date32).
- | (
- DataType::Int8 | DataType::Int16 | DataType::Int32,
- DataType::Timestamp(_, _),
- )
- // Float -> int / Double -> int (no integer branches under FLOAT/DOUBLE).
- | (
- DataType::Float32 | DataType::Float64,
- DataType::Int8 | DataType::Int16 | DataType::Int32 | DataType::Int64,
- )
- // Double -> float (narrowing).
- | (DataType::Float64, DataType::Float32)
- // Date -> Timestamp(LTZ). Spark allows Date -> TimestampNTZ only.
- | (DataType::Date32, DataType::Timestamp(_, Some(_)))
- // Timestamp -> Date.
- | (DataType::Timestamp(_, _), DataType::Date32)
- );
- if is_spark_rejected_conversion {
- let rejection = reject_on_non_empty_expr(
- child,
- cast.target_field(),
- input_field.name(),
- physical_type,
- target_type,
- );
- return Ok(Transformed::yes(rejection));
- }
-
- // Spark 3.x refuses to read a Parquet TimestampLTZ column as
- // TimestampNTZ (SPARK-36182); Spark 4.0 (SPARK-47447) lifted that.
- // The flag tracks Comet's per-Spark-version constant in
- // ShimCometConf. Deferred to runtime so empty files (SPARK-26709)
- // still pass. See #4219.
- //
- // This catches all LTZ physical encodings: TIMESTAMP_MICROS /
- // TIMESTAMP_MILLIS arrive as `Timestamp(_, Some(_))` directly, and
- // INT96 arrives as `Timestamp(_, Some("UTC"))` because `coerce_int96_tz`
- // attaches the UTC timezone (see `get_options`) instead of letting
- // `coerce_int96` strip it to a timezone-free `Timestamp(_, None)`.
- if !self.parquet_options.allow_timestamp_ltz_to_ntz
- && matches!(
- (physical_type, target_type),
- (
- DataType::Timestamp(_, Some(_)),
- DataType::Timestamp(_, None)
- )
- )
- {
- let rejection = reject_on_non_empty_expr(
- Arc::clone(&child),
- cast.target_field(),
- input_field.name(),
- physical_type,
- target_type,
- );
- return Ok(Transformed::yes(rejection));
- }
-
- // Scalar/complex mismatch (e.g. TIMESTAMP read as ARRAY<TIMESTAMP>):
- // Spark's vectorized reader rejects with
- // SchemaColumnConvertNotSupportedException (SPARK-45604). Same-shape
- // complex pairs and timestamp→timestamp / timestamp→int64 fall through
- // to CometCastColumnExpr below.
- let is_complex = |t: &DataType| {
- matches!(
- t,
- DataType::Struct(_) | DataType::List(_) | DataType::Map(_, _)
- )
- };
- if is_complex(physical_type) != is_complex(target_type) {
- return Err(parquet_schema_convert_err(
- input_field.name(),
- physical_type,
- target_type,
- ));
- }
-
// Leave DataFusion's `CastExpr` in place for a pure structural narrowing instead
// of swapping in `CometCastColumnExpr` (see `is_pure_structural_narrowing`'s doc
// comment for why that's safe and apache/datafusion-comet#4859 for why it matters).
@@ -1121,15 +1306,27 @@ impl SparkPhysicalExprAdapter {
return Ok(Transformed::no(expr));
}
- // Same-shape complex casts, timestamp tz relabel (e.g. Timestamp(us, None)
- // -> Timestamp(us, Some("UTC")) for INT96 reads), and Timestamp -> Int64
+ // Complex casts (including changes in list representation), timestamp tz relabel
+ // (e.g. Timestamp(us, None) -> Timestamp(us, Some("UTC")) for INT96 reads), and
+ // Timestamp -> Int64
// (Spark's `nanosAsLong`) need spark_parquet_convert: it handles nested
// field selection, metadata-only tz changes, and raw-value reinterpretation
// that Spark's Cast would otherwise convert incorrectly.
if matches!(
(physical_type, target_type),
(DataType::Struct(_), DataType::Struct(_))
- | (DataType::List(_), DataType::List(_))
+ | (
+ DataType::List(_)
+ | DataType::LargeList(_)
+ | DataType::FixedSizeList(_, _)
+ | DataType::ListView(_)
+ | DataType::LargeListView(_),
+ DataType::List(_)
+ | DataType::LargeList(_)
+ | DataType::FixedSizeList(_, _)
+ | DataType::ListView(_)
+ | DataType::LargeListView(_)
+ )
| (DataType::Map(_, _), DataType::Map(_, _))
| (DataType::Timestamp(_, _), DataType::Timestamp(_, _))
| (DataType::Timestamp(_, _), DataType::Int64)
@@ -1343,15 +1540,21 @@ mod test {
use crate::parquet::cast_column::CometCastColumnExpr;
use crate::parquet::parquet_support::SparkParquetOptions;
use crate::parquet::schema_adapter::{
- is_pure_structural_narrowing, SparkPhysicalExprAdapterFactory,
+ check_conversion, is_pure_structural_narrowing, ConversionCheck,
+ SparkPhysicalExprAdapterFactory,
};
+ use arrow::array::cast::AsArray;
use arrow::array::UInt32Array;
use arrow::array::{
- BinaryArray, Date32Array, Decimal128Array, Float32Array, Float64Array, Int32Array,
- Int64Array, StringArray, TimestampMicrosecondArray,
+ Array, ArrayRef, BinaryArray, Date32Array, Decimal128Array, FixedSizeListArray,
+ Float32Array, Float64Array, Int32Array, Int64Array, LargeListArray, ListArray, MapArray,
+ StringArray, StructArray, TimestampMicrosecondArray, TimestampMillisecondArray,
};
+ use arrow::buffer::OffsetBuffer;
use arrow::datatypes::SchemaRef;
- use arrow::datatypes::{DataType, Field, Fields, Schema};
+ use arrow::datatypes::{
+ DataType, Field, Fields, Int64Type, Schema, TimeUnit, TimestampMicrosecondType,
+ };
use arrow::record_batch::RecordBatch;
use datafusion::common::DataFusionError;
use datafusion::datasource::listing::PartitionedFile;
@@ -1361,7 +1564,7 @@ mod test {
use datafusion::execution::TaskContext;
use datafusion::physical_expr::expressions::Column;
use datafusion::physical_expr::PhysicalExpr;
- use datafusion::physical_plan::ExecutionPlan;
+ use datafusion::physical_plan::{ExecutionPlan, SendableRecordBatchStream};
use datafusion_comet_spark_expr::test_common::file_util::get_temp_filename;
use datafusion_comet_spark_expr::EvalMode;
use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory;
@@ -1931,12 +2134,14 @@ mod test {
Ok(())
}
- /// Create a Parquet file containing a single batch and then read the batch back using
- /// the specified required_schema. This will cause the PhysicalExprAdapter code to be used.
- async fn roundtrip(
+ /// Write `batch` to a temp Parquet file and execute a `DataSourceExec` over it with
+ /// `required_schema` and the Spark adapter configured from `options`, so the
+ /// `PhysicalExprAdapter` code runs exactly as it does in a native scan.
+ fn scan_parquet(
batch: &RecordBatch,
required_schema: SchemaRef,
- ) -> Result<RecordBatch, DataFusionError> {
+ options: SparkParquetOptions,
+ ) -> Result<SendableRecordBatchStream, DataFusionError> {
let filename = get_temp_filename();
let filename = filename.as_path().as_os_str().to_str().unwrap().to_string();
let file = File::create(&filename)?;
@@ -1946,17 +2151,13 @@ mod test {
let object_store_url = ObjectStoreUrl::local_filesystem();
- let mut spark_parquet_options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
- spark_parquet_options.allow_cast_unsigned_ints = true;
-
// Create expression adapter factory for Spark-compatible schema adaptation
- let expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory> = Arc::new(
- SparkPhysicalExprAdapterFactory::new(spark_parquet_options, None),
- );
+ let expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory> =
+ Arc::new(SparkPhysicalExprAdapterFactory::new(options, None));
let parquet_source = ParquetSource::new(required_schema);
- let files = FileGroup::new(vec![PartitionedFile::from_path(filename.to_string())?]);
+ let files = FileGroup::new(vec![PartitionedFile::from_path(filename)?]);
let file_scan_config =
FileScanConfigBuilder::new(object_store_url, Arc::new(parquet_source))
.with_file_groups(vec![files])
@@ -1964,11 +2165,801 @@ mod test {
.build();
let parquet_exec = DataSourceExec::new(Arc::new(file_scan_config));
+ parquet_exec.execute(0, Arc::new(TaskContext::default()))
+ }
- let mut stream = parquet_exec.execute(0, Arc::new(TaskContext::default()))?;
+ /// Create a Parquet file containing a single batch and then read the batch back using
+ /// the specified required_schema. This will cause the PhysicalExprAdapter code to be used.
+ async fn roundtrip(
+ batch: &RecordBatch,
+ required_schema: SchemaRef,
+ ) -> Result<RecordBatch, DataFusionError> {
+ let mut spark_parquet_options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ spark_parquet_options.allow_cast_unsigned_ints = true;
+ let mut stream = scan_parquet(batch, required_schema, spark_parquet_options)?;
stream.next().await.unwrap()
}
+ /// Build a one-column batch `s: struct<field>` holding `values`, for the nested
+ /// conversion tests (#5671).
+ fn struct_batch(field: Field, values: ArrayRef) -> Result<RecordBatch, DataFusionError> {
+ let fields = Fields::from(vec![field]);
+ let s = StructArray::try_new(fields.clone(), vec![values], None)?;
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "s",
+ DataType::Struct(fields),
+ true,
+ )]));
+ Ok(RecordBatch::try_new(schema, vec![Arc::new(s)])?)
+ }
+
+ /// Read schema `s: struct<fields>` for the nested conversion tests.
+ fn struct_schema(fields: Vec<Field>) -> SchemaRef {
+ Arc::new(Schema::new(vec![Field::new(
+ "s",
+ DataType::Struct(Fields::from(fields)),
+ true,
+ )]))
+ }
+
+ /// Read `batch` under `required_schema` with the default options, asserting the read is
+ /// rejected, and return the error message.
+ async fn nested_rejection_message(batch: &RecordBatch, required_schema: SchemaRef) -> String {
+ roundtrip(batch, required_schema)
+ .await
+ .expect_err("expected ParquetSchemaConvert for the nested conversion")
+ .to_string()
+ }
+
+ /// `INT64 -> int` inside a struct. Spark's vectorized reader runs `getUpdater` per leaf,
+ /// so the rejection of the top-level `parquet_long_read_as_int_errors` applies to `s.x`
+ /// too; Comet previously cast with `safe: true` and returned `{null}` for the
+ /// overflowing row. See #5671.
+ #[tokio::test]
+ async fn nested_long_read_as_int_errors() -> Result<(), DataFusionError> {
+ let batch = struct_batch(
+ Field::new("x", DataType::Int64, true),
+ Arc::new(Int64Array::from(vec![5_000_000_000i64, 1])),
+ )?;
+ let msg = nested_rejection_message(
+ &batch,
+ struct_schema(vec![Field::new("x", DataType::Int32, true)]),
+ )
+ .await;
+ assert!(
+ msg.contains("Column: [[s, x]]")
+ && msg.contains("Expected: int")
+ && msg.contains("Found: INT64"),
+ "unexpected error: {msg}"
+ );
+ Ok(())
+ }
+
+ /// `INT32 -> string` inside a struct (previously stringified to `"1"`, `"2"`).
+ #[tokio::test]
+ async fn nested_int_read_as_string_errors() -> Result<(), DataFusionError> {
+ let batch = struct_batch(
+ Field::new("x", DataType::Int32, true),
+ Arc::new(Int32Array::from(vec![1, 2])),
+ )?;
+ let msg = nested_rejection_message(
+ &batch,
+ struct_schema(vec![Field::new("x", DataType::Utf8, true)]),
+ )
+ .await;
+ assert!(
+ msg.contains("Column: [[s, x]]")
+ && msg.contains("Expected: string")
+ && msg.contains("Found: INT32"),
+ "unexpected error: {msg}"
+ );
+ Ok(())
+ }
+
+ /// Decimal precision narrowing inside a struct (previously nulled the value).
+ #[tokio::test]
+ async fn nested_decimal_narrowing_errors() -> Result<(), DataFusionError> {
+ let values = Decimal128Array::from(vec![12_345_678i128])
+ .with_precision_and_scale(10, 2)
+ .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))?;
+ let batch = struct_batch(
+ Field::new("d", DataType::Decimal128(10, 2), true),
+ Arc::new(values),
+ )?;
+ let msg = nested_rejection_message(
+ &batch,
+ struct_schema(vec![Field::new("d", DataType::Decimal128(5, 2), true)]),
+ )
+ .await;
+ assert!(
+ msg.contains("Column: [[s, d]]")
+ && msg.contains("Expected: decimal(5,2)")
+ && msg.contains("Found: INT64"),
+ "unexpected error: {msg}"
+ );
+ Ok(())
+ }
+
+ /// `BINARY -> int` inside a struct (previously parsed `"12"` and nulled `"abc"`).
+ #[tokio::test]
+ async fn nested_string_read_as_int_errors() -> Result<(), DataFusionError> {
+ let batch = struct_batch(
+ Field::new("x", DataType::Utf8, true),
+ Arc::new(StringArray::from(vec!["12", "abc"])),
+ )?;
+ let msg = nested_rejection_message(
+ &batch,
+ struct_schema(vec![Field::new("x", DataType::Int32, true)]),
+ )
+ .await;
+ assert!(
+ msg.contains("Column: [[s, x]]")
+ && msg.contains("Expected: int")
+ && msg.contains("Found: BINARY"),
+ "unexpected error: {msg}"
+ );
+ Ok(())
+ }
+
+ /// `INT32 -> array<int>` inside a struct (previously wrapped each value as `[1]`).
+ #[tokio::test]
+ async fn nested_int_read_as_list_errors() -> Result<(), DataFusionError> {
+ let batch = struct_batch(
+ Field::new("x", DataType::Int32, true),
+ Arc::new(Int32Array::from(vec![1, 2])),
+ )?;
+ let list_type = DataType::List(Arc::new(Field::new("element", DataType::Int32, true)));
+ let msg = nested_rejection_message(
+ &batch,
+ struct_schema(vec![Field::new("x", list_type, true)]),
+ )
+ .await;
+ assert!(
+ msg.contains("Column: [[s, x]]")
+ && msg.contains("Expected: array<int>")
+ && msg.contains("Found: INT32"),
+ "unexpected error: {msg}"
+ );
+ Ok(())
+ }
+
+ /// `array<int> -> int` inside a struct. Previously the unconverted list array was handed
+ /// to `StructArray::new`, which panicked with "Incorrect datatype for StructArray field";
+ /// this goes through the default adapter's failure path (`wrap_all_type_mismatches`).
+ #[tokio::test]
+ async fn nested_list_read_as_int_errors() -> Result<(), DataFusionError> {
+ let list = ListArray::new(
+ Arc::new(Field::new("element", DataType::Int32, true)),
+ OffsetBuffer::new(vec![0, 1, 2].into()),
+ Arc::new(Int32Array::from(vec![1, 2])),
+ None,
+ );
+ let batch = struct_batch(
+ Field::new("x", list.data_type().clone(), true),
+ Arc::new(list),
+ )?;
+ let msg = nested_rejection_message(
+ &batch,
+ struct_schema(vec![Field::new("x", DataType::Int32, true)]),
+ )
+ .await;
+ assert!(
+ msg.contains("Column: [[s, x]]") && msg.contains("Expected: int"),
+ "unexpected error: {msg}"
+ );
+ Ok(())
+ }
+
+ /// The walk descends through lists: `array<struct<x: INT64>>` read as
+ /// `array<struct<x: int>>` is rejected at the leaf, with the list element in the path.
+ #[tokio::test]
+ async fn nested_list_of_struct_long_read_as_int_errors() -> Result<(), DataFusionError> {
+ let element_fields = Fields::from(vec![Field::new("x", DataType::Int64, true)]);
+ let elements = StructArray::try_new(
+ element_fields.clone(),
+ vec![Arc::new(Int64Array::from(vec![5_000_000_000i64, 1]))],
+ None,
+ )?;
+ let element_field = Arc::new(Field::new(
+ "element",
+ DataType::Struct(element_fields),
+ true,
+ ));
+ let list = ListArray::new(
+ element_field,
+ OffsetBuffer::new(vec![0, 2].into()),
+ Arc::new(elements),
+ None,
+ );
+ let file_schema = Arc::new(Schema::new(vec![Field::new(
+ "a",
+ list.data_type().clone(),
+ true,
+ )]));
+ let batch = RecordBatch::try_new(file_schema, vec![Arc::new(list)])?;
+
+ let read_element = Field::new(
+ "element",
+ DataType::Struct(Fields::from(vec![Field::new("x", DataType::Int32, true)])),
+ true,
+ );
+ let required_schema = Arc::new(Schema::new(vec![Field::new(
+ "a",
+ DataType::List(Arc::new(read_element)),
+ true,
+ )]));
+ let msg = nested_rejection_message(&batch, required_schema).await;
+ assert!(
+ // The synthetic `list` segment assumes standard encoding; Arrow omits custom names.
+ msg.contains("Column: [[a, list, element, x]]")
+ && msg.contains("Expected: int")
+ && msg.contains("Found: INT64"),
+ "unexpected error: {msg}"
+ );
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn list_representations_preserve_missing_struct_fields() -> Result<(), DataFusionError> {
+ let fields = Fields::from(vec![
+ Field::new("old", DataType::Int64, true),
+ Field::new("keep", DataType::Int64, true),
+ ]);
+ let values = StructArray::try_new(
+ fields.clone(),
+ vec![
+ Arc::new(Int64Array::from(vec![Some(7), Some(8), None])),
+ Arc::new(Int64Array::from(vec![Some(2), Some(3), None])),
+ ],
+ Some(arrow::buffer::NullBuffer::from(vec![true, true, false])),
+ )?;
+ let item = Arc::new(Field::new("item", DataType::Struct(fields), true));
+ let list: ArrayRef = Arc::new(ListArray::try_new(
+ Arc::clone(&item),
+ OffsetBuffer::new(vec![0, 1, 2, 3].into()),
+ Arc::new(values),
+ Some(arrow::buffer::NullBuffer::from(vec![true, false, true])),
+ )?);
+ let requested_item = Arc::new(Field::new(
+ "item",
+ DataType::Struct(Fields::from(vec![
+ Field::new("missing", DataType::Int64, true),
+ Field::new("keep", DataType::Int64, true),
+ ])),
+ true,
+ ));
+ let representations = |item: &Arc<Field>| {
+ [
+ DataType::LargeList(Arc::clone(item)),
+ DataType::List(Arc::clone(item)),
+ DataType::FixedSizeList(Arc::clone(item), 1),
+ DataType::ListView(Arc::clone(item)),
+ DataType::LargeListView(Arc::clone(item)),
+ ]
+ };
+ for physical in representations(&item) {
+ let array = arrow::compute::cast(&list, &physical)?;
+ for nested in [false, true] {
+ let batch = if nested {
+ struct_batch(Field::new("a", physical.clone(), true), Arc::clone(&array))?
+ } else {
+ RecordBatch::try_new(
+ Arc::new(Schema::new(vec![Field::new("a", physical.clone(), true)])),
+ vec![Arc::clone(&array)],
+ )?
+ };
+ for target in representations(&requested_item) {
+ let field = Field::new("a", target.clone(), true);
+ let schema = if nested {
+ struct_schema(vec![field])
+ } else {
+ Arc::new(Schema::new(vec![field]))
+ };
+ // Write real Parquet, retaining Arrow's list representation metadata.
+ // Field IDs are disabled by roundtrip's default options.
+ let result = roundtrip(&batch, Arc::clone(&schema)).await?;
+ assert_eq!(result.schema(), schema);
+ let output = if nested {
+ result.column(0).as_struct().column(0)
+ } else {
+ result.column(0)
+ };
+ let output =
+ arrow::compute::cast(output, &DataType::List(Arc::clone(&requested_item)))?;
+ let output = output.as_list::<i32>();
+ assert_eq!(output.len(), 3);
+ assert!(output.is_null(1));
+ let first = output.value(0);
+ let first = first.as_struct();
+ assert!(first.column(0).is_null(0), "{physical:?} -> {target:?}, nested={nested}: missing field read old's value");
+ assert_eq!(first.column(1).as_primitive::<Int64Type>().value(0), 2);
+ assert!(output.value(2).is_null(0));
+ }
+ }
+ }
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn nested_list_offset_width_preserves_values() -> Result<(), DataFusionError> {
+ for large in [false, true] {
+ let values = vec![Some(vec![Some(5_000_000_000i64), None]), None, Some(vec![])];
+ let array: ArrayRef = if large {
+ Arc::new(LargeListArray::from_iter_primitive::<Int64Type, _, _>(
+ values,
+ ))
+ } else {
+ Arc::new(ListArray::from_iter_primitive::<Int64Type, _, _>(values))
+ };
+ let batch = struct_batch(Field::new("a", array.data_type().clone(), true), array)?;
+ for large_target in [false, true] {
+ let item = Arc::new(Field::new("item", DataType::Int64, true));
+ let target = if large_target {
+ DataType::LargeList(item)
+ } else {
+ DataType::List(item)
+ };
+ let schema = struct_schema(vec![Field::new("a", target, true)]);
+ let result = roundtrip(&batch, Arc::clone(&schema)).await?;
+ assert_eq!(result.schema(), schema);
+ let expected = arrow::compute::cast(batch.column(0), schema.field(0).data_type())?;
+ assert_eq!(result.column(0).to_data(), expected.to_data());
+ }
+ }
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn nested_list_view_conversion() -> Result<(), DataFusionError> {
+ let values = ListArray::from_iter_primitive::<Int64Type, _, _>(vec![
+ Some(vec![Some(1), Some(2)]),
+ Some(vec![Some(5_000_000_000), None]),
+ None,
+ Some(vec![]),
+ ]);
+ let item = Arc::new(Field::new("item", DataType::Int64, true));
+ for physical_type in [
+ DataType::ListView(Arc::clone(&item)),
+ DataType::LargeListView(Arc::clone(&item)),
+ ] {
+ let array = arrow::compute::cast(&values, &physical_type)?;
+ let batch = struct_batch(Field::new("a", physical_type, true), array)?;
+ let schema = struct_schema(vec![Field::new(
+ "a",
+ DataType::List(Arc::clone(&item)),
+ true,
+ )]);
+ let result = roundtrip(&batch, Arc::clone(&schema)).await?;
+ assert_eq!(result.schema(), schema);
+ assert_eq!(
+ result.column(0).as_struct().column(0).to_data(),
+ values.to_data()
+ );
+
+ let narrow_item = Arc::new(Field::new("item", DataType::Int32, true));
+ for target in [
+ DataType::List(Arc::clone(&narrow_item)),
+ DataType::ListView(Arc::clone(&narrow_item)),
+ DataType::LargeListView(narrow_item),
+ ] {
+ let msg = nested_rejection_message(
+ &batch,
+ struct_schema(vec![Field::new("a", target, true)]),
+ )
+ .await;
+ assert!(
+ msg.contains("Column: [[s, a, list, item]]")
+ && msg.contains("Expected: int")
+ && msg.contains("Found: INT64"),
+ "{msg}"
+ );
+ }
+ }
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn nested_fixed_size_list_preserves_values() -> Result<(), DataFusionError> {
+ let values = FixedSizeListArray::from_iter_primitive::<Int64Type, _, _>(
+ vec![
+ Some(vec![Some(1), Some(2)]),
+ None,
+ Some(vec![Some(5_000_000_000), None]),
+ ],
+ 2,
+ );
+ let batch = struct_batch(
+ Field::new("a", values.data_type().clone(), true),
+ Arc::new(values),
+ )?;
+ let item = Arc::new(Field::new("item", DataType::Int64, true));
+ for target in [
+ DataType::List(Arc::clone(&item)),
+ DataType::LargeList(Arc::clone(&item)),
+ DataType::FixedSizeList(item, 2),
+ ] {
+ let schema = struct_schema(vec![Field::new("a", target, true)]);
+ let result = roundtrip(&batch, Arc::clone(&schema)).await?;
+ assert_eq!(result.schema(), schema);
+ let expected = arrow::compute::cast(batch.column(0), schema.field(0).data_type())?;
+ assert_eq!(result.column(0).to_data(), expected.to_data());
+ }
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn nested_fixed_size_list_rejects_narrowing() -> Result<(), DataFusionError> {
+ let values = FixedSizeListArray::from_iter_primitive::<Int64Type, _, _>(
+ vec![Some(vec![Some(5_000_000_000), Some(1)])],
+ 2,
+ );
+ let batch = struct_batch(
+ Field::new("a", values.data_type().clone(), true),
+ Arc::new(values),
+ )?;
+ let item = Arc::new(Field::new("item", DataType::Int32, true));
+ for target in [
+ DataType::List(Arc::clone(&item)),
+ DataType::LargeList(Arc::clone(&item)),
+ DataType::FixedSizeList(item, 2),
+ ] {
+ let msg = nested_rejection_message(
+ &batch,
+ struct_schema(vec![Field::new("a", target, true)]),
+ )
+ .await;
+ assert!(
+ msg.contains("Column: [[s, a, list, item]]")
+ && msg.contains("Expected: int")
+ && msg.contains("Found: INT64"),
+ "{msg}"
+ );
+ }
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn nested_large_list_rejects_narrowing() -> Result<(), DataFusionError> {
+ let values =
+ LargeListArray::from_iter_primitive::<Int64Type, _, _>(vec![Some(vec![Some(
+ 5_000_000_000i64,
+ )])]);
+ let batch = struct_batch(
+ Field::new("a", values.data_type().clone(), true),
+ Arc::new(values),
+ )?;
+ for large_target in [false, true] {
+ let item = Arc::new(Field::new("item", DataType::Int32, true));
+ let target = if large_target {
+ DataType::LargeList(item)
+ } else {
+ DataType::List(item)
+ };
+ let msg = nested_rejection_message(
+ &batch,
+ struct_schema(vec![Field::new("a", target, true)]),
+ )
+ .await;
+ assert!(
+ msg.contains("Column: [[s, a, list, item]]")
+ && msg.contains("Expected: int")
+ && msg.contains("Found: INT64"),
+ "{msg}"
+ );
+ }
+ Ok(())
+ }
+
+ #[test]
+ fn nested_dictionary_containers_are_checked_by_value_type() -> Result<(), DataFusionError> {
+ let options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ let physical_struct = DataType::Struct(Fields::from(vec![
+ Field::new("x", DataType::Int64, true),
+ Field::new("unused", DataType::Int32, true),
+ ]));
+ let pruned = DataType::Struct(Fields::from(vec![Field::new("x", DataType::Int64, true)]));
+ let narrowed = DataType::Struct(Fields::from(vec![Field::new("x", DataType::Int32, true)]));
+ for (physical, target, invalid_target) in [
+ (physical_struct.clone(), pruned.clone(), narrowed.clone()),
+ (
+ DataType::LargeList(Arc::new(Field::new("item", physical_struct.clone(), true))),
+ DataType::List(Arc::new(Field::new("item", pruned.clone(), true))),
+ DataType::List(Arc::new(Field::new("item", narrowed.clone(), true))),
+ ),
+ (
+ map_type(physical_struct),
+ map_type(pruned),
+ map_type(narrowed),
+ ),
+ ] {
+ let dictionary = DataType::Dictionary(Box::new(DataType::Int32), Box::new(physical));
+ assert!(matches!(
+ check_conversion(&dictionary, &target, "a", &options)?,
+ ConversionCheck::Accept
+ ));
+ assert!(matches!(
+ check_conversion(&dictionary, &invalid_target, "a", &options)?,
+ ConversionCheck::RejectOnNonEmpty { .. }
+ ));
+ }
+ Ok(())
+ }
+
+ #[test]
+ fn nested_map_shape_mismatch_is_rejected() -> Result<(), DataFusionError> {
+ let options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ let valid = map_type(DataType::Int64);
+ let DataType::Map(entries, _) = &valid else {
+ unreachable!()
+ };
+ for invalid in [
+ DataType::Map(Arc::clone(entries), true),
+ DataType::Map(
+ Arc::new(Field::new("key_value", DataType::Int64, false)),
+ false,
+ ),
+ DataType::Map(
+ Arc::new(Field::new(
+ "key_value",
+ DataType::Struct(Fields::from(vec![Field::new("key", DataType::Utf8, false)])),
+ false,
+ )),
+ false,
+ ),
+ ] {
+ for (physical, target) in [(&valid, &invalid), (&invalid, &valid)] {
+ assert!(matches!(
+ check_conversion(physical, target, "m", &options)?,
+ ConversionCheck::Reject(_)
+ ));
+ }
+ }
+ Ok(())
+ }
+
+ /// Build the Arrow `Map<string, value_type>` type with Parquet's `key_value` / `key` /
+ /// `value` names, as Spark-written files surface it.
+ fn map_type(value_type: DataType) -> DataType {
+ let entries = Fields::from(vec![
+ Field::new("key", DataType::Utf8, false),
+ Field::new("value", value_type, true),
+ ]);
+ DataType::Map(
+ Arc::new(Field::new("key_value", DataType::Struct(entries), false)),
+ false,
+ )
+ }
+
+ /// The walk descends through maps: `map<string, INT64>` read as `map<string, int>` is
+ /// rejected at the value leaf, with the entries field in the path.
+ #[tokio::test]
+ async fn nested_map_value_long_read_as_int_errors() -> Result<(), DataFusionError> {
+ let DataType::Map(entries_field, _) = map_type(DataType::Int64) else {
+ unreachable!()
+ };
+ let DataType::Struct(entries_fields) = entries_field.data_type() else {
+ unreachable!()
+ };
+ let entries = StructArray::try_new(
+ entries_fields.clone(),
+ vec![
+ Arc::new(StringArray::from(vec!["k"])),
+ Arc::new(Int64Array::from(vec![5_000_000_000i64])),
+ ],
+ None,
+ )?;
+ let map = MapArray::try_new(
+ Arc::clone(&entries_field),
+ OffsetBuffer::new(vec![0, 1].into()),
+ entries,
+ None,
+ false,
+ )?;
+ let file_schema = Arc::new(Schema::new(vec![Field::new(
+ "m",
+ map.data_type().clone(),
+ true,
+ )]));
+ let batch = RecordBatch::try_new(file_schema, vec![Arc::new(map)])?;
+
+ let required_schema = Arc::new(Schema::new(vec![Field::new(
+ "m",
+ map_type(DataType::Int32),
+ true,
+ )]));
+ let msg = nested_rejection_message(&batch, required_schema).await;
+ assert!(
+ msg.contains("Column: [[m, key_value, value]]")
+ && msg.contains("Expected: int")
+ && msg.contains("Found: INT64"),
+ "unexpected error: {msg}"
+ );
+ Ok(())
+ }
+
+ /// Nested fields resolve with the same case-insensitive fold as the runtime convert, so
+ /// the rule applies to file field `X` read as `x` and the path names the file field.
+ #[tokio::test]
+ async fn nested_case_insensitive_field_match_applies_rules() -> Result<(), DataFusionError> {
+ let batch = struct_batch(
+ Field::new("X", DataType::Int64, true),
+ Arc::new(Int64Array::from(vec![1i64, 2])),
+ )?;
+ let msg = nested_rejection_message(
+ &batch,
+ struct_schema(vec![Field::new("x", DataType::Int32, true)]),
+ )
+ .await;
+ assert!(
+ msg.contains("Column: [[s, X]]") && msg.contains("Found: INT64"),
+ "unexpected error: {msg}"
+ );
+ Ok(())
+ }
+
+ /// Nested fields resolve by Parquet field id when `use_field_id` is set (names differ),
+ /// and the rule applies to the id-matched pair.
+ #[tokio::test]
+ async fn nested_field_id_match_applies_rules() -> Result<(), DataFusionError> {
+ let batch = struct_batch(
+ Field::new("a", DataType::Int64, true).with_metadata(id_meta("1")),
+ Arc::new(Int64Array::from(vec![1i64, 2])),
+ )?;
+ let required_schema = struct_schema(vec![
+ Field::new("b", DataType::Int32, true).with_metadata(id_meta("1"))
+ ]);
+ let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ options.use_field_id = true;
+ let mut stream = scan_parquet(&batch, required_schema, options)?;
+ let err = stream
+ .next()
+ .await
+ .unwrap()
+ .expect_err("expected ParquetSchemaConvert for the id-matched nested field");
+ let msg = err.to_string();
+ assert!(
+ msg.contains("Column: [[s, a]]")
+ && msg.contains("Expected: int")
+ && msg.contains("Found: INT64"),
+ "unexpected error: {msg}"
+ );
+ Ok(())
+ }
+
+ /// Disallowed widening (`INT32 -> bigint` with `allow_type_promotion` off) inside a
+ /// struct defers to `RejectOnNonEmpty`, like the top level: a non-empty file fails ...
+ #[tokio::test]
+ async fn nested_disallowed_widening_rejects_non_empty() -> Result<(), DataFusionError> {
+ let batch = struct_batch(
+ Field::new("x", DataType::Int32, true),
+ Arc::new(Int32Array::from(vec![1, 2, 3])),
+ )?;
+ let required_schema = struct_schema(vec![Field::new("x", DataType::Int64, true)]);
+ let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ options.allow_type_promotion = false;
+ let mut stream = scan_parquet(&batch, required_schema, options)?;
+ let err = stream
+ .next()
+ .await
+ .unwrap()
+ .expect_err("expected ParquetSchemaConvert for nested disallowed widening");
+ let msg = err.to_string();
+ assert!(
+ msg.contains("Column: [[s, x]]")
+ && msg.contains("Expected: bigint")
+ && msg.contains("Found: INT32"),
+ "unexpected error: {msg}"
+ );
+ Ok(())
+ }
+
+ /// ... while an empty file (no row groups, SPARK-26709) still reads.
+ #[tokio::test]
+ async fn nested_disallowed_widening_passes_for_empty_file() -> Result<(), DataFusionError> {
+ let batch = struct_batch(
+ Field::new("x", DataType::Int32, true),
+ Arc::new(Int32Array::from(Vec::<i32>::new())),
+ )?;
+ let required_schema = struct_schema(vec![Field::new("x", DataType::Int64, true)]);
+ let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ options.allow_type_promotion = false;
+ let mut stream = scan_parquet(&batch, required_schema, options)?;
+ while let Some(batch) = stream.next().await {
+ assert_eq!(batch?.num_rows(), 0);
+ }
+ Ok(())
+ }
+
+ /// Positive: `INT32 -> bigint` inside a struct still converts when type promotion is
+ /// allowed (Spark 4.x behaviour).
+ #[tokio::test]
+ async fn nested_int_widening_succeeds_with_type_promotion() -> Result<(), DataFusionError> {
+ let batch = struct_batch(
+ Field::new("x", DataType::Int32, true),
+ Arc::new(Int32Array::from(vec![1, 2, 3])),
+ )?;
+ let required_schema = struct_schema(vec![Field::new("x", DataType::Int64, true)]);
+ let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+ options.allow_type_promotion = true;
+ let mut stream = scan_parquet(&batch, required_schema, options)?;
+ let result = stream.next().await.unwrap()?;
+ let x = result
+ .column(0)
+ .as_struct()
+ .column(0)
+ .as_primitive::<Int64Type>();
+ assert_eq!(x.values().to_vec(), vec![1i64, 2, 3]);
+ Ok(())
+ }
+
+ /// Positive: TIMESTAMP_MILLIS inside a list still converts to microseconds.
+ #[tokio::test]
+ async fn nested_timestamp_millis_read_as_micros_succeeds() -> Result<(), DataFusionError> {
+ let list = ListArray::new(
+ Arc::new(Field::new(
+ "element",
+ DataType::Timestamp(TimeUnit::Millisecond, None),
+ true,
+ )),
+ OffsetBuffer::new(vec![0, 2, 3].into()),
+ Arc::new(TimestampMillisecondArray::from(vec![
+ 1_000i64, 2_000, 3_000,
+ ])),
+ None,
+ );
+ let file_schema = Arc::new(Schema::new(vec![Field::new(
+ "a",
+ list.data_type().clone(),
+ true,
+ )]));
+ let batch = RecordBatch::try_new(file_schema, vec![Arc::new(list)])?;
+ let required_schema = Arc::new(Schema::new(vec![Field::new(
+ "a",
+ DataType::List(Arc::new(Field::new(
+ "element",
+ DataType::Timestamp(TimeUnit::Microsecond, None),
+ true,
+ ))),
+ true,
+ )]));
+ let result = roundtrip(&batch, required_schema).await?;
+ let values = result
+ .column(0)
+ .as_list::<i32>()
+ .values()
+ .as_primitive::<TimestampMicrosecondType>();
+ assert_eq!(
+ values.values().to_vec(),
+ vec![1_000_000i64, 2_000_000, 3_000_000]
+ );
+ Ok(())
+ }
+
+ /// Positive: a requested nested field missing from the file still reads as null while
+ /// the present field is passed through.
+ #[tokio::test]
+ async fn nested_missing_field_reads_as_null() -> Result<(), DataFusionError> {
+ let batch = struct_batch(
+ Field::new("x", DataType::Int32, true),
+ Arc::new(Int32Array::from(vec![1, 2])),
+ )?;
+ let required_schema = struct_schema(vec![
+ Field::new("x", DataType::Int32, true),
+ Field::new("y", DataType::Int64, true),
+ ]);
+ let result = roundtrip(&batch, required_schema).await?;
+ let s = result.column(0).as_struct();
+ assert_eq!(
+ s.column(0)
+ .as_primitive::<arrow::datatypes::Int32Type>()
+ .values()
+ .to_vec(),
+ vec![1, 2]
+ );
+ assert_eq!(s.column(1).null_count(), 2);
+ Ok(())
+ }
+
#[tokio::test]
async fn parquet_duplicate_fields_case_insensitive() {
// Parquet file has columns "A", "B", "b" - reading "b" in case-insensitive mode
@@ -2402,8 +3393,8 @@ mod test {
#[test]
fn structural_narrowing_requires_unambiguous_exact_match() -> Result<(), DataFusionError> {
- use crate::parquet::cast_column::CometCastColumnExpr;
use datafusion::physical_expr::expressions::CastExpr;
+ use datafusion_comet_common::SparkError;
// #5707: an exact match must not hide a second case-insensitive match.
for (upper, lower) in [("ID", "id"), ("CAFÉ", "café")] {
@@ -2425,11 +3416,23 @@ mod test {
case_sensitive,
"{physical:?} -> {target:?}, case_sensitive={case_sensitive}"
);
- let rewritten = rewrite_events_column(physical.clone(), target.clone(), opts)?;
+ let rewritten = rewrite_events_column(physical.clone(), target.clone(), opts);
if case_sensitive {
- assert!(rewritten.downcast_ref::<CastExpr>().is_some());
+ assert!(rewritten?.downcast_ref::<CastExpr>().is_some());
} else {
- assert!(rewritten.downcast_ref::<CometCastColumnExpr>().is_some());
+ // Recursive validation now rejects the ambiguity during rewriting.
+ let error = rewritten.unwrap_err();
+ let DataFusionError::External(source) = &error else {
+ panic!("expected duplicate-field error, got {error}");
+ };
+ assert!(matches!(
+ source.downcast_ref::<SparkError>(),
+ Some(SparkError::DuplicateFieldCaseInsensitive {
+ required_field_name,
+ matched_fields,
+ }) if required_field_name == lower
+ && matched_fields == &format!("[{upper}, {lower}]")
+ ));
}
}
}
diff --git a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
index 2f434aac92..d412918b57 100644
--- a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
@@ -1598,6 +1598,95 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
+ test("native scan rejects nested Parquet conversions Spark rejects") {
+ // Regression guard for #5671. Spark's vectorized reader runs
+ // `ParquetVectorUpdaterFactory.getUpdater` on every leaf column, nested or not, so the
+ // conversions the top-level rejection tests above cover must also be rejected inside
+ // structs, arrays and maps instead of being silently cast (NULLs, parsed strings) or
+ // crashing the native reader.
+ withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "parquet") {
+ val cases = Seq(
+ // (written expression, read schema)
+ ("named_struct('x', cast(5000000000 as bigint))", "s struct<x:int>"),
+ ("named_struct('x', 1)", "s struct<x:string>"),
+ ("named_struct('d', cast('123456.78' as decimal(10,2)))", "s struct<d:decimal(5,2)>"),
+ ("named_struct('x', '12')", "s struct<x:int>"),
+ ("named_struct('x', 1)", "s struct<x:array<int>>"),
+ ("named_struct('x', array(1))", "s struct<x:int>"),
+ ("array(cast(5000000000 as bigint))", "s array<int>"),
+ ("map('k', cast(5000000000 as bigint))", "s map<string,int>"),
+ ("array(named_struct('x', cast(5000000000 as bigint)))", "s array<struct<x:int>>"))
+ cases.foreach { case (writeExpr, readSchema) =>
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark.sql(s"select $writeExpr as s").write.parquet(path)
+ withClue(s"$writeExpr read as $readSchema: ") {
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ intercept[SparkException] {
+ spark.read.schema(readSchema).parquet(path).collect()
+ }
+ }
+ val df = spark.read.schema(readSchema).parquet(path)
+ val plan = df.queryExecution.executedPlan
+ assert(
+ collect(plan) { case scan: CometNativeScanExec => scan }.nonEmpty,
+ s"Expected a native Comet scan before collecting:\n$plan")
+ val cometErr = intercept[SparkException](df.collect())
+ val chain = causeChain(cometErr)
+ assert(
+ chain.exists(
+ _.isInstanceOf[org.apache.spark.sql.execution.datasources.SchemaColumnConvertNotSupportedException]),
+ "expected SchemaColumnConvertNotSupportedException; chain was:\n" +
+ chain.map(t => s" ${t.getClass.getName}: ${t.getMessage}").mkString("\n"))
+ if (readSchema == "s array<struct<x:int>>") {
+ assert(chain.exists(t =>
+ Option(t.getMessage).exists(_.contains("[s, list, element, x]"))))
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("nested schema evolution follows Spark's per-version widening rules") {
+ // Companion to "schema evolution": `INT32 -> bigint` inside a struct is gated by the same
+ // per-Spark-version constant as the top level (see ShimCometConf), and accepted nested
+ // conversions keep working now that nested leaves are validated (#5671).
+ withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "parquet") {
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark
+ .sql("select named_struct('x', 1) as s union all select named_struct('x', 2)")
+ .write
+ .parquet(path)
+ val df = spark.read.schema("s struct<x:bigint, y:string>").parquet(path)
+ if (CometConf.COMET_SCHEMA_EVOLUTION_ENABLED) {
+ checkSparkAnswerAndOperator(df)
+ } else {
+ assertThrows[SparkException](df.collect())
+ }
+ }
+ }
+ }
+
+ test("nested TIMESTAMP_MILLIS columns read as timestamp") {
+ // Accepted nested conversion (TIMESTAMP_MILLIS -> TIMESTAMP_MICROS inside an array and a
+ // struct) must keep working now that nested leaves are validated (#5671).
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> "parquet",
+ SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> "TIMESTAMP_MILLIS") {
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark
+ .sql("select array(timestamp'2020-01-01 00:00:00.123') as a, " +
+ "named_struct('t', timestamp'2020-01-02 00:00:00.456') as s")
+ .write
+ .parquet(path)
+ checkSparkAnswerAndOperator(spark.read.parquet(path))
+ }
+ }
+ }
+
test("type widening: byte → short/int/long, short → int/long, int → long") {
// Widening of INT32 -> LONG is only allowed when Comet's type-promotion
// default permits it (Spark 4.x). See ShimCometConf.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org