Bring Lance's low-cost column evolution to the Flink connector (streaming-first, staged proposal)

[fightBoxing]

Background

The lance-spark connector recently exposed two SQL extensions built on top of Lance's native data-evolution APIs:

• ALTER TABLE a ADD COLUMNS c FROM view — add & backfill a new column without rewriting existing rows, powered by Fragment.addColumns().
• ALTER TABLE a UPDATE COLUMNS c FROM view — column-level batch update without full-row rewrite, powered by Fragment.updateColumns().

Reference: Lance + Spark SQL: 实现低成本加列回填数据与修改列 (official Lance & LanceDB post).

The underlying capability — column-file independence + row-address (_rowaddr + _fragid) addressing — lives in the Lance format layer, not in Spark. It should be reachable from Flink too. Today lance-flink does not expose it: LanceSink only supports APPEND / OVERWRITE at whole-row granularity, and both LanceCatalog / LanceNamespaceCatalog throw CatalogException in alterTable.

This issue proposes bringing the same underlying data-evolution capability to the Flink connector — but adapted to Flink's stream-first semantics, not by porting Spark's SQL surface.

Why not port lance-spark's SQL syntax to Flink

Before proposing anything, we want to be explicit about what we are not doing and why. A naive port of ALTER TABLE ... ADD/UPDATE COLUMNS FROM view to Flink SQL is tempting but a poor fit for at least four reasons:

1. User habits differ. Spark users routinely run interactive DDL in spark-sql / Zeppelin / notebooks — one-shot jobs, inspect result, next statement. Flink SQL is primarily used to declare long-running topologies (CREATE TABLE + INSERT INTO); interactive ALTER TABLE is rare in production.
2. ALTER TABLE in Flink is a metadata operation, not a data operation. It updates the Catalog but does not trigger a data-mutation job. Wiring a DDL statement to actually execute Fragment.addColumns() and write data cuts against Flink's architecture.
3. The batch pattern in the article assumes a finite target table. SELECT a._rowaddr, a._fragid, b.tag_0 FROM a INNER JOIN b ON a.biz_key = b.biz_key presumes a is a bounded, static dataset. In Flink, tables default to unbounded streams; the JOIN semantics (regular/temporal/interval), and even the stability of _rowaddr under streaming reads, all become open questions.
4. If a user just wants one-shot batch backfill, they should use Spark. Duplicating Spark's exact interactive experience in Flink would give users no reason to choose Flink.

So the design goal here is not "match lance-spark's SQL". It is: expose Lance's low-cost column evolution as a first-class primitive in the Flink runtime, then build the pipelines Flink can uniquely deliver on top.

Why this matters (Flink-specific value)

The one thing Flink can do that Spark cannot: streaming column backfill.

Consider an ML feature pipeline where new feature values arrive continuously (Kafka, CDC, upstream micro-service). Today's options are all bad:

• Rewrite the whole target table on every batch → prohibitive I/O.
• MERGE INTO on every micro-batch → whole-row rewrite, high write amplification.
• Buffer offline and run Spark daily → loses freshness.

With Lance's Fragment.updateColumns() being idempotent on _rowaddr, Flink can do something neither Spark nor MERGE INTO can:

Kafka(feature updates) → Flink stream → JOIN Lance dim view to get _rowaddr
                                     → per-checkpoint column-only commit → Lance

This is a genuine differentiator, and it maps naturally onto:

• Flink's exactly-once semantics (idempotent updateColumns on stable row addresses).
• The existing time-travel and vector-search story already shipped in this repo.
• ML feature freshness requirements that offline Spark cannot meet.

Batch feature backfill inside Flink is a secondary, opportunistic use case — worth supporting for users already on Flink, but not the primary justification.

Current gaps in lance-flink

Area	Status
Sink write modes	Only APPEND / OVERWRITE on RowData, no column-level API
_rowaddr / _fragid exposure	Not exposed — LanceDynamicTableSource implements SupportsProjectionPushDown / SupportsFilterPushDown / SupportsLimitPushDown / SupportsAggregatePushDown but not SupportsReadingMetadata
Catalog	Both LanceCatalog.alterTable and LanceNamespaceCatalog.alterTable explicitly throw CatalogException("...does not support altering...")
Lance Java SDK bindings	Fragment.addColumns() / Fragment.updateColumns() are available upstream but never referenced in this repo (verified: grep -r "addColumns|updateColumns" → 0 hits)

Proposed design (staged)

Four stages, ordered by Flink-fit rather than by mechanical parity with Spark. Each stage is independently shippable.

Stage 1 — Expose _rowaddr / _fragid as metadata columns (foundation)

