feat(dataset): add write_fragment_column and commit_column_writes

[wkalt]

Two-step API for writing a column's data to existing fragments:

• write_fragment_column writes one fragment's new column data as a
  standalone uncommitted data file and returns its DataReplacementGroup.
  The batch stream must match the fragment's physical row count.
• commit_column_writes publishes the accumulated replacements and the
  extended schema in one atomic Merge, tombstoning prior coverage of the
  replaced fields. Commit conflicts retry via the standard retry executor;
  a fragment rewritten or a field id reassigned concurrently fails the
  commit instead of publishing stale data.

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The staged per-fragment writer is a useful boundary, but the commit path currently permits invalid row alignment and reinterprets prewritten field IDs after a schema race, so it can publish unreadable or silently misnamed data.

Keep the split write/commit API, but bind staged results to one snapshot with centrally assigned schema IDs, validate each file against the planned fragment, and let conflicting data/schema commits force recomputation instead of replaying stale files. The existing Operation::Merge remains sufficient for atomic publication.

[lance-gatekeeper]

finish() returns FileWriteSummary::num_rows, but discarding it means the documented exact physical-row invariant is never enforced. A too-short file commits successfully and makes scans fail; a too-long file commits and its surplus row is silently ignored. Resolve the target fragment before writing, compare the summary to its physical row count, and reject/clean up a mismatch before returning a replacement.

Reproducer run on this head

I ran a standalone program against cafeade7fced472240ecf134a053f7207ae353d9 with:

CARGO_TARGET_DIR=/home/agent/tmp/pr8207-implementation.zcqSkO cargo run --manifest-path /home/agent/tmp/pr8207-repro.Bze75g/Cargo.toml

It created a two-row fragment, passed one-row [7] and three-row [7, 8, 9] streams to write_fragment_column, and committed each returned replacement. Both write and commit calls returned Ok. Expected: the write rejects both lengths before a commit is possible. Observed: the short case made validate() report Expected: 2 Got: 1 and scan fail with cannot read Ranges([0..2]) from columns with 1 rows; the long case made validate() report Expected: 2 Got: 3 while scan returned only [7, 8].

The exercised public-API sequence was:

let replacement = dataset
    .write_fragment_column(fragment_id, stream::iter([Ok(replacement_batch)]), &schema)
    .await?;
dataset
    .commit_column_writes(vec![replacement], &schema, None)
    .await?;
let validation = dataset.validate().await;
let scan = dataset.scan().try_into_batch().await;

[lance-gatekeeper]

Field-ID existence is not field equivalence. After a retry, a field that another commit assigned the same ID but a different name/type is treated as already present. The code then tombstones that concurrent field's file by ID and appends the stale file under the concurrent schema, returning success. Bind prepared writes to a centrally assigned snapshot schema and fail/recompute if a previously-new ID becomes occupied; only an exact field that existed in the original snapshot should qualify as an incremental replacement.

Reproducer run on this head

The same executed program opened two stale handles on a two-row dataset. Each derived max_field_id + 1 (ID 1) for different fields, first_value=[10,20] and second_value=[30,40], staged both files, then committed them sequentially so the second call entered this retry path.

Expected: the second commit reports a schema conflict and requires recomputation, or both named fields remain correct. Observed:

collision: derived_ids first=1 second=1
collision: second_commit=Ok(())
collision: final_fields=[("id", 0), ("first_value", 1)]
collision: validate_result=Ok(())
collision: scan_result=Ok rows=2 first_value=Some([Some(30), Some(40)]) second_value=None

The dataset validates while the first field now contains the second writer's bytes and the second field name is lost.

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The prior row-count and different-field-ID race defects are fixed. The commit boundary still does not preserve or validate the prepared write plan, so valid replacements can be dropped, mismatched files can publish an invalid version, and a conflict retry can overwrite newer values.

