[hfutatzhanghb]

Summary

• Add lance_ray.update_columns_from() for fragment-local updates of existing columns using Lance row addresses.
• Validate source metadata and schemas before grouping updates by fragment with Ray Data.
• Commit updates once against the source snapshot so conflicts cannot retry stale fragment metadata.

Background

Lance Spark exposes UPDATE COLUMNS FROM, which updates existing columns by matching row metadata. Unlike MERGE INTO, it uses Lance's fragment update path and avoids a global hash join and full-row rewrites.

Lance-Ray currently has additive data-evolution APIs (add_columns, add_columns_from, and merge_columns_from), but no equivalent fragment-level API for updating existing columns. This PR adds that primitive.

API

import lance_ray as lr

source = lr.read_lance("my_dataset.lance", with_metadata=True)
source = source.map_batches(modify_status, batch_format="pandas")

lr.update_columns_from(
    "my_dataset.lance",
    source,
    columns=["status"],
)

The source Ray Dataset must contain _rowaddr and every requested update column. _fragid is optional: it is derived from _rowaddr when absent and validated against the encoded fragment ID when supplied.

Unmatched source rows are ignored. This API only updates existing columns; it does not insert, delete, or upsert rows.

Changes

• lance_ray/io.py
  • add update_columns_from()
  • validate URI/namespace input, batch size, column names and types, row-address uniqueness, and fragment routing metadata
  • normalize integer _rowaddr / _fragid values to uint64
  • group source rows by _fragid and call fragment.update_columns(..., left_on="_rowaddr", right_on="_rowaddr")
  • preserve the Lance source version through normal Ray Data transforms when available
  • commit LanceOperation.Update once against the resolved source version and surface conflicts without retrying stale work
  • forward Ray remote arguments to map_groups()
• lance_ray/__init__.py: export update_columns_from.
• docs/src/data-evolution.md: document the API contract and conflict behavior.
• tests/test_update_columns_from.py: cover partial and filtered updates, metadata preservation, schema and row-address validation, source-version conflicts, remote arguments, batch sizing, and namespace mode.

Validation

• python -m ruff check lance_ray/io.py lance_ray/__init__.py tests/test_update_columns_from.py
• python -m ruff format --check lance_ray/io.py lance_ray/__init__.py tests/test_update_columns_from.py
• uv run mypy
• pytest tests/test_update_columns_from.py -q

[hfutatzhanghb]

@yanghua Hi, all checks have passed. Could you please take a look at this PR when free? Thanks very much !!!

[yanghua]

Left some comments.

[hfutatzhanghb]

Rationale for two intentionally unchanged review items

1. Two passes over the fragment object refs (ray.get)

   This is intentional. The first pass validates every referenced block, including Arrow types, fragment routing, null row addresses, and duplicate row addresses, before fragment.update_columns() is allowed to start writing update files. A single-pass reader would only discover an invalid later block after Lance had already consumed earlier batches. Caching all resolved Arrow tables would avoid the second lookup, but would make worker heap usage scale with all blocks routed to a fragment and would break the bounded-memory design. The implementation therefore retains object refs and performs a validation pass followed by the bounded update-reader pass.

2. Use of Dataset.to_arrow_refs()

   This API is marked as Ray DeveloperAPI, but it is the documented zero-copy API that exposes distributed Arrow block refs. It is present in the project minimum supported Ray version (2.41) and in the locked CI version (2.55). The alternatives do not improve compatibility: iter_internal_ref_bundles() is also DeveloperAPI, while get_internal_block_refs() is deprecated, and driver-side batch iteration would change the distributed-memory behavior. The call is intentionally kept at one documented location so a future Ray API change is isolated and caught by the version matrix/CI.

References: Ray to_arrow_refs documentation, Ray 2.41 implementation.

[hfutatzhanghb]

@yanghua Hi, sir. Please take a look at this PR again when have free time, Thanks!

[yanghua]

Will take a look today!

[zhangyue19921010]

Do we need to set update_mode="rewrite_columns" ?

[hfutatzhanghb]

Good catch — yes. This path uses LanceFragment.update_columns(), so the transaction has rewrite_columns semantics. I set update_mode="rewrite_columns" explicitly and added a regression test that captures the committed operation and verifies the mode. Fixed in 1c76f78.