Copilot commented on code in PR #3862:
URL: https://github.com/apache/iceberg-python/pull/3862#discussion_r3861532857
##########
pyiceberg/table/upsert_util.py:
##########
@@ -53,17 +53,86 @@ def has_duplicate_rows(df: pyarrow_table, join_cols:
list[str]) -> bool:
return len(df.select(join_cols).group_by(join_cols).aggregate([([],
"count_all")]).filter(pc.field("count_all") > 1)) > 0
+# How many values of a column are turned into Python objects at a time, when
PyArrow cannot
+# compare them itself
+_PYTHON_COMPARISON_SLICE = 10_000
+
+
+def _get_changed_struct_mask(source_column: pa.ChunkedArray, target_column:
pa.ChunkedArray) -> pa.ChunkedArray:
+ """Compare two struct columns field by field, which PyArrow can do even
though it cannot compare the structs."""
+ # `struct_field` carries the null of the struct into its fields, so the
fields of two null
+ # structs compare equal and only the struct itself decides for those rows
+ changed = pc.not_equal(pc.is_null(source_column),
pc.is_null(target_column))
+
+ for index in range(source_column.type.num_fields):
+ changed = pc.or_(changed,
_get_changed_mask(pc.struct_field(source_column, index),
pc.struct_field(target_column, index)))
+
+ return changed
+
+
+def _get_changed_mask(source_column: pa.ChunkedArray, target_column:
pa.ChunkedArray) -> pa.ChunkedArray:
+ """Return a boolean mask that flags the positions where the two columns
differ, treating two nulls as equal."""
+ try:
+ differs = pc.not_equal(source_column, target_column)
+ except (pa.ArrowNotImplementedError, pa.ArrowInvalid):
+ # PyArrow cannot compare columns with complex types
+ # See: https://github.com/apache/arrow/issues/35785
+ if pa.types.is_struct(source_column.type) and source_column.type ==
target_column.type:
+ return _get_changed_struct_mask(source_column, target_column)
+
+ # Two columns PyArrow refuses to compare may still hold the same
values in another type:
+ # a naive timestamp against a zoned one, or the string of a dataframe
against the
+ # large_string a scan reads. Comparing those in Python would call
every row changed, on
+ # every run, and would leave a struct out of the comparison by field
above. The types have
+ # to differ for this to make progress, the cast leaves them equal and
the next round
+ # settles it one way or the other
+ if source_column.type != target_column.type:
+ try:
+ return
_get_changed_mask(source_column.cast(target_column.type), target_column)
+ except pa.ArrowException:
+ # Whatever PyArrow makes of the cast, the comparison in Python
below still holds
+ pass
+
+ # A list or a map is left to be compared in Python, value by value. A
slice at a time,
+ # so that the objects of a whole column are never held at once
+ return pa.chunked_array(
+ [
+ [
+ source_val != target_val
+ for source_val, target_val in zip(
+ source_column.slice(offset,
_PYTHON_COMPARISON_SLICE).to_pylist(),
+ target_column.slice(offset,
_PYTHON_COMPARISON_SLICE).to_pylist(),
+ strict=True,
+ )
+ ]
+ for offset in range(0, len(source_column),
_PYTHON_COMPARISON_SLICE)
+ ]
+ or [[]],
+ type=pa.bool_(),
+ )
+
+ # `not_equal` is null as soon as either side is null, and a null differs
from a value
+ # but not from another null
+ return pc.fill_null(differs, pc.not_equal(pc.is_null(source_column),
pc.is_null(target_column)))
+
+
def get_rows_to_update(source_table: pa.Table, target_table: pa.Table,
join_cols: list[str]) -> pa.Table:
"""
Return a table with rows that need to be updated in the target table based
on the join columns.
The table is joined on the identifier columns, and then checked if there
are any updated rows.
Those are selected and everything is renamed correctly.
"""
- all_columns = set(source_table.column_names)
- join_cols_set = set(join_cols)
+ if set(source_table.column_names) != set(target_table.column_names):
+ raise ValueError(
+ f"Source table's field names are not matching the target's field
names: "
+ f"{source_table.column_names}, {target_table.column_names}"
+ )
Review Comment:
Using `set(...)` to validate field names can incorrectly pass when duplicate
column names exist (sets drop duplicates), which can lead to ambiguous
`table.column(name)` resolution and incorrect comparisons/writes. Prefer
validating with multiplicity preserved (e.g., compare
`Counter(source_table.column_names)` vs `Counter(target_table.column_names)`)
or use schema/field-based equality that doesn’t collapse duplicates.
##########
pyiceberg/table/upsert_util.py:
##########
@@ -53,17 +53,86 @@ def has_duplicate_rows(df: pyarrow_table, join_cols:
list[str]) -> bool:
return len(df.select(join_cols).group_by(join_cols).aggregate([([],
"count_all")]).filter(pc.field("count_all") > 1)) > 0
+# How many values of a column are turned into Python objects at a time, when
PyArrow cannot
+# compare them itself
+_PYTHON_COMPARISON_SLICE = 10_000
+
+
+def _get_changed_struct_mask(source_column: pa.ChunkedArray, target_column:
pa.ChunkedArray) -> pa.ChunkedArray:
+ """Compare two struct columns field by field, which PyArrow can do even
though it cannot compare the structs."""
+ # `struct_field` carries the null of the struct into its fields, so the
fields of two null
+ # structs compare equal and only the struct itself decides for those rows
+ changed = pc.not_equal(pc.is_null(source_column),
pc.is_null(target_column))
+
+ for index in range(source_column.type.num_fields):
+ changed = pc.or_(changed,
_get_changed_mask(pc.struct_field(source_column, index),
pc.struct_field(target_column, index)))
+
+ return changed
+
+
+def _get_changed_mask(source_column: pa.ChunkedArray, target_column:
pa.ChunkedArray) -> pa.ChunkedArray:
+ """Return a boolean mask that flags the positions where the two columns
differ, treating two nulls as equal."""
+ try:
+ differs = pc.not_equal(source_column, target_column)
+ except (pa.ArrowNotImplementedError, pa.ArrowInvalid):
+ # PyArrow cannot compare columns with complex types
+ # See: https://github.com/apache/arrow/issues/35785
+ if pa.types.is_struct(source_column.type) and source_column.type ==
target_column.type:
+ return _get_changed_struct_mask(source_column, target_column)
+
+ # Two columns PyArrow refuses to compare may still hold the same
values in another type:
+ # a naive timestamp against a zoned one, or the string of a dataframe
against the
+ # large_string a scan reads. Comparing those in Python would call
every row changed, on
+ # every run, and would leave a struct out of the comparison by field
above. The types have
+ # to differ for this to make progress, the cast leaves them equal and
the next round
+ # settles it one way or the other
+ if source_column.type != target_column.type:
+ try:
+ return
_get_changed_mask(source_column.cast(target_column.type), target_column)
+ except pa.ArrowException:
+ # Whatever PyArrow makes of the cast, the comparison in Python
below still holds
+ pass
+
+ # A list or a map is left to be compared in Python, value by value. A
slice at a time,
+ # so that the objects of a whole column are never held at once
+ return pa.chunked_array(
+ [
+ [
+ source_val != target_val
+ for source_val, target_val in zip(
+ source_column.slice(offset,
_PYTHON_COMPARISON_SLICE).to_pylist(),
+ target_column.slice(offset,
_PYTHON_COMPARISON_SLICE).to_pylist(),
+ strict=True,
+ )
+ ]
+ for offset in range(0, len(source_column),
_PYTHON_COMPARISON_SLICE)
+ ]
+ or [[]],
+ type=pa.bool_(),
+ )
+
+ # `not_equal` is null as soon as either side is null, and a null differs
from a value
+ # but not from another null
+ return pc.fill_null(differs, pc.not_equal(pc.is_null(source_column),
pc.is_null(target_column)))
+
+
def get_rows_to_update(source_table: pa.Table, target_table: pa.Table,
join_cols: list[str]) -> pa.Table:
"""
Return a table with rows that need to be updated in the target table based
on the join columns.
The table is joined on the identifier columns, and then checked if there
are any updated rows.
Those are selected and everything is renamed correctly.
"""
- all_columns = set(source_table.column_names)
- join_cols_set = set(join_cols)
+ if set(source_table.column_names) != set(target_table.column_names):
+ raise ValueError(
+ f"Source table's field names are not matching the target's field
names: "
+ f"{source_table.column_names}, {target_table.column_names}"
Review Comment:
The new error message is actionable but can be hard to read on wide tables
because it prints both full column lists. Consider reporting a concise diff
(e.g., `missing_in_source=[...]`, `extra_in_source=[...]`) and optionally
counts, while keeping the full lists only if needed for debugging.
##########
tests/table/test_upsert.py:
##########
@@ -29,9 +29,9 @@
from pyiceberg.io.pyarrow import schema_to_pyarrow
from pyiceberg.partitioning import PartitionField, PartitionSpec
from pyiceberg.schema import Schema
-from pyiceberg.table import Table, UpsertResult
+from pyiceberg.table import Table, UpsertResult, upsert_util
from pyiceberg.table.snapshots import Operation
Review Comment:
Importing `upsert_util` via `pyiceberg.table` makes the test depend on
package re-exports; this can be brittle if `__init__` changes. Importing the
module directly (e.g., `import pyiceberg.table.upsert_util as upsert_util`) is
more explicit and keeps the dependency local to what the test actually uses.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