Keep the split API, but make each prepared result snapshot- and schema-bound: validate dataset/read version, fragment identity, exact field schema, and grouped or unique fragment coverage before the atomic Merge. Relevant conflicts should require recomputation rather than replaying staged bytes.

[lance-gatekeeper]

Collecting by fragment ID silently discards every earlier replacement for that fragment. Staging separate columns is a valid use of these APIs, so either group and apply all files for a fragment or reject duplicate fragment IDs; last-entry-wins publishes nulls for the dropped columns.

Reproducer run on this head

I added a regression test on 9b29517fb7b597db218b53eede23744aa3a2e871 that staged first=[10,20] and second=[30,40] as separate files for the same fragment, then committed both with their union schema:

let first_schema = new_columns.project_by_ids(&[first_id], true);
let second_schema = new_columns.project_by_ids(&[second_id], true);
let first = dataset.write_fragment_column(fragment_id, stream::iter([Ok(first_batch)]), &first_schema).await?;
let second = dataset.write_fragment_column(fragment_id, stream::iter([Ok(second_batch)]), &second_schema).await?;
dataset.commit_column_writes(vec![first, second], &new_columns, None).await?;
dataset.validate().await?;
assert_eq!(read_values(&dataset, "first").await?, vec![Some(10), Some(20)]);

Run with:

CARGO_TARGET_DIR=/home/agent/tmp/verify-inputs.CFejAC/target cargo test -p lance test_commit_column_writes_preserves_multiple_files_for_same_fragment -- --nocapture

The commit and validate() both returned Ok, but the assertion observed first=[None,None]; only second survived.

[lance-gatekeeper]

The declared schema is checked only against the dataset schema; it is never checked against the staged files' field IDs. A caller can therefore publish a file/schema mismatch that Lance's own validator reports as corrupt. Bind the exact staged field schema to the prepared result and reject any mismatch before constructing the Merge.

Reproducer run on this head

On 9b29517fb7b597db218b53eede23744aa3a2e871, I staged field staged with ID 1, changed the commit schema to declare declared with ID 2, and ran:

let replacement = dataset
    .write_fragment_column(fragment_id, stream::iter([Ok(staged_batch)]), &staged_schema)
    .await?;
dataset
    .commit_column_writes(vec![replacement], &declared_schema, None)
    .await?;
dataset.validate().await?;

CARGO_TARGET_DIR=/home/agent/tmp/verify-inputs.CFejAC/target cargo test -p lance test_commit_column_writes_rejects_schema_not_matching_staged_file -- --nocapture

Expected the commit to reject the mismatch. It returned Ok; the follow-up validation failed with CorruptFile ... did not have any fields in common with the dataset schema.

[lance-gatekeeper]

Matching only name and data type does not make a staged write current or schema-equivalent. After a Merge conflict, execute_with_retry refreshes the dataset but reuses the old bytes, so a field introduced by the winning writer is mistaken for an incremental rewrite and silently overwritten; nullability differences are also accepted. Preserve the original snapshot/field identity in the prepared plan and require recomputation when relevant schema or fragment data changed, while comparing the complete field contract for intentional rewrites.

Reproducers run on this head

I opened two handles at version 1 and staged the same fresh ID/name/type with different values. The first committed [10,20]; the stale second handle then called the same API with [30,40]:

ds1.commit_column_writes(vec![first], &schema1, None).await?;
ds2.commit_column_writes(vec![stale], &schema2, None).await?;
ds1.checkout_latest().await?;
assert_eq!(read_values(&ds1, "value").await?, vec![10, 20]);

CARGO_TARGET_DIR=/home/agent/tmp/gate-snapshot.ZAqraJ/target cargo test -p lance test_commit_column_writes_rejects_stale_same_field_replacement -- --nocapture