Make LanceDynamicTableSource implement SupportsReadingMetadata, exposing:

Metadata key	Flink type	Semantics
rowaddr	BIGINT NOT NULL	Lance physical row address
fragid	INT NOT NULL	Fragment ID

Usage:

CREATE TABLE lance_users (
    id BIGINT,
    name STRING,
    rowaddr BIGINT METADATA FROM 'rowaddr' VIRTUAL,
    fragid  INT    METADATA FROM 'fragid'  VIRTUAL
) WITH ('connector' = 'lance', 'path' = '...');

Why first: pure additive change to a single source class; unblocks every downstream stage; also useful on its own for row-level upsert and diagnostic queries.
Scope: single class change + IT case. Low risk.

Stage 2 — Streaming LanceColumnUpdateSink (Flink's main differentiator)

Checkpoint-integrated sink that consumes a stream of (rowaddr, fragid, newValue…) records and commits column-only updates per checkpoint via Fragment.updateColumns().

Contract:

• Input RowData must carry rowaddr + fragid + one or more payload columns.
• Idempotent under exactly-once (Lance keys updates on _rowaddr; replay after failure is safe).
• Commits are batched at checkpoint boundaries to bound fragment growth.

Sketch:

DataStream<RowData> featureUpdates = /* Kafka → JOIN lance dim view → project (rowaddr, fragid, v) */;
featureUpdates.sinkTo(LanceColumnUpdateSink.forTable("lance_users")
        .targetColumns("tag_0")
        .mode(UPDATE));

Why this is the headline stage: this pipeline cannot be expressed by Spark. It is the concrete, differentiated value of bringing Lance data evolution to Flink.
Scope: new sink + checkpoint/commit protocol + IT case with fault-injection. Highest engineering investment, highest strategic value.

Stage 3 — Batch LanceColumnUpdateSink (opportunistic, for users already on Flink)

Same sink class as Stage 2 but exposed as a DynamicTableSink so it can be targeted by standard Flink Table API INSERT INTO:

Table view = tEnv.sqlQuery(
    "SELECT a.rowaddr, a.fragid, b.tag_0 " +
    "FROM lance_users a INNER JOIN feature_source b ON a.id = b.biz_key");

view.executeInsert("lance_column_sink");   // configured with mode=ADD, target-columns=tag_0

Positioning: this covers the same use case as lance-spark's ADD/UPDATE COLUMNS FROM, but for users whose whole pipeline is already on Flink and don't want to spin up Spark just to backfill one column. Not a reason to choose Flink over Spark on its own; a nice-to-have.
Scope: DynamicTableSink adapter + factory option wiring on top of Stage 2's runtime.

Stage 4 — Relax LanceCatalog.alterTable for the ADD-COLUMN case (lowest priority)

Allow ALTER TABLE t ADD COLUMN c <type> to succeed by calling Dataset.addColumns() with a NULL-filled column definition (schema-only add). Keep drop / rename / type change rejected.

Why last: Flink users rarely run interactive ALTER TABLE in production; the value is mostly cosmetic parity with other catalogs.
Scope: 2 catalog classes, gated by capability flag.

(Explicitly not proposed) SQL syntax extension ALTER TABLE ... ADD/UPDATE COLUMNS FROM

Adding this to Flink SQL requires either a Calcite parser plugin or a SqlDialect extension, plus multi-version maintenance across Flink 1.18/1.19/1.20. Given (a) the arguments above about engine fit and (b) Stage 3 already covering the same use case via Table API, we do not plan this and would push back on it unless there's clear community demand.

Scope of this issue

Umbrella / discussion ticket. If direction is accepted, each stage becomes its own tracked issue:

• Stage 1 — SupportsReadingMetadata for LanceDynamicTableSource
• Stage 2 — Streaming LanceColumnUpdateSink (the differentiator)
• Stage 3 — DynamicTableSink wrapper for batch backfill on Flink
• Stage 4 — LanceCatalog.alterTable — ADD COLUMN only
• (Deferred) SQL syntax extension

Open questions

1. Metadata column naming — rowaddr / fragid or underscore-prefixed _rowaddr / _fragid to stay consistent with lance-spark? Flink convention leans away from leading underscores.
2. _rowaddr stability under streaming reads — is _rowaddr guaranteed stable across snapshots between the JOIN read and the column-update commit? This is the correctness cornerstone of Stage 2 and needs an authoritative answer from the Lance team.
3. Streaming commit granularity (Stage 2) — per-checkpoint commit vs. time-based batching? How to bound fragment growth?
4. Compaction interaction (Stage 2) — do we auto-schedule compaction inside the sink, or expose a knob and let users run compaction externally? This is a known pre-existing pain point for lance-flink streaming writes.
5. Nullability semantics (Stage 3) — for ADD-column-from-view when some target rows have no matching source row, match Spark's INNER-JOIN-defaults-to-NULL behavior?
6. Type validation — plan-time schema check on target column vs. payload column, or runtime error?

References

• Article: Lance + Spark SQL: 实现低成本加列回填数据与修改列
• lance-spark: ALTER TABLE ... ADD COLUMNS FROM implementation
• Lance format docs — data evolution & row addresses
• Related in-repo files:
  • src/main/java/org/apache/flink/connector/lance/table/LanceDynamicTableSource.java (Stage 1 landing site)
  • src/main/java/org/apache/flink/connector/lance/LanceSink.java (Stage 2/3 template)
  • src/main/java/org/apache/flink/connector/lance/table/LanceCatalog.java (Stage 4 landing site)

Happy to send Stage 1 as a first small PR to validate the approach.

cc @LuciferYang @jackye1995

[LuciferYang]

Thanks for writing this up. I think the framing is right: don't port Spark's SQL surface, build what Flink can uniquely do. I also agree with deferring the parser extension. One substantive suggestion for Stage 2, then the open questions.

Stage 2 without row addresses

Open question 2 has an answer in the format docs, and it's no. docs/src/format/table/row_id_lineage.md: "Row addresses change when data is reorganized through compaction or updates." Compaction assigns fresh fragment ids and re-densifies row offsets. DELETE alone is safe, since it tombstones through a deletion vector and leaves other rows' offsets where they were.

The commit layer does catch the race, but not by pinning anything. In rust/lance/src/io/commit.rs the sweep checks out whatever read_version the caller passed, checks this commit against the transactions committed since it, and commits onto manifest.version + 1. Nothing verifies that read_version is the version the addresses were read at. A batch job's read-to-commit window is seconds. A streaming job's is unbounded, which turns a corner case into the normal case.

One way out is to not address rows at all. merge_insert matches on the on columns the caller names; _rowaddr lives inside the plan and is never a join input. lance-format/lance#8423 (open) adds write_mode=RewriteColumns, which patches the source columns into the fragments that already hold the matched rows. Fragment ids and row offsets don't change, so the addresses are preserved rather than relied on.

Kafka → Flink → mergeInsert(on = biz_key, write_mode = RewriteColumns) → Lance

If you go that way it reorders the stages. Stage 1 would stop being a prerequisite, though it's still worth doing for row-level upsert and diagnostics, and Stage 3's UPDATE mode becomes Stage 2 without the checkpoint protocol.

What that pipeline cannot express

RewriteColumns errors out instead of silently falling back, so its blocker list is effectively the sink's contract.

• Key plus payload, not whole rows. A source that covers every dataset column is rejected ("nothing to skip"), so a Table API sink handing over full rows won't qualify. Source columns also have to exist in the target schema already, which means Stage 3's ADD mode isn't this call: it needs a schema add (Dataset.addColumns, your Stage 4) first, then RewriteColumns to fill the column.
• Only keys already in the target. Inserts are rejected, so the join is an inner join and unmatched source rows disappear without an error. MergeInsertStats doesn't carry a dropped-row count either (it has inserted, updated, deleted, attempts, bytes, files), so the sink can't report the loss even if it wanted to. It has to guarantee the key exists or route unmatched keys to a separate append, and finding out which keys exist is a target lookup, so some of the dimension-table work comes back under a different name.
• No deletes. when_matched has to update and delete-by-source has to be Keep, so a changelog's -D records need a second call. A checkpoint then becomes two commits that aren't atomic together, and a failure between them leaves the checkpoint half applied.
• One update per key per batch. SourceDedupeBehavior defaults to Fail, so two updates to the same key inside one checkpoint fail the job, and the replayed batch fails the same way on restart. FirstSeen keeps the first row, which is backwards for a changelog. Keep-latest deduping has to happen before the call, either in a keyed operator upstream or in a keep-latest buffer inside the sink behind a keyed exchange. That changes your API sketch: the sink's input becomes (biz_key, payload…) plus an on(...) clause rather than (rowaddr, fragid, v).

Replays converge on the same values in the data columns and the commit is atomic, but they aren't no-ops. Each replay is a new version that rewrites the touched columns for every affected fragment, and with stable row ids it re-stamps those rows' _row_last_updated_at_version.

Costs, and open questions 3 and 4

The JOIN doesn't disappear either. It moves inside Lance, and it scans the whole target table on every checkpoint unless every join column carries an exact-equality scalar index, so an index on the business key is a performance prerequisite for Stage 2 rather than a nice-to-have. The connector also gives up control over that join: no dimension-table caching, no temporal-join semantics, and one bad record fails the whole batch instead of going to a side output.

• Cost shape. A replacement column file covers every row of each fragment it touches, so patching costs roughly (rows in the touched fragments) × (width of the source columns), while row-rewrite scales with the matched rows. Patching wins once the matched fraction exceeds about the fraction of each row's bytes the source columns occupy. #8423 has the measured numbers and admits there's no selectivity gate. A per-checkpoint stream matches very few rows, which is the wrong end of that curve; what rescues the ML feature case is wide payload columns, since a 1024-dim fp32 embedding is 4 KB on its own.
• Index invalidation. fields_modified prunes the patched fragments out of every index that covers a patched column, so a stream updating an indexed column erodes index coverage every checkpoint. That matters more here than it would elsewhere, given the vector-search story in this repo.

Compaction, your open question 4, is the one I'd want settled before Stage 2 ships. Compaction and a column update touching the same fragment conflict retryably in both directions, so background compaction and a per-checkpoint update stream end up spending each other's retry budget (withConflictRetries, default 10), while pulling opposite ways on cost since bigger fragments make sparse patches more expensive. My read is to keep compaction external and schedule it against a quiet window. That leaves open question 3 genuinely open: every checkpoint is a new version plus a column file per touched fragment, and the superseded files are garbage until cleanup runs, so a sink doing this needs a cleanup policy in its docs and not just a commit interval. For the record, concurrent Append doesn't conflict in either direction, so an append stream and a column-update stream can run side by side.

One routing detail worth knowing: Auto rewrites rows except when use_index is on (the default) and every join column has an exact-equality scalar index, in which case the legacy indexed route patches columns whatever you asked for. A partially indexed composite key doesn't qualify. #8423 makes both routes nameable, so a sink should set write_mode explicitly.

Open question 1: naming

_rowaddr and _rowid are format-layer system column names. _fragid isn't. lance-spark synthesizes it per fragment, because a fragment id is a field of a row address (its upper 32 bits) rather than a column you can query. Flink's METADATA FROM 'key' separates the key from the column name, so you can use the format-layer keys and let users name the columns whatever they like.

If you do Stage 1, expose _rowid too, since with stable row ids enabled it survives compaction and _rowaddr does not. Two caveats. With stable row ids off, which is the default, _rowid is the row address. And enable_stable_row_ids can only be set when the table is created, so the catalog's CREATE TABLE has to expose it before any of this is reachable.

The deferred SQL extension, and Stage 4

Agreed on deferring, and here's the shape of the tax. In lance-spark the .g4 is shared, but the parser and AstBuilder are copied into 5 modules, 2088 lines of them. The last grammar change I looked at (tag DDL) added 1405 lines across 22 files. 790 of those were in the shared base module, but 406 were the same edit repeated per engine version, including 4 identical AstBuilder changes. Across three Flink versions that's a recurring cost for syntax Stage 3 already covers.

On Stage 4: lance-spark#752 (support schema evolution DDLs) is still open, so the Spark-side semantics are in flux. Worth landing after it rather than against it.

I've left open questions 5 and 6 alone, since both are Stage 3 API-surface calls rather than format-layer ones. One thing worth noting: under RewriteColumns the NOT NULL question doesn't come up for UPDATE mode, because the columns the source omits keep the values they had.

Java bindings: three gaps, one structural

Dataset.mergeInsert is exposed and self-commits, with retry knobs. What's missing:

1. write_mode only reaches Python. It needs a setter and getter on MergeInsertParams plus an extract in the JNI layer.
2. source_dedupe_behavior isn't exposed at all, so a Java caller can't even choose FirstSeen. Given the default above, Stage 2 needs it.
3. execute_uncommitted and UncommittedMergeInsert are Rust-only; the JNI calls execute_reader, which commits. This is the structural one, because an exactly-once sink needs prepare and commit split at the checkpoint barrier and a self-committing call can't do that. Today's LanceSink is a RichSinkFunction plus CheckpointedFunction, so it isn't two-phase either, but Stage 2's exactly-once claim depends on this.

I'll send (1) and (2) as a follow-up to #8423 rather than widening it. (3) deserves its own issue on the Lance side. And to be explicit about what isn't landed: #8423 is still open and that follow-up is unsent, so Stage 2 as described depends on unmerged work in two places. Stage 1 depends on none of it, which is one more argument for sending it first even though it's no longer a prerequisite.