Skip to content

fix: apply Spark's Parquet conversion rules to nested struct/list/map fields #5681

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
peterxcli wants to merge 2 commits into
base: main
Choose a base branch
from
+1,080 −328
Open

fix: apply Spark's Parquet conversion rules to nested struct/list/map fields #5681

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
552db98
fix: apply Spark's Parquet conversion rules to nested struct/list/map…
peterxcli Sep 4, 2026
e441107
test: type nested list fixture as int
peterxcli Sep 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion native/core/src/execution/planner.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6396,7 +6396,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(
Expand Down
208 changes: 129 additions & 79 deletions native/core/src/parquet/parquet_support.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -173,8 +173,7 @@ fn parquet_convert_array(
use DataType::*;
let from_type = array.data_type();

// 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(),
Expand All @@ -190,12 +189,12 @@ fn parquet_convert_array(
parquet_options,
)?;

Ok(Arc::new(ListArray::new(
Ok(Arc::new(ListArray::try_new(
Arc::clone(to_inner_type),
list_arr.offsets().clone(),
cast_field,
list_arr.nulls().cloned(),
)))
)?))
}
(Timestamp(TimeUnit::Microsecond, None), Timestamp(TimeUnit::Microsecond, Some(tz))) => {
Ok(Arc::new(
Expand Down Expand Up @@ -234,7 +233,13 @@ fn parquet_convert_array(
_ 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}"
))),
}
}

Expand All @@ -246,6 +251,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(
Expand All @@ -256,77 +344,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(
Arc::clone(array.column(from_index)),
Expand All @@ -352,11 +374,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!(),
}
Expand Down Expand Up @@ -390,17 +412,17 @@ fn parquet_convert_map_to_map(
parquet_options,
)?;

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}"
Expand Down Expand Up @@ -642,6 +664,34 @@ mod tests {
prepare_object_store_with_configs(runtime_env, url, &HashMap::new())
}

/// 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() {
Expand Down
Loading
Loading