[hfutatzhanghb]

Summary

• Add lance_ray.merge_into, a fragment-parallel counterpart of pylance merge_insert(on).when_matched_update_all().when_not_matched_insert_all(), committed as one atomic LanceOperation.Update.
• Align semantics with native merge: join-all on duplicate target keys, preserve stable _rowid via RowIdSequence, and delete matched rows with LanceFragment.delete_rows using local physical offsets from _rowaddr.
• Match native index maintenance: the commit carries fields_modified=[], fields_for_preserving_frag_bitmap set to every field id in pre-order (nested leaves included), and update_mode="rewrite_rows", so no existing index treats a rewritten fragment as already indexed. This mirrors the full-schema rewrite path in rust/lance/src/dataset/write/merge_insert.rs.
• Deduplicate the source globally before planning: cast every block to the target schema, range-partition with a Ray Data sort on the join key, then drop adjacent duplicates per block. Casting first means keys that only become equal after conversion (string "01" against an integer column) collapse into one. Dictionary and float16 keys are sorted and compared on a decoded or widened helper column; the original column type is kept.
• Reject non-finite floating keys up front. pc.not_equal(NaN, NaN) is true and float("nan") does not collapse inside a Python set, so neither the dedupe nor the plan-side duplicate check would catch those rows, and repr(nan) is not a Lance SQL literal.
• Harden commit: drop the outer bump-read_version retry, treat a lost success ack as success (match new fragments by data-file identity), and fail when concurrent MoR rewrites deletion metadata.
• Cut shuffle object fan-out by decoupling num_partitions (plan shards) from num_workers (apply owners) and yielding only non-empty buckets.
• Consume source payloads one bucket at a time. The first apply pass still accumulates that owner's match row ids and offsets for every matched row, about 16 bytes per row, plus temporary memory while those integers are sorted. Row ids are released after the duplicate check. Offsets are released as each fragment's deletion file is written. Identities are copied off the bucket allocation so dropping the payload can free it.
• Support common scalar join keys (bool, int, float, string, date, timestamp, time, decimal, binary), fail-fast on nested types, and allocate helper column names (including the temporary sort key) from the target schema so they cannot collide with user fields. Boundaries: the Int64 minimum and every UInt64 value at or above 263 use a typed string literal, because a bare SQL number that does not fit in a signed 64-bit token is parsed as Float64; dates outside year 1–9999 are sorted and looked up as integer ticks; negative-scale decimals are sorted on their unscaled coefficient and looked up by multiplying that coefficient by a precision-1 power of ten at the same negative scale, so a full-precision coefficient does not overflow before the final cast. When 10(-scale) has more than 76 digits, as with decimal256(76, -76), that power is a product of smaller factors that each fit in Decimal256; Decimal256 values above 38 digits use arrow_cast. A join column name containing a backtick is rejected, because Lance 12 drops that field from its query schema and a filter cannot read it. An in-memory table with string_view or binary_view columns is cast to the target schema before Ray serializes it. Stable _rowid values stay UInt64, including ids at or above 2**63; insert rows carry a null placeholder instead of -1.
• Write new fragments with the target dataset's data_storage_version.
• Resolve the source without executing it: the temporal-sort-key decision comes from target_schema instead of Dataset.schema(), so the driver no longer triggers a limit(1) run, and every Ray call result is re-bound through _require_ray_dataset so type narrowing does not depend on the Ray stubs.

Notes and limitations (expanded in docs/src/merge_into.md)

• Object store: num_partitions bounds the deserialized source one apply worker holds, not cluster object-store use. After dedupe, materialize() plus to_arrow_refs() keep the full deduplicated source resident, the driver holds every bucket reference until commit, and join-all can copy one source row into several buckets.
• Concurrency: conflict detection is fragment-level and this call does not re-plan. Serialize every writer to the table for the whole call, including ordinary appends. An append that lands mid-merge is rebased inside LanceDataset.commit and both commits succeed, so a key it inserts that this merge also inserts survives twice.
• Failures: data files from write_fragments and deletion files already written by delete_rows are not removed when a later apply task fails or the commit conflicts. Reclaim them with LanceDataset.cleanup_old_versions(). The default does not delete unverified failed-transaction files younger than 7 days. Pass delete_unverified=True only when no other process is writing the dataset.
• Write options: file-size limits and base placement (target_bases / target_all_bases) stay at the Lance writer defaults, since merge_into exposes no write-option arguments.

Test plan

• tests/test_merge_into.py covers e2e merge, insert-only, update-only, empty-source no-op, more fragments than workers, more partitions than workers, join-all within one fragment and across fragments (with stable row ids), stable _rowid preservation, no nested-index reuse after a stable-row rewrite, scalar-index merges with filtered reads on the key column, float keys with and without a BTREE, bool keys, dictionary and float16 keys, a decimal256 lookup above 38 digits, rejection of a join column whose name contains a backtick, Int64 minimum keys with and without a BTREE, UInt64 keys at and above 263 with and without a BTREE, negative-scale decimal keys, including full-precision decimal128(38, -2), decimal256(76, -2), and decimal256(76, -76) with and without a BTREE, dates outside year 1–9999, UInt64 row ids at or above 263, and an in-memory string_view/binary_view source, a second merge over an existing deletion vector, storage-version preservation, merge-on-read behavior (deletion vector written and data files untouched; a fully updated fragment is removed), dedupe within and across chunks and for keys equal only after cast, nanosecond timestamp and time keys indexed and unindexed, non-finite float key rejection, empty temporal batches keeping one schema, backtick-quoted index field names, helper-name collisions, date/decimal/binary keys, directory-namespace e2e, lost-ack and failed-commit paths, and the driver validation matrix. Unit tests cover the _rowaddr split, bucket packing, delete offsets, packed match identities, schema field ids, and SQL literal rendering.
• uv run pytest tests/test_merge_into.py (CI tests job; not run in this checkout because its pylance.pth points at an unbuilt source tree)
• ruff check and ruff format --check clean; mypy reports no errors in lance_ray/merge_into.py or tests/test_merge_into.py
• Docs in docs/src/merge_into.md (num_workers / num_partitions best practices, notes and limitations), cross-checked against the implementation and the Lance 12.0.0 sources

[hfutatzhanghb]

Hi, @rgao25 . Have optimized based on your great work. PTAL when have free time, Thanks very much!

[hfutatzhanghb]

Hi, @yanghua . Could you please review this PR when have free time ? Thanks very much ! This PR is coworked by @rgao25 and @hfutatzhanghb .

[hfutatzhanghb]

Hi @yanghua @ddupg . Could you please review this PR when have free time? This PR is coworked by @rgao25 and @hfutatzhanghb . I think this is a very useful feature for large-scale updating.