diff --git a/Cargo.lock b/Cargo.lock index 8f8059a5e0e..996b05eb64c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4305,7 +4305,7 @@ checksum = "26bf8fc351c5ed29b5c2f0cbbac1b209b74f60ecd62e675a998df72c49af5204" [[package]] name = "deltalake" version = "0.32.3" -source = "git+https://github.com/feldera/delta-rs.git?rev=5b35c0988b07ab921d5eee5b8da8da55e4d65945#5b35c0988b07ab921d5eee5b8da8da55e4d65945" +source = "git+https://github.com/feldera/delta-rs.git?rev=297f2dd146f9947795f8b77f25db8da9daff2ade#297f2dd146f9947795f8b77f25db8da9daff2ade" dependencies = [ "buoyant_kernel", "ctor 0.10.1", @@ -4319,7 +4319,7 @@ dependencies = [ [[package]] name = "deltalake-aws" version = "0.15.0" -source = "git+https://github.com/feldera/delta-rs.git?rev=5b35c0988b07ab921d5eee5b8da8da55e4d65945#5b35c0988b07ab921d5eee5b8da8da55e4d65945" +source = "git+https://github.com/feldera/delta-rs.git?rev=297f2dd146f9947795f8b77f25db8da9daff2ade#297f2dd146f9947795f8b77f25db8da9daff2ade" dependencies = [ "async-trait", "aws-config", @@ -4345,7 +4345,7 @@ dependencies = [ [[package]] name = "deltalake-azure" version = "0.15.0" -source = "git+https://github.com/feldera/delta-rs.git?rev=5b35c0988b07ab921d5eee5b8da8da55e4d65945#5b35c0988b07ab921d5eee5b8da8da55e4d65945" +source = "git+https://github.com/feldera/delta-rs.git?rev=297f2dd146f9947795f8b77f25db8da9daff2ade#297f2dd146f9947795f8b77f25db8da9daff2ade" dependencies = [ "bytes", "deltalake-core", @@ -4358,7 +4358,7 @@ dependencies = [ [[package]] name = "deltalake-catalog-unity" version = "0.16.0" -source = "git+https://github.com/feldera/delta-rs.git?rev=5b35c0988b07ab921d5eee5b8da8da55e4d65945#5b35c0988b07ab921d5eee5b8da8da55e4d65945" +source = "git+https://github.com/feldera/delta-rs.git?rev=297f2dd146f9947795f8b77f25db8da9daff2ade#297f2dd146f9947795f8b77f25db8da9daff2ade" dependencies = [ "async-trait", "chrono", @@ -4385,7 +4385,7 @@ dependencies = [ [[package]] name = "deltalake-core" version = "0.32.3" -source = "git+https://github.com/feldera/delta-rs.git?rev=5b35c0988b07ab921d5eee5b8da8da55e4d65945#5b35c0988b07ab921d5eee5b8da8da55e4d65945" +source = "git+https://github.com/feldera/delta-rs.git?rev=297f2dd146f9947795f8b77f25db8da9daff2ade#297f2dd146f9947795f8b77f25db8da9daff2ade" dependencies = [ "arrow", "arrow-arith", @@ -4439,7 +4439,7 @@ dependencies = [ [[package]] name = "deltalake-derive" version = "1.0.0" -source = "git+https://github.com/feldera/delta-rs.git?rev=5b35c0988b07ab921d5eee5b8da8da55e4d65945#5b35c0988b07ab921d5eee5b8da8da55e4d65945" +source = "git+https://github.com/feldera/delta-rs.git?rev=297f2dd146f9947795f8b77f25db8da9daff2ade#297f2dd146f9947795f8b77f25db8da9daff2ade" dependencies = [ "convert_case 0.9.0", "itertools 0.14.0", @@ -4451,7 +4451,7 @@ dependencies = [ [[package]] name = "deltalake-gcp" version = "0.16.0" -source = "git+https://github.com/feldera/delta-rs.git?rev=5b35c0988b07ab921d5eee5b8da8da55e4d65945#5b35c0988b07ab921d5eee5b8da8da55e4d65945" +source = "git+https://github.com/feldera/delta-rs.git?rev=297f2dd146f9947795f8b77f25db8da9daff2ade#297f2dd146f9947795f8b77f25db8da9daff2ade" dependencies = [ "async-trait", "bytes", diff --git a/Cargo.toml b/Cargo.toml index 2985273bea9..56152b36bbc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -111,17 +111,20 @@ datafusion = "53.1" dbsp = { path = "crates/dbsp", version = "0.320.0" } dbsp_nexmark = { path = "crates/nexmark" } deadpool-postgres = "0.14.1" -# Feldera fork of delta-rs: upstream tag `rust-v0.32.3` + 2 patches, on branch -# `v0.32.3-feldera`. When bumping the pinned revision, preserve these patches: +# 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: # - 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` +# when reading UC-Uniform-over-Iceberg tables (columnMapping.mode=id). #6557. # Diff vs. upstream: -# https://github.com/delta-io/delta-rs/compare/rust-v0.32.3...feldera:delta-rs:v0.32.3-feldera -deltalake = { git = "https://github.com/feldera/delta-rs.git", rev = "5b35c0988b07ab921d5eee5b8da8da55e4d65945" } -deltalake-catalog-unity = { git = "https://github.com/feldera/delta-rs.git", rev = "5b35c0988b07ab921d5eee5b8da8da55e4d65945" } +# 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" } delta_kernel = { package = "buoyant_kernel", version = "0.22" } derive_more = { version = "1.0.0" } diff --git a/python/tests/platform/fixtures/uniform_iceberg.py b/python/tests/platform/fixtures/uniform_iceberg.py new file mode 100644 index 00000000000..78b4c2bd980 --- /dev/null +++ b/python/tests/platform/fixtures/uniform_iceberg.py @@ -0,0 +1,217 @@ +"""Build a UC-Uniform-over-Iceberg-shaped Delta table fixture with pyarrow. + +``tests.utils.ensure_delta_spark_fixture`` runs this as a subprocess +(``uv run --with pyarrow python uniform_iceberg.py ``). Unlike the +column-mapping fixture, this one cannot be written by Delta Spark: it reproduces +what a *native Iceberg writer* (Flink / pyiceberg, exposed as Delta by Unity +Catalog Uniform) puts on disk -- Parquet columns named by their **logical** names +carrying a Parquet ``field_id`` -- while the Delta log uses ``columnMapping.mode = +'id'``. We therefore emit the Parquet with pyarrow and hand-write the +``_delta_log``. + +The table mirrors the customer's ``cdc_raw`` shape: + +* ``id`` -- physical name ``col-100`` *diverges* from the on-disk ``id`` + (a nullable scalar resolved by field id). +* ``after`` -- a six-field nested struct whose children *also* diverge: each is + mapped to ``col-`` while the file names them logically + (``transaction__id`` ...). Under by-name resolution the struct cast + target (``col-104`` ...) shares no overlap with the file's logical + child names, the customer's "Cannot cast struct with 6 fields to 6 + fields because there is no field name overlap". +* ``op`` -- physical name ``col-102`` *diverges* and is **non-nullable**. + This is the column that fails an unpatched read with + ``Non-nullable column 'col-102' is missing from the physical schema``. + +A correct snapshot read resolves every column -- top level and the struct's +children -- by Parquet ``field_id``, so the diverging columns come back with their +real values rather than NULL (or a hard error for ``op`` / the struct cast). + +Bump ``FIXTURE_VERSION`` in ``test_delta_input_uniform_iceberg.py`` on any change +here, since a cached fixture is reused based on its path alone. +""" + +from __future__ import annotations + +import json +import os +import sys + + +# (logical child name, field id) for the six-field ``after`` struct. Children are +# mapped to diverging physical names ``col-`` in the Delta log below. +AFTER_CHILDREN = [ + ("transaction__id", 104), + ("transaction__merchant_id", 105), + ("transaction__merchant_name", 106), + ("transaction__time", 107), + ("transaction__status", 108), + ("transaction__amount", 109), +] + + +# Final logical rows expected from a snapshot read. Shared with the test via import +# so the two never drift. ``after`` is a nested row. +EXPECTED_ROWS = [ + { + "id": "txn-001", + "after": { + "transaction__id": "txn-001", + "transaction__merchant_id": "m-100", + "transaction__merchant_name": "Coffee Shop", + "transaction__time": "2026-01-01T00:00:00Z", + "transaction__status": "settled", + "transaction__amount": "12.50", + }, + "op": "c", + }, + { + "id": "txn-002", + "after": { + "transaction__id": "txn-002", + "transaction__merchant_id": "m-200", + "transaction__merchant_name": "Gas Station", + "transaction__time": "2026-01-02T00:00:00Z", + "transaction__status": "settled", + "transaction__amount": "40.00", + }, + "op": "c", + }, + { + "id": "txn-003", + "after": { + "transaction__id": "txn-003", + "transaction__merchant_id": "m-300", + "transaction__merchant_name": "Restaurant", + "transaction__time": "2026-01-03T00:00:00Z", + "transaction__status": "reversed", + "transaction__amount": "75.25", + }, + "op": "u", + }, +] + + +def _field_id(value: int) -> dict: + return {b"PARQUET:field_id": str(value).encode()} + + +def build(table_path: str) -> None: + """Create the UC-Uniform-over-Iceberg table at ``table_path``.""" + import pyarrow as pa + import pyarrow.parquet as pq + + after_type = pa.struct( + [ + pa.field(name, pa.string(), metadata=_field_id(field_id)) + for name, field_id in AFTER_CHILDREN + ] + ) + schema = pa.schema( + [ + pa.field("id", pa.string(), nullable=True, metadata=_field_id(100)), + pa.field("after", after_type, nullable=True, metadata=_field_id(103)), + pa.field("op", pa.string(), nullable=False, metadata=_field_id(102)), + ] + ) + table = pa.table( + { + "id": [r["id"] for r in EXPECTED_ROWS], + "after": [r["after"] for r in EXPECTED_ROWS], + "op": [r["op"] for r in EXPECTED_ROWS], + }, + schema=schema, + ) + + parquet_rel = "part-00000-uniform-iceberg.parquet" + log_dir = os.path.join(table_path, "_delta_log") + os.makedirs(log_dir, exist_ok=True) + # store_schema=False keeps the file Iceberg-shaped (no embedded Arrow schema); + # the field_ids are written into the Parquet schema regardless. + pq.write_table(table, os.path.join(table_path, parquet_rel), store_schema=False) + size = os.path.getsize(os.path.join(table_path, parquet_rel)) + + _write_log(log_dir, parquet_rel, size) + + +def _column_metadata(field_id: int, physical_name: str) -> dict: + return { + "delta.columnMapping.id": field_id, + "delta.columnMapping.physicalName": physical_name, + } + + +def _schema_string() -> str: + # Every field -- top level and the struct's children -- uses a diverging + # `col-` physical name, exactly as a native Iceberg writer's Uniform log + # does, so a correct read must resolve by field id at every level. + after_type = { + "type": "struct", + "fields": [ + { + "name": name, + "type": "string", + "nullable": True, + "metadata": _column_metadata(field_id, f"col-{field_id}"), + } + for name, field_id in AFTER_CHILDREN + ], + } + fields = [ + { + "name": "id", + "type": "string", + "nullable": True, + "metadata": _column_metadata(100, "col-100"), + }, + { + "name": "after", + "type": after_type, + "nullable": True, + "metadata": _column_metadata(103, "col-103"), + }, + { + "name": "op", + "type": "string", + "nullable": False, + "metadata": _column_metadata(102, "col-102"), + }, + ] + return json.dumps({"type": "struct", "fields": fields}) + + +def _write_log(log_dir: str, parquet_rel: str, size: int) -> None: + # Legacy column-mapping protocol (reader 2 / writer 5): no feature lists, + # which the delta kernel rejects below reader version 3. + protocol = {"protocol": {"minReaderVersion": 2, "minWriterVersion": 5}} + metadata = { + "metaData": { + "id": "uniform-iceberg-fixture-0000-0000-000000000000", + "format": {"provider": "parquet", "options": {}}, + "schemaString": _schema_string(), + "partitionColumns": [], + "configuration": { + "delta.columnMapping.mode": "id", + "delta.columnMapping.maxColumnId": "109", + }, + "createdTime": 1700000000000, + } + } + add = { + "add": { + "path": parquet_rel, + "partitionValues": {}, + "size": size, + "modificationTime": 1700000000000, + "dataChange": True, + } + } + with open(os.path.join(log_dir, "00000000000000000000.json"), "w") as f: + for entry in (protocol, metadata, add): + f.write(json.dumps(entry) + "\n") + + +if __name__ == "__main__": + if len(sys.argv) != 2: + raise SystemExit("usage: uniform_iceberg.py ") + build(sys.argv[1]) diff --git a/python/tests/platform/test_delta_input_uniform_iceberg.py b/python/tests/platform/test_delta_input_uniform_iceberg.py new file mode 100644 index 00000000000..a32766d348c --- /dev/null +++ b/python/tests/platform/test_delta_input_uniform_iceberg.py @@ -0,0 +1,139 @@ +"""Snapshot read of a UC-Uniform-over-Iceberg table (``columnMapping.mode='id'``). + +A Unity Catalog Uniform table exposes Iceberg data files as Delta. A native +Iceberg writer (Flink / pyiceberg) names the Parquet columns by their *logical* +names and identifies them by Parquet ``field_id``; the synthesized Delta log uses +``columnMapping.mode='id'`` with physical names ``col-``. A correct snapshot +read must resolve each column by field id, not by physical name -- otherwise a +column whose physical name (``col-``) is absent from the logically-named file +reads as NULL (nullable) or fails outright (non-nullable). + +This reproduces the customer's ``cdc_raw`` failure on two fronts: the non-nullable +top-level ``op`` (physical name ``col-102``) is missing from the physical schema +under by-name resolution, and the nested ``after`` struct's six children are +*themselves* mapped to ``col-`` -- so a by-name struct cast finds no overlap +with the file's logical child names ("Cannot cast struct with 6 fields to 6 fields +because there is no field name overlap"). A correct read resolves by field id at +every level. The fixture is written with pyarrow (Delta Spark cannot produce +logical-name-on-disk Parquet); the builder lives in ``fixtures/uniform_iceberg.py`` +and runs via ``uv run --with pyarrow``. +""" + +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.uniform_iceberg import EXPECTED_ROWS +from tests.utils import DeltaTestLocation, ensure_delta_spark_fixture + +TABLE = "t" +CONNECTOR = "delta_in" +# Bump to invalidate cached MinIO/local copies when the fixture definition changes. +FIXTURE_VERSION = "v1" + +# pyarrow builder that writes the logical-name Parquet + hand-written _delta_log. +# It runs in a subprocess (see ensure_delta_spark_fixture) with only pyarrow. +_FIXTURE_BUILDER = Path(__file__).parent / "fixtures" / "uniform_iceberg.py" + + +def _build_sql(loc: DeltaTestLocation) -> str: + connectors = json.dumps( + [ + { + "name": CONNECTOR, + "transport": { + "name": "delta_table_input", + "config": dict(loc.connector_config), + }, + } + ] + ).replace("'", "''") + # `op` is declared NOT NULL to match the customer; `after` is the customer's + # six-field nested ROW whose children are column-mapped to diverging `col-` + # names. Pre-fix, the snapshot fails resolving `op`/`col-102` and casting the + # `after` struct ("no field name overlap"). + return ( + f"CREATE TABLE {TABLE} (" + "id VARCHAR," + "after ROW(" + "transaction__id VARCHAR," + "transaction__merchant_id VARCHAR," + "transaction__merchant_name VARCHAR," + "transaction__time VARCHAR," + "transaction__status VARCHAR," + "transaction__amount VARCHAR" + ")," + "op VARCHAR NOT NULL" + f") WITH ('materialized' = 'true', 'connectors' = '{connectors}');" + ) + + +def _snapshot_rows(pipeline) -> list[dict]: + # Project the two diverging top-level scalars plus a nested child of `after`. + # The nested projection exercises field-id resolution *inside* the struct -- + # the case that fails the struct cast pre-fix, not just the top-level columns. + rows = pipeline.query( + f"SELECT id, op, after.transaction__merchant_name AS merchant FROM {TABLE}" + ) + return sorted( + ({"id": r["id"], "op": r["op"], "merchant": r["merchant"]} for r in rows), + key=lambda r: r["id"], + ) + + +def test_delta_input_uniform_iceberg_id_snapshot(pipeline_name): + """Snapshot read of a UC-Uniform-over-Iceberg table resolves columns by + Parquet field id at every level: the diverging non-nullable ``op`` (``col-102``) + comes back with real values instead of failing, and the nested ``after`` struct + -- whose children are themselves mapped to ``col-`` -- is reconstructed by + field id instead of failing the struct cast.""" + loc = DeltaTestLocation.create( + pipeline_name, + mode="snapshot", + stable_subpath=f"uniform_iceberg_id_{FIXTURE_VERSION}", + ) + try: + ensure_delta_spark_fixture( + loc, _FIXTURE_BUILDER, delta_spark_spec="pyarrow>=15" + ) + + 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) + + expected = sorted( + ( + { + "id": r["id"], + "op": r["op"], + "merchant": r["after"]["transaction__merchant_name"], + } + for r in EXPECTED_ROWS + ), + key=lambda r: r["id"], + ) + rows = _snapshot_rows(pipeline) + assert rows == expected, ( + "snapshot read must resolve column-mapped Uniform/Iceberg data by " + "field id at every level (top-level op/col-102 and the nested after " + f"struct's children); got {rows}" + ) + + pipeline.stop(force=True) + finally: + loc.cleanup()