Both commits returned Ok; the assertion observed [30,40], proving the retry overwrote the intervening commit. A second executed test made the winner's field non-nullable and the stale replacement nullable with [30,NULL]; that commit also returned Ok, and scanning the resulting version failed with Found unmasked nulls for non-nullable StructArray field "value".

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The previous three reproducers are fixed, but the prepared-write boundary still treats field IDs and fragment IDs as sufficient identity. That breaks recursive and physical schema contracts, and retries can publish new fragment state with missing or shadowed replacement values.

Keep the split API, but bind each prepared result to its full recursive staged schema and original dataset/version/fragment plan, then reject or recompute after relevant conflicts. Replacements should also use the existing overlay-tombstoning contract before the atomic Merge.

[lance-gatekeeper]

A set of top-level field IDs is not the staged schema. It accepts the same ID with a different physical type, which silently reinterprets stored bytes, while rejecting valid nested files because DataFile.fields contains leaf IDs. Carry or open the staged file's complete recursive schema and compare that contract; derive file coverage from the corresponding atomic fields rather than assuming top-level IDs appear in DataFile.fields.

Reproducers run on this head

For the type case, I staged Int32 [10,20], declared the same field ID as Float32, and committed through the public API.

CARGO_TARGET_DIR=/home/agent/tmp/verify-schema-pYKytT/target RUST_BACKTRACE=0 cargo test -p lance --test repro_column_schema repro_same_id_different_physical_type -- --nocapture

Expected: reject the staged/declared schema mismatch. Observed: commit and validate() returned Ok; scan returned Float32 [1.4e-44, 2.8e-44], the Int32 bit patterns interpreted as floats.

For a struct s: { x: Int32 }, the prepared schema had parent ID 1 and leaf ID 2, and the writer correctly returned DataFile.fields=[2].

CARGO_TARGET_DIR=/home/agent/tmp/verify-schema-pYKytT/target RUST_BACKTRACE=0 cargo test -p lance --test repro_column_schema repro_nested_struct_round_trip -- --nocapture

Expected: the writer's own result commits and round-trips. Observed: commit rejected leaf ID 2 as “not declared” because declared contains only top-level ID 1.

[lance-gatekeeper]

A retry after Append includes the new fragment in the global schema but has no replacement for it. For a new non-nullable column, this publishes a version that validates successfully and then panics while scanning the uncovered fragment. Bind the replacement plan to the original fragment set and require recomputation after an append, or otherwise prove that every current fragment has valid coverage before committing a non-nullable field.

Reproducer run on this head

I staged a non-nullable value=[10,20] for the only v1 fragment, concurrently appended an id=3 fragment at v2, then committed the stale replacement.

CARGO_TARGET_DIR=/home/agent/tmp/verify-schema-pYKytT/target RUST_BACKTRACE=0 cargo test -p lance --test repro_column_schema repro_concurrent_append_new_nonnullable_field -- --nocapture

The column commit returned Ok at v3 with two fragments and validate() also returned Ok. Scanning panicked at fragment.rs:567 because the appended fragment synthesized null for a field declared non-nullable: Column 'value' is declared as non-nullable but contains null values.

[lance-gatekeeper]

This replaces base-file coverage but leaves matching fragment.overlays live, so overlay values continue to shadow the newly committed replacement. Tombstone these fields in overlays with the established tombstone_overlay_fields helper, as DataReplacement and RewriteColumns already do.

Reproducer run on this head

I staged an existing-field replacement [30,40] at v1, concurrently committed an overlay [111,222] for both rows at v2, then let commit_column_writes retry and succeed at v3.

CARGO_TARGET_DIR=/home/agent/tmp/target-pr8207-overlay-d310 cargo test -p lance repro_commit_column_writes_tombstones_concurrent_overlay -- --nocapture

Expected: the later full replacement is authoritative and scans as [30,40]. Observed: scan returned [111,222]; adding the existing overlay-tombstoning call made the same regression pass.

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The recursive-schema, append-coverage, and overlay fixes now preserve those contracts, but existing-column writes are still not bound to the prepared fragment generation. A retry can therefore overwrite another writer's completed replacement.

