JingsongLi commented on PR #8826:
URL: https://github.com/apache/paimon/pull/8826#issuecomment-5207066215
I think this can be simplified by moving the abstraction one level down:
instead of introducing a specialized `update_by_transform` pipeline, provide a
generic, file-group-aligned row-id range processing primitive.
The ranges can be planned once from a pinned base snapshot and packed to
approximately `target_rows_per_range`, but boundaries should always align with
complete file groups (not individual physical files, because one logical row-id
range may be represented by a base file plus multiple data-evolution
delta/blob/vector files).
A possible low-level API is:
```python
with plan_row_id_ranges(
target="db.table",
catalog_options=options,
target_rows_per_range=1_000_000,
) as ranges:
for range_ctx in ranges:
source = range_ctx.read(
projection=["text"],
filter="language = 'en'",
)
updates = my_distributed_process(source)
range_ctx.update_by_row_id(
updates,
update_cols=["embedding"],
)
```
`RowIdRangeContext` could expose the stable public information and
operations, while keeping file metadata internal:
```python
class RowIdRangeContext:
snapshot_id
range_start
range_end
estimated_rows
sequence_number
def read(...): ...
def update_by_row_id(...): ...
```
A convenience wrapper could execute the same ranges sequentially:
```python
def process(ctx):
source = ctx.read(["text"])
updates = build_embeddings_distributed(source)
ctx.update_by_row_id(updates, ["embedding"])
process_row_id_ranges(
target="db.table",
catalog_options=options,
target_rows_per_range=1_000_000,
processor=process,
)
```
Here “sequentially” means multiple range execution/commit rounds within the
same Ray job and cluster, not submitting a brand-new Ray job per range. The
processor can construct its own Ray Dataset DAG, actor pool, GPU inference, or
even use another distributed service.
This model seems substantially simpler:
- one range follows a clear `read -> user processing -> write -> commit`
lifecycle;
- only one range has uncommitted staged files at a time;
- earlier committed ranges naturally remain visible after a later failure;
- no cross-range `groupby/map_groups`, result draining, error-as-data
protocol, or `on_group_result` callback is needed;
- commit ordering is deterministic;
- users control distributed transform execution instead of the API owning a
second scheduling layer.
If arbitrary distributed processing may reorder rows, the result needs a
stable association key. We either need to require row-count/order preservation,
or pass through an opaque internal row token. It is not possible to allow
arbitrary shuffle/repartition while also completely removing row identity from
the contract.
The commit layer still needs to protect correctness when a range reads a
pinned old snapshot: read/filter/output columns must participate in conflict
detection if concurrent non-conflicting writes are allowed. However, the exact
committed snapshot could be returned from an internal `commit(...)` result
(including the duplicate-after-uncertain-retry case), rather than adding an
operation-specific callback/state hook to the generic `TableCommit`.
Naming-wise, I suggest:
- `plan_row_id_ranges` for the low-level iterable planner;
- `process_row_id_ranges` for the managed sequential callback API;
- `target_rows_per_range` rather than `rows_per_commit`, because a generic
processor does not necessarily commit.
Then embedding backfill becomes one use of this primitive instead of
defining the primitive around transform updates.
--
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]
