fix: MERGE INTO on partitioned Iceberg tables (projection, TIMESTAMPTZ, pruning, manifest rewrite)

[rampage644]

Six commits that together let MERGE INTO work on real partitioned Iceberg targets, verified end-to-end against a 161 GB / 143M-row Snowplow events_hooli table on S3 Tables through a deployed Embucket Lambda. Supersedes and folds in #56.

The fixes

1. v2 manifest-list field renames + TIMESTAMPTZ date_transform + identity partition column filter (fe678ad) — three small wire-format / type-check fixes needed to even read real v2 Iceberg tables:

   • manifest_list.rs: declare count fields with #[serde(rename = "added_data_files_count", alias = "added_files_count")] so manifest lists written by current Apache Iceberg (>= 1.0, which renamed the v2 field names) deserialize cleanly. Regression test test_manifest_list_v2_apache_field_names.
   • pruning_statistics.rs::DateTransform: replace the hardcoded OneOf(Exact(...)) signature with TypeSignature::UserDefined + coerce_types so the date_transform UDF accepts timezone-aware timestamps and normalizes them to Timestamp(Microsecond, None) for the transform call.
   • table.rs::datafusion_partition_columns: skip identity-self-named partition fields so DataFusions parquet reader doesnt trip on the duplicate-column "expected N cols but got N+1" error.

2. refactor(table_scan): promote identity-self-named partition filter to table_scan (b940459) — introduce file_partition_fields / drop_partition_indices at the top of table_scan and thread them through every downstream consumer. Ensures the physical scan column set and the manifest pruners partition_column_names set agree (they used to diverge). Prerequisite for the rest of the stack.

3. Projection remap to combined-schema space (690b42b) — table_scan was passing the callers provider-schema projection straight through to FileScanConfig::with_projection, which interprets indices against [file_schema, table_partition_cols]. With synthetic transform partition columns (ts_day, id_bucket, ...) added by the physical layer, the two schemas diverged and the indices for __data_file_path / __manifest_file_path shifted. Fix: compute a combined_projection once, remapping caller indices to combined-schema positions, and use it in both the no-delete and equality-delete branches. 7 new regression tests in the table.rs mod tests (day / hour / month / year / bucket / truncate / identity-renamed partition transforms all stay green on the __data_file_path scan path).

4. Accept TIMESTAMPTZ for day/hour/month/year transforms (8d242d5) — transform_arrow in iceberg-rust/src/arrow/transform.rs only matched Timestamp(Microsecond, None) for day/hour/month/year and fell through for timezone-aware timestamps with Compute error: Failed to perform transform for datatype. Widen the match arms to Timestamp(Microsecond, _). Month/year additionally cast the array to Timestamp(Microsecond, None) before calling date_part to avoid a chrono-tz dependency. 4 new regression tests.

5. PruneDataFiles arrow_schema (cac01cd) — pass the full arrow_schema to PruneDataFiles::new instead of the narrow partition_schema. The second-stage pruner looks up each filter columns datatype via arrow_schema.field_with_name(...) to read per-file lower/upper bounds from the manifest; with only partition_schema (which is built from the reduced table_partition_cols), any column not in the Hive-style partition-key set — including identity-self-named partition columns that were dropped from table_partition_cols because theyre materialized in the parquet body — silently returned None and pruned nothing. With the full arrow schema, filters like event_name = ad_start_event now actually prune files. Fixes [iceberg-rust] identity-self-named partition columns are excluded from target-side filter pruning embucket#127. Regression test test_identity_self_named_partition_filter_prunes_files asserts a 3-partition identity-partitioned table drops to 1 file under a filter.

6. Manifest DataFile stats — resolve nested column ids + propagate deserialization errors (e7b9d06) — AvroMap::into_value_map used to validate lower_bounds / upper_bounds keys against only the top-level StructType::lookup, so any nested field id (e.g. a Snowplow contexts_com_snowplowanalytics_*.* leaf at id 479) returned ColumnNotInSchema. The error was then .unwrap()d inside from_existing_with_filters per-entry closure, panicking the tokio worker and SIGABRT-ing the whole Lambda on any MERGE that rewrote an existing manifest. Three changes:

   • New StructType::field_by_id(id) that walks nested Struct, List, Map types recursively.
   • into_value_map now resolves ids via field_by_id and silently skips unknown ids, matching Icebergs schema-evolution semantics (old stats on removed fields must not break readers).
   • The filter_map(...).unwrap() in from_existing_with_filter is replaced by an explicit for loop that propagates per-entry errors via ?, so any future deserialization edge case surfaces as a clean Error instead of a panic.
   • Two new regression tests: types::tests::field_by_id_finds_nested_fields (top-level + struct-of-struct + list + map<string, struct> + unknown id) and manifest::tests::into_value_map_accepts_nested_field_ids (nested decode + top-level decode + silent-skip).

Tests

• cargo test -p iceberg-rust --lib — 89 / 89 green.
• cargo test -p iceberg-rust-spec --lib — 88 / 88 green (includes the new nested-id tests).
• cargo test -p datafusion_iceberg --lib — 19 new/changed tests pass. Eight pre-existing failures on fix/v2-manifest-field-names (tokio-runtime, test_datafusion_table_insert_partitioned, materialized view tests) are unrelated and still pre-existing.

End-to-end proof on a patched Embucket Lambda

Deployed the combined build (this PR + Embucket/embucket#126) to embucket-demo-embucket-demo-ramp-1775514830 and ran, against the real 143M-row / 161 GB Snowplow events_hooli:

MERGE INTO demo.atomic.events_hooli AS t
USING demo.atomic.events_hooli_ident_matched AS s
  ON t.event_id = s.event_id
WHEN MATCHED THEN UPDATE SET etl_tags = <probe>;

• Source: 10 rows in an identity-partitioned scratch table (single event_name = ad_start_event partition).
• Target: events_hooli partitioned by day(collector_tstamp) + identity(event_name).
• Pre-fix progression: (a) Unsupported SQL statement: MERGE INTO on EXPLAIN paths → (b) Schema error: No field named event_name in the analyzer → (c) Input field name <col>_<transform> does not match with the projection expression __data_file_path at plan time → (d) Compute error: Failed to perform transform for datatype on TIMESTAMPTZ → (e) scan-time "expected N cols but got N+1" from the identity-partition duplicate → (f) partition pruning silently dropping nothing → (g) panicked at iceberg-rust/src/table/manifest.rs:549: Column 479 not in schema during manifest rewrite, SIGABRT. Each commit in this PR addresses one of these failure modes.
• Post-fix: MERGE commits in 10.05 s wall, MergeIntoSinkExec, metrics=[updated_rows=10, inserted_rows=0, deleted_rows=0], 10 rows carry the new tag, 0 rows bleed into other partitions.

EXPLAIN ANALYZE confirms plan-time pruning: the target DataSourceExec shows file_groups={6 groups} but every path in it sits under event_name=ad_start_event (one ~644 MB partition file split into 6 byte-range groups for parallel scan). Every other partition file in the table was pruned at plan time by iceberg-rusts manifest pruner before the physical plan was built. Scan output = 628,274 rows; HashJoin build = 10 rows with build_mem_used=1212 bytes; sink rewrites the partition file with the 10 updates applied in-place via MERGE COW.

Covered by 14 new regression tests plus this end-to-end verification against real Snowplow data on S3 Tables.