Repository navigation
Conversation
|
Hi, @raygao25 . May i have your contact information, i am very interested in this PR and i hope we can communicate with E-mail or Wechat. Could you leave your e-mail here? Thanks! |
hfutatzhanghb
left a comment
hfutatzhanghb
left a comment
There was a problem hiding this comment.
Bugbot review: five correctness findings, each recorded as an inline discussion thread.
| **namespace_kwargs, | ||
| ) | ||
| current_ids = {f.fragment_id for f in current.get_fragments()} | ||
| if not touched_fragment_ids.issubset(current_ids): |
There was a problem hiding this comment.
[P1] Detect changes to touched fragment metadata before rebasing
The retry gate only checks that every touched fragment ID still exists. A concurrent delete or update can keep the same fragment ID while replacing its metadata or deletion vector. Advancing read_version and resubmitting metadata built from the old snapshot can then overwrite that concurrent change and potentially make deleted rows visible again. Please compare each touched fragment against its baseline metadata before retrying, or keep the original read version and let Lance reject the conflict.
| storage_options=storage_options or None, | ||
| **namespace_kwargs, | ||
| ) | ||
| except Exception as exc: |
There was a problem hiding this comment.
[P1] Retry only known version conflicts
This catches every exception from commit, including ambiguous transport failures. If the transaction was committed but the success response was lost, retrying the same operation can apply it twice; insert-only merges are especially exposed because touched_fragment_ids is empty and the safety check always passes. Please retry only an explicit manifest/version conflict. Other errors should propagate unless the first transaction can be identified and checked idempotently.
| append_parts = replacement_parts + insert_parts | ||
| if append_parts: | ||
| append_table = pa.concat_tables(append_parts) | ||
| fragments = lance.fragment.write_fragments( |
There was a problem hiding this comment.
[P2] Preserve stable row IDs for replacement rows
For a dataset with stable row IDs, the plan phase captures the matched target _rowid, but the apply phase drops the helper column and writes replacement rows through write_fragments without attaching row_id_meta. Those logical rows can therefore receive new stable IDs after an update. Native Lance merge_insert captures the old IDs and writes them into the replacement fragment metadata. Please preserve matched row IDs, assign new IDs only to inserts, and add a test comparing _rowid before and after the merge.
| hits.column("_rowid").to_pylist(), | ||
| strict=False, | ||
| ): | ||
| rowinfo_of[key] = (rowaddr >> 32, rowid) |
There was a problem hiding this comment.
[P2] Handle every target row matching a join key
The lookup may return multiple target rows for the same join key, but this assignment stores only one tuple per key and silently overwrites earlier matches. Only the last target row is deleted and replaced, leaving other matching rows unchanged. Please either retain and route all matching row IDs, or explicitly reject non-unique target keys if uniqueness is part of the API contract.
|
|
||
| # Ship each row's target rowid with it so every bucket slice below | ||
| # carries the ids of the target rows it replaces (-1 on insert rows). | ||
| source_chunk = source_chunk.append_column( |
There was a problem hiding this comment.
[P3] Avoid colliding with a user column
The aligned source retains every target-schema field, so a target can legitimately contain a column named __merge_into_rowid. Appending the helper with the same name creates duplicate fields, and subsequent name-based column access or removal becomes ambiguous in PyArrow. Please generate a helper name that is absent from the target schema, or keep row IDs in a separate sidecar payload.
|
Hi, @yanghua . Could you please help update pylance version to 11.0.0 ? I think we will need RowIdSequence ability which was published in v11.0.0. Thanks very much! |
Hi, my email is r25gao@gmail.com. Happy to collaborate on this! |
Hi @hfutatzhanghb in case you missed the above. |
Hi, Ray. I have send you an E-mail with this id : hfutzhanghb@163.com |
Summary
Adds
lance_ray.merge_intofor large-scale, daily merge ingestions: match a source batch to a Lance table on a join key, update existing rows, and insert new ones, committed as a single atomic version.It is the distributed counterpart of pylance’s
LanceDataset.merge_insert(on).when_matched_update_all().when_not_matched_insert_all(), for Ray jobs where the source and target are too large to process on a single machine. Unlike lance-spark’sMERGE INTO, which shuffle-joins the source against the target, this path probes a scalar index (BTREE is preferred) to map each source key to its target fragment, then applies updates.How it works
key IN (...), served by a BTREE/scalar index when present), then shuffles rows to the worker that owns that fragment.LanceOperation.Updateon the driver. Concurrent appends rebase; a concurrent rewrite of a touched fragment fails rather than silently dropping the other write.A scalar index on the join key is strongly recommended for large targets.
Performance
Measured merging 1M source rows into a 2.5B-row table with a BTREE index on the join key, using 64 Ray workers (2 cpu each):
Peak memory ~5.5 GB.
Test plan
uv run pytest tests/test_merge_into.pyuv run ruff check lance_ray testsuv run ruff format --check lance_ray tests