Skip to content

Support distributed column updates for label and embedding backfills #5260

Description

@FANNG1

Problem

Two workflows on large Lance tables have no good path today, because Ray-based
column operations can only add columns, never overwrite an existing one.

1. Recompute the label column for a few days of data

A table holds months of rows with a date column. A labeling rule changes, or a
bug is found, and we need to recompute label for one week — say
date >= '2026-07-01' AND date < '2026-07-08' — leaving every other row and
every other column untouched.

Today the options are:

  • add_columns / add_columns_from / merge_columns_from — these only add new
    columns. label already exists, so they are not applicable.
  • Read the table into Ray, recompute, write a new dataset — this rewrites the
    entire table to update a few days of it, and drops the existing version
    lineage.

Neither is acceptable when the affected rows are a small fraction of a
multi-terabyte table.

2. Multi-stage image enrichment pipeline

The pipeline has three stages that must be independently re-runnable:

  1. Ingest — write images into Lance as blob columns.
  2. Embed — a GPU stage reads the image bytes and computes an embedding
    column.
  3. Label — a stage reads embedding and computes label.

The first execution of stages 2 and 3 creates new columns, which
add_columns_from handles. But re-running a stage is the normal case, not the
exception: an embedding model is upgraded, or a labeling threshold is tuned, and
we need to recompute embedding and label for the affected rows only — not
re-ingest the images, which are by far the most expensive part to store and
move.

Re-running a stage means overwriting a column that already exists, which nothing
in lance-ray covers.

Expected behavior

A distributed column-update operation that overwrites existing columns in place,
one Ray task per fragment:

import lance_ray as lr

# Use case 1: 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"],
)

# Use case 2: stage 2 of the image pipeline, on GPUs
lr.update_columns(
    "s3://bucket/images.lance",
    transform=embed_images,
    columns=["embedding"],
    read_columns=["image"],  # blob column, delivered to the transform as bytes
    ray_remote_args={"num_gpus": 1},
    concurrency=8,
)

Properties both use cases depend on:

  • No full-table rewrite. Rows do not move, _rowaddr stays stable, and
    columns that were not named keep their original data files.
  • Filter pushdown. Only matching rows receive new values; unmatched rows
    keep their old ones.
  • Fragment-level parallelism with Ray resource control, so the embedding
    stage can request GPUs and bound its concurrency.
  • Atomicity. The run commits as a single transaction or leaves the dataset
    untouched — a failure partway through must not leave some fragments updated
    and others not.
  • Blob columns readable as transform input. Stage 2 reads image bytes and
    writes a plain vector column.
  • Validation before any work starts. A wrong column name or an unsupported
    target type should fail on the driver, not after half the fragments have been
    rewritten.

Scope

  • Add a distributed update_columns operation that overwrites existing columns.
  • Support a Lance filter expression to restrict which rows are updated.
  • Support both URI and namespace (namespace_impl + table_id) table
    resolution, consistent with read_lance / write_lance.
  • Support reading blob columns as transform input.
  • Support ray_remote_args and concurrency for GPU stages.
  • Document the interaction with existing indexes: updating an indexed column may
    leave that index stale.

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions