feat: add update_columns_from API

docs/src/data-evolution.md

@@ -30,3 +30,114 @@ Add columns to an existing Lance dataset using Ray's distributed processing.
 - `concurrency`: Optional number of concurrent processes
 
 **Returns:** None
+
+## `update_columns_from`
+
+```python
+update_columns_from(
+    uri=None,
+    ds=None,
+    *,
+    columns,
+    read_version=None,
+    ray_remote_args=None,
+    storage_options=None,
+    namespace_impl=None,
+    namespace_properties=None,
+    table_id=None,
+    batch_size=1024,
+) -> int
+```
+
+Update existing columns in a Lance dataset using row metadata. The input Ray
+Dataset must contain `_rowaddr` and every column listed in `columns`. If
+`_fragid` is absent, it is derived from `_rowaddr`; if supplied, it must match
+the fragment ID encoded in `_rowaddr`.
+
+```python
+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"],
+)
+```
+
+Unmatched source rows are ignored. This API updates existing columns only; it
+does not insert, delete, or upsert rows. Transform functions must preserve
+`_rowaddr`. An empty source is treated as a no-op and emits a warning.
+
+Each `_rowaddr` must be a non-null integer value and may appear only once; it
+is normalized to `uint64` before routing. Names in `columns` must also be
+unique, and every source update column must have the same Arrow type as the
+corresponding target column.
+
+The final update operation is committed once against the dataset version that
+was originally read. Commit conflicts are returned to the caller; safely
+retrying requires rerunning the update pipeline from the latest dataset version.
+`read_lance()` retains that version and an irreversible SHA-256 digest of the
+stable dataset identity on the Lance datasource object. Ray's execution-plan
+logs only show `ReadLanceDatasource`; the digest is not part of the operator
+name. Renaming or combining lazy Dataset branches does not change the update
+base. Combining rows
+from different Lance datasets, or updating a different dataset than the one
+that was read, is rejected even when the version numbers match, but only while
+that source lineage remains present. If an operation replaces the lineage, such
+as materializing the Dataset, provide `read_version` explicitly; the update
+will not silently use the latest Lance version. An explicit `read_version`
+proves which target version is used, not which dataset produced the source
+rows, so after lineage is lost the caller must ensure that the source rows
+still correspond to that target dataset. If a reliable identity cannot be
+derived for an object-store dataset because no persistent UUID or non-sensitive
+backend discriminator is available, `update_columns_from` also requires an
+explicit `read_version` instead of performing unsafe automatic validation.
+
+Before fragment partitioning, the source is projected to `_rowaddr`, `_fragid`
+when present, and the columns listed in `columns`. Unused source columns are
+not routed into fragment update workers, and each fragment is streamed as
+bounded `RecordBatch` values rather than materialized as a full table.
+Partitions are sorted by `_rowaddr`, and duplicate addresses are rejected by a
+k-way merge of those sorted runs. The returned integer is the number of source
+rows that matched an existing target row.
+
+The commit is `LanceOperation.Update` with `update_mode="rewrite_columns"` and
+`updated_fragment_offsets`. On a dataset with stable row ids, those offsets are
+what lets Lance refresh `_row_last_updated_at_version` for the matched rows.
+
+The driver keeps every partition object reference until the function returns.
+Those objects are `ray.put` values owned by the partition workers and have no
+lineage, so the object store holds the partitioned copy for the whole update
+(about twice the source data at peak) and a lost worker raises `ObjectLostError`
+instead of recomputing the block. Fragment updates also wait until every
+partition task has finished. If one fragment task fails, files already written
+by the other tasks stay uncommitted and are not deleted, the same as
+`merge_columns_from`.
+
+An S3 or GCS endpoint that is set only through the process environment
+(`AWS_ENDPOINT_URL_S3`, `AWS_ENDPOINT_URL`, or `STORAGE_EMULATOR_HOST`) is part
+of the dataset identity. `storage_options` still take precedence over those
+variables. The identity does not include credentials from the endpoint URL.
+
+**Parameters:**
+
+- `uri`: Path to the Lance dataset. Either `uri` or namespace parameters are required.
+- `ds`: Ray Dataset containing `_rowaddr` and the columns to update. `_fragid`
+  is derived when absent and validated against `_rowaddr` when supplied.
+- `columns`: Existing columns to update.
+- `read_version`: Dataset version to update. Defaults to the unique Lance source
+  version retained in the Ray Dataset's logical lineage. Required when the
+  lineage is unavailable, such as after materializing the source. When lineage
+  is present, the source dataset identity digest must match the update target.
+  An explicit value is also required when a reliable object-store identity
+  cannot be determined.
+- `ray_remote_args`: Optional Ray Data task options.
+- `storage_options`: Optional storage configuration dictionary.
+- `namespace_impl`, `namespace_properties`, `table_id`: Namespace resolution arguments.
+- `batch_size`: Batch size for the update reader. Must be positive.
+
+**Returns:** The number of source rows that matched an existing target row.
+Unmatched row addresses are ignored. An empty source returns `0`.

lance_ray/__init__.py

@@ -21,6 +21,7 @@
     add_columns_from,
     merge_columns_from,
     read_lance,
+    update_columns_from,
     write_lance,
 )
 from .pool import clear_global_pool, get_global_pool, init_global_pool, set_global_pool
@@ -37,6 +38,7 @@
     "add_columns",
     "add_columns_from",
     "merge_columns_from",
+    "update_columns_from",
     "create_scalar_index",
     "create_index",
     "optimize_indices",