From 617dce982f07892871c24d5fa7d298b3e3afe9bb Mon Sep 17 00:00:00 2001 From: Swanand Mulay <73115739+swanandx@users.noreply.github.com> Date: Mon, 27 Jul 2026 21:25:04 +0530 Subject: [PATCH 1/8] use v0.32.3-feldera-unity-fix branch Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com> --- Cargo.lock | 14 +++++++------- Cargo.toml | 18 +++++++++++------- 2 files changed, 18 insertions(+), 14 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 264e19df8cf..2d29abc155e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4360,7 +4360,7 @@ checksum = "26bf8fc351c5ed29b5c2f0cbbac1b209b74f60ecd62e675a998df72c49af5204" [[package]] name = "deltalake" version = "0.32.3" -source = "git+https://github.com/feldera/delta-rs.git?rev=297f2dd146f9947795f8b77f25db8da9daff2ade#297f2dd146f9947795f8b77f25db8da9daff2ade" +source = "git+https://github.com/feldera/delta-rs.git?rev=4913d02fabc643bd898fcb72d5d2afb73c77ac29#4913d02fabc643bd898fcb72d5d2afb73c77ac29" dependencies = [ "buoyant_kernel", "ctor 0.10.1", @@ -4374,7 +4374,7 @@ dependencies = [ [[package]] name = "deltalake-aws" version = "0.15.0" -source = "git+https://github.com/feldera/delta-rs.git?rev=297f2dd146f9947795f8b77f25db8da9daff2ade#297f2dd146f9947795f8b77f25db8da9daff2ade" +source = "git+https://github.com/feldera/delta-rs.git?rev=4913d02fabc643bd898fcb72d5d2afb73c77ac29#4913d02fabc643bd898fcb72d5d2afb73c77ac29" dependencies = [ "async-trait", "aws-config", @@ -4400,7 +4400,7 @@ dependencies = [ [[package]] name = "deltalake-azure" version = "0.15.0" -source = "git+https://github.com/feldera/delta-rs.git?rev=297f2dd146f9947795f8b77f25db8da9daff2ade#297f2dd146f9947795f8b77f25db8da9daff2ade" +source = "git+https://github.com/feldera/delta-rs.git?rev=4913d02fabc643bd898fcb72d5d2afb73c77ac29#4913d02fabc643bd898fcb72d5d2afb73c77ac29" dependencies = [ "bytes", "deltalake-core", @@ -4413,7 +4413,7 @@ dependencies = [ [[package]] name = "deltalake-catalog-unity" version = "0.16.0" -source = "git+https://github.com/feldera/delta-rs.git?rev=297f2dd146f9947795f8b77f25db8da9daff2ade#297f2dd146f9947795f8b77f25db8da9daff2ade" +source = "git+https://github.com/feldera/delta-rs.git?rev=4913d02fabc643bd898fcb72d5d2afb73c77ac29#4913d02fabc643bd898fcb72d5d2afb73c77ac29" dependencies = [ "async-trait", "chrono", @@ -4440,7 +4440,7 @@ dependencies = [ [[package]] name = "deltalake-core" version = "0.32.3" -source = "git+https://github.com/feldera/delta-rs.git?rev=297f2dd146f9947795f8b77f25db8da9daff2ade#297f2dd146f9947795f8b77f25db8da9daff2ade" +source = "git+https://github.com/feldera/delta-rs.git?rev=4913d02fabc643bd898fcb72d5d2afb73c77ac29#4913d02fabc643bd898fcb72d5d2afb73c77ac29" dependencies = [ "arrow", "arrow-arith", @@ -4494,7 +4494,7 @@ dependencies = [ [[package]] name = "deltalake-derive" version = "1.0.0" -source = "git+https://github.com/feldera/delta-rs.git?rev=297f2dd146f9947795f8b77f25db8da9daff2ade#297f2dd146f9947795f8b77f25db8da9daff2ade" +source = "git+https://github.com/feldera/delta-rs.git?rev=4913d02fabc643bd898fcb72d5d2afb73c77ac29#4913d02fabc643bd898fcb72d5d2afb73c77ac29" dependencies = [ "convert_case 0.9.0", "itertools 0.14.0", @@ -4506,7 +4506,7 @@ dependencies = [ [[package]] name = "deltalake-gcp" version = "0.16.0" -source = "git+https://github.com/feldera/delta-rs.git?rev=297f2dd146f9947795f8b77f25db8da9daff2ade#297f2dd146f9947795f8b77f25db8da9daff2ade" +source = "git+https://github.com/feldera/delta-rs.git?rev=4913d02fabc643bd898fcb72d5d2afb73c77ac29#4913d02fabc643bd898fcb72d5d2afb73c77ac29" dependencies = [ "async-trait", "bytes", diff --git a/Cargo.toml b/Cargo.toml index 66df336a832..401b391fb92 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -113,20 +113,24 @@ dbsp = { path = "crates/dbsp", version = "0.326.0" } datafusion-functions-json = "0.53.1" dbsp_nexmark = { path = "crates/nexmark" } deadpool-postgres = "0.14.1" -# Feldera fork of delta-rs: upstream tag `rust-v0.32.3` + 3 patches, on branch -# `v0.32.3-feldera-fix`. When bumping the pinned revision, preserve these patches: +# Feldera fork of delta-rs: upstream tag `rust-v0.32.3` + 4 patches, on branch +# `v0.32.3-feldera-unity-fix`. When bumping the pinned revision, preserve these +# patches: # - 7f8bb445 "Disable unsupported feature check." — relax the writer-side # protocol-feature gate so we can write tables produced by newer clients. # - 5b35c098 "delta-rs: round-robin pre-split unpartitioned scans into # target_partitions FileGroups" — improves snapshot read parallelism for # unpartitioned tables; pairs with the DataFusion knobs we tune in adapters. -# - 9a58fbef "delta-rs: resolve column-mapped Uniform/Iceberg columns by -# Parquet field-id" — fixes `col- is missing from the physical schema` +# - 297f2dd1 "delta-rs: resolve column-mapped Uniform/Iceberg columns by +# Parquet field id." — fixes `col- is missing from the physical schema` # when reading UC-Uniform-over-Iceberg tables (columnMapping.mode=id). #6557. +# - 4913d02f "catalog-unity: refresh Unity Catalog S3 credentials before they +# expire" — the vended temporary credentials last an hour, so a long-running +# `uc://` pipeline failed with `ExpiredToken` without this. # Diff vs. upstream: -# https://github.com/delta-io/delta-rs/compare/rust-v0.32.3...feldera:delta-rs:v0.32.3-feldera-fix -deltalake = { git = "https://github.com/feldera/delta-rs.git", rev = "297f2dd146f9947795f8b77f25db8da9daff2ade" } -deltalake-catalog-unity = { git = "https://github.com/feldera/delta-rs.git", rev = "297f2dd146f9947795f8b77f25db8da9daff2ade" } +# https://github.com/delta-io/delta-rs/compare/rust-v0.32.3...feldera:delta-rs:v0.32.3-feldera-unity-fix +deltalake = { git = "https://github.com/feldera/delta-rs.git", rev = "4913d02fabc643bd898fcb72d5d2afb73c77ac29" } +deltalake-catalog-unity = { git = "https://github.com/feldera/delta-rs.git", rev = "4913d02fabc643bd898fcb72d5d2afb73c77ac29" } delta_kernel = { package = "buoyant_kernel", version = "0.22" } derive_more = { version = "1.0.0" } From 0368fd1038fe15dc20ab58fcf75c5b359488cf2e Mon Sep 17 00:00:00 2001 From: Swanand Mulay <73115739+swanandx@users.noreply.github.com> Date: Mon, 27 Jul 2026 23:56:41 +0530 Subject: [PATCH 2/8] connectors: delta fix uc:// follow reads (No suitable object store found) root_url() for a uc:// table has no trailing slash, so joining Add.path corrupted the object-store authority. Normalize the root before joining. Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com> --- .../src/integrated/delta_table/input.rs | 40 ++++++++++++------- 1 file changed, 25 insertions(+), 15 deletions(-) diff --git a/crates/adapters/src/integrated/delta_table/input.rs b/crates/adapters/src/integrated/delta_table/input.rs index 95311ee90c0..ee1cbd35e87 100644 --- a/crates/adapters/src/integrated/delta_table/input.rs +++ b/crates/adapters/src/integrated/delta_table/input.rs @@ -134,6 +134,19 @@ fn is_active_dv(dv: &DeletionVectorDescriptor) -> bool { dv.cardinality > 0 } +/// The table root as a base URL guaranteed to end in `/`, so string-joining a +/// relative `Add.path` yields a well-formed URL. delta-rs normalizes the location +/// to a trailing slash only when it has a path, so a root with none (`uc://cat.db.tbl`, +/// or a table at a bucket root) would otherwise collapse into the joined file name. +fn table_root_base(table: &DeltaTable) -> String { + let root = table.log_store().root_url().to_string(); + if root.ends_with('/') { + root + } else { + format!("{root}/") + } +} + /// Build the `DataFrame` that streams a CDC transaction to the circuit. /// /// Equivalent (in SQL) to: @@ -3201,11 +3214,11 @@ impl DeltaTableInputEndpointInner { /// [`Self::project_cdc_columns`] keeps), so they line up by position for the /// `EXCEPT ALL` in `build_cdc_dataframe` and never decode unused columns. /// - /// Plain files use the table's `root_url()`, not the synthetic - /// `delta-rs://` URL from `object_store_url()`. The latter folds the table - /// path into the URL host (slashes become dashes), which routes DataFusion's - /// object store but is malformed once joined with `Add.path`. - /// `start_input_endpoint` registers the store under `root_url()`. + /// Plain files are addressed under the table root (see [`table_root_base`]), + /// not the synthetic `delta-rs://` URL from `object_store_url()`. The latter + /// folds the table path into the URL host (slashes become dashes), which + /// routes DataFusion's object store but is malformed once joined with + /// `Add.path`. `start_input_endpoint` registers the store under `root_url()`. async fn cdc_side_dataframe( &self, table: &DeltaTable, @@ -3226,12 +3239,8 @@ impl DeltaTableInputEndpointInner { let mut dfs: Vec = Vec::new(); if !plain.is_empty() { - let log_store = table.log_store(); - let root_url = log_store.root_url(); - let files = plain - .iter() - .map(|p| format!("{}{}", root_url.as_str(), p)) - .collect(); + let base = table_root_base(table); + let files = plain.iter().map(|p| format!("{base}{p}")).collect(); let listing_table = Arc::new(self.create_parquet_table(files, description).await?); let df = self.datafusion.read_table(listing_table).map_err(|e| { anyhow!("internal error processing {description}; {REPORT_ERROR}; error reading Parquet files: {e}") @@ -3348,10 +3357,11 @@ impl DeltaTableInputEndpointInner { }) .await? } else { - // Address files via the table's real `root_url()` (e.g. `file:///...` - // or `s3://bucket/prefix/`). See `cdc_side_dataframe` for why we - // don't use `object_store_url()` here. - let full_path = format!("{}{}", table.log_store().root_url().as_str(), path); + // Address files under the table root (e.g. `file:///...`, + // `s3://bucket/prefix/`, or `uc://cat.db.tbl/`). See + // `cdc_side_dataframe` for why we don't use `object_store_url()`, + // and `table_root_base` for why the root is slash-normalized. + let full_path = format!("{}{}", table_root_base(table), path); Arc::new( self.create_parquet_table(vec![full_path], &description) .await?, From 304a78ea00db2a0ef3342860c6184fbd7255a887 Mon Sep 17 00:00:00 2001 From: Swanand Mulay <73115739+swanandx@users.noreply.github.com> Date: Tue, 28 Jul 2026 23:48:44 +0530 Subject: [PATCH 3/8] connectors: resolve nested column-mapped fields in follow/CDC reads The follow/CDC read builds its own Parquet ListingTable and renamed column-mapped fields to their physical (col-) names only at the top level. Nested struct children kept their logical names, so reading a mode=name table with a nested struct failed the Parquet struct cast ("Cannot cast struct ... no field name overlap") and dropped the rows, while a snapshot read (which delta-rs resolves) worked. Recurse the physical read schema into nested struct fields, and restore their logical names per batch in execute_df_inner (top-level names are still restored on the DataFrame by project_physical_to_logical). The per-batch relabel is keyed on a nested-only map, so flat and mapping-off tables are untouched. Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com> --- .../src/integrated/delta_table/input.rs | 407 +++++++++++++++++- .../fixtures/column_mapping_nested.py | 73 ++++ .../test_delta_input_column_mapping_nested.py | 100 +++++ 3 files changed, 562 insertions(+), 18 deletions(-) create mode 100644 python/tests/platform/fixtures/column_mapping_nested.py create mode 100644 python/tests/platform/test_delta_input_column_mapping_nested.py diff --git a/crates/adapters/src/integrated/delta_table/input.rs b/crates/adapters/src/integrated/delta_table/input.rs index ee1cbd35e87..ea1bdcd5f1b 100644 --- a/crates/adapters/src/integrated/delta_table/input.rs +++ b/crates/adapters/src/integrated/delta_table/input.rs @@ -7,8 +7,10 @@ use crate::integrated::delta_table::{delta_input_serde_config, register_storage_ use crate::transport::{InputEndpoint, InputQueue, InputReaderCommand, IntegratedInputEndpoint}; use crate::{ControllerError, InputConsumer, InputReader, PipelineState}; use anyhow::{Error as AnyError, Result as AnyResult, anyhow, bail}; -use arrow::array::BooleanArray; -use arrow::datatypes::{FieldRef, Schema as ArrowSchema, SchemaRef}; +use arrow::array::{Array, ArrayData, ArrayRef, BooleanArray, make_array}; +use arrow::datatypes::{ + DataType as ArrowDataType, Field as ArrowField, FieldRef, Schema as ArrowSchema, SchemaRef, +}; use chrono::{DateTime, Utc}; use datafusion::catalog::TableProvider; use datafusion::common::DataFusionError; @@ -147,6 +149,158 @@ fn table_root_base(table: &DeltaTable) -> String { } } +/// A field's physical (on-disk) name under column mapping, or its logical name +/// when unmapped. Delta stamps it into the Arrow field metadata at every level. +fn physical_name(field: &ArrowField) -> String { + field + .metadata() + .get(ColumnMetadataKey::ColumnMappingPhysicalName.as_ref()) + .cloned() + .unwrap_or_else(|| field.name().clone()) +} + +/// Returns a copy of `field` whose own name, and every nested field name (struct +/// children and list/map element fields, at any depth), is `rename`d. Nullability +/// and metadata carry over unchanged. +fn rename_fields(field: &FieldRef, rename: &dyn Fn(&ArrowField) -> String) -> FieldRef { + Arc::new( + ArrowField::new( + rename(field.as_ref()), + rename_nested_fields(field.data_type(), rename), + field.is_nullable(), + ) + .with_metadata(field.metadata().clone()), + ) +} + +/// Recurse [`rename_fields`] into every field a container type holds. Scalar +/// types are returned unchanged. +fn rename_nested_fields( + data_type: &ArrowDataType, + rename: &dyn Fn(&ArrowField) -> String, +) -> ArrowDataType { + let renamed = |field: &FieldRef| rename_fields(field, rename); + match data_type { + ArrowDataType::Struct(fields) => { + ArrowDataType::Struct(fields.iter().map(renamed).collect()) + } + ArrowDataType::List(field) => ArrowDataType::List(renamed(field)), + ArrowDataType::LargeList(field) => ArrowDataType::LargeList(renamed(field)), + ArrowDataType::FixedSizeList(field, len) => { + ArrowDataType::FixedSizeList(renamed(field), *len) + } + ArrowDataType::Map(field, sorted) => ArrowDataType::Map(renamed(field), *sorted), + other => other.clone(), + } +} + +/// Returns a copy of `field` named as it appears on disk at every level: each +/// column-mapped name becomes its physical name, unmapped names carry over. +fn field_to_physical(field: &FieldRef) -> FieldRef { + rename_fields(field, &physical_name) +} + +/// Maps each nested column-mapped field's physical name to its logical name, +/// descending through struct children and list/map element fields at any depth. +/// Top-level fields are excluded. Empty unless the table nests column-mapped +/// fields. +fn nested_physical_to_logical(schema: &ArrowSchema) -> HashMap { + fn collect(data_type: &ArrowDataType, map: &mut HashMap) { + for field in child_fields(data_type) { + let physical = physical_name(field); + if physical != *field.name() { + map.insert(physical, field.name().clone()); + } + collect(field.data_type(), map); + } + } + let mut map = HashMap::new(); + for field in schema.fields() { + collect(field.data_type(), &mut map); + } + map +} + +/// The fields a container type holds directly: a struct's children, or the sole +/// element field of a list/map. Scalar types hold none. +fn child_fields(data_type: &ArrowDataType) -> Vec<&FieldRef> { + match data_type { + ArrowDataType::Struct(fields) => fields.iter().collect(), + ArrowDataType::List(field) + | ArrowDataType::LargeList(field) + | ArrowDataType::FixedSizeList(field, _) + | ArrowDataType::Map(field, _) => vec![field], + _ => vec![], + } +} + +/// Rebuild `array` with every struct/list/map field name substituted through +/// `map`, at any nesting depth, reusing the underlying buffers. Only field-name +/// metadata changes; the physical layout is untouched. Names absent from `map` +/// carry over unchanged. +fn relabel_array(array: &ArrayRef, map: &HashMap) -> AnyResult { + Ok(make_array(relabel_array_data(array.to_data(), map)?)) +} + +/// Recursive core of [`relabel_array`], operating on the raw [`ArrayData`] tree. +fn relabel_array_data(data: ArrayData, map: &HashMap) -> AnyResult { + let relabeled_type = relabel_data_type(data.data_type(), map); + let children: Vec = data + .child_data() + .iter() + .map(|child| relabel_array_data(child.clone(), map)) + .collect::>()?; + // Relabeling only renames fields; buffers, offsets, lengths, and null bitmaps + // carry over untouched, so the built data is structurally identical. `build` + // only errors on a real layout mismatch, which would be a bug here. + data.into_builder() + .data_type(relabeled_type) + .child_data(children) + .build() + .map_err(|e| anyhow!("relabeling column-mapped field names failed: {e}")) +} + +/// Substitute nested field names in `data_type` through `map`. Names absent from +/// `map`, and scalar types, are left unchanged. +fn relabel_data_type(data_type: &ArrowDataType, map: &HashMap) -> ArrowDataType { + rename_nested_fields(data_type, &|field| { + map.get(field.name()) + .cloned() + .unwrap_or_else(|| field.name().clone()) + }) +} + +/// Translate a batch's nested field names physical-to-logical. Top-level names +/// are left as-is (already logical); only names nested inside a struct, list, or +/// map are rewritten. +fn relabel_nested_columns( + batch: &RecordBatch, + map: &HashMap, +) -> AnyResult { + let columns: Vec = batch + .columns() + .iter() + .map(|c| relabel_array(c, map)) + .collect::>()?; + let fields: Vec = batch + .schema() + .fields() + .iter() + .zip(&columns) + .map(|(f, c)| { + Arc::new( + ArrowField::new(f.name(), c.data_type().clone(), f.is_nullable()) + .with_metadata(f.metadata().clone()), + ) + }) + .collect(); + RecordBatch::try_new( + Arc::new(ArrowSchema::new(fields).with_metadata(batch.schema().metadata().clone())), + columns, + ) + .map_err(|e| anyhow!("relabeling column-mapped field names failed: {e}")) +} + /// Build the `DataFrame` that streams a CDC transaction to the circuit. /// /// Equivalent (in SQL) to: @@ -2513,6 +2667,18 @@ impl DeltaTableInputEndpointInner { self.consumer .update_connector_health(ConnectorHealth::healthy()); + // Nested struct fields are read under physical names; restore their + // logical names per batch (top-level names are already logical). Empty, + // a no-op, unless the table nests column-mapped fields. + let nested_map = nested_physical_to_logical( + self.schema_snapshot() + .snapshot() + .map_err(|e| format!("error accessing Delta table snapshot: {e}"))? + .snapshot() + .arrow_schema() + .as_ref(), + ); + let mut num_batches = 0; let mut total_records = 0usize; @@ -2597,7 +2763,11 @@ impl DeltaTableInputEndpointInner { )); } }; - // info!("schema: {}", batch.schema()); + let batch = if nested_map.is_empty() { + batch + } else { + relabel_nested_columns(&batch, &nested_map).map_err(|e| e.to_string())? + }; num_batches += 1; total_records += batch.num_rows(); @@ -3051,12 +3221,13 @@ impl DeltaTableInputEndpointInner { .collect()) } - /// Returns the Arrow schema to use when reading the raw data files, named as - /// they appear on disk: the table's logical schema restricted to the columns - /// `keep` accepts (in schema order, so unions with another read side line - /// up), with each column-mapped field renamed to its physical (`col-`) - /// name so DataFusion's by-name matching finds them. The same as the kept - /// logical schema when column mapping is off. + /// Arrow schema for reading the raw data files, named as they appear on disk: + /// the logical schema restricted to columns `keep` accepts (in schema order, + /// so read sides line up), with each column-mapped field renamed to its + /// physical (`col-`) name so DataFusion matches by name. It recurses + /// into nested struct, list, and map fields. Logical names are restored afterwards: + /// top level in `project_physical_to_logical`, nested in `relabel_nested_columns`. + /// The kept logical schema when column mapping is off. fn physical_read_schema(&self, keep: impl Fn(&str) -> bool) -> AnyResult { let schema_table = self.schema_snapshot(); let logical = schema_table @@ -3064,19 +3235,11 @@ impl DeltaTableInputEndpointInner { .map_err(|e| anyhow!("error accessing Delta table snapshot: {e}"))? .snapshot() .arrow_schema(); - let pairs = self.column_mapping()?; - let to_physical: HashMap<&str, &str> = pairs - .iter() - .map(|(l, p)| (l.as_str(), p.as_str())) - .collect(); let fields: Vec = logical .fields() .iter() .filter(|f| keep(f.name())) - .map(|f| match to_physical.get(f.name().as_str()) { - Some(physical) => Arc::new(f.as_ref().clone().with_name(*physical)), - None => Arc::clone(f), - }) + .map(field_to_physical) .collect(); Ok(Arc::new( ArrowSchema::new(fields).with_metadata(logical.metadata().clone()), @@ -3606,3 +3769,211 @@ mod is_skippable_tests { )); } } + +#[cfg(test)] +mod column_mapping_tests { + use super::{field_to_physical, nested_physical_to_logical, relabel_nested_columns}; + use arrow::array::{ArrayRef, ListArray, RecordBatch, StringArray, StructArray}; + use arrow::buffer::OffsetBuffer; + use arrow::datatypes::{DataType, Field, Fields, Schema}; + use std::collections::HashMap; + use std::sync::Arc; + + /// A column-mapped field: logical `name`, physical name in its metadata. + fn mapped(name: &str, data_type: DataType, physical: &str) -> Field { + Field::new(name, data_type, true).with_metadata(HashMap::from([( + "delta.columnMapping.physicalName".to_string(), + physical.to_string(), + )])) + } + + /// A `struct<..>` data type from mapped fields. + fn struct_of(fields: Vec) -> DataType { + DataType::Struct(Fields::from(fields)) + } + + /// A `list` data type. The `element` field itself is not column + /// mapped, matching how Delta stores list elements. + fn list_of(element: DataType) -> DataType { + DataType::List(Arc::new(Field::new("element", element, true))) + } + + /// The children of a struct-typed field, or panic. + fn struct_children(field: &Field) -> &Fields { + match field.data_type() { + DataType::Struct(children) => children, + other => panic!("expected struct, got {other:?}"), + } + } + + /// The element field of a list-typed field, or panic. + fn list_element(field: &Field) -> &Field { + match field.data_type() { + DataType::List(element) => element, + other => panic!("expected list, got {other:?}"), + } + } + + // The read side: nested struct children must be renamed to physical names, + // else the Parquet read fails the struct cast. + #[test] + fn read_schema_renames_nested_fields() { + let after = DataType::Struct(Fields::from(vec![ + mapped("id", DataType::Utf8, "col-id"), + mapped("amount", DataType::Utf8, "col-amount"), + ])); + let physical = field_to_physical(&Arc::new(mapped("after", after, "col-after"))); + + assert_eq!(physical.name(), "col-after"); + let DataType::Struct(children) = physical.data_type() else { + panic!("`after` must stay a struct"); + }; + assert_eq!(children[0].name(), "col-id"); + assert_eq!(children[1].name(), "col-amount"); + } + + // The write side: the read batch arrives with logical top-level names but + // physical nested names; relabeling must restore logical nested names while + // preserving the data, else nested fields silently read as NULL. + #[test] + fn relabel_restores_nested_names_and_preserves_data() { + let ids: ArrayRef = Arc::new(StringArray::from(vec!["t1", "t2"])); + let amounts: ArrayRef = Arc::new(StringArray::from(vec!["10", "20"])); + let after: ArrayRef = Arc::new(StructArray::from(vec![ + ( + Arc::new(Field::new("col-id", DataType::Utf8, true)), + ids.clone(), + ), + ( + Arc::new(Field::new("col-amount", DataType::Utf8, true)), + amounts.clone(), + ), + ])); + let batch = RecordBatch::try_from_iter(vec![("after", after)]).unwrap(); + + let map = nested_physical_to_logical(&Schema::new(vec![mapped( + "after", + DataType::Struct(Fields::from(vec![ + mapped("id", DataType::Utf8, "col-id"), + mapped("amount", DataType::Utf8, "col-amount"), + ])), + "col-after", + )])); + let relabeled = relabel_nested_columns(&batch, &map).unwrap(); + + let DataType::Struct(children) = relabeled.schema().field(0).data_type().clone() else { + panic!("`after` must stay a struct"); + }; + assert_eq!(children[0].name(), "id"); + assert_eq!(children[1].name(), "amount"); + + let after = relabeled + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(after.column(0).as_ref(), ids.as_ref()); + assert_eq!(after.column(1).as_ref(), amounts.as_ref()); + } + + // Structs in structs: the rename must reach every level. + #[test] + fn read_schema_renames_struct_in_struct() { + let inner = struct_of(vec![mapped("leaf", DataType::Utf8, "col-leaf")]); + let outer = struct_of(vec![mapped("inner", inner, "col-inner")]); + let physical = field_to_physical(&Arc::new(mapped("outer", outer, "col-outer"))); + + assert_eq!(physical.name(), "col-outer"); + let inner = &struct_children(&physical)[0]; + assert_eq!(inner.name(), "col-inner"); + assert_eq!(struct_children(inner)[0].name(), "col-leaf"); + } + + // Arrays in structs: the rename descends through the list element. + #[test] + fn read_schema_renames_array_in_struct() { + let outer = struct_of(vec![mapped("items", list_of(DataType::Utf8), "col-items")]); + let physical = field_to_physical(&Arc::new(mapped("outer", outer, "col-outer"))); + + // The list element itself carries no mapping, so it keeps its name; only + // the struct field wrapping the list is renamed. + let items = &struct_children(&physical)[0]; + assert_eq!(items.name(), "col-items"); + assert!(matches!(items.data_type(), DataType::List(_))); + } + + // Structs in arrays: the rename descends into the list element's struct. + #[test] + fn read_schema_renames_struct_in_array() { + let element = struct_of(vec![mapped("id", DataType::Utf8, "col-id")]); + let physical = field_to_physical(&Arc::new(mapped("items", list_of(element), "col-items"))); + + assert_eq!(physical.name(), "col-items"); + let element = list_element(&physical); + assert_eq!(struct_children(element)[0].name(), "col-id"); + } + + // Structs of structs of arrays of structs: the deepest leaf must be renamed. + #[test] + fn read_schema_renames_struct_of_struct_of_array_of_struct() { + let leaf = struct_of(vec![mapped("amount", DataType::Utf8, "col-amount")]); + let mid = struct_of(vec![mapped("rows", list_of(leaf), "col-rows")]); + let outer = struct_of(vec![mapped("mid", mid, "col-mid")]); + let physical = field_to_physical(&Arc::new(mapped("outer", outer, "col-outer"))); + + let mid = &struct_children(&physical)[0]; + assert_eq!(mid.name(), "col-mid"); + let rows = &struct_children(mid)[0]; + assert_eq!(rows.name(), "col-rows"); + let leaf = list_element(rows); + assert_eq!(struct_children(leaf)[0].name(), "col-amount"); + } + + // Structs in arrays of structs, end to end: relabeling must restore logical + // names inside the list element and preserve the leaf data. + #[test] + fn relabel_restores_names_inside_array_of_structs() { + // A list> with two lists over three elements. + let ids: ArrayRef = Arc::new(StringArray::from(vec!["t1", "t2", "t3"])); + let element_values: ArrayRef = Arc::new(StructArray::from(vec![( + Arc::new(Field::new("col-id", DataType::Utf8, true)), + ids.clone(), + )])); + let element_field = Arc::new(Field::new( + "element", + element_values.data_type().clone(), + true, + )); + let items: ArrayRef = Arc::new(ListArray::new( + element_field, + OffsetBuffer::new(vec![0, 2, 3].into()), + element_values, + None, + )); + let batch = RecordBatch::try_from_iter(vec![("items", items)]).unwrap(); + + let map = nested_physical_to_logical(&Schema::new(vec![mapped( + "items", + list_of(struct_of(vec![mapped("id", DataType::Utf8, "col-id")])), + "col-items", + )])); + let relabeled = relabel_nested_columns(&batch, &map).unwrap(); + + let schema = relabeled.schema(); + let element = list_element(schema.field(0)); + assert_eq!(struct_children(element)[0].name(), "id"); + + // The leaf data survives the relabel unchanged. + let list = relabeled + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let element = list + .values() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(element.column(0).as_ref(), ids.as_ref()); + } +} diff --git a/python/tests/platform/fixtures/column_mapping_nested.py b/python/tests/platform/fixtures/column_mapping_nested.py new file mode 100644 index 00000000000..74bba5c7149 --- /dev/null +++ b/python/tests/platform/fixtures/column_mapping_nested.py @@ -0,0 +1,73 @@ +"""Build a column-mapped Delta table with a nested struct column via PySpark. + +``tests.utils.ensure_delta_spark_fixture`` runs this as a subprocess +(``uv run --with "delta-spark>=4.2,<5" python column_mapping_nested.py ``). +With ``delta.columnMapping.mode = 'name'`` every field is stored on disk under a +physical name (``col-``), including the nested ``after`` struct's children. +PySpark is currently the only writer that can produce column mapping. + +Models a ``uc://`` CDC table with a nested ``after`` struct. History (one commit +per step): v0 CREATE, v1 INSERT two rows, v2 INSERT one row. No schema evolution, +so a follow replay from v0 converges to a snapshot. +""" + +from __future__ import annotations + +import sys + + +# Rows expected from a read of the whole table. Imported by the test so the two +# never drift. +EXPECTED_ROWS = [ + {"after": {"transaction__id": "t1", "transaction__amount": "10"}, "op": "c"}, + {"after": {"transaction__id": "t2", "transaction__amount": "20"}, "op": "c"}, + {"after": {"transaction__id": "t3", "transaction__amount": "30"}, "op": "u"}, +] + + +def build(table_path: str) -> None: + from delta import configure_spark_with_delta_pip + from pyspark.sql import SparkSession + + builder = ( + SparkSession.builder.appName("feldera-column-mapping-nested-fixture") + .master("local[2]") + .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") + .config( + "spark.sql.catalog.spark_catalog", + "org.apache.spark.sql.delta.catalog.DeltaCatalog", + ) + .config("spark.ui.showConsoleProgress", "false") + ) + spark = configure_spark_with_delta_pip(builder).getOrCreate() + spark.sparkContext.setLogLevel("ERROR") + try: + t = f"delta.`{table_path}`" + props = ( + "TBLPROPERTIES ('delta.columnMapping.mode' = 'name'," + " 'delta.minReaderVersion' = '2'," + " 'delta.minWriterVersion' = '5')" + ) + spark.sql( + f"CREATE TABLE {t} (" + " after STRUCT," + " op STRING" + f") USING delta {props}" + ) + spark.sql( + f"INSERT INTO {t} VALUES " + "(named_struct('transaction__id','t1','transaction__amount','10'),'c')," + "(named_struct('transaction__id','t2','transaction__amount','20'),'c')" + ) + spark.sql( + f"INSERT INTO {t} VALUES " + "(named_struct('transaction__id','t3','transaction__amount','30'),'u')" + ) + finally: + spark.stop() + + +if __name__ == "__main__": + if len(sys.argv) != 2: + raise SystemExit("usage: column_mapping_nested.py ") + build(sys.argv[1]) diff --git a/python/tests/platform/test_delta_input_column_mapping_nested.py b/python/tests/platform/test_delta_input_column_mapping_nested.py new file mode 100644 index 00000000000..79f39d63e34 --- /dev/null +++ b/python/tests/platform/test_delta_input_column_mapping_nested.py @@ -0,0 +1,100 @@ +"""Follow a column-mapped table with a nested struct column. + +With ``delta.columnMapping.mode = 'name'`` the nested ``after`` struct's children +are stored on disk under physical names (``col-``). The follow path reads +each commit as raw Parquet and must resolve those names at every level. This +guards the regression where only top-level columns were renamed, so nested +children stayed physical and the read dropped the rows (see +``fixtures/column_mapping_nested.py``). +""" + +from __future__ import annotations + +import json +from pathlib import Path + +from feldera import PipelineBuilder +from feldera.runtime_config import RuntimeConfig +from feldera.testutils import FELDERA_TEST_NUM_HOSTS, FELDERA_TEST_NUM_WORKERS + +from tests import TEST_CLIENT +from tests.platform.fixtures.column_mapping_nested import EXPECTED_ROWS +from tests.utils import DeltaTestLocation, ensure_delta_spark_fixture + +TABLE = "t" +CONNECTOR = "delta_in" +# Bump to invalidate cached copies when the fixture definition changes. +FIXTURE_VERSION = "v1" + +_FIXTURE_BUILDER = Path(__file__).parent / "fixtures" / "column_mapping_nested.py" + + +def _build_sql(loc: DeltaTestLocation) -> str: + config = dict(loc.connector_config) + # Replay v1 + v2 (v0 is the empty CREATE) through the follow path. + config.update({"version": 0, "end_version": 2}) + connectors = json.dumps( + [{"name": CONNECTOR, "transport": {"name": "delta_table_input", "config": config}}] + ).replace("'", "''") + return ( + f"CREATE TABLE {TABLE} (" + "after ROW(transaction__id VARCHAR, transaction__amount VARCHAR)," + "op VARCHAR" + f") WITH ('materialized' = 'true', 'connectors' = '{connectors}');" + ) + + +def test_delta_input_column_mapping_nested_follow(pipeline_name): + """Follow a nested column-mapped (``mode = 'name'``) table: the nested struct + children resolve to their logical names instead of being dropped.""" + loc = DeltaTestLocation.create( + pipeline_name, + mode="snapshot_and_follow", + stable_subpath=f"column_mapping_nested_{FIXTURE_VERSION}", + ) + try: + ensure_delta_spark_fixture(loc, _FIXTURE_BUILDER) + + pipeline = PipelineBuilder( + TEST_CLIENT, + pipeline_name, + sql=_build_sql(loc), + runtime_config=RuntimeConfig( + workers=FELDERA_TEST_NUM_WORKERS, + hosts=FELDERA_TEST_NUM_HOSTS, + logging="debug", + ), + ).create_or_replace() + pipeline.start() + pipeline.wait_for_completion(force_stop=False, timeout_s=600) + + # Project a nested child to exercise resolution inside the struct. + rows = sorted( + ( + {"id": r["id"], "amount": r["amount"], "op": r["op"]} + for r in pipeline.query( + "SELECT after.transaction__id AS id," + f" after.transaction__amount AS amount, op FROM {TABLE}" + ) + ), + key=lambda r: r["id"], + ) + expected = sorted( + ( + { + "id": r["after"]["transaction__id"], + "amount": r["after"]["transaction__amount"], + "op": r["op"], + } + for r in EXPECTED_ROWS + ), + key=lambda r: r["id"], + ) + assert rows == expected, ( + "follow must resolve the nested column-mapped struct children to " + f"their logical names; got {rows}" + ) + + pipeline.stop(force=True) + finally: + loc.cleanup() From 34469349f489ad87de371f686b9fab4231f0f6c5 Mon Sep 17 00:00:00 2001 From: Swanand Mulay <73115739+swanandx@users.noreply.github.com> Date: Wed, 29 Jul 2026 02:06:41 +0530 Subject: [PATCH 4/8] connector: read uc:// follow files via object store, not ListingTable A uc:// location is path-less, so the URL-addressed ListingTable built from root_url() + Add.path reads empty and silently drops the commit. Route uc:// follow reads (and DV files) through the delta-rs object store directly, the reader snapshot and CDC-masked reads already use. Other schemes keep the ListingTable path unchanged. Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com> --- .../src/integrated/delta_table/input.rs | 47 ++++++++++--------- 1 file changed, 26 insertions(+), 21 deletions(-) diff --git a/crates/adapters/src/integrated/delta_table/input.rs b/crates/adapters/src/integrated/delta_table/input.rs index ea1bdcd5f1b..97636cc89de 100644 --- a/crates/adapters/src/integrated/delta_table/input.rs +++ b/crates/adapters/src/integrated/delta_table/input.rs @@ -136,6 +136,12 @@ fn is_active_dv(dv: &DeletionVectorDescriptor) -> bool { dv.cardinality > 0 } +/// A `uc://` location is path-less, so a `ListingTable` built from `root_url() + +/// Add.path` reads empty. Such tables must read through the object store directly. +fn requires_direct_object_store_read(table: &DeltaTable) -> bool { + table.log_store().root_url().scheme() == "uc" +} + /// The table root as a base URL guaranteed to end in `/`, so string-joining a /// relative `Add.path` yields a well-formed URL. delta-rs normalizes the location /// to a trailing slash only when it has a path, so a root with none (`uc://cat.db.tbl`, @@ -3508,28 +3514,27 @@ impl DeltaTableInputEndpointInner { ) -> AnyResult<()> { let description = format!("file '{path}'"); - // An active deletion vector routes the file through a streaming provider - // that masks the deleted rows, restricted to `used_columns` so unread - // columns are never decoded; otherwise the regular `ListingTable` path - // applies. - let provider: Arc = - if let Some(dv) = deletion_vector.filter(|d| is_active_dv(d)) { - let bitmap = self.decode_dv(table, Some(dv), &description).await?; - self.file_provider(table, path, bitmap, ReadMode::NotInBitmap, |name| { - used_columns.contains(&name) - }) - .await? - } else { - // Address files under the table root (e.g. `file:///...`, - // `s3://bucket/prefix/`, or `uc://cat.db.tbl/`). See - // `cdc_side_dataframe` for why we don't use `object_store_url()`, - // and `table_root_base` for why the root is slash-normalized. - let full_path = format!("{}{}", table_root_base(table), path); - Arc::new( - self.create_parquet_table(vec![full_path], &description) - .await?, - ) + // DV files, and uc:// tables (whose path-less location a ListingTable + // can't resolve), read through the object store directly. An empty bitmap + // reads every row. Other schemes use the ListingTable path. + let provider: Arc = if requires_direct_object_store_read(table) + || deletion_vector.is_some_and(is_active_dv) + { + let bitmap = match deletion_vector.filter(|d| is_active_dv(d)) { + Some(dv) => self.decode_dv(table, Some(dv), &description).await?, + None => RoaringTreemap::new(), }; + self.file_provider(table, path, bitmap, ReadMode::NotInBitmap, |name| { + used_columns.contains(&name) + }) + .await? + } else { + let full_path = format!("{}{}", table_root_base(table), path); + Arc::new( + self.create_parquet_table(vec![full_path], &description) + .await?, + ) + }; self.emit_provider( provider, From 3fb17acce77e18638d3122ccb8633e83666ae880 Mon Sep 17 00:00:00 2001 From: Swanand Mulay <73115739+swanandx@users.noreply.github.com> Date: Wed, 29 Jul 2026 02:52:49 +0530 Subject: [PATCH 5/8] delta: resolve follow-read columns by Parquet field id columnMapping.mode=id tables (UniForm over Iceberg) write data files with logical column names plus a Parquet field id, while the Delta log assigns diverging col- physical names. The follow reader matched columns by name, so it could not find col- in the file, null-filled, and dropped the data (or failed on a non-nullable column). Match file columns to the read schema by field id (PARQUET:field_id on the file, delta.columnMapping.id on the schema), rebuilding nested struct/list/map children and casting leaves. Falls back to name when a field id is absent, so mode=name and unmapped tables are unaffected. Mirrors the snapshot path's field-id resolution. Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com> --- .../integrated/delta_table/deletion_vector.rs | 243 +++++++++++++++--- 1 file changed, 202 insertions(+), 41 deletions(-) diff --git a/crates/adapters/src/integrated/delta_table/deletion_vector.rs b/crates/adapters/src/integrated/delta_table/deletion_vector.rs index 8242e6c3a53..a83d5b5e4da 100644 --- a/crates/adapters/src/integrated/delta_table/deletion_vector.rs +++ b/crates/adapters/src/integrated/delta_table/deletion_vector.rs @@ -11,9 +11,11 @@ //! to one batch. use anyhow::{Result as AnyResult, anyhow}; -use arrow::array::{ArrayRef, new_null_array}; +use arrow::array::{ + Array, ArrayRef, LargeListArray, ListArray, MapArray, StructArray, new_null_array, +}; use arrow::compute::cast; -use arrow::datatypes::SchemaRef; +use arrow::datatypes::{DataType, Field, Fields, SchemaRef}; use arrow::record_batch::RecordBatch; use async_stream::try_stream; use datafusion::catalog::TableProvider; @@ -34,6 +36,7 @@ use parquet::arrow::ProjectionMask; use parquet::arrow::arrow_reader::{RowSelection, RowSelector}; use parquet::arrow::async_reader::{ParquetObjectReader, ParquetRecordBatchStreamBuilder}; use roaring::RoaringTreemap; +use std::collections::{HashMap, HashSet}; use std::fmt; use std::sync::Arc; @@ -179,44 +182,138 @@ impl fmt::Debug for MaskedParquetPartition { } } -/// Project `batch` onto `logical_schema`: match columns by name, cast when the -/// file and logical Arrow types differ, and null-fill columns the file lacks. -/// The cast reconciles nested-field metadata/names (e.g. `List` vs -/// `List`) and safe widening (e.g. `Int32` to `Int64`); -/// a genuinely incompatible pair makes `cast` fail and this returns an error. -/// This mirrors DataFusion's default `SchemaAdapter`, which we cannot use -/// here: DataFusion 53 deprecates it for an adapter that only works inside -/// its own scan operators. +/// A field's Parquet field id. The data file stamps `PARQUET:field_id`; the Delta +/// read schema carries `delta.columnMapping.id`. Either identifies the same column. +fn field_id(field: &Field) -> Option<&str> { + field + .metadata() + .get("PARQUET:field_id") + .or_else(|| field.metadata().get("delta.columnMapping.id")) + .map(String::as_str) +} + +/// Index a field list by field id, skipping fields without one. +fn field_index_by_id(fields: &Fields) -> HashMap<&str, usize> { + fields + .iter() + .enumerate() + .filter_map(|(i, f)| field_id(f).map(|id| (id, i))) + .collect() +} + +/// Rebuild `array` to `target`, pairing nested struct/list/map children by field +/// id (falling back to position) and casting leaves whose type differs. Leaf +/// arrays are reused. This bridges a file that names columns logically (Iceberg, +/// `columnMapping.mode=id`) to the read schema's physical (`col-`) names. +fn realign_array(array: &ArrayRef, target: &DataType) -> Result { + match target { + DataType::Struct(target_fields) => { + let source = array + .as_any() + .downcast_ref::() + .ok_or_else(|| realign_type_error("struct", array))?; + let src_idx_by_id = field_index_by_id(source.fields()); + let children = target_fields + .iter() + .enumerate() + .map(|(pos, tf)| { + let idx = field_id(tf) + .and_then(|id| src_idx_by_id.get(id).copied()) + .unwrap_or(pos); + let child = source.columns().get(idx).ok_or_else(|| { + DataFusionError::Internal(format!( + "field-id realign found no source child for '{}'", + tf.name() + )) + })?; + realign_array(child, tf.data_type()) + }) + .collect::, _>>()?; + Ok(Arc::new(StructArray::try_new( + target_fields.clone(), + children, + source.nulls().cloned(), + )?)) + } + DataType::List(inner) => { + let source = array + .as_any() + .downcast_ref::() + .ok_or_else(|| realign_type_error("list", array))?; + let values = realign_array(source.values(), inner.data_type())?; + Ok(Arc::new(ListArray::try_new( + inner.clone(), + source.offsets().clone(), + values, + source.nulls().cloned(), + )?)) + } + DataType::LargeList(inner) => { + let source = array + .as_any() + .downcast_ref::() + .ok_or_else(|| realign_type_error("large list", array))?; + let values = realign_array(source.values(), inner.data_type())?; + Ok(Arc::new(LargeListArray::try_new( + inner.clone(), + source.offsets().clone(), + values, + source.nulls().cloned(), + )?)) + } + DataType::Map(entries, sorted) => { + let source = array + .as_any() + .downcast_ref::() + .ok_or_else(|| realign_type_error("map", array))?; + let rebuilt = + realign_array(&(Arc::new(source.entries().clone()) as ArrayRef), entries.data_type())?; + let entries_arr = rebuilt + .as_any() + .downcast_ref::() + .expect("map entries realign to a struct") + .clone(); + Ok(Arc::new(MapArray::try_new( + entries.clone(), + source.offsets().clone(), + entries_arr, + source.nulls().cloned(), + *sorted, + )?)) + } + _ if array.data_type() == target => Ok(Arc::clone(array)), + _ => Ok(cast(array, target)?), + } +} + +fn realign_type_error(expected: &str, array: &ArrayRef) -> DataFusionError { + DataFusionError::Internal(format!( + "field-id realign expected a {expected} array, got {}", + array.data_type() + )) +} + +/// Project `batch` onto `logical_schema`, matching columns by field id (falling +/// back to name), rebuilding nested shapes and casting leaves, and null-filling +/// columns the file lacks. Field-id matching handles `columnMapping.mode=id` +/// tables, whose files name columns logically rather than by physical `col-`. /// -/// Partition columns also come out NULL (Delta stores them in -/// `partitionValues`, not in the file). This is the same pre-existing -/// limitation as the connector's `ListingTable` path. +/// Partition columns come out NULL (Delta stores them in `partitionValues`, not +/// in the file), a pre-existing limitation of the connector's Parquet reader. fn project_to_logical( batch: &RecordBatch, logical_schema: &SchemaRef, ) -> Result { let num_rows = batch.num_rows(); + let file_schema = batch.schema(); + let file_idx_by_id = field_index_by_id(file_schema.fields()); let mut columns: Vec = Vec::with_capacity(logical_schema.fields().len()); for field in logical_schema.fields().iter() { - let col = match batch.schema().column_with_name(field.name()) { - Some((idx, file_field)) => { - if file_field.data_type() == field.data_type() { - Arc::clone(batch.column(idx)) - } else { - cast(batch.column(idx), field.data_type()).map_err(|e| { - DataFusionError::External( - format!( - "deletion-vector reader: cannot adapt file column '{}' \ - ({:?}) to Delta logical type {:?}: {e}", - field.name(), - file_field.data_type(), - field.data_type(), - ) - .into(), - ) - })? - } - } + let source = field_id(field) + .and_then(|id| file_idx_by_id.get(id).copied()) + .or_else(|| file_schema.index_of(field.name()).ok()); + let col = match source { + Some(idx) => realign_array(batch.column(idx), field.data_type())?, None => new_null_array(field.data_type(), num_rows), }; columns.push(col); @@ -229,22 +326,26 @@ fn project_to_logical( }) } -/// Build the [`ProjectionMask`] selecting the root file columns that -/// `logical_schema` names; the rest are never decoded. +/// Build the [`ProjectionMask`] selecting the root file columns `logical_schema` +/// wants, matched by field id (falling back to name); the rest are never decoded. fn logical_projection_mask( builder: &ParquetRecordBatchStreamBuilder, logical_schema: &SchemaRef, ) -> ProjectionMask { - // Root Arrow fields map one-to-one, in order, to root Parquet columns, so - // this index mapping is infallible. A file column the logical schema does - // not name is pruned on purpose (not an error worth logging); a logical - // column the file lacks is null-filled later in `project_to_logical`. + let want_ids: HashSet<&str> = logical_schema + .fields() + .iter() + .filter_map(|f| field_id(f)) + .collect(); let roots = builder .schema() .fields() .iter() .enumerate() - .filter(|(_, field)| logical_schema.column_with_name(field.name()).is_some()) + .filter(|(_, field)| { + field_id(field).is_some_and(|id| want_ids.contains(id)) + || logical_schema.column_with_name(field.name()).is_some() + }) .map(|(idx, _)| idx); ProjectionMask::roots(builder.parquet_schema(), roots) } @@ -343,13 +444,73 @@ impl PartitionStream for MaskedParquetPartition { #[cfg(test)] mod tests { use super::*; - use arrow::array::{Array, Int32Array, Int64Array, StringArray}; - use arrow::datatypes::{DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema}; + use arrow::array::{Array, Int32Array, Int64Array, StringArray, StructArray}; + use arrow::datatypes::{ + DataType as ArrowDataType, Field as ArrowField, Fields as ArrowFields, Schema as ArrowSchema, + }; use datafusion::prelude::SessionContext; use deltalake::{DeltaTableBuilder, ensure_table_uri}; use proptest::prelude::*; use tempfile::TempDir; + fn with_id(field: ArrowField, key: &str, id: &str) -> ArrowField { + field.with_metadata(HashMap::from([(key.to_string(), id.to_string())])) + } + + // A columnMapping.mode=id file names columns logically (`op`, `after`) and + // carries `PARQUET:field_id`; the Delta read schema uses physical `col-` + // names and `delta.columnMapping.id`. project_to_logical must pair them by + // field id, not name, else it null-fills and drops the data. + #[test] + fn project_to_logical_matches_by_field_id() { + let file_after: ArrayRef = Arc::new(StructArray::from(vec![( + Arc::new(with_id( + ArrowField::new("transaction__id", ArrowDataType::Utf8, true), + "PARQUET:field_id", + "2", + )), + Arc::new(StringArray::from(vec!["t1"])) as ArrayRef, + )])); + let file_op: ArrayRef = Arc::new(StringArray::from(vec!["INSERT"])); + let file_schema = Arc::new(ArrowSchema::new(vec![ + with_id( + ArrowField::new("after", file_after.data_type().clone(), true), + "PARQUET:field_id", + "1", + ), + with_id(ArrowField::new("op", ArrowDataType::Utf8, false), "PARQUET:field_id", "8"), + ])); + let batch = RecordBatch::try_new(file_schema, vec![file_after, file_op]).unwrap(); + + let read_schema = Arc::new(ArrowSchema::new(vec![ + with_id( + ArrowField::new( + "col-1", + ArrowDataType::Struct(ArrowFields::from(vec![with_id( + ArrowField::new("col-2", ArrowDataType::Utf8, true), + "delta.columnMapping.id", + "2", + )])), + true, + ), + "delta.columnMapping.id", + "1", + ), + with_id( + ArrowField::new("col-8", ArrowDataType::Utf8, false), + "delta.columnMapping.id", + "8", + ), + ])); + + let out = project_to_logical(&batch, &read_schema).unwrap(); + assert_eq!(out.schema().field(1).name(), "col-8"); + let op = out.column(1).as_any().downcast_ref::().unwrap(); + assert_eq!(op.value(0), "INSERT", "op resolved by field id, not null-filled"); + let after = out.column(0).as_any().downcast_ref::().unwrap(); + assert_eq!(after.column(0).as_any().downcast_ref::().unwrap().value(0), "t1"); + } + /// Expand a [`RowSelection`] into the row positions it selects. fn selected_rows(selection: &RowSelection) -> Vec { let mut rows = Vec::new(); From e0f0da863ceafeb6431c0fd2d2f22ef773ddf431 Mon Sep 17 00:00:00 2001 From: Swanand Mulay <73115739+swanandx@users.noreply.github.com> Date: Wed, 29 Jul 2026 03:40:32 +0530 Subject: [PATCH 6/8] delta: harden field-id realign to coerce container types realign_array recursed into list/largelist/map and downcast to the exact container the read schema named, so a file whose list kind differed (Delta List vs a file's LargeList) failed with an internal error. Rebuild only structs by field id (where column-mapped names must line up) and route everything else, including container coercion, through cast. Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com> --- .../integrated/delta_table/deletion_vector.rs | 190 +++++++++--------- .../test_delta_input_column_mapping_nested.py | 7 +- 2 files changed, 98 insertions(+), 99 deletions(-) diff --git a/crates/adapters/src/integrated/delta_table/deletion_vector.rs b/crates/adapters/src/integrated/delta_table/deletion_vector.rs index a83d5b5e4da..7f48314560b 100644 --- a/crates/adapters/src/integrated/delta_table/deletion_vector.rs +++ b/crates/adapters/src/integrated/delta_table/deletion_vector.rs @@ -11,9 +11,7 @@ //! to one batch. use anyhow::{Result as AnyResult, anyhow}; -use arrow::array::{ - Array, ArrayRef, LargeListArray, ListArray, MapArray, StructArray, new_null_array, -}; +use arrow::array::{Array, ArrayRef, StructArray, new_null_array}; use arrow::compute::cast; use arrow::datatypes::{DataType, Field, Fields, SchemaRef}; use arrow::record_batch::RecordBatch; @@ -201,96 +199,47 @@ fn field_index_by_id(fields: &Fields) -> HashMap<&str, usize> { .collect() } -/// Rebuild `array` to `target`, pairing nested struct/list/map children by field -/// id (falling back to position) and casting leaves whose type differs. Leaf -/// arrays are reused. This bridges a file that names columns logically (Iceberg, -/// `columnMapping.mode=id`) to the read schema's physical (`col-`) names. +/// Convert a file column `array` to the `target` type the read schema expects. +/// +/// Under column mapping a struct's field names differ between the file and the +/// schema (a file may use logical names, the schema uses `col-`), so a struct +/// is rebuilt: each target child takes the source child with the same field id +/// (by position if none matches). Non-struct types (scalars, lists, maps) have no +/// such names to match, so `cast` handles them, including type and container +/// differences like `List` vs `LargeList`. fn realign_array(array: &ArrayRef, target: &DataType) -> Result { - match target { - DataType::Struct(target_fields) => { - let source = array - .as_any() - .downcast_ref::() - .ok_or_else(|| realign_type_error("struct", array))?; - let src_idx_by_id = field_index_by_id(source.fields()); - let children = target_fields - .iter() - .enumerate() - .map(|(pos, tf)| { - let idx = field_id(tf) - .and_then(|id| src_idx_by_id.get(id).copied()) - .unwrap_or(pos); - let child = source.columns().get(idx).ok_or_else(|| { - DataFusionError::Internal(format!( - "field-id realign found no source child for '{}'", - tf.name() - )) - })?; - realign_array(child, tf.data_type()) - }) - .collect::, _>>()?; - Ok(Arc::new(StructArray::try_new( - target_fields.clone(), - children, - source.nulls().cloned(), - )?)) - } - DataType::List(inner) => { - let source = array - .as_any() - .downcast_ref::() - .ok_or_else(|| realign_type_error("list", array))?; - let values = realign_array(source.values(), inner.data_type())?; - Ok(Arc::new(ListArray::try_new( - inner.clone(), - source.offsets().clone(), - values, - source.nulls().cloned(), - )?)) - } - DataType::LargeList(inner) => { - let source = array - .as_any() - .downcast_ref::() - .ok_or_else(|| realign_type_error("large list", array))?; - let values = realign_array(source.values(), inner.data_type())?; - Ok(Arc::new(LargeListArray::try_new( - inner.clone(), - source.offsets().clone(), - values, - source.nulls().cloned(), - )?)) - } - DataType::Map(entries, sorted) => { - let source = array - .as_any() - .downcast_ref::() - .ok_or_else(|| realign_type_error("map", array))?; - let rebuilt = - realign_array(&(Arc::new(source.entries().clone()) as ArrayRef), entries.data_type())?; - let entries_arr = rebuilt - .as_any() - .downcast_ref::() - .expect("map entries realign to a struct") - .clone(); - Ok(Arc::new(MapArray::try_new( - entries.clone(), - source.offsets().clone(), - entries_arr, - source.nulls().cloned(), - *sorted, - )?)) - } - _ if array.data_type() == target => Ok(Arc::clone(array)), - _ => Ok(cast(array, target)?), - } -} - -fn realign_type_error(expected: &str, array: &ArrayRef) -> DataFusionError { - DataFusionError::Internal(format!( - "field-id realign expected a {expected} array, got {}", - array.data_type() - )) + let DataType::Struct(target_fields) = target else { + return if array.data_type() == target { + Ok(Arc::clone(array)) + } else { + Ok(cast(array, target)?) + }; + }; + let Some(source) = array.as_any().downcast_ref::() else { + return Ok(cast(array, target)?); + }; + let src_idx_by_id = field_index_by_id(source.fields()); + let children = target_fields + .iter() + .enumerate() + .map(|(pos, tf)| { + let idx = field_id(tf) + .and_then(|id| src_idx_by_id.get(id).copied()) + .unwrap_or(pos); + let child = source.columns().get(idx).ok_or_else(|| { + DataFusionError::Internal(format!( + "field-id realign found no source child for '{}'", + tf.name() + )) + })?; + realign_array(child, tf.data_type()) + }) + .collect::, _>>()?; + Ok(Arc::new(StructArray::try_new( + target_fields.clone(), + children, + source.nulls().cloned(), + )?)) } /// Project `batch` onto `logical_schema`, matching columns by field id (falling @@ -446,7 +395,8 @@ mod tests { use super::*; use arrow::array::{Array, Int32Array, Int64Array, StringArray, StructArray}; use arrow::datatypes::{ - DataType as ArrowDataType, Field as ArrowField, Fields as ArrowFields, Schema as ArrowSchema, + DataType as ArrowDataType, Field as ArrowField, Fields as ArrowFields, + Schema as ArrowSchema, }; use datafusion::prelude::SessionContext; use deltalake::{DeltaTableBuilder, ensure_table_uri}; @@ -457,6 +407,26 @@ mod tests { field.with_metadata(HashMap::from([(key.to_string(), id.to_string())])) } + // The read schema's list kind may differ from the file's (Delta `List` vs a + // file's `LargeList`); realign must coerce the container instead of failing. + #[test] + fn realign_array_coerces_list_containers() { + use arrow::array::{LargeListBuilder, StringBuilder}; + let mut b = LargeListBuilder::new(StringBuilder::new()); + b.values().append_value("a"); + b.values().append_value("b"); + b.append(true); + b.values().append_value("c"); + b.append(true); + let source: ArrayRef = Arc::new(b.finish()); + + let target = + ArrowDataType::List(Arc::new(ArrowField::new("item", ArrowDataType::Utf8, true))); + let out = realign_array(&source, &target).unwrap(); + assert_eq!(out.data_type(), &target); + assert_eq!(out.len(), 2); + } + // A columnMapping.mode=id file names columns logically (`op`, `after`) and // carries `PARQUET:field_id`; the Delta read schema uses physical `col-` // names and `delta.columnMapping.id`. project_to_logical must pair them by @@ -478,7 +448,11 @@ mod tests { "PARQUET:field_id", "1", ), - with_id(ArrowField::new("op", ArrowDataType::Utf8, false), "PARQUET:field_id", "8"), + with_id( + ArrowField::new("op", ArrowDataType::Utf8, false), + "PARQUET:field_id", + "8", + ), ])); let batch = RecordBatch::try_new(file_schema, vec![file_after, file_op]).unwrap(); @@ -505,10 +479,30 @@ mod tests { let out = project_to_logical(&batch, &read_schema).unwrap(); assert_eq!(out.schema().field(1).name(), "col-8"); - let op = out.column(1).as_any().downcast_ref::().unwrap(); - assert_eq!(op.value(0), "INSERT", "op resolved by field id, not null-filled"); - let after = out.column(0).as_any().downcast_ref::().unwrap(); - assert_eq!(after.column(0).as_any().downcast_ref::().unwrap().value(0), "t1"); + let op = out + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!( + op.value(0), + "INSERT", + "op resolved by field id, not null-filled" + ); + let after = out + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!( + after + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + "t1" + ); } /// Expand a [`RowSelection`] into the row positions it selects. diff --git a/python/tests/platform/test_delta_input_column_mapping_nested.py b/python/tests/platform/test_delta_input_column_mapping_nested.py index 79f39d63e34..79d5bdff367 100644 --- a/python/tests/platform/test_delta_input_column_mapping_nested.py +++ b/python/tests/platform/test_delta_input_column_mapping_nested.py @@ -34,7 +34,12 @@ def _build_sql(loc: DeltaTestLocation) -> str: # Replay v1 + v2 (v0 is the empty CREATE) through the follow path. config.update({"version": 0, "end_version": 2}) connectors = json.dumps( - [{"name": CONNECTOR, "transport": {"name": "delta_table_input", "config": config}}] + [ + { + "name": CONNECTOR, + "transport": {"name": "delta_table_input", "config": config}, + } + ] ).replace("'", "''") return ( f"CREATE TABLE {TABLE} (" From db274932c5c5634edcd6f5a5341a8a6f3f1392a3 Mon Sep 17 00:00:00 2001 From: Swanand Mulay <73115739+swanandx@users.noreply.github.com> Date: Wed, 29 Jul 2026 04:30:35 +0530 Subject: [PATCH 7/8] delta: null-fill missing struct child in field-id realign realign_array fell back to the positional child when a target struct child's field id had no match in the file. That could bind an unrelated sibling column, or, when the position was out of range, fail with an internal error. A column-mapped struct that gains a nested child through schema evolution hits both. Match nested children by field id only, and null-fill a child the file lacks, mirroring how project_to_logical handles a missing top-level column. Positional matching now applies solely to the unmapped case, where neither side carries a field id. Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com> --- .../integrated/delta_table/deletion_vector.rs | 74 +++++++++++++++---- 1 file changed, 60 insertions(+), 14 deletions(-) diff --git a/crates/adapters/src/integrated/delta_table/deletion_vector.rs b/crates/adapters/src/integrated/delta_table/deletion_vector.rs index 7f48314560b..adf4d4aa99d 100644 --- a/crates/adapters/src/integrated/delta_table/deletion_vector.rs +++ b/crates/adapters/src/integrated/delta_table/deletion_vector.rs @@ -203,10 +203,12 @@ fn field_index_by_id(fields: &Fields) -> HashMap<&str, usize> { /// /// Under column mapping a struct's field names differ between the file and the /// schema (a file may use logical names, the schema uses `col-`), so a struct -/// is rebuilt: each target child takes the source child with the same field id -/// (by position if none matches). Non-struct types (scalars, lists, maps) have no -/// such names to match, so `cast` handles them, including type and container -/// differences like `List` vs `LargeList`. +/// is rebuilt: each target child takes the source child with the same field id, or +/// the child at the same position when neither side carries an id (unmapped). A +/// target child the file lacks is null-filled, matching how [`project_to_logical`] +/// handles a missing top-level column. Non-struct types (scalars, lists, maps) +/// have no such names to match, so `cast` handles them, including type and +/// container differences like `List` vs `LargeList`. fn realign_array(array: &ArrayRef, target: &DataType) -> Result { let DataType::Struct(target_fields) = target else { return if array.data_type() == target { @@ -223,16 +225,16 @@ fn realign_array(array: &ArrayRef, target: &DataType) -> Result src_idx_by_id.get(id).copied(), + None => Some(pos), + }; + match idx.and_then(|i| source.columns().get(i)) { + Some(child) => realign_array(child, tf.data_type()), + None => Ok(new_null_array(tf.data_type(), source.len())), + } }) .collect::, _>>()?; Ok(Arc::new(StructArray::try_new( @@ -505,6 +507,50 @@ mod tests { ); } + // A struct child the file lacks (e.g. a field added to the struct after the + // file was written) must null-fill, not error or grab a wrong-id sibling. + #[test] + fn realign_array_null_fills_missing_struct_child() { + let source: ArrayRef = Arc::new(StructArray::from(vec![( + Arc::new(with_id( + ArrowField::new("id", ArrowDataType::Utf8, true), + "PARQUET:field_id", + "2", + )), + Arc::new(StringArray::from(vec!["t1", "t2"])) as ArrayRef, + )])); + + // Target wants both id 2 (present) and id 3 (absent from the file). + let target = ArrowDataType::Struct(ArrowFields::from(vec![ + with_id( + ArrowField::new("col-2", ArrowDataType::Utf8, true), + "delta.columnMapping.id", + "2", + ), + with_id( + ArrowField::new("col-3", ArrowDataType::Utf8, true), + "delta.columnMapping.id", + "3", + ), + ])); + + let out = realign_array(&source, &target).unwrap(); + let out = out.as_any().downcast_ref::().unwrap(); + let present = out + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(present.value(0), "t1"); + let missing = out + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(missing.len(), 2); + assert!(missing.is_null(0) && missing.is_null(1)); + } + /// Expand a [`RowSelection`] into the row positions it selects. fn selected_rows(selection: &RowSelection) -> Vec { let mut rows = Vec::new(); From 3029b6d3e676ba9ea5c84229f010889d82f8baf0 Mon Sep 17 00:00:00 2001 From: Swanand Mulay <73115739+swanandx@users.noreply.github.com> Date: Thu, 30 Jul 2026 04:22:23 +0530 Subject: [PATCH 8/8] delta: name the column and file in follow-read conversion errors A failed conversion surfaced Arrow's bare cast error, which names neither the column nor the file: "Casting from Utf8 to FixedSizeBinary(16) not supported". Under column mapping the column is named col- on disk, so the user had no way to tell which column of which table failed. Thread the file and a dotted column path through realign_array, so a nested leaf reports 'after.col-2' rather than the top-level column, and report the file's type alongside the type the Delta table declares. Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com> --- .../integrated/delta_table/deletion_vector.rs | 105 +++++++++++++++--- 1 file changed, 89 insertions(+), 16 deletions(-) diff --git a/crates/adapters/src/integrated/delta_table/deletion_vector.rs b/crates/adapters/src/integrated/delta_table/deletion_vector.rs index adf4d4aa99d..c185f953afc 100644 --- a/crates/adapters/src/integrated/delta_table/deletion_vector.rs +++ b/crates/adapters/src/integrated/delta_table/deletion_vector.rs @@ -199,7 +199,25 @@ fn field_index_by_id(fields: &Fields) -> HashMap<&str, usize> { .collect() } +fn conversion_error( + file: &str, + column: &str, + from: &DataType, + to: &DataType, + cause: impl fmt::Display, +) -> DataFusionError { + DataFusionError::External( + format!( + "Delta file reader: cannot read column '{column}' of file '{file}': the file stores \ + it as {from:?}, which is not convertible to the {to:?} that the Delta table's \ + schema declares: {cause}" + ) + .into(), + ) +} + /// Convert a file column `array` to the `target` type the read schema expects. +/// `file` and `column` (dotted for a nested child) locate it in errors. /// /// Under column mapping a struct's field names differ between the file and the /// schema (a file may use logical names, the schema uses `col-`), so a struct @@ -209,16 +227,25 @@ fn field_index_by_id(fields: &Fields) -> HashMap<&str, usize> { /// handles a missing top-level column. Non-struct types (scalars, lists, maps) /// have no such names to match, so `cast` handles them, including type and /// container differences like `List` vs `LargeList`. -fn realign_array(array: &ArrayRef, target: &DataType) -> Result { +fn realign_array( + array: &ArrayRef, + target: &DataType, + file: &str, + column: &str, +) -> Result { + let cast_to_target = || { + cast(array, target) + .map_err(|e| conversion_error(file, column, array.data_type(), target, e)) + }; let DataType::Struct(target_fields) = target else { return if array.data_type() == target { Ok(Arc::clone(array)) } else { - Ok(cast(array, target)?) + cast_to_target() }; }; let Some(source) = array.as_any().downcast_ref::() else { - return Ok(cast(array, target)?); + return cast_to_target(); }; let src_idx_by_id = field_index_by_id(source.fields()); let children = target_fields @@ -232,16 +259,22 @@ fn realign_array(array: &ArrayRef, target: &DataType) -> Result Some(pos), }; match idx.and_then(|i| source.columns().get(i)) { - Some(child) => realign_array(child, tf.data_type()), + Some(child) => realign_array( + child, + tf.data_type(), + file, + &format!("{column}.{}", tf.name()), + ), None => Ok(new_null_array(tf.data_type(), source.len())), } }) .collect::, _>>()?; - Ok(Arc::new(StructArray::try_new( - target_fields.clone(), - children, - source.nulls().cloned(), - )?)) + // The children match `target_fields` by construction, so this rejects only a + // null-filled child of a NOT NULL target field, i.e. a column the file lacks + // that the table requires. + StructArray::try_new(target_fields.clone(), children, source.nulls().cloned()) + .map(|s| Arc::new(s) as ArrayRef) + .map_err(|e| conversion_error(file, column, array.data_type(), target, e)) } /// Project `batch` onto `logical_schema`, matching columns by field id (falling @@ -254,6 +287,7 @@ fn realign_array(array: &ArrayRef, target: &DataType) -> Result Result { let num_rows = batch.num_rows(); let file_schema = batch.schema(); @@ -264,15 +298,19 @@ fn project_to_logical( .and_then(|id| file_idx_by_id.get(id).copied()) .or_else(|| file_schema.index_of(field.name()).ok()); let col = match source { - Some(idx) => realign_array(batch.column(idx), field.data_type())?, + Some(idx) => realign_array(batch.column(idx), field.data_type(), file, field.name())?, None => new_null_array(field.data_type(), num_rows), }; columns.push(col); } RecordBatch::try_new(Arc::clone(logical_schema), columns).map_err(|e| { DataFusionError::External( - format!("deletion-vector reader: projected batch rejected by logical schema: {e}") - .into(), + format!( + "Delta file reader: file '{file}' does not satisfy the Delta table's schema: {e}. \ + A column the file lacks is read as NULL, which the table rejects when it \ + declares the column NOT NULL." + ) + .into(), ) }) } @@ -380,7 +418,7 @@ impl PartitionStream for MaskedParquetPartition { let batch = batch.map_err(|e| DataFusionError::External( format!("error reading Parquet file '{path}': {e}").into()))?; if batch.num_rows() > 0 { - yield project_to_logical(&batch, &logical_schema)?; + yield project_to_logical(&batch, &logical_schema, path.as_ref())?; } } }; @@ -405,6 +443,9 @@ mod tests { use proptest::prelude::*; use tempfile::TempDir; + /// Stands in for the data file the reader is decoding; it appears in errors. + const TEST_FILE: &str = "part-00000.parquet"; + fn with_id(field: ArrowField, key: &str, id: &str) -> ArrowField { field.with_metadata(HashMap::from([(key.to_string(), id.to_string())])) } @@ -424,7 +465,7 @@ mod tests { let target = ArrowDataType::List(Arc::new(ArrowField::new("item", ArrowDataType::Utf8, true))); - let out = realign_array(&source, &target).unwrap(); + let out = realign_array(&source, &target, TEST_FILE, "items").unwrap(); assert_eq!(out.data_type(), &target); assert_eq!(out.len(), 2); } @@ -479,7 +520,7 @@ mod tests { ), ])); - let out = project_to_logical(&batch, &read_schema).unwrap(); + let out = project_to_logical(&batch, &read_schema, TEST_FILE).unwrap(); assert_eq!(out.schema().field(1).name(), "col-8"); let op = out .column(1) @@ -534,7 +575,7 @@ mod tests { ), ])); - let out = realign_array(&source, &target).unwrap(); + let out = realign_array(&source, &target, TEST_FILE, "after").unwrap(); let out = out.as_any().downcast_ref::().unwrap(); let present = out .column(0) @@ -551,6 +592,38 @@ mod tests { assert!(missing.is_null(0) && missing.is_null(1)); } + // A user hitting a type mismatch sees only the physical `col-` name on + // disk, so the error has to name the column, the file, and both types. Arrow's + // bare cast error carries none of that. + #[test] + fn conversion_error_names_column_file_and_types() { + let file_child: ArrayRef = Arc::new(StringArray::from(vec!["not-a-timestamp"])); + let source: ArrayRef = Arc::new(StructArray::from(vec![( + Arc::new(with_id( + ArrowField::new("amount", ArrowDataType::Utf8, true), + "PARQUET:field_id", + "2", + )), + file_child, + )])); + // Utf8 to a fixed-size binary is not a cast Arrow supports. + let target = ArrowDataType::Struct(ArrowFields::from(vec![with_id( + ArrowField::new("col-2", ArrowDataType::FixedSizeBinary(16), true), + "delta.columnMapping.id", + "2", + )])); + + let err = realign_array(&source, &target, TEST_FILE, "after") + .expect_err("Utf8 does not cast to FixedSizeBinary") + .to_string(); + + // The nested child, not just the top-level column. + assert!(err.contains("'after.col-2'"), "{err}"); + assert!(err.contains(TEST_FILE), "{err}"); + assert!(err.contains("Utf8"), "{err}"); + assert!(err.contains("FixedSizeBinary(16)"), "{err}"); + } + /// Expand a [`RowSelection`] into the row positions it selects. fn selected_rows(selection: &RowSelection) -> Vec { let mut rows = Vec::new();