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" } diff --git a/crates/adapters/src/integrated/delta_table/deletion_vector.rs b/crates/adapters/src/integrated/delta_table/deletion_vector.rs index 8242e6c3a53..c185f953afc 100644 --- a/crates/adapters/src/integrated/delta_table/deletion_vector.rs +++ b/crates/adapters/src/integrated/delta_table/deletion_vector.rs @@ -11,9 +11,9 @@ //! to one batch. use anyhow::{Result as AnyResult, anyhow}; -use arrow::array::{ArrayRef, new_null_array}; +use arrow::array::{Array, ArrayRef, 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 +34,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,72 +180,161 @@ 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() +} + +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. /// -/// 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. +/// 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, 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, + 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 { + cast_to_target() + }; + }; + let Some(source) = array.as_any().downcast_ref::() else { + return cast_to_target(); + }; + let src_idx_by_id = field_index_by_id(source.fields()); + let children = target_fields + .iter() + .enumerate() + .map(|(pos, tf)| { + // With an id, match by id only: falling back to position would risk + // grabbing an unrelated column. Without one (unmapped), use position. + let idx = match field_id(tf) { + Some(id) => 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(), + file, + &format!("{column}.{}", tf.name()), + ), + None => Ok(new_null_array(tf.data_type(), source.len())), + } + }) + .collect::, _>>()?; + // 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 +/// 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 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, + file: &str, ) -> 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(), 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(), ) }) } -/// 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) } @@ -328,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())?; } } }; @@ -343,13 +433,197 @@ 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; + /// 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())])) + } + + // 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, TEST_FILE, "items").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 + // 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, TEST_FILE).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" + ); + } + + // 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, TEST_FILE, "after").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)); + } + + // 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(); diff --git a/crates/adapters/src/integrated/delta_table/input.rs b/crates/adapters/src/integrated/delta_table/input.rs index 95311ee90c0..97636cc89de 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; @@ -134,6 +136,177 @@ 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`, +/// 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}/") + } +} + +/// 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: @@ -2500,6 +2673,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; @@ -2584,7 +2769,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(); @@ -3038,12 +3227,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 @@ -3051,19 +3241,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()), @@ -3201,11 +3383,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 +3408,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}") @@ -3336,27 +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 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); - 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, @@ -3596,3 +3774,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..79d5bdff367 --- /dev/null +++ b/python/tests/platform/test_delta_input_column_mapping_nested.py @@ -0,0 +1,105 @@ +"""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()