JingsongLi commented on code in PR #9239:
URL: https://github.com/apache/paimon/pull/9239#discussion_r3790927830
##########
paimon-python/pypaimon/write/table_update.py:
##########
@@ -225,40 +225,50 @@ 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)
Review Comment:
**[P1] Stream callable updates by complete row-id file group**
Requiring a predicate limits syntax but not match cardinality, so this still
materializes every matched row and the callback output at once. Callable
execution can be streamed if batching becomes part of the public contract:
callbacks may be invoked zero, one, or multiple times, receive a subset of
matched rows, must be deterministic, side-effect-free, and row-local, and must
return one value per input row. Please reuse the pinned scan plan and file
metadata, process one complete `first_row_id` logical file group per invocation
(rather than arbitrary record batches, which could create overlapping delta
files for the same row-id range), accumulate normal `CommitMessage`s, commit
once atomically, and abort all staged files if a later group fails. This bounds
peak memory by the largest logical file group and can extend the file-group
path in #9240 to callables. Whole-result transforms such as ranking or global
normalization should use a separate two-pass API.
--
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]