Keep the split API, but retain and validate each target fragment's prepare-time file identity—or require recomputation for every relevant fragment conflict—before replaying staged bytes.

[lance-gatekeeper]

A matching field schema does not prove that an existing-column rewrite is still current. When two handles stage that field from one snapshot, the first Merge commits; the second gets a retryable conflict, refreshes, passes this check because the schema is unchanged, and then tombstones the first writer's file on the same fragment ID. The stale commit returns Ok and silently wins. Bind each replacement to the prepare-time fragment/file state and reject or recompute when that state changed.

Reproducer

On 12aa6202bcde3d37e03b0a0b6cfc7e1ef1a3b91a, I added an isolated regression that opened two handles on [10,20], staged existing-field replacements [30,40] and [50,60], committed them in that order, and scanned the result.

cd /home/agent/tmp/pr8207-preexisting-race
CARGO_TARGET_DIR=/home/agent/tmp/pr8207-preexisting-race-target cargo test -p lance reproduce_commit_column_writes_stale_preexisting_rewrite_wins -- --nocapture

The test passed while asserting the defect: the second commit returned Ok and the final values were [50,60]. The required snapshot contract is for the stale commit to fail/recompute and leave [30,40] intact.

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The backing-file guard closes the stale-rewrite race when existing values are physically present, but the same lost update remains for a nullable field that is legitimately uncovered on a fragment.

Keep this approach, but bind every targeted existing field to its exact prepare-time backing state—including absence—before replaying staged bytes.

[lance-gatekeeper]

Absence must be part of this binding. Because this only records a field when a backing path exists, an existing nullable field with no file coverage on the target fragment produces no expectation to compare during retry. Two handles can then fill that fragment; the stale retry sees an empty binding list, tombstones the winner, and returns Ok. Record an expected Option<backing identity> for every targeted existing field and require the current option to match exactly.

Reproducer

On 7c0bb5fee8fef449e883cdb27f9a9039b0198c5e, I created two one-row fragments, committed nullable value=[10] only for fragment 0, then cloned two handles and staged value=[30] and [50] for the still-uncovered fragment 1.

cd /home/agent/tmp/pr8207-stale-fix-7c0bb5
CARGO_TARGET_DIR=/home/agent/tmp/pr8207-preexisting-race-target cargo test -p lance reproduce_stale_rewrite_when_prepare_backing_is_absent -- --nocapture

The first replacement committed, the stale second replacement also returned Ok, and the final scan was [10,50]. The stale commit must fail/recompute and leave the winner's [10,30] intact.

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The absence binding now protects base-file state, but target-field freshness still excludes overlays even though the commit tombstones them.

Bind the complete prepare-time target coverage—including base-file absence/presence and relevant overlay identities/coverage—and reject any change on retry before publishing the staged bytes.

[lance-gatekeeper]

These bindings inspect only prepare_fragment.files, while replace_column_coverage later tombstones fragment.overlays. A concurrent DataOverlay makes Merge retry but leaves the base-file path unchanged, so this check accepts the stale replacement and erases the overlay winner. Include the relevant prepare-time overlay identity/coverage in this compare-and-swap state (or reject a target-field overlay change during retry).

Reproducer

I ran an isolated regression on a5219dc51c8985a6220d04c651b69794e011ad7f beside the existing commit_overlay helper:

1. A stale handle stages [50; 6] for val on fragment 0.
2. A concurrent handle commits a dense overlay [30; 6] for the same fragment and field.
3. The stale handle calls commit_column_writes.
4. The test asserts that the stale call errors and the scan keeps [30; 6].

cd /home/agent/tmp/pr8207-overlay-a521
CARGO_TARGET_DIR=/home/agent/tmp/pr8207-preexisting-race-target cargo test --locked -p lance reproduce_stale_column_write_erases_concurrent_overlay -- --nocapture

Expected: the stale call errors and the overlay winner remains. Observed: the stale call returned Ok(()); the targeted fragment scanned as [50; 6] instead of [30; 6].

[lance-gatekeeper]

❌ Gate recommendation: request changes.

Target-field freshness still excludes overlays, so a concurrent winner can be silently erased.

Bind the full prepare-time target coverage—base absence/presence plus live overlay path, version, and coverage—and compare it on every retry before tombstoning.

[lance-gatekeeper]

Only base data files feed this binding; relevant prepare_fragment.overlays are omitted. A concurrent DataOverlay leaves the base path unchanged, Merge retries, this guard passes, and replace_column_coverage then tombstones the overlay winner. Bind field-selective overlay identity (live field association, path/version/coverage), or reject a target-field overlay change during retry while ignoring unrelated and tombstoned overlay fields.

Reproducer

On 6bd7da5f318573704c06087ba04e3b75643e7124, I added an isolated regression beside the existing commit_overlay helper:

1. A stale handle stages [50; 6] for val on fragment 0.
2. A concurrent handle commits a dense overlay [30; 6] for the same fragment and field.
3. The stale handle calls commit_column_writes.
4. The test asserts that the stale call errors and the scan keeps [30; 6].

cd /home/agent/tmp/pr8207-overlay-6bd
CARGO_TARGET_DIR=/home/agent/tmp/pr8207-overlay-target cargo test --locked -p lance reproduce_stale_column_write_erases_concurrent_overlay -- --nocapture

Expected: the stale call errors and the overlay winner remains. Observed: the stale call returned Ok(()); the targeted fragment scanned as [50; 6] instead of [30; 6].

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The new path guard catches normal overlay appends, but paths alone are not complete overlay identity. A retry can still erase a concurrent target-field coverage change that retains the overlay file path.

Bind each target field to overlay coverage and committed version as well as path—or reject any target-field overlay metadata change—before replaying staged bytes.

[lance-gatekeeper]

Overlay paths do not identify manifest-resident coverage or committed_version. A public Dataset::commit(Operation::Merge) can change the effective coverage of an existing overlay while retaining its file path; after that concurrent commit, this equality still passes, the retry tombstones the overlay, and the stale replacement silently wins. Include target-field coverage and version in FieldBacking (or compare an equivalent field-projected overlay identity) so this forces recomputation.

Reproducer

I added this regression beside the existing overlay helper on b78c7389d1320e5bd5a3598fa15e594c6c4c6127: create an overlay at offset 0, stage a full column replacement, use the public Merge commit to move the same overlay path to offset 1, then commit the stale replacement.

let mut stale = commit_overlay(
    create_base_dataset().await,
    "same_path_overlay",
    0,
    &[1],
    OverlayCoverage::dense(RoaringBitmap::from_iter([0u32])),
    vec![Arc::new(Int32Array::from(vec![777])) as ArrayRef],
).await;
let replacement = stale
    .write_fragment_column(0, replacement_stream, &age_schema)
    .await
    .unwrap();

let concurrent = stale.clone();
let read_version = concurrent.version().version;
let schema = concurrent.schema().clone();
let mut fragments: Vec<_> = concurrent
    .get_fragments()
    .into_iter()
    .map(|fragment| fragment.metadata().clone())
    .collect();
fragments[0].overlays[0].coverage =
    OverlayCoverage::dense(RoaringBitmap::from_iter([1u32]));
Dataset::commit(
    WriteDestination::Dataset(Arc::new(concurrent)),
    Operation::Merge { fragments, schema },
    Some(read_version),
    None,
    None,
    Arc::new(Default::default()),
    false,
).await.unwrap();

let result = stale.commit_column_writes(vec![replacement], None).await;
assert!(result.is_err());

Run with:

CARGO_TARGET_DIR=/home/agent/tmp/pr8207-overlay-target cargo test --locked -p lance repro_stale_column_write_rejects_same_path_overlay_coverage_change -- --nocapture

