diff --git a/native/core/src/parquet/cast_column.rs b/native/core/src/parquet/cast_column.rs index 8c72455f4db..288e7f02622 100644 --- a/native/core/src/parquet/cast_column.rs +++ b/native/core/src/parquet/cast_column.rs @@ -24,7 +24,9 @@ use arrow::{ record_batch::RecordBatch, }; -use crate::parquet::parquet_support::{spark_parquet_convert, SparkParquetOptions}; +use crate::parquet::parquet_support::{ + spark_parquet_convert_with_mapping, FieldMapping, SparkParquetOptions, +}; use datafusion::common::format::DEFAULT_CAST_OPTIONS; use datafusion::common::{DataFusionError, Result as DataFusionResult}; use datafusion::logical_expr::ColumnarValue; @@ -153,8 +155,11 @@ pub struct CometCastColumnExpr { /// Options forwarded to [`cast_column`]. cast_options: CastOptions<'static>, /// Spark parquet options for complex nested type conversions. - /// When present, enables `spark_parquet_convert` as a fallback. + /// When present, enables the nested conversion as a fallback. parquet_options: Option, + /// Which file field supplies each requested nested field, resolved once per file and + /// reused for every batch. Set together with `parquet_options`. + field_mapping: Option>, } // Manually derive `PartialEq`/`Hash` as `Arc` does not @@ -166,6 +171,7 @@ impl PartialEq for CometCastColumnExpr { && self.target_field.eq(&other.target_field) && self.cast_options.eq(&other.cast_options) && self.parquet_options.eq(&other.parquet_options) + && self.field_mapping.eq(&other.field_mapping) } } @@ -176,6 +182,7 @@ impl Hash for CometCastColumnExpr { self.target_field.hash(state); self.cast_options.hash(state); self.parquet_options.hash(state); + self.field_mapping.hash(state); } } @@ -214,12 +221,19 @@ impl CometCastColumnExpr { target_field, cast_options: cast_options.unwrap_or(DEFAULT_CAST_OPTIONS), parquet_options: None, + field_mapping: None, }) } - /// Set Spark parquet options to enable complex nested type conversions. - pub fn with_parquet_options(mut self, options: SparkParquetOptions) -> Self { + /// Enable nested type conversions with Spark parquet options and the field mapping + /// resolved for this expression's physical and target types. + pub fn with_parquet_options( + mut self, + options: SparkParquetOptions, + field_mapping: Arc, + ) -> Self { self.parquet_options = Some(options); + self.field_mapping = Some(field_mapping); self } } @@ -270,12 +284,22 @@ impl PhysicalExpr for CometCastColumnExpr { let input_physical_field = self.input_physical_field.data_type(); let target_field = self.target_field.data_type(); + // Relabeling only swaps metadata, so it is right when every requested field reads + // the file field at its own position. A mapping that reorders fields (ids resolved + // to other positions) has to go through the nested conversion below. + let positional = self + .field_mapping + .as_ref() + .is_none_or(|mapping| mapping.is_positional()); + match (input_physical_field, target_field) { // Nested types that differ only in field names (e.g., List element named // "item" vs "element", or Map entries named "key_value" vs "entries"). // Re-label the array so the DataType metadata matches the logical schema. (physical, logical) - if physical != logical && types_differ_only_in_field_names(physical, logical) => + if positional + && physical != logical + && types_differ_only_in_field_names(physical, logical) => { match value { ColumnarValue::Array(array) => { @@ -285,16 +309,17 @@ impl PhysicalExpr for CometCastColumnExpr { other => Ok(other), } } - // Fallback: use spark_parquet_convert for complex nested type conversions - // (e.g., List → List, Map field selection, etc.) - _ => { - if let Some(parquet_options) = &self.parquet_options { - let converted = spark_parquet_convert(value, target_field, parquet_options)?; - Ok(converted) - } else { - Ok(value) - } - } + // Fallback: nested conversion through the resolved mapping + // (e.g., List -> List, Map field selection, etc.) + _ => match (&self.parquet_options, &self.field_mapping) { + (Some(parquet_options), Some(mapping)) => spark_parquet_convert_with_mapping( + value, + target_field, + mapping, + parquet_options, + ), + _ => Ok(value), + }, } } @@ -318,8 +343,8 @@ impl PhysicalExpr for CometCastColumnExpr { Arc::clone(&self.target_field), Some(self.cast_options.clone()), )?; - if let Some(opts) = &self.parquet_options { - new_expr = new_expr.with_parquet_options(opts.clone()); + if let (Some(opts), Some(mapping)) = (&self.parquet_options, &self.field_mapping) { + new_expr = new_expr.with_parquet_options(opts.clone(), Arc::clone(mapping)); } Ok(Arc::new(new_expr)) } @@ -332,12 +357,166 @@ impl PhysicalExpr for CometCastColumnExpr { #[cfg(test)] mod tests { use super::*; + use crate::parquet::parquet_support::resolve_field_mapping; use arrow::array::{ Array, Int32Array, StringArray, TimestampMicrosecondArray, TimestampMillisecondArray, }; use arrow::datatypes::{Field, Fields}; use datafusion::physical_expr::expressions::Column; use datafusion_comet_spark_expr::EvalMode; + use parquet::arrow::PARQUET_FIELD_ID_META_KEY; + use std::collections::HashMap; + + fn int_field_with_id(name: &str, id: i32) -> Field { + Field::new(name, DataType::Int32, true).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + id.to_string(), + )])) + } + + /// File struct `x` (id 1) = 42, `y` (id 2) = 43; requested struct names them the same + /// but swaps the ids. Names and types match, so only the positional gate keeps the + /// relabel shortcut from firing: the mapping reads by id and the result must be + /// `x` = 43, `y` = 42. + #[test] + fn test_swapped_field_ids_bypass_relabel_shortcut() { + let physical_fields = + Fields::from(vec![int_field_with_id("x", 1), int_field_with_id("y", 2)]); + let logical_fields = + Fields::from(vec![int_field_with_id("x", 2), int_field_with_id("y", 1)]); + + let input_field = Arc::new(Field::new( + "s", + DataType::Struct(physical_fields.clone()), + true, + )); + let target_field = Arc::new(Field::new( + "s", + DataType::Struct(logical_fields.clone()), + true, + )); + + let columns: Vec = vec![ + Arc::new(Int32Array::from(vec![42])), + Arc::new(Int32Array::from(vec![43])), + ]; + let struct_arr = StructArray::new(physical_fields, columns, None); + let schema = Schema::new(vec![Arc::clone(&input_field)]); + let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(struct_arr)]).unwrap(); + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.use_field_id = true; + let mapping = Arc::new( + resolve_field_mapping(input_field.data_type(), target_field.data_type(), &opts) + .unwrap(), + ); + assert!(!mapping.is_positional()); + + let col_expr: Arc = Arc::new(Column::new("s", 0)); + let cast_expr = CometCastColumnExpr::try_new(col_expr, input_field, target_field, None) + .unwrap() + .with_parquet_options(opts, mapping); + + let ColumnarValue::Array(arr) = cast_expr.evaluate(&batch).unwrap() else { + panic!("expected array result"); + }; + assert_eq!(arr.data_type(), &DataType::Struct(logical_fields)); + let result = arr.as_any().downcast_ref::().unwrap(); + let x = result + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let y = result + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(x.value(0), 43); + assert_eq!(y.value(0), 42); + } + + /// Companion guard: without any field ids the relabel shortcut must keep + /// handling name-only differences, whether or not id read mode is enabled. + #[test] + fn test_relabel_shortcut_kept_for_name_only_differences_without_ids() { + // Physical: s { col: List(Field("item", Int32)) } + // Logical: s { col: List(Field("element", Int32)) } + let physical_list_field = Arc::new(Field::new("item", DataType::Int32, true)); + let logical_list_field = Arc::new(Field::new("element", DataType::Int32, true)); + let physical_fields = Fields::from(vec![Field::new( + "col", + DataType::List(Arc::clone(&physical_list_field)), + true, + )]); + let logical_fields = Fields::from(vec![Field::new( + "col", + DataType::List(logical_list_field), + true, + )]); + + let input_field = Arc::new(Field::new( + "s", + DataType::Struct(physical_fields.clone()), + true, + )); + let target_field = Arc::new(Field::new( + "s", + DataType::Struct(logical_fields.clone()), + true, + )); + + let values = Int32Array::from(vec![1, 2, 3]); + let list = ListArray::new( + physical_list_field, + arrow::buffer::OffsetBuffer::new(vec![0, 2, 3].into()), + Arc::new(values), + None, + ); + let struct_arr = StructArray::new(physical_fields, vec![Arc::new(list) as ArrayRef], None); + let schema = Schema::new(vec![Arc::clone(&input_field)]); + let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(struct_arr)]).unwrap(); + + let col_expr: Arc = Arc::new(Column::new("s", 0)); + + // Without parquet options the fallback arm would return the value + // unchanged, so a relabeled result proves the shortcut itself fired. + let plain_expr = CometCastColumnExpr::try_new( + Arc::clone(&col_expr), + Arc::clone(&input_field), + Arc::clone(&target_field), + None, + ) + .unwrap(); + + // Enabling id read mode without any id metadata must not disable the + // shortcut either: the resolved mapping is positional. + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.use_field_id = true; + let mapping = Arc::new( + resolve_field_mapping(input_field.data_type(), target_field.data_type(), &opts) + .unwrap(), + ); + assert!(mapping.is_positional()); + let id_mode_expr = CometCastColumnExpr::try_new(col_expr, input_field, target_field, None) + .unwrap() + .with_parquet_options(opts, mapping); + + for cast_expr in [plain_expr, id_mode_expr] { + let result = cast_expr.evaluate(&batch).unwrap(); + let ColumnarValue::Array(arr) = result else { + panic!("expected array result"); + }; + assert_eq!(arr.data_type(), &DataType::Struct(logical_fields.clone())); + let result_struct = arr.as_any().downcast_ref::().unwrap(); + let result_list = result_struct + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(result_list.len(), 2); + } + } #[test] fn test_rejects_millisecond_logical_timestamp() { @@ -385,7 +564,10 @@ mod tests { let expr: Arc = Arc::new(Column::new("ts", 0)); let cast_expr = CometCastColumnExpr::try_new(expr, input_field, target_field, None) .unwrap() - .with_parquet_options(SparkParquetOptions::new(eval_mode, "UTC", false)); + .with_parquet_options( + SparkParquetOptions::new(eval_mode, "UTC", false), + Arc::new(FieldMapping::Leaf), + ); let input = TimestampMillisecondArray::from(vec![Some(1_234), Some(-1_234), None]) .with_timezone_opt(source_tz.clone()); diff --git a/native/core/src/parquet/eager_page_index_reader_factory.rs b/native/core/src/parquet/eager_page_index_reader_factory.rs index 22a94d04a19..a5454f3a88f 100644 --- a/native/core/src/parquet/eager_page_index_reader_factory.rs +++ b/native/core/src/parquet/eager_page_index_reader_factory.rs @@ -45,7 +45,14 @@ //! //! Filed upstream as apache/datafusion#23978. Revert this once the opener merges its deferred //! page-index load back into `FileMetadataCache` instead of bypassing it. +//! +//! The metadata fetch is also the one per-file hook DataFusion runs unconditionally, so the +//! factory validates requested Parquet field ids there; see [`FieldIdCheck`]. +use crate::parquet::parquet_support::{ + schema_holds_field_ids, validate_field_mapping, SparkParquetOptions, +}; +use arrow::datatypes::SchemaRef; use bytes::Bytes; use datafusion::common::Result as DFResult; use datafusion::datasource::physical_plan::parquet::metadata::DFParquetMetadata; @@ -57,19 +64,23 @@ use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet; use datafusion_datasource::PartitionedFile; use futures::future::BoxFuture; use futures::{FutureExt, TryFutureExt}; +use object_store::path::Path; use object_store::{ObjectStore, ObjectStoreExt}; use parquet::arrow::arrow_reader::ArrowReaderOptions; use parquet::arrow::async_reader::AsyncFileReader; +use parquet::arrow::parquet_to_arrow_schema; use parquet::errors::ParquetError; use parquet::file::metadata::{PageIndexPolicy, ParquetMetaData}; +use std::collections::HashMap; use std::fmt::Debug; use std::ops::Range; -use std::sync::Arc; +use std::sync::{Arc, Mutex, MutexGuard, PoisonError, Weak}; #[derive(Debug)] pub struct EagerPageIndexReaderFactory { store: Arc, metadata_cache: Arc, + field_id_check: Option>, } impl EagerPageIndexReaderFactory { @@ -77,7 +88,75 @@ impl EagerPageIndexReaderFactory { Self { store, metadata_cache, + field_id_check: None, + } + } + + /// Validate the ids `requested_schema` carries against each file's schema as its footer + /// loads. Installs nothing when field id matching is off or the schema carries no id, so + /// ordinary reads pay nothing. + pub fn with_field_id_check( + mut self, + requested_schema: SchemaRef, + parquet_options: &SparkParquetOptions, + ) -> Self { + if parquet_options.use_field_id && schema_holds_field_ids(&requested_schema) { + self.field_id_check = Some(Arc::new(FieldIdCheck { + requested_schema, + parquet_options: parquet_options.clone(), + validated: Mutex::new(HashMap::new()), + })); } + self + } +} + +/// Validates requested field ids for files the expression adapter never sees: DataFusion's +/// opener creates the adapter only when a predicate is pushed or the file schema differs from +/// the requested one, so a metadata-free file whose schema equals it is read positionally +/// (comet#5801). Resolves the mapping the adapter resolves, so both raise the same error. +#[derive(Debug)] +struct FieldIdCheck { + requested_schema: SchemaRef, + parquet_options: SparkParquetOptions, + /// Files already validated, keyed by path to the metadata they were checked against, so a + /// footer served from `FileMetadataCache` is not rechecked on every open. + validated: Mutex>>, +} + +impl FieldIdCheck { + fn validate( + &self, + location: &Path, + metadata: &Arc, + ) -> parquet::errors::Result<()> { + if self.is_validated(location, metadata) { + return Ok(()); + } + // The same conversion the opener applies, so field ids land in field metadata under + // `PARQUET:field_id` and the mapping resolves against the schema the adapter would see. + let file_metadata = metadata.file_metadata(); + let file_schema = parquet_to_arrow_schema( + file_metadata.schema_descr(), + file_metadata.key_value_metadata(), + )?; + validate_field_mapping(&file_schema, &self.requested_schema, &self.parquet_options) + .map_err(|e| ParquetError::External(Box::new(e)))?; + self.lock() + .insert(location.clone(), Arc::downgrade(metadata)); + Ok(()) + } + + fn is_validated(&self, location: &Path, metadata: &Arc) -> bool { + self.lock() + .get(location) + .is_some_and(|seen| std::ptr::eq(Weak::as_ptr(seen), Arc::as_ptr(metadata))) + } + + fn lock(&self) -> MutexGuard<'_, HashMap>> { + self.validated + .lock() + .unwrap_or_else(PoisonError::into_inner) } } @@ -101,6 +180,7 @@ impl ParquetFileReaderFactory for EagerPageIndexReaderFactory { partitioned_file, metadata_cache: Arc::clone(&self.metadata_cache), metadata_size_hint, + field_id_check: self.field_id_check.clone(), })) } } @@ -114,6 +194,7 @@ struct EagerPageIndexReader { partitioned_file: PartitionedFile, metadata_cache: Arc, metadata_size_hint: Option, + field_id_check: Option>, } impl AsyncFileReader for EagerPageIndexReader { @@ -154,6 +235,7 @@ impl AsyncFileReader for EagerPageIndexReader { let metadata_cache = Arc::clone(&self.metadata_cache); let store = Arc::clone(&self.store); let metadata_size_hint = self.metadata_size_hint; + let field_id_check = self.field_id_check.clone(); async move { let file_decryption_properties = options .and_then(|o| o.file_decryption_properties()) @@ -164,7 +246,7 @@ impl AsyncFileReader for EagerPageIndexReader { options.map(|o| o.column_index_policy()) }; - DFParquetMetadata::new(store.as_ref(), &object_meta) + let metadata = DFParquetMetadata::new(store.as_ref(), &object_meta) .with_decryption_properties(file_decryption_properties) .with_file_metadata_cache(Some(metadata_cache)) .with_metadata_size_hint(metadata_size_hint) @@ -176,7 +258,11 @@ impl AsyncFileReader for EagerPageIndexReader { "Failed to fetch metadata for file {}: {e}", object_meta.location, )) - }) + })?; + if let Some(check) = &field_id_check { + check.validate(&object_meta.location, &metadata)?; + } + Ok(metadata) } .boxed() } diff --git a/native/core/src/parquet/parquet_exec.rs b/native/core/src/parquet/parquet_exec.rs index 8796cb23244..236b478d9a3 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -166,11 +166,16 @@ pub(crate) fn init_datasource_exec( // TODO: metadata I/O is invisible in metrics. `fetch_metadata` reads via `ObjectStore::get_ranges`, // bypassing the `get_bytes` path where `bytes_scanned` is counted. A byte-counting ObjectStore // wrapper would surface it. + // + // The factory also validates the requested field ids against each file's footer. The + // expression adapter below does the same, but DataFusion's opener skips it for a file whose + // schema equals the requested one when no predicate is pushed (#5801). let runtime_env = session_ctx.runtime_env(); let store = runtime_env.object_store(&object_store_url)?; let metadata_cache = runtime_env.cache_manager.get_file_metadata_cache(); parquet_source = parquet_source.with_parquet_file_reader_factory(Arc::new( - EagerPageIndexReaderFactory::new(store, metadata_cache), + EagerPageIndexReaderFactory::new(store, metadata_cache) + .with_field_id_check(Arc::clone(&required_schema), &spark_parquet_options), )); // Route data filters through `try_pushdown_filters` rather than calling @@ -297,13 +302,18 @@ fn get_options( mod tests { use super::*; use arrow::array::Int32Array; + use arrow::array::{Array, AsArray, Int64Array, StructArray}; + use arrow::datatypes::Fields; use arrow::datatypes::{DataType, Field, Schema}; use arrow::record_batch::RecordBatch; + use datafusion::common::DataFusionError; use datafusion::datasource::physical_plan::parquet::metadata::CachedParquetMetaData; use datafusion::physical_plan::ExecutionPlan; use datafusion_comet_spark_expr::test_common::file_util::get_temp_filename; use futures::StreamExt; - use parquet::arrow::ArrowWriter; + use parquet::arrow::arrow_reader::{ArrowReaderMetadata, ArrowReaderOptions}; + use parquet::arrow::arrow_writer::ArrowWriterOptions; + use parquet::arrow::{ArrowWriter, PARQUET_FIELD_ID_META_KEY}; use parquet::file::properties::{EnabledStatistics, WriterProperties}; use std::fs::File; @@ -437,4 +447,217 @@ mod tests { "cached metadata must include the page index" ); } + + /// A nullable Int64 field carrying a Parquet field id. + fn field_with_id(name: &str, id: i32) -> Field { + Field::new(name, DataType::Int64, true).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + id.to_string(), + )])) + } + + /// One row of `s`, with `s` itself carrying id 10. + fn struct_batch(x_id: i32, y_id: i32) -> RecordBatch { + let children = Fields::from(vec![field_with_id("x", x_id), field_with_id("y", y_id)]); + let struct_field = Field::new("s", DataType::Struct(children.clone()), true).with_metadata( + HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "10".to_string())]), + ); + let values: Vec> = vec![ + Arc::new(Int64Array::from(vec![42])), + Arc::new(Int64Array::from(vec![43])), + ]; + let column = Arc::new(StructArray::new(children, values, None)) as Arc; + RecordBatch::try_new(Arc::new(Schema::new(vec![struct_field])), vec![column]).unwrap() + } + + /// `struct_batch` with a leading `a: 7 (id 20)` column, for projected reads. + fn two_column_batch(x_id: i32, y_id: i32) -> RecordBatch { + let s = struct_batch(x_id, y_id); + let fields = vec![ + Arc::new(field_with_id("a", 20)), + Arc::clone(&s.schema().fields()[0]), + ]; + let a = Arc::new(Int64Array::from(vec![7])) as Arc; + let columns = vec![a, Arc::clone(s.column(0))]; + RecordBatch::try_new(Arc::new(Schema::new(fields)), columns).unwrap() + } + + /// The schema holding only column `index` of `batch`. + fn column_schema(batch: &RecordBatch, index: usize) -> SchemaRef { + Arc::new(Schema::new(vec![Arc::clone( + &batch.schema().fields()[index], + )])) + } + + /// Write `batch` to a Parquet file with no key-value metadata at all. arrow-rs then derives + /// a file schema equal to the batch schema, and DataFusion's opener skips the expression + /// adapter on a predicate-free scan of it. Both are asserted here so the scan tests below + /// exercise that path rather than the adapter. + fn write_bare_parquet(batch: &RecordBatch) -> PartitionedFile { + let filename = get_temp_filename() + .as_path() + .as_os_str() + .to_str() + .unwrap() + .to_string(); + let file = File::create(&filename).unwrap(); + let options = ArrowWriterOptions::new().with_skip_arrow_metadata(true); + let mut writer = ArrowWriter::try_new_with_options(file, batch.schema(), options).unwrap(); + writer.write(batch).unwrap(); + writer.close().unwrap(); + + let reader_metadata = + ArrowReaderMetadata::load(&File::open(&filename).unwrap(), ArrowReaderOptions::new()) + .unwrap(); + assert!( + reader_metadata + .metadata() + .file_metadata() + .key_value_metadata() + .is_none(), + "the file must carry no key-value metadata" + ); + assert_eq!( + reader_metadata.schema().as_ref(), + batch.schema().as_ref(), + "the file schema must equal the requested schema so the opener skips the adapter" + ); + PartitionedFile::from_path(filename).unwrap() + } + + /// Scan a metadata-free file of `batch` with its own schema requested and no filter. + async fn scan_bare_file( + batch: &RecordBatch, + use_field_id: bool, + ) -> Result, DataFusionError> { + scan_bare_file_projected(batch, batch.schema(), None, None, use_field_id).await + } + + /// Scan a metadata-free file of `batch` with no filter, wired the way the planner wires a + /// Spark scan: `data_schema` is the full table schema and `projection` selects the + /// `required_schema` columns from it. + async fn scan_bare_file_projected( + batch: &RecordBatch, + required_schema: SchemaRef, + data_schema: Option, + projection: Option>, + use_field_id: bool, + ) -> Result, DataFusionError> { + let file = write_bare_parquet(batch); + let session_ctx = Arc::new(SessionContext::new()); + let scan = init_datasource_exec( + required_schema, + data_schema, + None, + ObjectStoreUrl::local_filesystem(), + vec![vec![file]], + projection, + None, + None, + "UTC", + true, + false, + false, + false, + &session_ctx, + false, + use_field_id, + false, + ) + .map_err(|e| DataFusionError::Execution(e.to_string()))?; + let mut stream = scan.execute(0, session_ctx.task_ctx())?; + let mut batches = Vec::new(); + while let Some(batch) = stream.next().await { + batches.push(batch?); + } + Ok(batches) + } + + /// The `(x, y)` values of the single struct row in `batches`. + fn struct_values(batches: &[RecordBatch]) -> (i64, i64) { + assert_eq!(batches.len(), 1); + let s = batches[0].column(0).as_struct(); + ( + s.column(0) + .as_primitive::() + .value(0), + s.column(1) + .as_primitive::() + .value(0), + ) + } + + /// Regression test for #5801: with no key-value metadata the file schema equals the + /// requested schema, so DataFusion's opener never creates the expression adapter that + /// validates field ids. The reader factory must reject requested id 1 matching both `x` + /// and `y` the way the adapter does, instead of reading the struct positionally. + #[tokio::test] + async fn duplicate_struct_field_id_rejected_when_opener_skips_adapter() { + let err = scan_bare_file(&struct_batch(1, 1), true) + .await + .expect_err("requested id 1 matches two file fields and must error"); + let msg = err.to_string(); + assert!( + msg.contains("_LEGACY_ERROR_TEMP_2094") && msg.contains("id=1"), + "expected duplicate field id error, got: {msg}" + ); + } + + #[tokio::test] + async fn unique_struct_field_ids_read_when_opener_skips_adapter() { + let batches = scan_bare_file(&struct_batch(1, 2), true).await.unwrap(); + assert_eq!(struct_values(&batches), (42, 43)); + } + + /// Without field id matching Spark clips by name, so duplicate ids in the file are not an + /// error and the factory must not run the check. + #[tokio::test] + async fn duplicate_struct_field_ids_ignored_when_field_id_matching_off() { + let batches = scan_bare_file(&struct_batch(1, 1), false).await.unwrap(); + assert_eq!(struct_values(&batches), (42, 43)); + } + + /// The planner passes the full table schema as `data_schema` and projects the required + /// columns from it, so the opener compares the file against `data_schema`. The check must + /// still fire for a requested struct under that wiring. + #[tokio::test] + async fn duplicate_struct_field_id_rejected_under_projected_data_schema() { + let batch = two_column_batch(1, 1); + let err = scan_bare_file_projected( + &batch, + column_schema(&batch, 1), + Some(batch.schema()), + Some(vec![1]), + true, + ) + .await + .expect_err("requested id 1 matches two file fields and must error"); + let msg = err.to_string(); + assert!( + msg.contains("_LEGACY_ERROR_TEMP_2094") && msg.contains("id=1"), + "expected duplicate field id error, got: {msg}" + ); + } + + /// Spark's `clipParquetSchema` validates only the requested columns, so a duplicate id + /// inside a struct the read does not project is not an error. + #[tokio::test] + async fn unrequested_duplicate_struct_field_ids_read_under_projected_data_schema() { + let batch = two_column_batch(1, 1); + let batches = scan_bare_file_projected( + &batch, + column_schema(&batch, 0), + Some(batch.schema()), + Some(vec![0]), + true, + ) + .await + .unwrap(); + assert_eq!(batches.len(), 1); + assert_eq!(batches[0].num_columns(), 1); + let a = batches[0] + .column(0) + .as_primitive::(); + assert_eq!(a.value(0), 7); + } } diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index a461e8a16c0..ae7454a9ca0 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -17,10 +17,12 @@ use crate::execution::operators::ExecutionError; use crate::parquet::name_fold::fold_names; -use arrow::array::{FixedSizeBinaryArray, ListArray, MapArray, StringArray}; +use arrow::array::{ + FixedSizeBinaryArray, LargeListArray, ListArray, MapArray, OffsetSizeTrait, StringArray, +}; use arrow::buffer::NullBuffer; use arrow::compute::can_cast_types; -use arrow::datatypes::{FieldRef, Fields}; +use arrow::datatypes::{Field, FieldRef, Fields, Schema}; use arrow::{ array::{ cast::AsArray, new_null_array, types::TimestampMicrosecondType, @@ -148,17 +150,32 @@ impl SparkParquetOptions { /// Spark-compatible cast implementation. Defers to DataFusion's cast where that is known /// to be compatible, and returns an error when a not supported and not DF-compatible cast -/// is requested. +/// is requested. Resolves the nested field mapping for this one value; a per-file caller +/// resolves once and uses [`spark_parquet_convert_with_mapping`] for every batch. pub fn spark_parquet_convert( arg: ColumnarValue, data_type: &DataType, parquet_options: &SparkParquetOptions, +) -> DataFusionResult { + let mapping = + resolve_field_mapping(&arg.data_type(), data_type, parquet_options).map_err(spark_error)?; + spark_parquet_convert_with_mapping(arg, data_type, &mapping, parquet_options) +} + +/// [`spark_parquet_convert`] with a mapping already resolved for the value's type. +pub(crate) fn spark_parquet_convert_with_mapping( + arg: ColumnarValue, + data_type: &DataType, + mapping: &FieldMapping, + parquet_options: &SparkParquetOptions, ) -> DataFusionResult { match arg { - ColumnarValue::Array(array) => Ok(ColumnarValue::Array(parquet_convert_array( + ColumnarValue::Array(array) => Ok(ColumnarValue::Array(convert_array( array, data_type, + mapping, parquet_options, + None, )?)), ColumnarValue::Scalar(scalar) => { // Note that normally CAST(scalar) should be fold in Spark JVM side. However, for @@ -166,7 +183,7 @@ pub fn spark_parquet_convert( // here. let array = scalar.to_array()?; let scalar = ScalarValue::try_from_array( - &parquet_convert_array(array, data_type, parquet_options)?, + &convert_array(array, data_type, mapping, parquet_options, None)?, 0, )?; Ok(ColumnarValue::Scalar(scalar)) @@ -174,17 +191,286 @@ pub fn spark_parquet_convert( } } -fn parquet_convert_array( - array: ArrayRef, +/// Wrap a [`SparkError`] the way every native operator surfaces it to the JVM. +pub(crate) fn spark_error(error: SparkError) -> DataFusionError { + DataFusionError::External(Box::new(error)) +} + +/// Outcome of matching one requested id or name against a struct's file fields: the last +/// file field that matched and whether more than one did. A plain `Copy` value, so resolving +/// a wide struct allocates nothing per id or per name; the matched names are only gathered +/// when an ambiguity is reported. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct FieldMatch { + pub(crate) index: usize, + pub(crate) ambiguous: bool, +} + +impl FieldMatch { + pub(crate) fn new(index: usize, ambiguous: bool) -> Self { + Self { index, ambiguous } + } + + /// The first file field carrying this id or name. + pub(crate) fn first(index: usize) -> Self { + Self::new(index, false) + } + + /// A further file field carrying the same id or name: the later index wins, as Spark's + /// `toMap` does for exact names, and the entry turns ambiguous. + pub(crate) fn also(self, index: usize) -> Self { + Self::new(index, true) + } +} + +/// Record file field `index` under `key`, keeping the entry `Copy`-sized however many fields +/// share the key. +pub(crate) fn record_field_match( + matches: &mut HashMap, + key: K, + index: usize, +) { + matches + .entry(key) + .and_modify(|m| *m = m.also(index)) + .or_insert_with(|| FieldMatch::first(index)); +} + +/// Comma-joined names of the fields carrying `id`, for the duplicate-id error message. +pub(crate) fn field_names_with_id(fields: &Fields, id: i32) -> String { + fields + .iter() + .filter(|f| field_id(f) == Some(id)) + .map(|f| f.name().as_str()) + .collect::>() + .join(", ") +} + +/// Which file field supplies each requested field, resolved once per file and reused for +/// every batch. Follows the requested type as Spark's `clipParquetSchema` does: a struct +/// lists one source per requested field, a list (large or not) or map carries the mapping +/// of its element or key and value types, and anything else is a leaf converted by type. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub(crate) enum FieldMapping { + Struct(Vec), + List(Box), + Map(Box, Box), + Leaf, +} + +/// The file field behind one requested struct field. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub(crate) struct StructFieldSource { + /// Index of the file field supplying the requested field; `None` null-fills it. + pub(crate) from_index: Option, + /// Mapping of the requested field's own type. + pub(crate) nested: FieldMapping, +} + +impl FieldMapping { + /// Mapping of a list's element type. A `Leaf` list converts its elements by type alone, + /// as the adapter hands one to every column whose type holds no struct. + fn list_element(&self) -> DataFusionResult<&FieldMapping> { + match self { + FieldMapping::List(inner) => Ok(inner), + FieldMapping::Leaf => Ok(&FieldMapping::Leaf), + other => Err(DataFusionError::Internal(format!( + "list column resolved to a non-list field mapping: {other:?}" + ))), + } + } + + /// Mappings of a map's key and value types; see [`FieldMapping::list_element`]. + fn map_entries(&self) -> DataFusionResult<(&FieldMapping, &FieldMapping)> { + match self { + FieldMapping::Map(key, value) => Ok((key, value)), + FieldMapping::Leaf => Ok((&FieldMapping::Leaf, &FieldMapping::Leaf)), + other => Err(DataFusionError::Internal(format!( + "map column resolved to a non-map field mapping: {other:?}" + ))), + } + } + + /// True when every requested field reads the file field at its own position, so a + /// metadata-only relabel of the file array already yields the requested layout. + pub(crate) fn is_positional(&self) -> bool { + match self { + FieldMapping::Struct(sources) => sources + .iter() + .enumerate() + .all(|(i, s)| s.from_index == Some(i) && s.nested.is_positional()), + FieldMapping::List(inner) => inner.is_positional(), + FieldMapping::Map(key, value) => key.is_positional() && value.is_positional(), + FieldMapping::Leaf => true, + } + } +} + +/// True when a field of `schema`, at any nesting depth, carries a Parquet field id. +pub(crate) fn schema_holds_field_ids(schema: &Schema) -> bool { + schema.fields().iter().any(|f| field_holds_id(f)) +} + +fn field_holds_id(field: &Field) -> bool { + field_id(field).is_some() + || match field.data_type() { + DataType::Struct(fields) => fields.iter().any(|f| field_holds_id(f)), + DataType::List(f) | DataType::LargeList(f) | DataType::Map(f, _) => field_holds_id(f), + _ => false, + } +} + +/// Resolve every requested root field against `file_schema` the way the expression adapter +/// does, keeping only the ambiguity Spark reports. DataFusion's opener creates the adapter +/// only when a predicate is pushed or the file schema differs from the requested one, so the +/// reader factory runs this on every footer it loads to cover the files the adapter never sees. +pub(crate) fn validate_field_mapping( + file_schema: &Schema, + requested_schema: &Schema, + parquet_options: &SparkParquetOptions, +) -> Result<(), SparkError> { + // `ParquetMissingFieldIds` needs no counterpart here: a file with no ids differs from an + // id-bearing requested schema in field metadata, so the opener runs the adapter for it. + resolve_field_mapping( + &DataType::Struct(file_schema.fields().clone()), + &DataType::Struct(requested_schema.fields().clone()), + parquet_options, + ) + .map(|_| ()) +} + +/// Resolve how `to_type` reads from `from_type`, recursing through struct, list, and map +/// types. Raises the ambiguity Spark reports from `clipParquetGroupFields` when a requested +/// id or case-insensitive name matches more than one file field at any level. +pub(crate) fn resolve_field_mapping( + from_type: &DataType, to_type: &DataType, parquet_options: &SparkParquetOptions, -) -> DataFusionResult { - parquet_convert_array_impl(array, to_type, parquet_options, None) +) -> Result { + use DataType::*; + match (from_type, to_type) { + (Struct(from_fields), Struct(to_fields)) => { + resolve_struct_mapping(from_fields, to_fields, parquet_options) + } + (List(from_item), List(to_item)) | (LargeList(from_item), LargeList(to_item)) => { + Ok(FieldMapping::List(Box::new(resolve_field_mapping( + from_item.data_type(), + to_item.data_type(), + parquet_options, + )?))) + } + (Map(from_entries, from_ordered), Map(to_entries, to_ordered)) + if from_ordered == to_ordered => + { + match (from_entries.data_type(), to_entries.data_type()) { + (Struct(from_kv), Struct(to_kv)) if from_kv.len() == 2 && to_kv.len() == 2 => { + let key = resolve_field_mapping( + from_kv[0].data_type(), + to_kv[0].data_type(), + parquet_options, + )?; + let value = resolve_field_mapping( + from_kv[1].data_type(), + to_kv[1].data_type(), + parquet_options, + )?; + Ok(FieldMapping::Map(Box::new(key), Box::new(value))) + } + _ => Ok(FieldMapping::Leaf), + } + } + _ => Ok(FieldMapping::Leaf), + } } -fn parquet_convert_array_impl( +/// Match `to` (requested) struct fields to `from` (file) fields. Mirrors Spark's +/// `clipParquetGroupFields`: when the requested struct carries Parquet field ids anywhere, +/// id-bearing requested fields match only by id and the rest by name; otherwise every field +/// matches by name. +fn resolve_struct_mapping( + from_fields: &Fields, + to_fields: &Fields, + parquet_options: &SparkParquetOptions, +) -> Result { + let should_match_by_id = + parquet_options.use_field_id && to_fields.iter().any(|f| field_id(f).is_some()); + + let mut id_matches: HashMap = HashMap::new(); + if should_match_by_id { + for (i, field) in from_fields.iter().enumerate() { + if let Some(id) = field_id(field) { + record_field_match(&mut id_matches, id, i); + } + } + } + + // Fold the file and requested names once via the same `toLowerCase(Locale.ROOT)` the + // top-level schema adapter uses, so nested case-insensitive matching agrees with it. + 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()); + + let mut name_matches: HashMap<&str, FieldMatch> = HashMap::new(); + for (i, folded) in from_folded.iter().enumerate() { + record_field_match(&mut name_matches, folded.as_str(), i); + } + + let mut sources = 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)) { + // A missing id match is a missing column, never a name match. + (true, Some(id)) => match id_matches.get(&id) { + Some(m) if m.ambiguous => { + return Err(SparkError::DuplicateFieldByFieldId { + required_id: id, + matched_fields: field_names_with_id(from_fields, id), + }); + } + Some(m) => Some(m.index), + None => None, + }, + _ => match name_matches.get(to_folded[to_pos].as_str()) { + // Spark's `caseInsensitiveParquetFieldMap` rejects a requested name that folds + // onto more than one file field. In case-sensitive mode the fold is identity, so + // a collision means byte-identical siblings and the later one wins silently, + // as with Spark's `caseSensitiveParquetFieldMap` built by `toMap`. + Some(m) if m.ambiguous && !parquet_options.case_sensitive => { + let matched: Vec<&str> = from_folded + .iter() + .zip(from_fields.iter()) + .filter(|(folded, _)| *folded == &to_folded[to_pos]) + .map(|(_, f)| f.name().as_str()) + .collect(); + return Err(SparkError::duplicate_field_case_insensitive( + to_field.name(), + &matched, + )); + } + Some(m) => Some(m.index), + None => None, + }, + }; + let nested = match from_index { + Some(i) => resolve_field_mapping( + from_fields[i].data_type(), + to_field.data_type(), + parquet_options, + )?, + None => FieldMapping::Leaf, + }; + sources.push(StructFieldSource { from_index, nested }); + } + Ok(FieldMapping::Struct(sources)) +} + +/// Convert `array` to `to_type` through its resolved `mapping`. `parent_nulls` masks the rows +/// hidden beneath null ancestors, so only values Spark reads are checked for overflow. +fn convert_array( array: ArrayRef, to_type: &DataType, + mapping: &FieldMapping, parquet_options: &SparkParquetOptions, parent_nulls: Option<&NullBuffer>, ) -> DataFusionResult { @@ -203,25 +489,35 @@ fn parquet_convert_array_impl( // Try Comet specific handlers first, then arrow-rs cast if supported, // return uncasted data otherwise - match (from_type, to_type) { - (Struct(_), Struct(_)) => Ok(parquet_convert_struct_to_struct( + match (from_type, to_type, mapping) { + (Struct(_), Struct(to_fields), FieldMapping::Struct(sources)) => convert_struct( array.as_struct(), - from_type, - to_type, + to_fields, + sources, parquet_options, visible.as_ref(), - )?), - (List(_), List(to_inner_type)) => { + ), + // A struct always resolves to a struct mapping; anything else is a planning bug and + // must not fall through to a silent pass-through of the file's struct. + (Struct(_), Struct(_), other) => Err(DataFusionError::Internal(format!( + "struct column resolved to a non-struct field mapping: {other:?}" + ))), + (List(_), List(to_inner_type), _) => { + let inner = mapping.list_element()?; let list_arr: &ListArray = array.as_list(); let child_visibility = if checked_timestamp_overflow { repeated_visibility( - list_arr.value_offsets(), list_arr.values().len(), visible.as_ref()) + list_arr.value_offsets(), + list_arr.values().len(), + visible.as_ref(), + ) } else { None }; - let cast_field = parquet_convert_array_impl( + let cast_field = convert_array( Arc::clone(list_arr.values()), to_inner_type.data_type(), + inner, parquet_options, child_visibility.as_ref(), )?; @@ -233,10 +529,36 @@ fn parquet_convert_array_impl( list_arr.nulls().cloned(), ))) } - ( - Timestamp(TimeUnit::Millisecond, _), - Timestamp(TimeUnit::Microsecond, target_tz), - ) if checked_timestamp_overflow => { + (LargeList(_), LargeList(to_inner_type), _) => { + let inner = mapping.list_element()?; + let list_arr: &LargeListArray = array.as_list(); + let child_visibility = if checked_timestamp_overflow { + repeated_visibility( + list_arr.value_offsets(), + list_arr.values().len(), + visible.as_ref(), + ) + } else { + None + }; + let cast_field = convert_array( + Arc::clone(list_arr.values()), + to_inner_type.data_type(), + inner, + parquet_options, + child_visibility.as_ref(), + )?; + + Ok(Arc::new(LargeListArray::new( + Arc::clone(to_inner_type), + list_arr.offsets().clone(), + cast_field, + list_arr.nulls().cloned(), + ))) + } + (Timestamp(TimeUnit::Millisecond, _), Timestamp(TimeUnit::Microsecond, target_tz), _) + if checked_timestamp_overflow => + { // Spark's Parquet reader calls the checked `millisToMicros` conversion for both // direct and dictionary values, independent of CAST evaluation mode: // https://github.com/apache/spark/blob/v4.2.0/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java#L817-L833 @@ -248,15 +570,17 @@ fn parquet_convert_array_impl( let millis = array.as_primitive::(); // Ignore values hidden by null ancestors or by sliced list/map offsets. // Restore the original child validity: required fields must remain non-null. - let micros = arrow::array::TimestampMillisecondArray::new( - millis.values().clone(), visible) - .try_unary::<_, TimestampMicrosecondType, _>(|value| value.mul_checked(1_000))?; + let micros = + arrow::array::TimestampMillisecondArray::new(millis.values().clone(), visible) + .try_unary::<_, TimestampMicrosecondType, _>(|value| value.mul_checked(1_000))?; let micros = arrow::array::TimestampMicrosecondArray::new( - micros.values().clone(), millis.nulls().cloned()) - .with_timezone_opt(target_tz.clone()); + micros.values().clone(), + millis.nulls().cloned(), + ) + .with_timezone_opt(target_tz.clone()); Ok(Arc::new(micros)) } - (Timestamp(TimeUnit::Microsecond, None), Timestamp(TimeUnit::Microsecond, Some(tz))) => { + (Timestamp(TimeUnit::Microsecond, None), Timestamp(TimeUnit::Microsecond, Some(tz)), _) => { Ok(Arc::new( array .as_primitive::() @@ -264,12 +588,21 @@ fn parquet_convert_array_impl( .with_timezone(Arc::clone(tz)), )) } - (Map(_, ordered_from), Map(_, ordered_to)) if ordered_from == ordered_to => - parquet_convert_map_to_map(array.as_map(), to_type, parquet_options, *ordered_to, visible.as_ref(), checked_timestamp_overflow) - , + (Map(_, ordered_from), Map(_, ordered_to), _) if ordered_from == ordered_to => { + let (key, value) = mapping.map_entries()?; + parquet_convert_map_to_map( + array.as_map(), + to_type, + key, + value, + parquet_options, + visible.as_ref(), + checked_timestamp_overflow, + ) + } // Iceberg stores UUIDs as 16-byte fixed binary but Spark expects string representation. // Arrow doesn't support casting FixedSizeBinary to Utf8, so we handle it manually. - (FixedSizeBinary(16), Utf8) => { + (FixedSizeBinary(16), Utf8, _) => { let binary_array = array .as_any() .downcast_ref::() @@ -279,9 +612,8 @@ fn parquet_convert_array_impl( .iter() .map(|opt_bytes| { opt_bytes.map(|bytes| { - let uuid = uuid::Uuid::from_bytes( - bytes.try_into().expect("Expected 16 bytes") - ); + let uuid = + uuid::Uuid::from_bytes(bytes.try_into().expect("Expected 16 bytes")); uuid.to_string() }) }) @@ -313,161 +645,109 @@ fn has_timestamp_unit(data_type: &DataType, unit: TimeUnit) -> bool { } // List/map values can include entries outside a slice or beneath a null parent. -fn repeated_visibility( - offsets: &[i32], +fn repeated_visibility( + offsets: &[O], len: usize, nulls: Option<&NullBuffer>, ) -> Option { - 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)) } /// Read the Parquet field id stored under arrow-rs's `PARQUET_FIELD_ID_META_KEY`. -fn field_id(field: &arrow::datatypes::Field) -> Option { +pub(crate) fn field_id(field: &arrow::datatypes::Field) -> Option { field .metadata() .get(PARQUET_FIELD_ID_META_KEY) .and_then(|v| v.parse::().ok()) } -/// Cast between struct types based on logic in +/// Build the requested struct from the file struct, reading each requested field from the +/// file field at its resolved source index. Based on /// `org.apache.spark.sql.catalyst.expressions.Cast#castStruct`. -fn parquet_convert_struct_to_struct( +fn convert_struct( array: &StructArray, - from_type: &DataType, - to_type: &DataType, + to_fields: &Fields, + sources: &[StructFieldSource], parquet_options: &SparkParquetOptions, parent_nulls: Option<&NullBuffer>, ) -> DataFusionResult { - 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 = 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> = HashMap::new(); - for (i, folded) in from_folded.iter().enumerate() { - folded_to_indices - .entry(folded.as_str()) - .or_default() - .push(i); - } + if sources.len() != to_fields.len() { + return Err(DataFusionError::Internal(format!( + "struct field mapping has {} sources for {} requested fields", + sources.len(), + to_fields.len() + ))); + } - let mut field_overlap = false; - let mut cast_fields: Vec = 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, - }, - }; - - if let Some(from_index) = from_index { - cast_fields.push(parquet_convert_array_impl( - Arc::clone(array.column(from_index)), - to_field.data_type(), - parquet_options, - parent_nulls, - )?); - field_overlap = true; - } else { - cast_fields.push(new_null_array(to_field.data_type(), array.len())); - } + let mut field_overlap = false; + let mut cast_fields: Vec = Vec::with_capacity(to_fields.len()); + for (to_field, source) in to_fields.iter().zip(sources) { + match source.from_index { + Some(from_index) => { + // The mapping is resolved once per file against the physical schema; a batch + // whose struct carries fewer children than that schema must error, not panic. + let child = array.columns().get(from_index).ok_or_else(|| { + DataFusionError::Internal(format!( + "struct field {} maps to file child {from_index}, but the struct has {} \ + child(ren)", + to_field.name(), + array.num_columns() + )) + })?; + cast_fields.push(convert_array( + Arc::clone(child), + to_field.data_type(), + &source.nested, + parquet_options, + parent_nulls, + )?); + field_overlap = true; } - - // When the file's struct contains none of the requested fields, the - // returned validity buffer depends on Spark's - // `spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing` (SPARK-53535, - // Spark 4.1+). Legacy mode marks the whole column null; the new default - // preserves the file's parent-row nullness so non-null parents materialize - // as a struct of all-null fields. - let nulls = - if !field_overlap && parquet_options.return_null_struct_if_all_fields_missing { - Some(NullBuffer::new_null(array.len())) - } else { - array.nulls().cloned() - }; - - Ok(Arc::new(StructArray::new( - to_fields.clone(), - cast_fields, - nulls, - ))) + None => cast_fields.push(new_null_array(to_field.data_type(), array.len())), } - _ => unreachable!(), } + + // When the file's struct contains none of the requested fields, the + // returned validity buffer depends on Spark's + // `spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing` (SPARK-53535, + // Spark 4.1+). Legacy mode marks the whole column null; the new default + // preserves the file's parent-row nullness so non-null parents materialize + // as a struct of all-null fields. + let nulls = if !field_overlap && parquet_options.return_null_struct_if_all_fields_missing { + Some(NullBuffer::new_null(array.len())) + } else { + array.nulls().cloned() + }; + + Ok(Arc::new(StructArray::new( + to_fields.clone(), + cast_fields, + nulls, + ))) } /// Cast a map type to another map type. The same as arrow-cast except we recursively call our own -/// parquet_convert_array +/// convert_array with the resolved key and value mappings. fn parquet_convert_map_to_map( from: &MapArray, to_data_type: &DataType, + key_mapping: &FieldMapping, + value_mapping: &FieldMapping, parquet_options: &SparkParquetOptions, - to_ordered: bool, parent_nulls: Option<&NullBuffer>, checked_timestamp_overflow: bool, ) -> Result { match to_data_type { - DataType::Map(entries_field, _) => { + DataType::Map(entries_field, to_ordered) => { let key_field = key_field(entries_field).ok_or(DataFusionError::Internal( "map is missing key field".to_string(), ))?; @@ -480,15 +760,17 @@ fn parquet_convert_map_to_map( } else { None }; - let key_array = parquet_convert_array_impl( + let key_array = convert_array( Arc::clone(from.keys()), key_field.data_type(), + key_mapping, parquet_options, child_visibility.as_ref(), )?; - let value_array = parquet_convert_array_impl( + let value_array = convert_array( Arc::clone(from.values()), value_field.data_type(), + value_mapping, parquet_options, child_visibility.as_ref(), )?; @@ -502,7 +784,7 @@ fn parquet_convert_map_to_map( from.entries().nulls().cloned(), ), from.nulls().cloned(), - to_ordered, + *to_ordered, ))) } dt => Err(DataFusionError::Internal(format!( @@ -789,9 +1071,23 @@ mod tests { } } + /// Convert one array through the public entry point, resolving its mapping. + fn parquet_convert_array( + array: arrow::array::ArrayRef, + to_type: &arrow::datatypes::DataType, + parquet_options: &crate::parquet::parquet_support::SparkParquetOptions, + ) -> datafusion::common::Result { + use crate::parquet::parquet_support::spark_parquet_convert; + use datafusion::physical_plan::ColumnarValue; + match spark_parquet_convert(ColumnarValue::Array(array), to_type, parquet_options)? { + ColumnarValue::Array(array) => Ok(array), + ColumnarValue::Scalar(_) => unreachable!("array input yields an array"), + } + } + #[test] fn test_millis_to_micros_overflow_checked_in_nested_fields() { - use crate::parquet::parquet_support::{parquet_convert_array, SparkParquetOptions}; + use crate::parquet::parquet_support::SparkParquetOptions; use arrow::array::{Array, ArrayRef, StructArray, TimestampMillisecondArray}; use arrow::datatypes::{DataType, Field, Fields, TimeUnit}; use datafusion_comet_spark_expr::EvalMode; @@ -856,7 +1152,7 @@ mod tests { #[test] fn test_millis_to_micros_preserves_unchanged_siblings() { - use super::{parquet_convert_array, SparkParquetOptions}; + use crate::parquet::parquet_support::SparkParquetOptions; use arrow::array::{ cast::AsArray, Array, ArrayRef, Int32Array, ListArray, MapArray, StructArray, TimestampMicrosecondArray, TimestampMillisecondArray, @@ -954,7 +1250,7 @@ mod tests { #[test] fn test_millis_to_micros_nested_visibility() { - use super::{parquet_convert_array, SparkParquetOptions}; + use crate::parquet::parquet_support::SparkParquetOptions; use arrow::array::{ Array, ArrayRef, ListArray, MapArray, StructArray, TimestampMicrosecondArray, TimestampMillisecondArray, @@ -1144,4 +1440,391 @@ mod tests { assert_eq!(path, Path::from(expected_path)); } } + + mod struct_field_matching { + use super::parquet_convert_array; + use crate::parquet::parquet_support::{ + resolve_field_mapping, FieldMapping, FieldMatch, SparkParquetOptions, + }; + use arrow::array::{Array, ArrayRef, Int32Array, LargeListArray, StructArray}; + use arrow::datatypes::{DataType, Field, Fields}; + use datafusion_comet_spark_expr::EvalMode; + + /// The per-id lookup entry is a plain `Copy` value: the second field sharing an id + /// only flips the ambiguity flag, so resolving a wide struct allocates no vector + /// per id. + #[test] + fn field_match_records_ambiguity_without_allocating() { + fn assert_copy() {} + assert_copy::(); + + let first = FieldMatch::first(3); + assert_eq!(first, FieldMatch::new(3, false)); + let again = first.also(5); + assert_eq!(again, FieldMatch::new(5, true)); + assert!(again.ambiguous); + } + + /// Every requested id resolves to exactly one file field: the resolved mapping is + /// positional and carries one source per requested field. + #[test] + fn resolve_mapping_by_id_is_positional_for_unique_ids() { + let fields: Vec = (0..256) + .map(|i| field_with_id(&format!("c{i}"), i)) + .collect(); + let from_type = DataType::Struct(Fields::from(fields.clone())); + let to_type = DataType::Struct(Fields::from(fields)); + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.use_field_id = true; + + let mapping = resolve_field_mapping(&from_type, &to_type, &opts).unwrap(); + assert!(mapping.is_positional()); + let FieldMapping::Struct(sources) = &mapping else { + panic!("expected a struct mapping"); + }; + assert_eq!(sources.len(), 256); + assert!(sources + .iter() + .enumerate() + .all(|(i, s)| s.from_index == Some(i))); + } + + /// Requested ids in a different order than the file resolve by id, so the mapping + /// is not positional and a metadata-only relabel would read the wrong columns. + #[test] + fn resolve_mapping_by_id_reorders_swapped_ids() { + let from_type = DataType::Struct(Fields::from(vec![ + field_with_id("x", 1), + field_with_id("y", 2), + ])); + let to_type = DataType::Struct(Fields::from(vec![ + field_with_id("x", 2), + field_with_id("y", 1), + ])); + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.use_field_id = true; + + let mapping = resolve_field_mapping(&from_type, &to_type, &opts).unwrap(); + assert!(!mapping.is_positional()); + let FieldMapping::Struct(sources) = &mapping else { + panic!("expected a struct mapping"); + }; + assert_eq!(sources[0].from_index, Some(1)); + assert_eq!(sources[1].from_index, Some(0)); + } + + /// A large list element resolves like a list element: swapped ids inside it make the + /// mapping non-positional and the conversion reads each field by id. + #[test] + fn resolve_mapping_recurses_into_large_list_element() { + let from_elem = Fields::from(vec![field_with_id("x", 1), field_with_id("y", 2)]); + let to_elem = Fields::from(vec![field_with_id("x", 2), field_with_id("y", 1)]); + let from_field = Arc::new(Field::new("item", DataType::Struct(from_elem), true)); + let to_field = Arc::new(Field::new("item", DataType::Struct(to_elem), true)); + let from_type = DataType::LargeList(Arc::clone(&from_field)); + let to_type = DataType::LargeList(to_field); + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.use_field_id = true; + + let mapping = resolve_field_mapping(&from_type, &to_type, &opts).unwrap(); + assert!(!mapping.is_positional()); + + let element = struct_of( + vec![field_with_id("x", 1), field_with_id("y", 2)], + vec![42, 43], + ); + let list = LargeListArray::new( + from_field, + arrow::buffer::OffsetBuffer::new(vec![0i64, 1].into()), + element, + None, + ); + let result = parquet_convert_array(Arc::new(list), &to_type, &opts).unwrap(); + assert_eq!(result.data_type(), &to_type); + let values = result + .as_any() + .downcast_ref::() + .unwrap() + .values() + .as_any() + .downcast_ref::() + .unwrap() + .clone(); + let x = values + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let y = values + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(x.value(0), 43); + assert_eq!(y.value(0), 42); + } + + /// A duplicated requested id nested under a list element is rejected at resolution + /// time, mirroring Spark's `clipParquetListType` recursing into `matchIdField`. + #[test] + fn resolve_mapping_rejects_duplicate_id_inside_list_element() { + let from_elem = DataType::Struct(Fields::from(vec![ + field_with_id("x", 1), + field_with_id("y", 1), + ])); + let to_elem = DataType::Struct(Fields::from(vec![field_with_id("x", 1)])); + let from_type = DataType::List(Arc::new(Field::new("item", from_elem, true))); + let to_type = DataType::List(Arc::new(Field::new("element", to_elem, true))); + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.use_field_id = true; + + let err = resolve_field_mapping(&from_type, &to_type, &opts).unwrap_err(); + let msg = err.to_string(); + assert!( + msg.contains("_LEGACY_ERROR_TEMP_2094") && msg.contains("[x, y]"), + "unexpected error: {msg}" + ); + } + use parquet::arrow::PARQUET_FIELD_ID_META_KEY; + use std::collections::HashMap; + use std::sync::Arc; + + fn field_with_id(name: &str, id: i32) -> Field { + Field::new(name, DataType::Int32, true).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + id.to_string(), + )])) + } + + fn struct_of(fields: Vec, values: Vec) -> ArrayRef { + let arrays: Vec = values + .into_iter() + .map(|v| Arc::new(Int32Array::from(vec![Some(v)])) as ArrayRef) + .collect(); + Arc::new(StructArray::new(Fields::from(fields), arrays, None)) + } + + /// Two physical struct fields share field ID 1 and the logical struct requests that + /// ID: Spark's `matchIdField` raises `foundDuplicateFieldInFieldIdLookupModeError` + /// (`_LEGACY_ERROR_TEMP_2094`) rather than silently reading the first match. + #[test] + fn requested_duplicate_field_id_errors() { + let from = struct_of( + vec![field_with_id("x", 1), field_with_id("y", 1)], + vec![42, 43], + ); + let to_type = DataType::Struct(Fields::from(vec![field_with_id("f", 1)])); + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.use_field_id = true; + + let err = parquet_convert_array(from, &to_type, &opts).unwrap_err(); + let msg = err.to_string(); + assert!( + msg.contains("_LEGACY_ERROR_TEMP_2094") && msg.contains("id=1"), + "unexpected error: {msg}" + ); + } + + /// Companion to `requested_duplicate_field_id_errors`: a duplicated file ID that no + /// requested field looks up must stay harmless (Spark only raises inside + /// `matchIdField`, i.e. for requested IDs). + #[test] + fn unrequested_duplicate_field_id_reads_fine() { + let from = struct_of( + vec![ + field_with_id("x", 1), + field_with_id("y", 1), + field_with_id("z", 2), + ], + vec![42, 43, 44], + ); + let to_type = DataType::Struct(Fields::from(vec![field_with_id("f", 2)])); + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.use_field_id = true; + + let result = parquet_convert_array(from, &to_type, &opts).unwrap(); + let result_struct = result.as_any().downcast_ref::().unwrap(); + let col = result_struct + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(col.value(0), 44); + } + + /// Two physical struct fields carry the IDENTICAL name in case-sensitive mode. + /// Spark's `caseSensitiveParquetFieldMap` is built with `.toMap`, where the later + /// entry wins silently; the exact-name lookup here must do the same rather than + /// return the first field. + #[test] + fn duplicate_exact_names_resolve_to_the_last_field() { + let from = struct_of( + vec![ + Field::new("d", DataType::Int32, true), + Field::new("d", DataType::Int32, true), + ], + vec![1, 2], + ); + let to_type = + DataType::Struct(Fields::from(vec![Field::new("d", DataType::Int32, true)])); + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.case_sensitive = true; + + let result = parquet_convert_array(from, &to_type, &opts).unwrap(); + let result_struct = result.as_any().downcast_ref::().unwrap(); + let col = result_struct + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(col.value(0), 2); + } + + /// Two file children differ only by case and the requested name folds onto both: + /// Spark's `caseInsensitiveParquetFieldMap` raises `_LEGACY_ERROR_TEMP_2093` rather + /// than picking either, and case-sensitive mode reads the exact match. + #[test] + fn case_insensitive_ambiguous_names_error_but_exact_match_reads() { + let from = struct_of( + vec![ + Field::new("A", DataType::Int32, true), + Field::new("a", DataType::Int32, true), + ], + vec![1, 2], + ); + let to_type = + DataType::Struct(Fields::from(vec![Field::new("a", DataType::Int32, true)])); + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.case_sensitive = false; + let err = parquet_convert_array(Arc::clone(&from), &to_type, &opts).unwrap_err(); + let msg = err.to_string(); + assert!( + msg.contains("duplicate field") && msg.contains("A") && msg.contains("a"), + "unexpected error: {msg}" + ); + + opts.case_sensitive = true; + let result = parquet_convert_array(from, &to_type, &opts).unwrap(); + let result_struct = result.as_any().downcast_ref::().unwrap(); + let col = result_struct + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(col.value(0), 2); + } + + /// A list or map column without any struct inside gets a `Leaf` mapping from the + /// adapter; its elements still convert through the checked millis-to-micros path. + #[test] + fn leaf_mapping_converts_list_and_map_elements() { + use crate::parquet::parquet_support::spark_parquet_convert_with_mapping; + use arrow::array::{ListArray, MapArray, TimestampMillisecondArray}; + use arrow::buffer::OffsetBuffer; + use arrow::datatypes::TimeUnit; + use datafusion::physical_plan::ColumnarValue; + + let millis: ArrayRef = Arc::new(TimestampMillisecondArray::from(vec![i64::MAX])); + let ms_field = Arc::new(Field::new( + "item", + DataType::Timestamp(TimeUnit::Millisecond, None), + true, + )); + let us_field = Arc::new(Field::new( + "item", + DataType::Timestamp(TimeUnit::Microsecond, None), + true, + )); + let list: ArrayRef = Arc::new(ListArray::new( + Arc::clone(&ms_field), + OffsetBuffer::new(vec![0, 1].into()), + Arc::clone(&millis), + None, + )); + let entries = StructArray::new( + Fields::from(vec![ + Field::new( + "key", + DataType::Timestamp(TimeUnit::Millisecond, None), + false, + ), + Field::new("value", DataType::Int32, true), + ]), + vec![Arc::clone(&millis), Arc::new(Int32Array::from(vec![1]))], + None, + ); + let map: ArrayRef = Arc::new(MapArray::new( + Arc::new(Field::new("entries", entries.data_type().clone(), false)), + OffsetBuffer::new(vec![0, 1].into()), + entries, + None, + false, + )); + let map_target = DataType::Map( + Arc::new(Field::new( + "entries", + DataType::Struct(Fields::from(vec![ + Field::new( + "key", + DataType::Timestamp(TimeUnit::Microsecond, None), + false, + ), + Field::new("value", DataType::Int32, true), + ])), + false, + )), + false, + ); + let opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + for (array, target) in [(list, DataType::List(us_field)), (map, map_target)] { + let result = spark_parquet_convert_with_mapping( + ColumnarValue::Array(array), + &target, + &FieldMapping::Leaf, + &opts, + ); + assert!(result.is_err(), "overflow must be reported for {target:?}"); + } + } + + /// A mapping resolved against a wider struct than the array actually carries must + /// surface as an error naming the field, not as an index panic in the executor. + #[test] + fn mapping_index_beyond_struct_children_errors() { + use crate::parquet::parquet_support::{ + spark_parquet_convert_with_mapping, StructFieldSource, + }; + use datafusion::physical_plan::ColumnarValue; + + let from = struct_of(vec![Field::new("x", DataType::Int32, true)], vec![1]); + let to_type = + DataType::Struct(Fields::from(vec![Field::new("y", DataType::Int32, true)])); + let mapping = FieldMapping::Struct(vec![StructFieldSource { + from_index: Some(1), + nested: FieldMapping::Leaf, + }]); + let opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + + let err = spark_parquet_convert_with_mapping( + ColumnarValue::Array(from), + &to_type, + &mapping, + &opts, + ) + .unwrap_err(); + let msg = err.to_string(); + assert!( + msg.contains("y") && msg.contains("1 child"), + "unexpected error: {msg}" + ); + } + } } diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs index 004cf93ebf9..cf9cdbd541c 100644 --- a/native/core/src/parquet/schema_adapter.rs +++ b/native/core/src/parquet/schema_adapter.rs @@ -17,7 +17,10 @@ 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::{ + field_names_with_id, record_field_match, resolve_field_mapping, spark_error, + spark_parquet_convert, FieldMapping, FieldMatch, SparkParquetOptions, +}; use arrow::array::new_empty_array; use arrow::compute::can_cast_types; use arrow::datatypes::{DataType, Field, FieldRef, Schema, SchemaRef}; @@ -180,6 +183,25 @@ fn is_pure_structural_narrowing( } } +/// True when two root fields of `schema` carry the same exact name. +fn has_duplicate_names(schema: &SchemaRef) -> bool { + let mut seen: HashSet<&str> = HashSet::with_capacity(schema.fields().len()); + schema + .fields() + .iter() + .any(|f| !seen.insert(f.name().as_str())) +} + +/// Per root field of `schema`, whether a later field carries the same exact name. +fn shadowed_by_later_duplicate(schema: &SchemaRef) -> Vec { + let mut seen: HashSet<&str> = HashSet::with_capacity(schema.fields().len()); + let mut shadowed = vec![false; schema.fields().len()]; + for (i, f) in schema.fields().iter().enumerate().rev() { + shadowed[i] = !seen.insert(f.name().as_str()); + } + shadowed +} + /// Remap physical schema field names to match logical schema field names. Mirrors Spark's /// `clipParquetGroupFields`: prefer ID match for any logical field that carries a /// `PARQUET:field_id`, fall back to case-insensitive name match otherwise. @@ -205,30 +227,23 @@ fn remap_physical_schema( ))); } - // Build id -> all matching physical field names. We need the full list so we can mirror - // Spark's `_LEGACY_ERROR_TEMP_2094` "Found duplicate field(s)" error when an ID-bearing - // logical field would resolve to more than one physical field. - let mut id_to_phys_names: HashMap> = HashMap::new(); + // Index every physical field id once. Spark's `matchIdField` raises + // `_LEGACY_ERROR_TEMP_2094` "Found duplicate field(s)" when an ID-bearing logical field + // resolves to more than one physical field; the matched names are only gathered then. if should_match_by_id { - for pf in physical_schema.fields() { + let mut id_matches: HashMap = HashMap::new(); + for (i, pf) in physical_schema.fields().iter().enumerate() { if let Some(id) = parse_field_id(pf) { - id_to_phys_names - .entry(id) - .or_default() - .push(pf.name().clone()); + record_field_match(&mut id_matches, id, i); } } for lf in logical_schema.fields() { if let Some(id) = parse_field_id(lf) { - if let Some(matches) = id_to_phys_names.get(&id) { - if matches.len() > 1 { - return Err(DataFusionError::External(Box::new( - SparkError::DuplicateFieldByFieldId { - required_id: id, - matched_fields: matches.join(", "), - }, - ))); - } + if id_matches.get(&id).is_some_and(|m| m.ambiguous) { + return Err(spark_error(SparkError::DuplicateFieldByFieldId { + required_id: id, + matched_fields: field_names_with_id(physical_schema.fields(), id), + })); } } } @@ -254,30 +269,55 @@ fn remap_physical_schema( let logical_folded = fold_schema_names(logical_schema, case_sensitive); let physical_folded = fold_schema_names(physical_schema, case_sensitive); - // Folded names of ID-bearing logical fields whose ID is not present in the file. Any physical - // field that shares one of these names must be renamed to something the - // `DefaultPhysicalExprAdapter` cannot name-match, otherwise the read would silently fall - // through to a name match. Spark's `matchIdField` solves the same problem with - // `generateFakeColumnName` (see `ParquetReadSupport.scala`). - let unmatched_id_logical_folded: HashSet = if should_match_by_id { + // Folded names of ID-bearing logical fields. Spark's `matchIdField` resolves these + // strictly by ID and never falls back to a name match, so a physical field that carries + // such a name WITHOUT being the ID match (its ID is absent, different, or the logical ID + // matched a different physical field) must be renamed to something the + // `DefaultPhysicalExprAdapter` cannot name-match; otherwise the read would silently + // resolve the wrong column instead of null-filling. Spark's `matchIdField` solves the + // same problem with `generateFakeColumnName` (see `ParquetReadSupport.scala`). + let id_logical_folded: HashSet = if should_match_by_id { logical_schema .fields() .iter() .enumerate() - .filter_map(|(j, lf)| { - parse_field_id(lf).and_then(|id| { - if id_to_phys_names.contains_key(&id) { - None - } else { - Some(logical_folded[j].clone()) - } - }) - }) + .filter(|(_, lf)| parse_field_id(lf).is_some()) + .map(|(j, _)| logical_folded[j].clone()) .collect() } else { HashSet::new() }; + + // Fake names must never collide with a real column from either schema on the folded names + // downstream lookups use, or a requested column differing only by case counts as present + // and loses its default. The lowercase candidate is its own fold; the counter is bumped + // past reserved names, and the set is built on the first fake name so flat reads skip it. + let mut reserved_names: Option> = None; let mut fake_counter: usize = 0; + let mut next_fake_name = || { + let reserved = reserved_names.get_or_insert_with(|| { + logical_folded + .iter() + .chain(physical_folded.iter()) + .map(String::as_str) + .collect() + }); + loop { + fake_counter += 1; + let candidate = format!("__comet_unmatched_field_id_{}", fake_counter); + if !reserved.contains(candidate.as_str()) { + return candidate; + } + } + }; + + // Physical fields whose exact name recurs later in the file. In case-sensitive mode the + // later one wins, as with Spark's `toMap`, so the earlier ones must not stay name-matchable. + let shadowed = if case_sensitive { + shadowed_by_later_duplicate(physical_schema) + } else { + Vec::new() + }; let mut name_map: HashMap = HashMap::new(); let remapped_fields: Vec = physical_schema @@ -305,21 +345,26 @@ fn remap_physical_schema( } } - // Block accidental name match for ID-bearing logical fields whose ID is missing - // from the file. Mirrors Spark's `generateFakeColumnName` in `matchIdField`. - if should_match_by_id - && unmatched_id_logical_folded.contains(&physical_folded[phys_idx]) - { - fake_counter += 1; - let fake_name = format!("__comet_unmatched_field_id_{}", fake_counter); + // A field shadowed by a later exact duplicate takes a fake name so the downstream + // exact-name lookup lands on the last one, matching Spark's `toMap`. + if shadowed.get(phys_idx).copied().unwrap_or(false) { return Arc::new( - Field::new(fake_name, field.data_type().clone(), field.is_nullable()) - .with_metadata(field.metadata().clone()), + Field::new( + next_fake_name(), + field.data_type().clone(), + field.is_nullable(), + ) + .with_metadata(field.metadata().clone()), ); } - // Name match. Spark's `matchIdField` does not fall through to a name match for - // ID-bearing logical fields, so skip those when the schema is ID-bearing. + // Name match. Spark resolves every non-ID-bearing logical field by name + // (`matchCaseSensitiveField` / `matchCaseInsensitiveField` in + // `clipParquetGroupFields`) even when field-ID matching is on; only ID-bearing + // logical fields skip the name fallback. Case-sensitive mode needs no rename + // here (the downstream adapter's exact-name lookup already hits); the + // case-insensitive lookup rewrites the physical name, and a successful match + // claims the field before the shield below can hide it. if !case_sensitive { let logical_field = logical_schema .fields() @@ -342,9 +387,28 @@ fn remap_physical_schema( .with_metadata(field.metadata().clone()), ); } + return Arc::clone(field); } } + // Shield: any remaining physical field whose name would hit an ID-bearing + // logical field downstream gets a fake name (Spark's `generateFakeColumnName` + // equivalent). ID-bearing logical fields resolve strictly by ID, so a name hit + // on one would read the wrong column instead of null-filling it or leaving it + // to its real ID match. The folded comparison mirrors the matcher that would + // otherwise hit: identity fold in case-sensitive mode, the JVM lowercase fold + // otherwise. + if should_match_by_id && id_logical_folded.contains(&physical_folded[phys_idx]) { + return Arc::new( + Field::new( + next_fake_name(), + field.data_type().clone(), + field.is_nullable(), + ) + .with_metadata(field.metadata().clone()), + ); + } + Arc::clone(field) }) .collect(); @@ -493,7 +557,11 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory { let case_sensitive = self.parquet_options.case_sensitive; let should_match_by_id = self.parquet_options.use_field_id && schema_has_field_ids(&logical_file_schema); - let needs_remap = !case_sensitive || should_match_by_id; + // Duplicate exact root names need the remap too: the default adapter's `index_of` + // returns the first match, while Spark's root-level `caseSensitiveParquetFieldMap` is + // the same last-wins `toMap` used for nested groups. + let needs_remap = + !case_sensitive || should_match_by_id || has_duplicate_names(&physical_file_schema); let (adapted_physical_schema, logical_to_physical_names, original_physical_dup_check) = if needs_remap { let (remapped, logical_to_physical) = remap_physical_schema( @@ -555,6 +623,16 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory { None }; + // Resolve every nested struct once per file, the way Spark's `clipParquetSchema` + // recurses through struct, list, and map types while clipping the file schema. Each + // batch reuses the result, and an ambiguity inside it is raised from `rewrite` for the + // columns a read references, whether or not a cast is emitted for them. + let nested_mappings = resolve_nested_mappings( + &logical_file_schema, + &adapted_physical_schema, + &self.parquet_options, + ); + let default_factory = DefaultPhysicalExprAdapterFactory; let default_adapter = default_factory.create( Arc::clone(&logical_file_schema), @@ -572,10 +650,76 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory { id_resolved_logical_folded, logical_folded, physical_folded, + nested_mappings, })) } } +/// Per logical field name, the mapping of its nested type against its physical counterpart, +/// or the ambiguity Spark reports for it. Only fields whose type holds a struct are listed. +type NestedMappings = HashMap, SparkError>>; + +fn type_holds_struct(data_type: &DataType) -> bool { + match data_type { + DataType::Struct(_) => true, + DataType::List(f) | DataType::LargeList(f) | DataType::Map(f, _) => { + type_holds_struct(f.data_type()) + } + _ => false, + } +} + +/// Resolve the nested mapping of every logical field whose type holds a struct and that has +/// a physical counterpart. Returns `None` when no field qualifies, so flat reads build +/// nothing here. Ambiguities are kept per field rather than raised: Spark validates only +/// the fields a read requests, and `rewrite` sees which ones those are. +fn resolve_nested_mappings( + logical_schema: &SchemaRef, + physical_schema: &SchemaRef, + parquet_options: &SparkParquetOptions, +) -> Option { + let mut physical_by_name: Option> = None; + let mut mappings = NestedMappings::new(); + for logical_field in logical_schema.fields() { + if !type_holds_struct(logical_field.data_type()) { + continue; + } + // The last physical field wins an exact-name tie, as Spark's `toMap` does; the remap + // has already hidden the earlier ones from the default adapter's lookup. + let by_name = physical_by_name.get_or_insert_with(|| { + physical_schema + .fields() + .iter() + .enumerate() + .map(|(i, pf)| (pf.name().as_str(), i)) + .collect::>() + }); + let Some(&physical_index) = by_name.get(logical_field.name().as_str()) else { + continue; + }; + let resolved = resolve_field_mapping( + physical_schema.field(physical_index).data_type(), + logical_field.data_type(), + parquet_options, + ) + .map(Arc::new); + mappings.insert(logical_field.name().clone(), resolved); + } + (!mappings.is_empty()).then_some(mappings) +} + +/// Names of every `Column` referenced by `expr`, in traversal order. +fn referenced_column_names(expr: &Arc) -> Vec { + let mut names: Vec = Vec::new(); + let _ = Arc::clone(expr).transform(|e| { + if let Some(col) = e.downcast_ref::() { + names.push(col.name().to_string()); + } + Ok(Transformed::no(e)) + }); + names +} + /// Spark-compatible physical expression adapter. /// /// This adapter rewrites expressions at planning time to: @@ -618,40 +762,50 @@ struct SparkPhysicalExprAdapter { /// `physical_file_schema` field names pre-folded once, parallel to /// `physical_file_schema.fields()`. See `logical_folded`. physical_folded: Vec, + /// Nested field mappings resolved once in `create` (see `resolve_nested_mappings`), + /// handed to every `CometCastColumnExpr` built here. `None` for schemas without structs. + nested_mappings: Option, } impl PhysicalExprAdapter for SparkPhysicalExprAdapter { fn rewrite(&self, expr: Arc) -> DataFusionResult> { - // In case-insensitive mode, check if any Column in this expression references - // a field with multiple case-insensitive matches in the physical schema. - // Only the columns actually referenced trigger the error (not the whole schema). - if let Some((orig_physical, folded_to_indices)) = &self.original_physical_dup_check { - // Collect referenced column names, then fold them in one JVM crossing rather than one - // per Column node. Physical names were already folded once in `create()`. - let mut col_names: Vec = Vec::new(); - let _ = Arc::::clone(&expr).transform(|e| { - if let Some(col) = e.downcast_ref::() { - col_names.push(col.name().to_string()); - } - Ok(Transformed::no(e)) - }); - let col_refs: Vec<&str> = col_names.iter().map(|s| s.as_str()).collect(); - let col_folded = fold_names(&col_refs, false); - for (name, folded) in col_names.iter().zip(&col_folded) { - // Fields resolved by Parquet field id are selected by id before names are - // compared, so an id-resolved column must not trip the name-ambiguity check - // (mirrors Spark's `matchIdField`, which never raises the duplicate-field error). - if self - .id_resolved_logical_folded - .as_ref() - .is_some_and(|ids| ids.contains(folded)) - { - continue; + // Only the columns this expression references are checked, as Spark validates only + // the fields a read requests. Referenced names are collected once for both checks. + if self.original_physical_dup_check.is_some() || self.nested_mappings.is_some() { + let col_names = referenced_column_names(&expr); + + // In case-insensitive mode, a referenced column with more than one + // case-insensitive match in the physical schema is ambiguous. Names are folded in + // one JVM crossing; physical names were already folded once in `create()`. + if let Some((orig_physical, folded_to_indices)) = &self.original_physical_dup_check { + let col_refs: Vec<&str> = col_names.iter().map(|s| s.as_str()).collect(); + let col_folded = fold_names(&col_refs, false); + for (name, folded) in col_names.iter().zip(&col_folded) { + // Fields resolved by Parquet field id are selected by id before names are + // compared, so an id-resolved column must not trip the name-ambiguity check + // (mirrors Spark's `matchIdField`, which never raises the duplicate-field error). + if self + .id_resolved_logical_folded + .as_ref() + .is_some_and(|ids| ids.contains(folded)) + { + continue; + } + if let Some(err) = + check_column_duplicate(name, folded, folded_to_indices, orig_physical) + { + return Err(spark_error(err)); + } } - if let Some(err) = - check_column_duplicate(name, folded, folded_to_indices, orig_physical) - { - return Err(DataFusionError::External(Box::new(err))); + } + + // An ambiguity inside a referenced column's nested type surfaces here, so it is + // raised for every read of that column and not only when a cast is emitted. + if let Some(nested) = &self.nested_mappings { + for name in &col_names { + if let Some(Err(err)) = nested.get(name.as_str()) { + return Err(spark_error(err.clone())); + } } } } @@ -728,6 +882,7 @@ impl SparkPhysicalExprAdapter { let Some(physical_field) = self.physical_file_schema.fields().get(column.index()) else { return Ok(expr); }; + let mapping = self.field_mapping_for(column.name())?; Ok(Arc::new( CometCastColumnExpr::try_new( @@ -736,10 +891,24 @@ impl SparkPhysicalExprAdapter { Arc::new(logical_field.clone()), None, )? - .with_parquet_options(self.parquet_options.clone()), + .with_parquet_options(self.parquet_options.clone(), mapping), )) } + /// The once-per-file mapping for the logical field named `logical_name`, or a leaf + /// mapping for a field whose type holds no struct. + fn field_mapping_for(&self, logical_name: &str) -> DataFusionResult> { + match self + .nested_mappings + .as_ref() + .and_then(|mappings| mappings.get(logical_name)) + { + Some(Ok(mapping)) => Ok(Arc::clone(mapping)), + Some(Err(err)) => Err(spark_error(err.clone())), + None => Ok(Arc::new(FieldMapping::Leaf)), + } + } + /// Wrap ALL Column expressions that have type mismatches with CometCastColumnExpr. /// This is the fallback path when the default adapter fails (e.g., for complex /// nested type casts like List or Map). Uses `spark_parquet_convert` @@ -810,7 +979,10 @@ impl SparkPhysicalExprAdapter { Arc::clone(logical_field), None, )? - .with_parquet_options(self.parquet_options.clone()), + .with_parquet_options( + self.parquet_options.clone(), + self.field_mapping_for(logical_field.name())?, + ), ); return Ok(Transformed::yes(cast_expr)); } else if column.index() != phys_idx { @@ -857,7 +1029,10 @@ impl SparkPhysicalExprAdapter { Arc::clone(cast.target_field()), None, )? - .with_parquet_options(self.parquet_options.clone()), + .with_parquet_options( + self.parquet_options.clone(), + self.field_mapping_for(cast.target_field().name())?, + ), ); return Ok(Transformed::yes(comet_cast)); } @@ -1141,6 +1316,7 @@ impl SparkPhysicalExprAdapter { | (DataType::Timestamp(_, _), DataType::Timestamp(_, _)) | (DataType::Timestamp(_, _), DataType::Int64) ) { + let field_mapping = self.field_mapping_for(cast.target_field().name())?; let comet_cast: Arc = Arc::new( CometCastColumnExpr::try_new( child, @@ -1148,7 +1324,7 @@ impl SparkPhysicalExprAdapter { Arc::clone(cast.target_field()), None, )? - .with_parquet_options(self.parquet_options.clone()), + .with_parquet_options(self.parquet_options.clone(), field_mapping), ); return Ok(Transformed::yes(comet_cast)); } @@ -1354,7 +1530,7 @@ mod test { }; use arrow::array::UInt32Array; use arrow::array::{ - BinaryArray, Date32Array, Decimal128Array, Float32Array, Float64Array, Int32Array, + Array, BinaryArray, Date32Array, Decimal128Array, Float32Array, Float64Array, Int32Array, Int64Array, StringArray, TimestampMicrosecondArray, }; use arrow::datatypes::SchemaRef; @@ -1369,12 +1545,15 @@ mod test { use datafusion::physical_expr::expressions::Column; use datafusion::physical_expr::PhysicalExpr; use datafusion::physical_plan::ExecutionPlan; + use datafusion::scalar::ScalarValue; use datafusion_comet_spark_expr::test_common::file_util::get_temp_filename; use datafusion_comet_spark_expr::EvalMode; use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory; use futures::StreamExt; use parquet::arrow::ArrowWriter; use parquet::arrow::PARQUET_FIELD_ID_META_KEY; + use parquet::file::metadata::KeyValue; + use parquet::file::properties::WriterProperties; use parquet::variant::VariantType; use std::collections::HashMap; use std::fs::File; @@ -2035,6 +2214,117 @@ mod test { ); } + /// Two root columns share the exact name `d` in case-sensitive mode. Spark's root-level + /// `caseSensitiveParquetFieldMap` is the same `toMap` used for nested groups, so the later + /// file column wins; the rewritten column must bind to it, as the nested rule already does. + #[test] + fn duplicate_root_names_bind_to_the_last_column() -> Result<(), DataFusionError> { + use datafusion::physical_expr::expressions::Column; + let logical = Arc::new(Schema::new(vec![Field::new("d", DataType::Int32, false)])); + let physical = Arc::new(Schema::new(vec![ + Field::new("d", DataType::Int32, false), + Field::new("d", DataType::Int32, false), + ])); + let batch = RecordBatch::try_new( + Arc::clone(&physical), + vec![ + Arc::new(Int32Array::from(vec![1, 2, 3])), + Arc::new(Int32Array::from(vec![10, 20, 30])), + ], + )?; + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.case_sensitive = true; + let adapter = SparkPhysicalExprAdapterFactory::new(opts, None) + .create(Arc::clone(&logical), Arc::clone(&physical))?; + let rewritten = adapter.rewrite(Arc::new(Column::new("d", 0)))?; + let values = rewritten.evaluate(&batch)?.into_array(batch.num_rows())?; + let values = values.as_any().downcast_ref::().unwrap(); + assert_eq!(values.values(), &[10, 20, 30]); + Ok(()) + } + + /// Root duplicates under field-id matching: the id match selects the earlier `d`, so the + /// later `d` must not shadow it, and the later one is shielded from a name match instead. + #[test] + fn duplicate_root_names_defer_to_field_id_match() -> Result<(), DataFusionError> { + use datafusion::physical_expr::expressions::Column; + let logical = Arc::new(Schema::new(vec![ + Field::new("d", DataType::Int32, false).with_metadata(id_meta("1")) + ])); + let physical = Arc::new(Schema::new(vec![ + Field::new("d", DataType::Int32, false).with_metadata(id_meta("1")), + Field::new("d", DataType::Int32, false).with_metadata(id_meta("2")), + ])); + let batch = RecordBatch::try_new( + Arc::clone(&physical), + vec![ + Arc::new(Int32Array::from(vec![1, 2, 3])), + Arc::new(Int32Array::from(vec![10, 20, 30])), + ], + )?; + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.case_sensitive = true; + opts.use_field_id = true; + let adapter = SparkPhysicalExprAdapterFactory::new(opts, None) + .create(Arc::clone(&logical), Arc::clone(&physical))?; + let rewritten = adapter.rewrite(Arc::new(Column::new("d", 0)))?; + let values = rewritten.evaluate(&batch)?.into_array(batch.num_rows())?; + let values = values.as_any().downcast_ref::().unwrap(); + assert_eq!(values.values(), &[1, 2, 3]); + Ok(()) + } + + /// Scan-level companion to `duplicate_root_names_bind_to_the_last_column`: a file whose + /// root holds two `d` columns reads the later one for a requested `d`. + #[tokio::test] + async fn parquet_duplicate_root_names_read_the_last_column() -> Result<(), DataFusionError> { + let file_schema = Arc::new(Schema::new(vec![ + Field::new("d", DataType::Int32, false), + Field::new("d", DataType::Int32, false), + ])); + let batch = RecordBatch::try_new( + Arc::clone(&file_schema), + vec![ + Arc::new(Int32Array::from(vec![1, 2, 3])), + Arc::new(Int32Array::from(vec![10, 20, 30])), + ], + )?; + let filename = get_temp_filename(); + let filename = filename.as_path().as_os_str().to_str().unwrap().to_string(); + let file = File::create(&filename)?; + let mut writer = ArrowWriter::try_new(file, Arc::clone(&file_schema), None)?; + writer.write(&batch)?; + writer.close()?; + + let required_schema = Arc::new(Schema::new(vec![Field::new("d", DataType::Int32, false)])); + let mut spark_parquet_options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + spark_parquet_options.case_sensitive = true; + let expr_adapter_factory: Arc = Arc::new( + SparkPhysicalExprAdapterFactory::new(spark_parquet_options, None), + ); + let parquet_source = ParquetSource::new(required_schema); + let files = FileGroup::new(vec![PartitionedFile::from_path(filename)?]); + let file_scan_config = FileScanConfigBuilder::new( + ObjectStoreUrl::local_filesystem(), + Arc::new(parquet_source), + ) + .with_file_groups(vec![files]) + .with_expr_adapter(Some(expr_adapter_factory)) + .build(); + let parquet_exec = DataSourceExec::new(Arc::new(file_scan_config)); + let mut stream = parquet_exec.execute(0, Arc::new(TaskContext::default()))?; + let batch = stream.next().await.unwrap()?; + let values = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(values.values(), &[10, 20, 30]); + Ok(()) + } + /// Crate-level check of the case-insensitive remap. Under `cargo test` there is no attached /// JVM, so `fold_names` uses the ASCII fallback; ASCII casing still distinguishes match from /// no-match, so this documents that `remap_physical_schema` renames the physical field to the @@ -2093,6 +2383,182 @@ mod test { assert_eq!(remapped.field(0).name(), "a"); } + /// Build a nullable Int64 field carrying a Parquet field ID. + fn field_with_id(name: &str, id: i32) -> Field { + Field::new(name, DataType::Int64, true).with_metadata(id_meta(&id.to_string())) + } + + /// Write a Parquet file from `file_schema`/`columns`, then scan it with + /// `required_schema` through the Spark expression adapter and return the first batch. + async fn scan_with_adapter( + file_schema: SchemaRef, + columns: Vec>, + required_schema: SchemaRef, + spark_parquet_options: SparkParquetOptions, + ) -> Result { + scan_with_defaults( + file_schema, + columns, + required_schema, + spark_parquet_options, + None, + ) + .await + } + + /// `scan_with_adapter` with column defaults for fields missing from the file. + async fn scan_with_defaults( + file_schema: SchemaRef, + columns: Vec>, + required_schema: SchemaRef, + spark_parquet_options: SparkParquetOptions, + default_values: Option>, + ) -> Result { + let batch = RecordBatch::try_new(Arc::clone(&file_schema), columns).unwrap(); + + let filename = get_temp_filename(); + let filename = filename.as_path().as_os_str().to_str().unwrap().to_string(); + let file = File::create(&filename).unwrap(); + // Spark stamps key-value metadata into every file it writes. arrow-rs folds that into + // the file schema's metadata, so the file schema never compares equal to the requested + // schema and DataFusion always runs the expression adapter, as it does for real reads. + let props = WriterProperties::builder() + .set_key_value_metadata(Some(vec![KeyValue::new( + "org.apache.spark.version".to_string(), + "3.5.0".to_string(), + )])) + .build(); + let mut writer = ArrowWriter::try_new(file, file_schema, Some(props)).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + + let expr_adapter_factory: Arc = Arc::new( + SparkPhysicalExprAdapterFactory::new(spark_parquet_options, default_values), + ); + + let object_store_url = ObjectStoreUrl::local_filesystem(); + let parquet_source = ParquetSource::new(required_schema); + let files = FileGroup::new(vec![PartitionedFile::from_path(filename).unwrap()]); + let file_scan_config = + FileScanConfigBuilder::new(object_store_url, Arc::new(parquet_source)) + .with_file_groups(vec![files]) + .with_expr_adapter(Some(expr_adapter_factory)) + .build(); + + let parquet_exec = DataSourceExec::new(Arc::new(file_scan_config)); + let mut stream = parquet_exec + .execute(0, Arc::new(TaskContext::default())) + .unwrap(); + stream.next().await.unwrap() + } + + /// File: one column `κ` (U+03BA) with field ID 2 holding 7. Required: `Κ` (U+039A, + /// field ID 1) and ID-less `κ`; case-sensitive, field-ID reading on. Spark routes `Κ` + /// through `matchIdField` (no ID 1 in the file -> null-filled behind a faked REQUESTED + /// name) and resolves `κ` by exact name through `matchCaseSensitiveField`, reading the + /// real column: the result is (NULL, 7), never (NULL, NULL). + #[tokio::test] + async fn parquet_field_id_miss_null_fills_but_exact_name_sibling_still_reads() { + let file_schema = Arc::new(Schema::new(vec![field_with_id("\u{3BA}", 2)])); + let col = Arc::new(Int64Array::from(vec![7])) as Arc; + let required_schema = Arc::new(Schema::new(vec![ + field_with_id("\u{39A}", 1), + Field::new("\u{3BA}", DataType::Int64, true), + ])); + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.case_sensitive = true; + opts.use_field_id = true; + + let batch = scan_with_adapter(file_schema, vec![col], required_schema, opts) + .await + .unwrap(); + assert_eq!(batch.num_rows(), 1); + let capital = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert!(capital.is_null(0)); + let small = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + assert!(!small.is_null(0)); + assert_eq!(small.value(0), 7); + } + + /// Case-insensitive variant of the Kappa scenario. Spark's `matchCaseInsensitiveField` + /// resolves the ID-less requested `κ` through the `toLowerCase(Locale.ROOT)`-keyed map + /// of the file's fields, which holds the physical `κ`; the unmatched-ID requested `Κ` + /// is null-filled and never blocks that lookup. Same (NULL, 7) result as the + /// case-sensitive read. + #[tokio::test] + async fn parquet_field_id_miss_case_insensitive_sibling_still_reads() { + let file_schema = Arc::new(Schema::new(vec![field_with_id("\u{3BA}", 2)])); + let col = Arc::new(Int64Array::from(vec![7])) as Arc; + let required_schema = Arc::new(Schema::new(vec![ + field_with_id("\u{39A}", 1), + Field::new("\u{3BA}", DataType::Int64, true), + ])); + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.case_sensitive = false; + opts.use_field_id = true; + + let batch = scan_with_adapter(file_schema, vec![col], required_schema, opts) + .await + .unwrap(); + assert_eq!(batch.num_rows(), 1); + let capital = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert!(capital.is_null(0)); + let small = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + assert!(!small.is_null(0)); + assert_eq!(small.value(0), 7); + } + + /// File: a stray ID-less column literally named `A` = [10, 20] FIRST, then `a` with + /// field ID 1 = [1, 2]. Required: `A` with field ID 1, case-insensitive, field-ID + /// reading on. Spark's `matchIdField` resolves requested `A` to physical `a` by ID; the + /// stray `A` is never requested, and no case-insensitive duplicate error fires because + /// ID-routed requested fields never enter the name lookup. Expect [1, 2] -- neither the + /// stray column's data nor a spurious duplicate-field error. + #[tokio::test] + async fn parquet_field_id_match_beats_stray_column_with_requested_name() { + let file_schema = Arc::new(Schema::new(vec![ + Field::new("A", DataType::Int64, true), + field_with_id("a", 1), + ])); + let stray = Arc::new(Int64Array::from(vec![10, 20])) as Arc; + let matched = Arc::new(Int64Array::from(vec![1, 2])) as Arc; + let required_schema = Arc::new(Schema::new(vec![field_with_id("A", 1)])); + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.case_sensitive = false; + opts.use_field_id = true; + + let batch = scan_with_adapter(file_schema, vec![stray, matched], required_schema, opts) + .await + .unwrap(); + assert_eq!(batch.num_rows(), 2); + let values = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(values.value(0), 1); + assert_eq!(values.value(1), 2); + } + /// Field-id precedence in the case-insensitive duplicate check: an explicit `ω` (id 2) /// reading a file that holds both `ω` (id 2) and `Ω` (id 1) must resolve by id (Spark's /// `matchIdField` selects the id before ever comparing names) rather than raising a @@ -2373,12 +2839,13 @@ mod test { assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); } + /// #5707: an exact match must not hide a second case-insensitive match. The mapping is + /// resolved once per file in `rewrite`, so case-insensitive mode raises the duplicate + /// there, as Spark's `clipParquetSchema` does, and case-sensitive mode keeps the plain cast. #[test] fn structural_narrowing_requires_unambiguous_exact_match() -> Result<(), DataFusionError> { - use crate::parquet::cast_column::CometCastColumnExpr; use datafusion::physical_expr::expressions::CastExpr; - // #5707: an exact match must not hide a second case-insensitive match. for (upper, lower) in [("ID", "id"), ("CAFÉ", "café")] { let physical = struct_type(vec![(upper, DataType::Int64), (lower, DataType::Int64)]); let target = struct_type(vec![(lower, DataType::Int64)]); @@ -2398,11 +2865,15 @@ 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::().is_some()); + assert!(rewritten?.downcast_ref::().is_some()); } else { - assert!(rewritten.downcast_ref::().is_some()); + let err = rewritten.expect_err("ambiguous case-insensitive match"); + assert!( + err.to_string().contains("duplicate field"), + "{physical:?} -> {target:?}: {err}" + ); } } } @@ -2643,4 +3114,88 @@ mod test { let target = struct_type(vec![("id", DataType::Int64)]); assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); } + + /// A requested column named like the shield's placeholder must still receive its + /// configured default. File: `k` (id 2). Required: `k` (id 1) and an id-less + /// `__COMET_UNMATCHED_FIELD_ID_1` with default 7, case-insensitive, field-id reading on. + /// The file's `k` is not the id match for requested `k`, so it is hidden behind a + /// placeholder name; that placeholder must not fold onto the requested column, or the + /// missing-column check treats it as present and the default is lost. + #[tokio::test] + async fn parquet_shield_placeholder_never_folds_onto_requested_column() { + let file_schema = Arc::new(Schema::new(vec![field_with_id("k", 2)])); + let col = Arc::new(Int64Array::from(vec![1])) as Arc; + let required_schema = Arc::new(Schema::new(vec![ + field_with_id("k", 1), + Field::new("__COMET_UNMATCHED_FIELD_ID_1", DataType::Int64, true), + ])); + let defaults = HashMap::from([( + Column::new("__COMET_UNMATCHED_FIELD_ID_1", 1), + ScalarValue::Int64(Some(7)), + )]); + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.case_sensitive = false; + opts.use_field_id = true; + + let batch = scan_with_defaults( + file_schema, + vec![col], + required_schema, + opts, + Some(defaults), + ) + .await + .unwrap(); + assert_eq!(batch.num_rows(), 1); + let k = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert!( + k.is_null(0), + "requested k (id 1) has no id match in the file" + ); + let defaulted = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + assert!(!defaulted.is_null(0), "configured default must apply"); + assert_eq!(defaulted.value(0), 7); + } + + /// File and requested schema are identical: `s` holding `x` and `y` that both carry + /// field id 1. No column needs conversion, so no cast is ever emitted, yet Spark's + /// `clipParquetSchema` rejects the read because requested id 1 resolves to two file + /// fields. The validation must therefore run when the file schema is mapped, not + /// only inside a cast. + #[tokio::test] + async fn parquet_duplicate_struct_field_id_rejected_without_cast() { + let child_fields = + arrow::datatypes::Fields::from(vec![field_with_id("x", 1), field_with_id("y", 1)]); + let struct_field = Field::new("s", DataType::Struct(child_fields.clone()), true) + .with_metadata(id_meta("10")); + let file_schema = Arc::new(Schema::new(vec![struct_field.clone()])); + let required_schema = Arc::new(Schema::new(vec![struct_field])); + let children: Vec> = vec![ + Arc::new(Int64Array::from(vec![42])), + Arc::new(Int64Array::from(vec![43])), + ]; + let col = Arc::new(arrow::array::StructArray::new(child_fields, children, None)) + as Arc; + + let mut opts = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + opts.use_field_id = true; + + let err = scan_with_adapter(file_schema, vec![col], required_schema, opts) + .await + .expect_err("requested id 1 matches two file fields and must error"); + let msg = err.to_string(); + assert!( + msg.contains("_LEGACY_ERROR_TEMP_2094") && msg.contains("id=1"), + "expected duplicate field id error, got: {msg}" + ); + } } diff --git a/native/jni-bridge/src/errors.rs b/native/jni-bridge/src/errors.rs index 3e72f6e8048..745513d78c9 100644 --- a/native/jni-bridge/src/errors.rs +++ b/native/jni-bridge/src/errors.rs @@ -573,7 +573,12 @@ fn throw_exception(env: &mut Env, error: &CometError, backtrace: Option) // FAILED_READ_FILE / FileNotFound via the structured SparkError channel. Anything else // falls back to generic handling. CometError::DataFusion { msg: _, source } => { - if let Some(spark_error) = try_classify_file_read_error(source) { + if let Some(json_message) = spark_error_json_in_chain(source) { + env.throw_new( + jni::jni_str!("org/apache/comet/exceptions/CometQueryExecutionException"), + JNIString::new(json_message), + ) + } else if let Some(spark_error) = try_classify_file_read_error(source) { throw_spark_error_as_json(env, &spark_error) } else { throw_generic_exception(env, error, backtrace) @@ -603,6 +608,24 @@ fn typed_jvm_exception(error: &(dyn std::error::Error + 'static)) -> Option Option { + let mut cause = Some(error); + while let Some(error) = cause { + if let Some(spark_error) = error.downcast_ref::() { + return Some(spark_error.to_json()); + } + if let Some(spark_error) = error.downcast_ref::() { + return Some(spark_error.to_json()); + } + cause = error.source(); + } + None +} + /// Generic fallback throw for an error that isn't a structured `SparkError`. Recognises a /// file-not-found arriving through non-typed wrapping paths and duplicate-field errors; otherwise /// throws the error's natural JVM exception (with the captured backtrace when available). @@ -1370,6 +1393,32 @@ mod tests { } } + /// A `SparkError` the reader factory raises from its metadata fetch is wrapped by the parquet + /// reader and DataFusion's opener, yet must surface with its own Spark error class. + #[test] + fn spark_error_inside_parquet_error_keeps_its_class() { + let spark_error = SparkError::DuplicateFieldByFieldId { + required_id: 1, + matched_fields: "x, y".to_string(), + }; + let expected = spark_error.to_json(); + let e = DataFusionError::Context( + "opening file".to_string(), + Box::new(DataFusionError::ParquetError(Box::new( + parquet::errors::ParquetError::External(Box::new(spark_error)), + ))), + ); + assert_eq!(spark_error_json_in_chain(&e), Some(expected)); + } + + #[test] + fn plain_parquet_error_has_no_spark_error_in_chain() { + let e = DataFusionError::ParquetError(Box::new(parquet::errors::ParquetError::General( + "corrupt footer".to_string(), + ))); + assert_eq!(spark_error_json_in_chain(&e), None); + } + #[test] fn classify_parquet_error_is_file_read() { let e = DataFusionError::ParquetError(Box::new(parquet::errors::ParquetError::General( 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 2f434aac92c..c61c80c7ec6 100644 --- a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala +++ b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala @@ -31,9 +31,13 @@ import org.scalactic.source.Position import org.scalatest.Tag import org.apache.arrow.vector.types.pojo.{ArrowType, DictionaryEncoding, Field => ArrowField, FieldType, Schema => ArrowSchema} +import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.Path +import org.apache.parquet.example.data.Group import org.apache.parquet.example.data.simple.SimpleGroup -import org.apache.parquet.hadoop.example.ExampleParquetWriter +import org.apache.parquet.hadoop.ParquetWriter +import org.apache.parquet.hadoop.api.WriteSupport +import org.apache.parquet.hadoop.example.{ExampleParquetWriter, GroupWriteSupport} import org.apache.parquet.io.api.Binary import org.apache.parquet.schema.MessageTypeParser import org.apache.spark.SparkException @@ -1810,6 +1814,19 @@ abstract class ParquetReadSuite extends CometTestBase { } } + // parquet-mr's example writer stamps `writer.model.name` into the footer. This builder + // leaves the key-value metadata empty, so the file schema arrow-rs derives compares equal + // to a matching requested schema. + private class BareGroupWriterBuilder(path: Path) + extends ParquetWriter.Builder[Group, BareGroupWriterBuilder](path) { + override protected def self(): BareGroupWriterBuilder = this + + override protected def getWriteSupport(conf: Configuration): WriteSupport[Group] = + new GroupWriteSupport() { + override def getName: String = null + } + } + private def withId(id: Int) = new MetadataBuilder().putLong(ParquetUtils.FIELD_ID_METADATA_KEY, id).build() @@ -2024,6 +2041,98 @@ abstract class ParquetReadSuite extends CometTestBase { } } + // Spark's `clipParquetSchema` runs `matchIdField` at every nesting level while clipping the + // file schema, so a struct child id duplicated in the file is rejected even when the read + // schema is identical to the file schema and no column needs any conversion. + test("duplicate field id inside a struct is rejected without a cast") { + withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") { + withTempPath { dir => + val schema = + new StructType() + .add( + "s", + new StructType() + .add("x", LongType, true, withId(1)) + .add("y", LongType, true, withId(1)), + true, + withId(2)) + + val writeData = Seq(Row(Row(42L, 43L))) + spark + .createDataFrame(spark.sparkContext.parallelize(writeData), schema) + .write + .mode("overwrite") + .parquet(dir.getCanonicalPath) + + val cause = intercept[SparkException] { + spark.read.schema(schema).parquet(dir.getCanonicalPath).collect() + }.getCause + assert( + cause.isInstanceOf[RuntimeException] && + cause.getMessage.contains("Found duplicate field(s)")) + } + } + } + + // DataFusion's opener hands a file to the expression adapter only when a predicate is pushed + // or the file schema differs from the requested one. Spark-written files always carry + // key-value metadata that arrow-rs folds into the file schema, so they always differ; a + // file with none is read positionally unless the reader factory validates the ids (#5801). + test("duplicate field id inside a struct is rejected without key-value metadata") { + withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") { + withTempDir { dir => + val path = new Path(dir.toURI.toString, "part-r-0.parquet") + val schema = MessageTypeParser.parseMessageType(""" + |message root { + | optional group s = 2 { + | optional int64 x = 1; + | optional int64 y = 1; + | } + |} + |""".stripMargin) + val conf = spark.sessionState.newHadoopConf() + GroupWriteSupport.setSchema(schema, conf) + val writer = new BareGroupWriterBuilder(path).withConf(conf).build() + val record = new SimpleGroup(schema) + val nested = record.addGroup(0) + nested.add(0, 42L) + nested.add(1, 43L) + writer.write(record) + writer.close() + + val footerReader = org.apache.parquet.hadoop.ParquetFileReader + .open(org.apache.parquet.hadoop.util.HadoopInputFile.fromPath(path, conf)) + try { + assert(footerReader.getFooter.getFileMetaData.getKeyValueMetaData.isEmpty) + } finally { + footerReader.close() + } + + val readSchema = + new StructType() + .add( + "s", + new StructType() + .add("x", LongType, true, withId(1)) + .add("y", LongType, true, withId(1)), + true, + withId(2)) + val df = spark.read.schema(readSchema).parquet(path.toString) + // Spark's own reader raises the same error, so make sure the native scan is what runs. + val scans = stripAQEPlan(df.queryExecution.executedPlan).collect { + case scan: CometNativeScanExec => scan + } + assert(scans.nonEmpty, "expected CometNativeScanExec in the plan") + val cause = intercept[SparkException] { + df.collect() + }.getCause + assert( + cause.isInstanceOf[RuntimeException] && + cause.getMessage.contains("Found duplicate field(s)")) + } + } + } + // Verbatim port of Spark `ParquetFieldIdIOSuite.test("read parquet file without ids")`, // for the same reason as the duplicate-id test above. test("read parquet file without ids") {