JingsongLi commented on code in PR #9239:
URL: https://github.com/apache/paimon/pull/9239#discussion_r3789614798
##########
paimon-python/pypaimon/write/table_update.py:
##########
@@ -228,17 +228,17 @@ def _update_by_predicate(
) -> List[CommitMessage]:
"""Shared implementation for SQL-like ``UPDATE ... WHERE ...``.
- ``predicate`` identifies the target rows. ``assignments`` maps target
- column names to literal values. The method reads matching ``_ROW_ID``
- values, builds an Arrow update table, then delegates to the existing
- row-id update path.
+ ``predicate`` identifies the target rows. Assignment values may be
+ literals or callables receiving the matched rows as an Arrow table.
"""
self._validate_predicate_update(assignments)
scan_table = self._matched_update_scan_table()
read_builder = scan_table.new_read_builder()
if predicate is not None:
read_builder.with_filter(predicate)
+ if predicate is not None or any(
+ callable(value) for value in assignments.values()):
Review Comment:
**[P1] Avoid materializing every column for full-table callables**
With `predicate=None`, any callable takes this branch, so even an assignment
that reads one scalar column projects every field and `to_arrow(splits)`
retains the whole table. Because `blob-as-descriptor` defaults to false and
BLOB deferral is disabled for predicate/limit-free reads, this also resolves
every BLOB payload before the callback runs. On production-scale or multimodal
tables this makes a single-column update require unbounded driver memory and
full BLOB I/O. Please require declared input columns and execute bounded
batches/ranges, or reject unbounded callable updates.
##########
paimon-python/pypaimon/write/table_update.py:
##########
@@ -313,19 +312,22 @@ def _validate_predicate_update(self, assignments:
Mapping[str, Any]):
def _build_predicate_update_table(
self,
- row_ids,
assignments: Mapping[str, Any],
- row_count: int,
+ matched: pa.Table,
) -> pa.Table:
table_schema = PyarrowFieldParser.from_paimon_schema(
self.table.table_schema.fields
)
- arrays = [row_ids]
+ arrays = [matched[SpecialFields.ROW_ID.name]]
fields = [pa.field(SpecialFields.ROW_ID.name, pa.int64())]
for col, value in assignments.items():
+ if callable(value):
+ value = value(matched)
target_field = table_schema.field(col)
arrays.append(
- self._assignment_to_array(value, target_field.type, row_count)
+ self._assignment_to_array(
+ value, target_field.type, matched.num_rows
Review Comment:
**[P2] Preserve chunked callable outputs**
Arrow kernels and identity transforms over `rows[col]` normally return a
`ChunkedArray`, but `_assignment_to_array` immediately calls
`combine_chunks()`. That fails for otherwise valid multi-chunk string/list
values whose aggregate 32-bit offsets exceed 2 GiB. I reproduced `ArrowInvalid:
offset overflow` on this head with two valid list chunks totaling 2.2B
children. Please keep/cast chunks independently and pass the chunked result
through `Table.from_arrays`, with a large-offset regression test.
##########
paimon-python/pypaimon/write/table_update.py:
##########
@@ -252,9 +252,8 @@ def _update_by_predicate(
return []
update_table = self._build_predicate_update_table(
- matched[SpecialFields.ROW_ID.name],
assignments,
- matched.num_rows,
+ matched,
Review Comment:
**[P1] Pin the updater to the snapshot that produced `matched`**
A callable can run for an arbitrary time after this table is read, but
`TableUpdateByRowId(self.table, ...)` is created afterward and records the
then-latest snapshot as `check_from_snapshot`. If another writer changes the
same row/column during the callback, the callback still returns a value derived
from the old snapshot while the newer commit is treated as the base, so the
stale update commits without a conflict. On this head I reproduced `age: 35 ->
concurrent 100 -> final 36`, with scan snapshot 2 but message baseline 3.
Please pin the updater/file metadata and conflict baseline to this scan plan's
snapshot before invoking user code, and add a concurrent-callback regression
test.
##########
paimon-python/pypaimon/write/table_update.py:
##########
@@ -313,19 +312,22 @@ def _validate_predicate_update(self, assignments:
Mapping[str, Any]):
def _build_predicate_update_table(
self,
- row_ids,
assignments: Mapping[str, Any],
- row_count: int,
+ matched: pa.Table,
) -> pa.Table:
table_schema = PyarrowFieldParser.from_paimon_schema(
self.table.table_schema.fields
)
- arrays = [row_ids]
+ arrays = [matched[SpecialFields.ROW_ID.name]]
fields = [pa.field(SpecialFields.ROW_ID.name, pa.int64())]
for col, value in assignments.items():
+ if callable(value):
+ value = value(matched)
Review Comment:
**[P1] Track callable read dependencies in conflict detection**
The callable may read any column in `matched`, but the generated
delta/conflict metadata contains only `assignments.keys()`. A concurrent change
to a read dependency therefore does not conflict. Two transforms staged from
the same snapshot—`name = f(id)` and `id = g(name)`—both committed on this head
and left `(id=5, name='id-1')`, which neither serial order can produce. This
remains reproducible even when the messages are forced back to the correct
read-snapshot baseline. Please require/track callable `read_columns` and
include them in conflict detection, or conservatively conflict on concurrent
updates to the matched row range.
--
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]