Expected: the stale commit errors and the winner remains [0, 777, 20, 30, 40, 50] on fragment 0. Observed: the stale commit returned Ok and the fragment read [50, 50, 50, 50, 50, 50].

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The overlay coverage/version race is fixed, but the retry witness is still incomplete: same-path changes to base or overlay data-file routing can alter effective target values without changing the stored identity.

Project the full semantic data-file identity for each target field—path, base, file version, and physical column mapping—alongside overlay precedence and coverage before replaying staged bytes.

[lance-gatekeeper]

A path is not the full physical backing identity. Readers also use base_id, file version, and the target field to column_indices mapping; OverlayBacking omits those DataFile components as well. A concurrent public Merge can therefore change the values selected under the same path, this witness still compares equal, and the stale retry silently overwrites the winner. Store a field-projected data-file witness for both base and overlay entries (path, base_id, file version, and mapped column index), in addition to overlay version and coverage.

Reproducer

I added an isolated regression on b70e82c568f931cbb734e3e1690d3e7eef0bef3a beside create_base_dataset: stage age=[50; 6], then use the public Merge commit to swap the existing two-column base file mapping while retaining its path.

let replacement = stale
    .write_fragment_column(0, futures::stream::iter([Ok(staged)]), &age_schema)
    .await
    .unwrap();

let concurrent = stale.clone();
let read_version = concurrent.version().version;
let schema = concurrent.schema().clone();
let mut fragments: Vec<_> = concurrent
    .get_fragments()
    .into_iter()
    .map(|fragment| fragment.metadata().clone())
    .collect();
assert_eq!(fragments[0].files[0].fields.as_ref(), &[0, 1]);
assert_eq!(fragments[0].files[0].column_indices.as_ref(), &[0, 1]);
fragments[0].files[0].column_indices = Arc::from([1, 0]);
let winner = Dataset::commit(
    WriteDestination::Dataset(Arc::new(concurrent)),
    Operation::Merge { fragments, schema },
    Some(read_version),
    None,
    None,
    Arc::new(Default::default()),
    false,
).await.unwrap();
assert_eq!(read_age(&winner).await, &[0, 1, 2, 3, 4, 5]);

let result = stale.commit_column_writes(vec![replacement], None).await;
assert!(result.is_err());

Run with:

CARGO_TARGET_DIR=/home/agent/tmp/pr8207-overlay-target cargo test --locked -p lance repro_stale_column_write_rejects_same_path_base_mapping_change -- --nocapture

Expected: the stale commit errors, preserving the concurrent age=[0, 1, 2, 3, 4, 5] mapping on fragment 0. Observed: it returned Ok and the fragment read age=[50, 50, 50, 50, 50, 50].

[lance-gatekeeper]

✅ Gate recommendation: approve.

The staged column-write path now binds every targeted field to its complete prepare-time physical backing, including absence, base and overlay routing, file version, column mapping, and coverage. The focused stale-write regressions pass, so retries reject relevant concurrent changes before replaying staged bytes.

[lance-gatekeeper]

✅ Gate recommendation: approve.

The typed column-write errors preserve the previously verified staging and retry contracts, while the table-driven scenarios continue to cover schema, fragment, overlay, and non-nullability races. Bounded footer reads and cached staged-file sizes do not change atomic publication semantics.

[lance-gatekeeper]

❌ Gate recommendation: request changes.

New-column staging must preserve the reserved virtual-column contract. The current helper and commit path can publish a stored system-column name that validates but cannot be projected.

A viable revision should reject every is_system_column name when building and committing staged schemas, matching the ordinary insert boundary.

[lance-gatekeeper]

This check covers only fields already stored in the dataset, so virtual system names such as _rowid pass. commit_column_writes accepts the staged schema too; the resulting version validates, but projecting _rowid fails because the scanner adds a second synthetic UInt64 field with the same name. Reject lance_core::is_system_column(field.name()) here and at commit validation, matching the insert path so callers who build their own staging schema cannot bypass the guard.

