JingsongLi commented on code in PR #9239:
URL: https://github.com/apache/paimon/pull/9239#discussion_r3790847572
##########
paimon-python/pypaimon/write/table_update.py:
##########
@@ -225,40 +225,61 @@ def _update_by_predicate(
predicate: Optional[Predicate],
assignments: Mapping[str, Any],
commit_identifier: int,
+ read_columns: Optional[Sequence[str]] = None,
) -> 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)
+ has_callable = any(callable(value) for value in assignments.values())
+ read_columns = tuple(read_columns or ())
+ self._validate_predicate_update(
+ predicate, assignments, read_columns, has_callable
+ )
scan_table = self._matched_update_scan_table()
read_builder = scan_table.new_read_builder()
if predicate is not None:
read_builder.with_filter(predicate)
- read_builder.with_projection(
- list(scan_table.field_names) + [SpecialFields.ROW_ID.name]
- )
+ if has_callable:
+ projection = list(dict.fromkeys(read_columns))
+ projection.append(SpecialFields.ROW_ID.name)
+ read_builder.with_projection(projection)
else:
read_builder.with_projection([SpecialFields.ROW_ID.name])
scan = read_builder.new_scan()
- splits = scan.plan_for_write().splits()
+ plan = scan.plan_for_write()
+ splits = plan.splits()
matched = read_builder.new_read().to_arrow(splits)
if matched.num_rows == 0:
return []
update_table = self._build_predicate_update_table(
- matched[SpecialFields.ROW_ID.name],
assignments,
- matched.num_rows,
+ matched,
)
- return TableUpdateByRowId(
+ snapshot_id = plan.snapshot_id if plan.snapshot_id is not None else -1
+ files_info = TableUpdateByRowId._files_info_from_splits(
+ snapshot_id, splits
+ )
+ messages = TableUpdateByRowId(
self.table, self.commit_user, commit_identifier,
+ _precomputed_files_info=files_info,
).update_columns(update_table, list(assignments.keys()))
+ predicate_cols = (
+ self._predicate_fields(predicate) if predicate is not None else
set()
+ )
+ conflict_cols = list(dict.fromkeys(
+ list(assignments.keys())
+ + list(read_columns)
+ + [col for col in self.table.field_names if col in predicate_cols]
+ ))
+ for message in messages:
+ for file in message.new_files:
+ file._conflict_cols = conflict_cols
Review Comment:
**[P2] Keep callable conflict semantics aligned with Java**
Could we avoid introducing `_conflict_cols` here? Java's
`RowIdColumnConflictChecker` derives conflicts solely from
`DataFileMeta.writeCols`, which matches Paimon's documented snapshot-isolation
guarantee: concurrent writes to different columns may coexist, while
overlapping physical writes conflict. Treating predicate and read columns as
conflict columns gives this Python API a stronger serializable-like contract,
diverges from Java, and introduces transient transaction metadata that must
survive rewrite, retry, and message transport. Please keep the pinned-snapshot
fix, rely on physical `write_cols` for conflict detection, remove
`_conflict_cols` and the corresponding checker special case, and adjust the
docs to say that callable results are computed from the pinned snapshot rather
than promising conflicts on every predicate/read-column change.
--
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]
