Skip to content

feat: distributed update_columns for overwriting existing columns - #5261

Open
FANNG1 wants to merge 11 commits into
lance-format:mainfrom
FANNG1:feat/distributed-update-columns
Open

feat: distributed update_columns for overwriting existing columns#5261
FANNG1 wants to merge 11 commits into
lance-format:mainfrom
FANNG1:feat/distributed-update-columns

Conversation

@FANNG1

@FANNG1 FANNG1 commented Jul 28, 2026

Copy link
Copy Markdown
Contributor

Closes #5260.

Adds lance_ray.update_columns(), a distributed operation that overwrites
existing columns
of a Lance dataset. Until now add_columns,
add_columns_from and merge_columns_from could only add new columns, so
re-running an enrichment stage — recomputing labels for a date range, or
regenerating embeddings after a model upgrade — meant rewriting the whole table.

import lance_ray as lr

# Recompute one week of labels
lr.update_columns(
    "s3://bucket/events.lance",
    transform=recompute_label,
    columns=["label"],
    filter="date >= '2026-07-01' AND date < '2026-07-08'",
    read_columns=["embedding"],
)

# Regenerate embeddings from stored image blobs, on GPUs
lr.update_columns(
    "s3://bucket/images.lance",
    transform=embed_images,
    columns=["embedding"],
    read_columns=["image"],
    ray_remote_args={"num_gpus": 1},
    concurrency=8,
)

How it works

A dataset snapshot is pinned on the driver, then each fragment is processed by
exactly one Ray task that scans, filters, transforms and rewrites only its own
fragment. Keeping scan and rewrite together in one task avoids the Ray Data
shuffle and regrouping that merge_columns_from needs for externally created
data.

The rewrite uses Lance's RewriteColumns update mode, so rows do not move,
_rowaddr stays stable, and columns that were not named keep their original
data files. The driver collects every worker's fragment metadata and issues a
single LanceOperation.Update commit — a failure in any fragment aborts the run
without a transaction, so the dataset is never left half-updated.

Transform results are streamed to Lance as a RecordBatchReader rather than
materialized per fragment.

API notes

columns takes column names, not a schema. update_columns only overwrites
columns that already exist, so each target's Arrow type and nullability are
fixed by the dataset — a caller-supplied schema could only either repeat what
the dataset already says or be an error. Taking names keeps the redundancy out
of the signature, and avoids the trap where pa.schema([("price", pa.float64())])
silently defaults to nullable=True and mismatches a non-nullable target.

The transform result is reordered to columns and cast to the dataset's types
with safe=True. One limit is documented rather than papered over: Arrow's safe
cast range-checks integers and time units but not float narrowing, so
returning a float64 for a float32 column rounds (and overflows to inf)
silently. The docstring and docs/src/data-evolution.md say so and recommend
producing the column's own type when precision matters.

Everything checkable is checked on the driver before any Ray task starts —
unknown column, metadata column, nested path, duplicate, struct target, blob
write target, bare-string columns — so a bad call cannot rewrite half the
fragments before failing.

Limitations

  • Stable row IDs are rejected. pylance does not expose the matched row
    offsets Lance needs to advance _row_last_updated_at_version, so the change
    would be invisible to CDF consumers (Update op with UpdateMode.RewriteColumns does not advance _row_last_updated_at_version lance#6734). Raises
    NotImplementedError rather than committing something wrong.
  • Struct columns are not supported as update targets yet.
  • Blob columns can be read by the transform as raw bytes, but are not valid
    update targets.
  • The physical rewrite is fragment-wide, so a very sparse filter is a poor
    fit — matching rows should be reasonably concentrated.
  • Indexes may go stale when an indexed column is updated; rebuild before
    relying on them.

Tests

50 tests in tests/test_update_columns.py covering: multi-fragment and
multi-batch updates, filters and partial fragment coverage, multi-column
updates, the no-op path when a filter matches nothing, the transform contract
(row count, row order, output columns, BatchUDF, casting, non-nullable
targets), physical correctness (_rowaddr preservation, untouched data files,
time travel, deleted rows), transaction behavior (stale snapshot, concurrent
append rebase), leaf field ids for list and fixed-size-list columns, blob v1
and v2 input, namespace-only resolution, and driver-side rejection with a
dataset fingerprint asserting nothing was written.

ruff check is clean. ruff format --check reports the same three files as
main; none are touched here.

https://claude.ai/code/session_016cKDCcCvDW9YVxT7Ea7Qtb

@github-actions github-actions Bot added the enhancement New feature or request label Jul 28, 2026
Ray workers cannot import the test module -- it imports pytest, which is
absent from the worker environment -- so every transform that referenced a
test-module global failed to deserialize.

Move the two helpers that cross the Ray boundary into _ray_test_support,
which exists precisely to be importable from workers, and replace the bare
asserts inside two transforms with explicit raises: pytest rewrites `assert`
into calls on `_pytest`, which the pickled closure then drags along.

Claude-Session: https://claude.ai/code/session_016cKDCcCvDW9YVxT7Ea7Qtb
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

enhancement New feature or request

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Support distributed column updates for label and embedding backfills

1 participant