[python][ray] Add distributed read_by_row_id for data-evolution tables#8465
Merged
Conversation
The read-side mirror of update_by_row_id: read columns (including blob) of a data-evolution table straight from a Ray Dataset that carries _ROW_ID, without reading or joining the whole target. Each row id is routed to the data file owning it via the manifest first-row-id index, and only those files -- and only the matched rows, via IndexedSplit row-range slicing (native row-index pushdown for blob format) -- are read. Pairs with bucket_join, which produces the row ids. Returns a ray.data.Dataset of (*projection, _ROW_ID). Requires a target with data-evolution.enabled + row-tracking.enabled; deletion-vectors tables are rejected for now (a DV-deleted row still lives in its data file, so row-id slicing cannot tell it apart without extra reads).
…split merge Self-review fixes: - _read_group emitted a wrong-shaped block ([_ROW_ID] only) for an empty group, which would clash with the projected schema of non-empty blocks. Emit a correctly-typed empty block built from the projection schema instead. - Add a test that update_by_row_id splits a row's range across the original file and a column delta, then read_by_row_id merges them (updated column from the delta, untouched column from the original) -- exercises the multi-file merge path the other tests skip.
…; compress ranges Address review feedback on apache#8465: - Empty source against a non-empty target returned a schema-less Dataset (a groupby over zero rows yields zero groups, so the read map never runs). Short- circuit an empty source to a typed empty Dataset so the output always carries (projection..., _ROW_ID). Add a test. - Add row_id_col (default _ROW_ID) so a bucket_join locator's "row_id" column can be used directly without a rename step. Add a test. - Compress contiguous row ids into ranges via Range.to_ranges instead of one Range(r, r) per row, cutting memory/CPU for dense groups.
…r reuse Address second review pass on apache#8465: - Extract _read_output_schema() and use it for both empty-result paths (distributed_read_by_row_id and read_by_row_id._empty_result) so the empty-read schema can't drift from a non-empty one. - Comment that TableUpdateByRowId is reused purely as a read-only planner (its constructor only scans the manifest -- no staging dirs, locks or commits; verified it writes nothing and has no resources to release). - Comment that _read_group uses the group only for row ids and that output columns come from the fresh projected read, so the injected frid column never leaks.
Spell out in the docstring that the API has SQL-IN-like lookup semantics: one row per distinct matched row id (duplicates deduplicated), non-row_id_col source columns dropped, input order not preserved, empty source -> empty typed Dataset.
- Trim inline comments to be concise. - Add a 'Read By Row Id' section to docs/docs/pypaimon/ray-data.md mirroring the Update By Row Id section (usage, parameters, returns, lookup/set semantics).
…ts/docs Address review feedback: - Vectorize the foreign-row-id check with a single searchsorted over the sorted+merged valid ranges (O(rows log ranges) instead of rows x ranges), matching distributed_delete_apply. Add a foreign-row-id test. - Trim comments and the internal docstrings to the essential "why". - Document that projection is top-level only, and that a materialized row_ids source is preferred (the emptiness check reads one block up front).
The per-group read used the unpinned table while the planner routed against the base-snapshot-pinned scan_table; align the read to scan_table (as the delete path does) so files and the read snapshot can't drift under a concurrent commit. Reads are of explicit pinned splits, so this is defensive, but consistent.
QuakeWang
approved these changes
Jul 5, 2026
QuakeWang
left a comment
QuakeWang
left a comment
Member
There was a problem hiding this comment.
LGTM overall. I left one minor non-blocking comment about preserving Arrow field nullability for empty-result schemas.
…ult schema _read_output_schema rebuilt fields from type only, dropping nullability, so an empty result exposed all-nullable fields while a non-empty read has a non-nullable _ROW_ID (and preserves NOT NULL columns) -- the empty/non-empty schemas drifted. Keep full.field(col) for table columns and build _ROW_ID as non-nullable. Verified the empty schema now equals a non-empty read's; add a regression test. (Thanks @QuakeWang.) Also drop non-essential test comments.
Contributor
|
+1 |
JingsongLi
pushed a commit
that referenced
this pull request
Jul 6, 2026
…#8468) Two follow-ups to #8465. - `dynamic_options`: a dict applied via table.copy (as a normal read does), so callers can override read options on a target passed by name — mainly `{"blob-as-descriptor": "true"}` to return small BlobDescriptor bytes instead of materialising large blobs, resolved later with map_with_blobs. Distributed path unchanged. - `Lazy Dataset`: two upfront checks executed the source at call time (so a bucket_join source ran twice). Now the schema is read with fetch_if_missing=False and the empty-source precheck is dropped — limit(1).count() runs only in the empty-target branch (like update_by_row_id), schema preserved by unioning a typed-empty block.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
Add Ray support for distributed
read_by_row_idon data-evolution tables.This is the read-side counterpart of
update_by_row_id. It allows a Ray Dataset carrying target_ROW_IDs, for example produced bybucket_join, to read projected columns from the target table without scanning or joining the whole target table.This is useful for workflows such as:
_ROW_IDs withbucket_joinupdate_by_row_idTests
ray_read_by_row_id_test.py