Skip to content

Commit 10b74c3

Browse files
AyushPatel101Ayush Patel
andauthored
fix: cast join cols to canonical Arrow types in Table.merge() (#11)
pa.Table.join is strict about the small-vs-large variant pair (string vs large_string, binary vs large_binary). Both variants are semantically the same Iceberg type but pyarrow.join refuses to bridge them. ArrowScan and a user-built dataframe can disagree on which variant they produce - pandas->arrow typically yields the small variant while ArrowScan returns whichever variant the parquet metadata implies. This caused merge() to fail at the anti-join with: ArrowInvalid: Incompatible data types for corresponding join field keys: FieldRef.Name(<col>) of type string and FieldRef.Name(<col>) of type large_string The original merge() tests only exercised int32 join columns, which is the only type with no small/large variant pair, so this slipped through. Fix: before the anti-join, cast the join columns on both target_data and the source df to the canonical Iceberg-derived Arrow form via ``self.table_metadata.schema().as_arrow()``. Mirrors what ArrowProjectionVisitor._cast_if_needed does internally per pa.Array, applied at the pa.Table level. Source NULL and NaN values in join columns continue to be rejected by the In predicate constructor (which forbids NULL and NaN literals). Target rows with NULL/NaN keys are correctly preserved by the anti-join since their keys cannot match any source key under SQL three-valued logic / IEEE 754 semantics. Tests: tests/table/test_merge.py covers string/binary variant drift on join cols (every direction), composite keys with drift, edge values (empty, unicode, embedded NULs, long strings, high-bit bytes), NULL/NaN handling, and row-preservation / atomicity invariants. Co-authored-by: Ayush Patel <Ayush.Patel@imc.com>
1 parent 4b7d642 commit 10b74c3

2 files changed

Lines changed: 1701 additions & 2 deletions

File tree

pyiceberg/table/__init__.py

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -963,6 +963,22 @@ def merge(
963963
format_version=self.table_metadata.format_version,
964964
)
965965

966+
# Cast join cols on both sides to the canonical iceberg-derived
967+
# Arrow types (StringType -> large_string, BinaryType -> large_binary,
968+
# etc.) before building the In filter and the anti-join. pa.Table.join
969+
# is strict about the string/large_string and binary/large_binary
970+
# variant pair even though both are semantically the same Iceberg
971+
# type; the source and target can disagree on physical encoding
972+
# depending on how each was constructed (pandas->arrow typically
973+
# produces the small variant; parquet metadata can dictate either).
974+
# Normalize both to the iceberg-derived form so the join is well-defined.
975+
iceberg_arrow_schema = self.table_metadata.schema().as_arrow()
976+
977+
def _cast_join_cols(t: pa.Table) -> pa.Table:
978+
return t.cast(pa.schema([iceberg_arrow_schema.field(f.name) if f.name in join_cols else f for f in t.schema]))
979+
980+
df = _cast_join_cols(df)
981+
966982
# Step 1: Build per-column In filters for file pruning.
967983
# O(sum of cardinalities) instead of O(product).
968984
# Over-approximates the match set, which is fine - row-level
@@ -991,7 +1007,7 @@ def merge(
9911007
row_filter=ALWAYS_TRUE,
9921008
case_sensitive=True,
9931009
)
994-
target_data = arrow_scan.to_table(tasks)
1010+
target_data = _cast_join_cols(arrow_scan.to_table(tasks))
9951011
source_keys = df.select(join_cols)
9961012

9971013
kept_rows = target_data.join(source_keys, keys=join_cols, join_type="left anti")

0 commit comments

Comments
 (0)