Reproducer run on this head

I added this regression beside the new column-write tests on 49ea13aef2284d442e3c9edc799cf41b5f7d6d0c:

let mut dataset = id_dataset(2, 1024).await;
let arrow = Arc::new(ArrowSchema::new(vec![ArrowField::new(
    ROW_ID,
    DataType::Int32,
    true,
)]));
let schema = dataset.new_column_schema(arrow.as_ref()).unwrap();
let fragment_id = dataset.get_fragments()[0].id() as u64;
let staged = arrow_array::record_batch!((ROW_ID, Int32, [7, 8])).unwrap();
let replacement = dataset
    .write_fragment_column(
        fragment_id,
        futures::stream::iter([Ok(staged)]),
        &schema,
    )
    .await
    .unwrap();
dataset.commit_column_writes(vec![replacement], None).await.unwrap();
dataset.validate().await.unwrap();
let projected = dataset.scan().project(&[ROW_ID]);
assert!(projected.is_ok());

Run with:

CARGO_TARGET_DIR=/home/agent/tmp/pr8207-overlay-target cargo test --locked -p lance reproduce_new_column_schema_accepts_reserved_rowid -- --nocapture

Expected: the reserved name is rejected before staging or commit, or at minimum projection remains valid. Observed: schema creation, staging, commit, and validate() all returned Ok; projection failed with Duplicate field name "_rowid" in schema, showing both the stored Int32 field and synthetic UInt64 field.

[lance-gatekeeper]

✅ Gate recommendation: approve.

Reserved virtual-column names are now rejected both by the schema helper and at the commit boundary, so callers cannot publish the stored/synthetic collision through either supported or hand-built staging schemas. The prior atomicity and stale-write protections remain intact.

[wjones127]

question: why is this info collected after the fact? Seems like it could be collected up front instead.

[wjones127]

praise: this is nice. At some point I want to refactor our errors to be like this. And also move off of snafu and onto thiserror.

[wjones127]

suggestion: I think it would be totally fine to make write_fragment_column return immutable structs that could be passed to the commit function. That would avoid the need for all this validation.

[Xuanwo]

Thanks for digging into this and fighting with lance-gatekeeper 😆. there’s a lot of careful work here, especially around streaming, file metadata, cleanup, and the stale-write / compaction cases.

One thing I keep coming back to is whether we want to introduce a second column-writing path alongside Updater / FileFragment::add_columns. The individual pieces make sense, but together this gives us another physical writer, staging contract, schema-numbering API, error model, retry job, and commit entry point.

That means we’ll have two paths maintaining many of the same invariants around field IDs, deletions, file versions, schema evolution, tombstoning, and concurrent commits. They already have slightly different row semantics: Updater works with live rows and restores deleted positions, while write_fragment_column expects every physical row from the caller. I’m a little worried these two paths will drift over time.

From what I can tell, FileFragment::add_columns is already pretty close to the composition we need here: it works at the fragment level, accepts streamed or precomputed input, and returns (Fragment, Schema). The coordinator can then collect those results and commit them through Operation::Merge, which is also how the existing distributed LanceFragment.merge_columns flow is structured.

Would it make sense to build this around that path instead?

• Workers would produce updated fragments through FileFragment::add_columns.
• If Updater's scan and row-alignment work is too expensive for already-computed physical batches, we could add a direct fast path under the fragment API while keeping the same (Fragment, Schema) contract.
• The coordinator would commit the collected fragments with Operation::Merge.
• Any missing stale-fragment, field-backing, tombstoning, or conflict checks could live in the shared fragment / Merge machinery.
• Scheduling, checkpoints, recomputation, and resume policy would stay in our enterprise.

The main thing I’d like to understand is what requirement the existing path cannot express. If avoiding Updater is the key reason, that seems reasonable. I’d just prefer to make that a fast path behind the existing fragment-level abstraction instead of introducing a parallel Dataset-level staging and commit protocol.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!