diff --git a/docs/src/format/table/.pages b/docs/src/format/table/.pages
index 16c20058608..5b0cb0e95e6 100644
--- a/docs/src/format/table/.pages
+++ b/docs/src/format/table/.pages
@@ -6,4 +6,5 @@ nav:
- Layout: layout.md
- Branch & Tag: branch_tag.md
- Row ID & Lineage: row_id_lineage.md
+ - Data Overlay Files: data_overlay_file.md
- MemTable & WAL: mem_wal.md
diff --git a/docs/src/format/table/data_overlay_file.md b/docs/src/format/table/data_overlay_file.md
new file mode 100644
index 00000000000..5540f860018
--- /dev/null
+++ b/docs/src/format/table/data_overlay_file.md
@@ -0,0 +1,390 @@
+# Data Overlay Files
+
+!!! note "Overlay files require feature flag 64 (data overlay files)"
+
+ A reader or writer that does not understand overlay files must refuse a
+ dataset that uses them. Silently ignoring an overlay would return stale base
+ values, which is a correctness bug rather than a degraded experience.
+
+Overlay files supply new values for a subset of `(row offset, field)` cells
+within a fragment **without rewriting the fragment's base data files**. They make
+updates cheap when only a small fraction of rows and/or columns change: instead
+of rewriting whole columns or moving rows to a new fragment, a writer appends a
+small file carrying just the changed cells.
+
+This is Lance's third mechanism for changing data in place, alongside
+[deletion files](index.md#deletion-files) (which remove rows) and
+[data evolution](index.md#data-evolution) (which adds or rewrites whole columns).
+An overlay changes individual cells.
+
+## Concepts
+
+### Coverage and resolution
+
+Each overlay declares which cells it provides through a **coverage** bitmap (or,
+for sparse overlays, one bitmap per field). The bitmaps index **physical row
+offsets** — positions in the base data files, counting deleted rows — so they are
+stable across deletions, exactly like deletion vectors.
+
+To resolve a cell `(offset, field)` on read, walk the fragment's overlays from
+**newest to oldest**. The first overlay that covers `(offset, field)` wins; its
+value is used. If no overlay covers the cell, the value falls through to the base
+data file (or is `NULL` if no base data file holds that field).
+
+Precedence among overlays is determined by:
+
+1. `committed_version` — higher wins (see [Versioning](#versioning-and-ordering)).
+2. Position in `DataFragment.overlays` as a tiebreaker — a later entry is newer.
+
+A covered offset whose value is `NULL` overrides the cell **to** `NULL`. This is
+distinct from an offset that is simply absent from the bitmap, which falls
+through to the base. Coverage, not value-nullness, decides whether an overlay
+applies.
+
+### Interaction with deletions
+
+Deletions take precedence over overlays. If a row offset is marked deleted in the
+fragment's deletion file, any overlay value for that offset is dead and is
+ignored, regardless of commit order. This keeps the invariant simple: a deletion
+is the final word on a row, so a concurrent overlay against a row that was
+deleted needs no special conflict handling — its values are merely inert.
+
+### Physical layout
+
+An overlay's data file stores **one value column per field**, in the order of
+`data_file.fields`. It does **not** store a row-offset key column. The position of
+a covered offset's value within its column is the **rank** of that offset in the
+field's coverage bitmap — the number of set bits below it. For a Roaring bitmap
+this is an O(1) operation, so random access to any cell is a rank computation
+followed by a single value fetch, with no offset column to read and no binary
+search.
+
+Because different fields may cover different offset sets, the value columns of a
+single sparse overlay may have **different lengths**. The Lance file format
+permits columns of differing item counts within one file, so a sparse overlay is
+representable as a single file. (See [Writer support](#writer-support) for the
+current implementation status.)
+
+### Dense vs. sparse overlays
+
+A single overlay is one of two shapes:
+
+- **Dense (rectangular).** One `shared_offset_bitmap` applies to every field. Every
+ covered offset has a value for every field. This is the common case for a plain
+ `UPDATE`, where one `SET` list is applied to one set of rows.
+- **Sparse.** A `FieldCoverage` carries one bitmap per field, used when different
+ fields cover different offset sets — for example a `MERGE` with multiple
+ `WHEN MATCHED` branches, where different rows update different columns. A dense
+ overlay would have to widen to the bounding rectangle and fill the untouched
+ cells with their current values (post-images), which for wide columns such as
+ embeddings means re-storing data that did not change. A sparse overlay stores
+ exactly the changed cells.
+
+A writer may always express a non-rectangular update as **multiple dense overlays
+in one transaction** (one per coverage group) instead of a single sparse overlay.
+
+## Protobuf
+
+
+DataOverlayFile protobuf message
+
+```protobuf
+%%% proto.message.DataOverlayFile %%%
+```
+
+
+
+
+FieldCoverage protobuf message
+
+```protobuf
+%%% proto.message.FieldCoverage %%%
+```
+
+
+
+## Versioning and ordering
+
+Overlays reuse the dataset version as their ordering clock rather than
+introducing a separate generation counter.
+
+`committed_version` is the dataset version at which an overlay **became
+effective** — the version of the commit that introduced it, **not** the version
+it was read from. It is stamped at commit time and re-stamped if the commit is
+retried, in the same way as the created-at / last-updated-at version sequences.
+
+This single value drives every ordering decision:
+
+- **Overlay vs. overlay** (read precedence): higher `committed_version` wins.
+- **Overlay vs. index** (query correctness): an index records the
+ `dataset_version` it was built from. An index whose `dataset_version >=
+ committed_version` already incorporates the overlay. An overlay whose
+ `committed_version > index.dataset_version` is newer than the index and its
+ cells must be excluded from index results and re-evaluated.
+- **Scheduler signal**: the gap between an overlay's `committed_version` and an
+ index's `dataset_version`, or between an overlay and the base, is a staleness
+ measure the compaction scheduler can use.
+
+!!! note "Why effective version, not read version"
+
+ Suppose an overlay reads version 5 and commits at version 6, while an index
+ is built reading version 5 (before the overlay) and commits at version 7 with
+ `dataset_version = 5`. If the overlay stored its *read* version (5), the test
+ `5 > 5` is false, the row would not be excluded, and the index — which never
+ saw the overlay — would return a stale result. Storing the *effective*
+ version (6) makes `6 > 5` true, the cell is excluded and re-evaluated, and the
+ result is correct.
+
+## Index integration
+
+Building an index over a fragment that has overlays does **not** require dropping
+the fragment from the index's coverage. The fragment stays indexed, and the query
+path reconciles overlays at query time using an **exclusion set**.
+
+The exclusion set for an index on field `F` is the union of the coverage bitmaps,
+restricted to field `F`, of every overlay whose `committed_version >
+index.dataset_version`. The exclusion is **field-aware**: an overlay that touches
+only unrelated columns does not exclude anything from the index on `F`.
+
+The query then proceeds as:
+
+1. Run the index search as usual, producing candidate rows.
+2. Remove any candidate in the exclusion set. (Its indexed value may be stale.)
+3. **Re-evaluate** the excluded rows against their current values — the same flat
+ path already used for the unindexed tail of fragments. For a scalar predicate
+ this re-applies the filter; for a vector query it re-scores the row's current
+ vector. Rows that still match are added back to the result.
+
+Step 3 is what makes exclusion correct rather than merely safe: removing a row
+from index candidates without re-evaluating it would silently drop a row that
+should match under its new value.
+
+### Correctness invariant
+
+> For every indexed field `F` and every row offset `o` in a fragment the index
+> covers, the index's entry for `(o, F)` is trusted unless `o` is excluded.
+> `o` is excluded iff some overlay with `committed_version > index.dataset_version`
+> covers `(o, F)`.
+
+The write and compaction paths together preserve this:
+
+- **Writes** change a cell only by adding an overlay, and that overlay's
+ `committed_version` exceeds the version of any pre-existing index — so the
+ change is always covered by an exclusion.
+- **Compaction** may remove an overlay only if the index no longer relies on it
+ (see below).
+
+## Compaction
+
+Overlays accumulate read cost — every overlay is a bitmap to test and a possible
+file to open. Compaction bounds that cost in two modes:
+
+- **Overlay → overlay.** Merge several overlays into fewer, computing the
+ post-image per `(offset, field)` by walking the merged overlays newest-first.
+ The merged overlay takes the **maximum** `committed_version` of its inputs, so
+ the exclusion semantics are preserved. Indexes are unaffected. This is cheap and
+ does not touch the base.
+- **Overlay → base.** Fold overlays into a fresh base data file, computing the
+ post-image for every covered cell, then clear the overlays. The base is
+ complete, so every post-image is well defined. Overlay offsets are physical, so
+ they cannot survive a rewrite that reorders rows; folding therefore materializes
+ values rather than carrying overlays forward.
+
+!!! warning "Folding an indexed field must update its index"
+
+ An overlay→base fold removes the overlay, which removes the exclusion signal
+ that kept an index correct. Folding an overlay that covers an indexed field
+ `F` is therefore equivalent to a column rewrite of `F` and must, in the same
+ commit, either rebuild the index to a `dataset_version` at least the folded
+ overlay's `committed_version`, or remove the fragment from the index's
+ coverage so the rows fall to the flat path. Otherwise the index would serve
+ stale values with no overlay to exclude them. This is the same rule that
+ already governs rewriting a column that an index is built on.
+
+When a fragment with overlays is compacted by a row-rewriting operation
+(`RewriteRows`, which produces new fragments with new row addresses), the
+overlays are folded into the new base as part of the rewrite, and existing
+[fragment-reuse remapping](row_id_lineage.md) handles the row-address changes as
+it does today.
+
+## Row lineage
+
+An overlay write updates the `last_updated_at_version` of every covered row, so
+change-data-feed and time-travel queries observe the update. Because overlays are
+addressed by physical offset, they do **not** require stable row IDs to be
+enabled; lineage updates apply only when those features are on.
+
+## Worked example
+
+A table `users` with stable row IDs enabled and these fields:
+
+| field id | name | type |
+|----------|-----------|-------------------------|
+| 1 | id | `int32` (primary key) |
+| 2 | name | `utf8` |
+| 3 | age | `int32` |
+| 4 | embedding | `fixed_size_list`|
+
+Created at version 1 as a single fragment `0` with one base data file
+`data/file0.lance` holding all four columns. `physical_rows = 4`:
+
+| offset | id | name | age | embedding |
+|--------|----|-------|-----|------------------|
+| 0 | 1 | Alice | 30 | … |
+| 1 | 2 | Bob | 25 | … |
+| 2 | 3 | Carol | 40 | … |
+| 3 | 4 | Dave | 22 | … |
+
+A BTree scalar index on `age` is built at version 1, covering fragment `0`
+(`dataset_version = 1`).
+
+### Step 1 — write an overlay
+
+```sql
+UPDATE users SET age = 26 WHERE id = 2; -- Bob, offset 1
+```
+
+This touches one field (`age`) for one row, so the writer emits a dense overlay
+and commits it as version 2. Fragment `0` gains:
+
+```text
+DataOverlayFile {
+ data_file: { path: "data/overlay-.lance", fields: [3], column_indices: [0] }
+ coverage: shared_offset_bitmap = {1}
+ committed_version: 2
+}
+```
+
+The overlay file stores a single `age` column with one value, `[26]`, at
+rank `{1}.rank(1) = 0`. `last_updated_at_version[1]` is set to 2.
+
+### Step 2 — read
+
+`SELECT id, age FROM users` reads base ages `[30, 25, 40, 22]`. For `age`
+(field 3), the overlay covers offset 1, so `age[1]` is replaced with the overlay
+value at position `{1}.rank(1) = 0` → `26`. Result ages: `[30, 26, 40, 22]`.
+
+### Step 3 — index query
+
+```sql
+SELECT * FROM users WHERE age = 26;
+```
+
+The `age` index was built at `dataset_version = 1`; the overlay's
+`committed_version` is 2. Since `2 > 1`, the overlay's coverage for `age`, `{1}`,
+is the exclusion set for this query.
+
+- The index (built at v1) holds Bob's *old* `age = 25`, so a lookup for `26`
+ returns nothing from the index.
+- Offset 1 is in the exclusion set, so it is re-evaluated on the flat path. Its
+ current `age` (26, via the overlay) matches `age = 26`, so Bob is returned.
+
+The mirror case `WHERE age = 25` shows exclusion preventing a stale hit: the index
+returns offset 1 (stale `25`), but offset 1 is excluded, re-evaluated to `26`, and
+correctly dropped.
+
+### Step 4 — a second, non-rectangular write
+
+```sql
+MERGE INTO users USING staged ON users.id = staged.id
+WHEN MATCHED AND staged.kind = 'rename' THEN UPDATE SET name = staged.name -- Carol(2), Dave(3)
+WHEN MATCHED AND staged.kind = 'revec' THEN UPDATE SET embedding = staged.embedding -- Bob(1)
+```
+
+`name` is updated for offsets `{2, 3}` and `embedding` for offset `{1}` — different
+fields over different rows. This is a sparse overlay, committed as version 3:
+
+```text
+DataOverlayFile {
+ data_file: { path: "data/overlay-.lance", fields: [2, 4], column_indices: [0, 1] }
+ coverage: field_coverage { offset_bitmaps: [ {2,3}, {1} ] }
+ // name (field 2) ^ ^ embedding (field 4)
+ committed_version: 3
+}
+```
+
+The file's `name` column has **two** values (`["Caroline", "David"]`, at
+ranks 0 and 1 of `{2,3}`) and its `embedding` column has **one** value (at rank 0
+of `{1}`) — columns of different lengths in one file.
+
+### Step 5 — read after the second write
+
+`SELECT name, age, embedding FROM users` resolves each field independently,
+newest overlay first:
+
+- `name`: the v3 overlay covers `{2,3}` → `["Alice", "Bob", "Caroline", "David"]`.
+- `age`: the v3 overlay does not cover `age`; the v2 overlay still applies at
+ offset 1 → `[30, 26, 40, 22]`.
+- `embedding`: the v3 overlay covers `{1}` → Bob's vector is the new one, others
+ from base.
+
+Overlays from different versions coexist and apply per field.
+
+### Step 6 — compaction (overlay → base)
+
+The scheduler folds both overlays into fragment `0` at version 4, computing
+post-images for `age`, `name`, and `embedding`, and writing a new base data file
+`data/file1.lance` with those columns. In the old file, fields 2, 3, and 4 are
+tombstoned (`-2`); field 1 (`id`) remains. The fragment's `overlays` list is
+cleared. Row addresses are preserved (a column rewrite, not a row rewrite), so
+stable row IDs and the deletion vector are untouched.
+
+Because the fold removed the overlay that was excluding offset 1 from the `age`
+index, the same commit must reconcile that index: either rebuild it at
+`dataset_version >= 2`, or drop fragment `0` from its coverage so `age` queries
+fall to the flat path. After a rebuild at version 4, no overlay remains and the
+`age` index directly returns `26` for Bob with no exclusion needed.
+
+## Guidance
+
+!!! note "This section is a stub."
+
+ The following are implementation considerations, not part of the on-disk
+ specification.
+
+### When to overlay vs. rewrite a column vs. move rows
+
+*(To be expanded.)* The choice between appending an overlay, rewriting a full
+column (data evolution), and moving updated rows to a new fragment depends on the
+fraction of rows changed, the fraction of columns changed, column width, the
+presence of indexes on the changed columns, and the accumulated overlay read
+cost. Roughly: few rows changed favors overlays; most rows in a few columns
+favors a column rewrite; most columns changed favors moving rows to a new
+fragment.
+
+### Writer support
+
+*(To be expanded.)* Dense (rectangular) overlays write with the existing
+equal-length file writer today. Sparse overlays stored as a **single** file
+require the writer to emit columns of independent lengths, which the current v2
+writer does not yet do (it advances all columns from one global row counter).
+Until that support lands, a writer can express a sparse update as multiple dense
+overlays in one transaction.
+
+### Scheduling compaction
+
+*(To be expanded.)* The overlay→overlay and overlay→base modes have very
+different costs; a cost/benefit scheduler decides when each is worthwhile, using
+the version gap as a staleness signal.
+
+### Open questions
+
+*(To be resolved.)*
+
+- **Per-fragment vs. per-table overlays.** Overlays are attached per fragment.
+ Should there be a table-level overlay concept, and how would it interact with
+ fragment-level row addressing?
+- **Relationship to LSM.** Overlays plus compaction resemble an LSM tree (newest
+ layer wins, periodic merge). How far should that analogy be taken, and what do
+ we deliberately do differently given Lance's random-access requirements?
+- **Coverage bitmap spill.** Coverage bitmaps live inline in the manifest. Very
+ large coverage (an overlay touching many rows) may warrant external spill, as
+ the row-ID and last-updated-at sequences already do above a size threshold.
+
+## Related specifications
+
+- [Table format overview](index.md)
+- [Transactions](transaction.md)
+- [Row ID & Lineage](row_id_lineage.md)
+- [Index Formats](../index/index.md)
+- [Format Versioning](versioning.md)
diff --git a/docs/src/format/table/index.md b/docs/src/format/table/index.md
index 94ea4b90dc9..ce4d0b26613 100644
--- a/docs/src/format/table/index.md
+++ b/docs/src/format/table/index.md
@@ -168,6 +168,35 @@ However, this invalidates row addresses and requires rebuilding indices, which c
+## Data Overlay Files
+
+!!! note "Overlay files require feature flag 64 (data overlay files)"
+
+Overlay files supply new values for a subset of `(row offset, field)` cells within
+a fragment without rewriting the base data files. They make updates cheap when only
+a small percentage of rows and/or columns change: a writer appends a small file
+carrying just the changed cells instead of rewriting whole columns or moving rows
+to a new fragment.
+
+On read, each cell is resolved by consulting the fragment's overlays from newest to
+oldest; the first overlay covering that `(offset, field)` wins, otherwise the value
+falls through to the base data file. Indices keep covering the fragment and reconcile
+overlays at query time through a field-aware exclusion set.
+
+For the full specification — coverage and resolution rules, dense vs. sparse layout,
+versioning, index integration, compaction, and a worked example — see the
+[Data Overlay Files Specification](data_overlay_file.md).
+
+
+DataOverlayFile protobuf message
+
+```protobuf
+%%% proto.message.DataOverlayFile %%%
+```
+
+
+
+
## Related Specifications
### Storage Layout
diff --git a/protos/table.proto b/protos/table.proto
index d298809d5d8..cc8b477a6a6 100644
--- a/protos/table.proto
+++ b/protos/table.proto
@@ -113,6 +113,11 @@ message Manifest {
// * 2: row ids are stable and stored as part of the fragment metadata.
// * 4: use v2 format (deprecated)
// * 8: table config is present
+ // * 16: data files use multiple base paths (shallow clone / multi-base)
+ // * 32: the transaction file under _transactions is not written (inline only)
+ // * 64: data overlay files are present (see DataOverlayFile). Readers that do
+ // not understand overlays must refuse the dataset, since ignoring an overlay
+ // would silently return stale base values.
uint64 reader_feature_flags = 9;
// Feature flags for writers.
@@ -311,6 +316,15 @@ message DataFragment {
repeated DataFile files = 2;
+ // Optional overlay files for this fragment, which supply new values for a
+ // subset of cells without rewriting the base data files. This MUST be empty
+ // if the data overlay files feature flag (64) is not set in the manifest.
+ //
+ // Order is significant: a later entry is newer than an earlier one. When two
+ // overlays cover the same (offset, field) and share a `committed_version`, the
+ // later entry wins. See DataOverlayFile for the full resolution rules.
+ repeated DataOverlayFile overlays = 11;
+
// File that indicates which rows, if any, should be considered deleted.
DeletionFile deletion_file = 3;
@@ -433,6 +447,66 @@ message DataFile {
optional uint32 base_id = 7;
} // DataFile
+// An overlay file supplies new values for a subset of (row offset, field) cells
+// within a fragment, without rewriting the fragment's base data files. It is
+// used for efficient updates when only a small fraction of rows and/or columns
+// change.
+//
+// On read, a cell is resolved by consulting the fragment's overlays from newest
+// to oldest: the first overlay that covers that (offset, field) wins; if none
+// cover it, the value falls through to the base data file. Because deletions
+// take precedence over overlays, an overlay value for an offset that is also
+// marked deleted is dead and is ignored.
+//
+// The overlay's data file does NOT store a row-offset key column. Within a value
+// column, the position of a covered offset's value is the rank (0-based count of
+// set bits below it) of that offset within the field's coverage bitmap. Because
+// fields may cover different offset sets, the value columns of a single overlay
+// data file may have different lengths (which the Lance file format permits).
+message DataOverlayFile {
+ // The data file storing the overlay's new cell values, one value column per
+ // field in `data_file.fields`. No row-offset key column is stored.
+ DataFile data_file = 1;
+
+ // Which (offset, field) cells this overlay provides values for.
+ oneof coverage {
+ // A single 32-bit Roaring bitmap of physical row offsets that applies to
+ // every field in `data_file.fields` (a "dense" / rectangular overlay).
+ // Every covered offset has a value for every field. This is the common case
+ // for a plain UPDATE, where one SET list is applied to one set of rows.
+ bytes shared_offset_bitmap = 2;
+ // Per-field coverage for a "sparse" overlay, used when different fields cover
+ // different offset sets (e.g. a MERGE with multiple WHEN MATCHED branches).
+ FieldCoverage field_coverage = 4;
+ }
+
+ // The dataset version at which this overlay became effective: the version of
+ // the commit that introduced it, NOT the version it was read from. It is
+ // stamped at commit time and re-stamped if the commit is retried, in the same
+ // way as the created-at / last-updated-at version sequences.
+ //
+ // This drives two orderings:
+ // * Versus index builds: an index whose `dataset_version` >= this value
+ // already incorporates this overlay. Otherwise the overlay's covered cells
+ // are excluded from index results for the affected fields and re-evaluated
+ // against their current values (see the Data Overlay Files specification).
+ // * Versus other overlays: when two overlays cover the same (offset, field),
+ // the one with the higher `committed_version` wins. Overlays that share a
+ // `committed_version` are ordered by their position in
+ // `DataFragment.overlays`, where a later entry is newer and wins.
+ uint64 committed_version = 3;
+}
+
+// Per-field coverage for a sparse overlay.
+message FieldCoverage {
+ // One entry per field in the overlay's `data_file.fields`, in the same order.
+ // Each is a 32-bit Roaring bitmap of the physical row offsets covered for that
+ // field. An offset present in a field's bitmap but mapped to a NULL value
+ // means the cell is overridden to NULL (distinct from an offset that is absent,
+ // which falls through to the base data file).
+ repeated bytes offset_bitmaps = 1;
+}
+
// Deletion File
//
// The path of the deletion file is constructed as:
diff --git a/protos/transaction.proto b/protos/transaction.proto
index e72e95025a4..bfc0eee354b 100644
--- a/protos/transaction.proto
+++ b/protos/transaction.proto
@@ -315,6 +315,44 @@ message Transaction {
repeated DataReplacementGroup replacements = 1;
}
+ // Overlay files to append to a single fragment, in order (the last entry is
+ // newest). The overlays are appended to the fragment's existing `overlays`
+ // list; they do not replace it, so overlays written by concurrent commits are
+ // preserved.
+ message DataOverlayGroup {
+ uint64 fragment_id = 1;
+ // Each DataOverlayFile.committed_version is left 0 by the writer and stamped
+ // to the new dataset version at commit time (re-stamped on retry), in the
+ // same way as the created-at / last-updated-at version sequences. The fields
+ // touched are read from each overlay's `data_file.fields`.
+ repeated DataOverlayFile overlays = 2;
+ }
+
+ // Attach overlay files to fragments, supplying new values for a subset of
+ // (row offset, field) cells without rewriting the fragments' base data files.
+ // See the DataOverlayFile message in table.proto and the Data Overlay Files
+ // specification for resolution, coverage, and versioning rules.
+ //
+ // Conflict semantics (intentionally permissive, like DataReplacement). Against
+ // a concurrent operation that touches one of the same fragments:
+ // * Another DataOverlay (any fields): COMPATIBLE. Overlays stack; when two
+ // overlays cover the same (offset, field) the one with the higher
+ // `committed_version` wins, so independent backfills never conflict.
+ // * Append / new fragments: COMPATIBLE.
+ // * Delete: COMPATIBLE. A deletion takes precedence over an overlay, so an
+ // overlay value for a deleted offset is inert (no special handling needed).
+ // * DataReplacement or column-rewrite (Update with REWRITE_COLUMNS) of the
+ // same field: COMPATIBLE. Both preserve physical row addresses, so overlay
+ // offsets stay valid; the overlay is newer and wins its covered cells, and
+ // the version gate excludes those cells from any rebuilt index.
+ // * Row-rewrite, compaction, or an overlay->base fold of the fragment:
+ // CONFLICT. These change physical row addresses or consume the overlays, so
+ // the overlay's offsets are no longer valid. The writer must re-read the new
+ // fragment, recompute, and retry.
+ message DataOverlay {
+ repeated DataOverlayGroup groups = 1;
+ }
+
// Update the merged generations in MemWAL index.
// This operation is used during merge-insert to atomically record which
// generations have been merged to the base table.
@@ -346,6 +384,7 @@ message Transaction {
UpdateMemWalState update_mem_wal_state = 112;
Clone clone = 113;
UpdateBases update_bases = 114;
+ DataOverlay data_overlay = 115;
}
// Fields 200/202 (`blob_append` / `blob_overwrite`) previously represented blob dataset ops.
diff --git a/rust/lance-file/src/reader.rs b/rust/lance-file/src/reader.rs
index c454f73819e..98135b60f7e 100644
--- a/rust/lance-file/src/reader.rs
+++ b/rust/lance-file/src/reader.rs
@@ -491,6 +491,87 @@ impl FileReader {
self.num_rows
}
+ /// The number of rows stored in a single physical column.
+ ///
+ /// For ordinary (rectangular) files every column has the same length, equal
+ /// to [`num_rows`](Self::num_rows). Files written with
+ /// [`FileWriter::write_column`](crate::writer::FileWriter::write_column)
+ /// may have columns of differing lengths; this returns the length of one
+ /// such column, derived by summing its pages' row counts. Returns `None` if
+ /// `column_index` is out of bounds.
+ pub fn column_num_rows(&self, column_index: usize) -> Option {
+ self.metadata
+ .column_metadatas
+ .get(column_index)
+ .map(|col| col.pages.iter().map(|page| page.length).sum())
+ }
+
+ // Number of physical columns a field (and its descendants) contributes to a
+ // projection, mirroring `ReaderProjection::from_field_ids_helper`. Used to
+ // partition a projection's flat `column_indices` by top-level field.
+ fn count_projected_columns(field: &Field, is_structural: bool) -> usize {
+ let contributes = !is_structural
+ || field.children.is_empty()
+ || field.is_blob()
+ || field.is_packed_struct();
+ let recurse = !is_structural || (!field.is_blob() && !field.is_packed_struct());
+ let mut count = contributes as usize;
+ if recurse {
+ for child in &field.children {
+ count += Self::count_projected_columns(child, is_structural);
+ }
+ }
+ count
+ }
+
+ // The top-level row count of each projected top-level field, in projection
+ // order. A field's length is the page-row sum of its root (first) physical
+ // column, which equals the field's top-level row count for primitive,
+ // struct, and list fields in both v2.0 and v2.1. Ordinary files have equal
+ // lengths across columns; files written with `write_column` may not.
+ fn projected_field_lengths(&self, projection: &ReaderProjection) -> Vec {
+ let is_structural = self.metadata.version() >= LanceFileVersion::V2_1;
+ let mut lengths = Vec::with_capacity(projection.schema.fields.len());
+ let mut cursor = 0usize;
+ for field in &projection.schema.fields {
+ let n = Self::count_projected_columns(field, is_structural);
+ if n == 0 {
+ continue;
+ }
+ let Some(&root_column) = projection.column_indices.get(cursor) else {
+ break;
+ };
+ lengths.push(self.column_num_rows(root_column as usize).unwrap_or(0));
+ cursor += n;
+ }
+ lengths
+ }
+
+ // The reader combines a projection's columns into rectangular batches, so
+ // they must all have the same length. Returns that common length, or a
+ // descriptive error (naming the per-column lengths) when they differ.
+ // Ordinary files always pass; only files written with `write_column` whose
+ // columns ended up unequal can fail, and those must be read separately.
+ fn verify_uniform_lengths(projection: &ReaderProjection, field_lengths: &[u64]) -> Result {
+ let first = field_lengths.first().copied().unwrap_or(0);
+ if field_lengths.iter().all(|&len| len == first) {
+ return Ok(first);
+ }
+ let columns = projection
+ .schema
+ .fields
+ .iter()
+ .map(|f| f.name.as_str())
+ .zip(field_lengths.iter().copied())
+ .map(|(name, len)| format!("{name}={len}"))
+ .collect::>()
+ .join(", ");
+ Err(Error::invalid_input(format!(
+ "cannot read columns of differing lengths together ({columns}); \
+ read each column (or equal-length group) separately"
+ )))
+ }
+
pub fn metadata(&self) -> &Arc {
&self.metadata
}
@@ -1221,11 +1302,18 @@ impl FileReader {
) -> Result + Send>>> {
let projection = projection.unwrap_or_else(|| self.base_projection.clone());
Self::validate_projection(&projection, &self.metadata)?;
+ // All projected columns must share a length: the reader combines them
+ // into rectangular batches. Ordinary files satisfy this (every column
+ // has `self.num_rows` rows); files written with `FileWriter::write_column`
+ // may not, and such columns must be read separately. `read_len` is that
+ // common length, which `RangeFull`/`RangeFrom` resolve against (rather
+ // than `self.num_rows`, the file's longest column).
+ let field_lengths = self.projected_field_lengths(&projection);
+ let read_len = Self::verify_uniform_lengths(&projection, &field_lengths)?;
let verify_bound = |params: &ReadBatchParams, bound: u64, inclusive: bool| {
- if bound > self.num_rows || bound == self.num_rows && inclusive {
+ if bound > read_len || (bound == read_len && inclusive) {
Err(Error::invalid_input(format!(
- "cannot read {:?} from file with {} rows",
- params, self.num_rows
+ "cannot read {params:?} from columns with {read_len} rows"
)))
} else {
Ok(())
@@ -1267,13 +1355,8 @@ impl FileReader {
}
ReadBatchParams::RangeFrom(range) => {
verify_bound(¶ms, range.start as u64, true)?;
- self.read_range(
- range.start as u64..self.num_rows,
- batch_size,
- projection,
- filter,
- )
- .await
+ self.read_range(range.start as u64..read_len, batch_size, projection, filter)
+ .await
}
ReadBatchParams::RangeTo(range) => {
verify_bound(¶ms, range.end as u64, false)?;
@@ -1281,7 +1364,7 @@ impl FileReader {
.await
}
ReadBatchParams::RangeFull => {
- self.read_range(0..self.num_rows, batch_size, projection, filter)
+ self.read_range(0..read_len, batch_size, projection, filter)
.await
}
}
@@ -1476,11 +1559,18 @@ impl FileReader {
) -> Result> {
let projection = projection.unwrap_or_else(|| self.base_projection.clone());
Self::validate_projection(&projection, &self.metadata)?;
+ // All projected columns must share a length: the reader combines them
+ // into rectangular batches. Ordinary files satisfy this (every column
+ // has `self.num_rows` rows); files written with `FileWriter::write_column`
+ // may not, and such columns must be read separately. `read_len` is that
+ // common length, which `RangeFull`/`RangeFrom` resolve against (rather
+ // than `self.num_rows`, the file's longest column).
+ let field_lengths = self.projected_field_lengths(&projection);
+ let read_len = Self::verify_uniform_lengths(&projection, &field_lengths)?;
let verify_bound = |params: &ReadBatchParams, bound: u64, inclusive: bool| {
- if bound > self.num_rows || bound == self.num_rows && inclusive {
+ if bound > read_len || (bound == read_len && inclusive) {
Err(Error::invalid_input(format!(
- "cannot read {:?} from file with {} rows",
- params, self.num_rows
+ "cannot read {params:?} from columns with {read_len} rows"
)))
} else {
Ok(())
@@ -1521,7 +1611,7 @@ impl FileReader {
ReadBatchParams::RangeFrom(range) => {
verify_bound(¶ms, range.start as u64, true)?;
self.read_range_blocking(
- range.start as u64..self.num_rows,
+ range.start as u64..read_len,
batch_size,
projection,
filter,
@@ -1532,7 +1622,7 @@ impl FileReader {
self.read_range_blocking(0..range.end as u64, batch_size, projection, filter)
}
ReadBatchParams::RangeFull => {
- self.read_range_blocking(0..self.num_rows, batch_size, projection, filter)
+ self.read_range_blocking(0..read_len, batch_size, projection, filter)
}
}
}
diff --git a/rust/lance-file/src/writer.rs b/rust/lance-file/src/writer.rs
index 12bd50df6fe..96d282e978b 100644
--- a/rust/lance-file/src/writer.rs
+++ b/rust/lance-file/src/writer.rs
@@ -6,7 +6,7 @@ use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
-use arrow_array::RecordBatch;
+use arrow_array::{ArrayRef, RecordBatch};
use arrow_data::ArrayData;
use bytes::{Buf, BufMut, Bytes, BytesMut};
@@ -221,6 +221,11 @@ pub struct FileWriter {
field_id_to_column_indices: Vec<(u32, u32)>,
num_columns: u32,
rows_written: u64,
+ // The number of rows written for each top-level field (i.e. each entry in
+ // `column_writers`). With `write_batch` every field advances together and
+ // these are all equal, but `write_column` advances one field at a time, so
+ // a single file may end up with columns of differing item counts.
+ field_rows_written: Vec,
global_buffers: Vec<(u64, u64)>,
schema_metadata: HashMap,
options: FileWriterOptions,
@@ -277,6 +282,7 @@ impl FileWriter {
column_metadata: Vec::new(),
num_columns: 0,
rows_written: 0,
+ field_rows_written: Vec::new(),
field_id_to_column_indices: Vec::new(),
global_buffers: Vec::new(),
schema_metadata: HashMap::new(),
@@ -467,6 +473,7 @@ impl FileWriter {
BatchEncoder::try_new(&schema, encoding_strategy.as_ref(), &encoding_options)?;
self.num_columns = encoder.num_columns();
+ self.field_rows_written = vec![0; encoder.field_encoders.len()];
self.column_writers = encoder.field_encoders;
self.column_metadata = vec![initial_column_metadata(); self.num_columns as usize];
self.field_id_to_column_indices = encoder.field_id_to_column_index;
@@ -490,13 +497,14 @@ impl FileWriter {
batch: &RecordBatch,
external_buffers: &mut OutOfLineBuffers,
) -> Result>> {
- self.schema
+ let items = self
+ .schema
.as_ref()
.unwrap()
.fields
.iter()
- .zip(self.column_writers.iter_mut())
- .map(|(field, column_writer)| {
+ .enumerate()
+ .map(|(field_idx, field)| {
let array =
batch
.column_by_name(&field.name)
@@ -507,19 +515,53 @@ impl FileWriter {
)
.into(),
))?;
+ Ok((field_idx, array.clone()))
+ })
+ .collect::>>()?;
+ self.encode_columns(&items, external_buffers)
+ }
+
+ // Encode a set of `(field index, array)` pairs, each advancing only its own
+ // column. The returned tasks must be written before the per-field row
+ // counters are advanced (see `advance_columns`).
+ fn encode_columns(
+ &mut self,
+ items: &[(usize, ArrayRef)],
+ external_buffers: &mut OutOfLineBuffers,
+ ) -> Result>> {
+ // Snapshot the starting row number of each field before borrowing the
+ // column writers mutably below.
+ let row_numbers = items
+ .iter()
+ .map(|(field_idx, _)| self.field_rows_written[*field_idx])
+ .collect::>();
+ items
+ .iter()
+ .zip(row_numbers)
+ .map(|((field_idx, array), row_number)| {
let repdef = RepDefBuilder::default();
let num_rows = array.len() as u64;
- column_writer.maybe_encode(
+ self.column_writers[*field_idx].maybe_encode(
array.clone(),
external_buffers,
repdef,
- self.rows_written,
+ row_number,
num_rows,
)
})
.collect::>>()
}
+ // Advance the per-field row counters after a set of columns has been
+ // written, keeping `rows_written` (the file's logical length) in sync as the
+ // maximum column length.
+ fn advance_columns(&mut self, items: &[(usize, ArrayRef)]) {
+ for (field_idx, array) in items {
+ self.field_rows_written[*field_idx] += array.len() as u64;
+ }
+ self.rows_written = self.field_rows_written.iter().copied().max().unwrap_or(0);
+ }
+
/// Schedule a batch of data to be written to the file
///
/// Note: the future returned by this method may complete before the data has been fully
@@ -557,18 +599,93 @@ impl FileWriter {
.flatten()
.collect::>();
- self.rows_written = match self.rows_written.checked_add(batch.num_rows() as u64) {
- Some(rows_written) => rows_written,
- None => {
- return Err(Error::invalid_input_source(format!("cannot write batch with {} rows because {} rows have already been written and Lance files cannot contain more than 2^64 rows", num_rows, self.rows_written).into()));
- }
- };
+ // `write_batch` advances every field by the same amount, keeping all
+ // columns equal length. Guard against overflowing the row counter.
+ if self.rows_written.checked_add(num_rows).is_none() {
+ return Err(Error::invalid_input_source(format!("cannot write batch with {} rows because {} rows have already been written and Lance files cannot contain more than 2^64 rows", num_rows, self.rows_written).into()));
+ }
+ for field_rows in self.field_rows_written.iter_mut() {
+ *field_rows += num_rows;
+ }
+ self.rows_written = self.field_rows_written.iter().copied().max().unwrap_or(0);
self.write_pages(encoding_tasks).await?;
Ok(())
}
+ /// Write a single column, advancing only that column's row counter.
+ ///
+ /// Unlike [`write_batch`](Self::write_batch), which advances every column
+ /// from a single shared row counter, this method advances one column
+ /// independently. Used across calls it produces a single file whose columns
+ /// may have different item counts — the physical layout used by sparse data
+ /// overlay files, where each field covers a different set of rows.
+ ///
+ /// `column_index` refers to a top-level field in the writer's schema (the
+ /// same order as the schema's fields). A column may be written across
+ /// multiple calls; its values are appended. A field that is never written
+ /// ends up as a zero-length column. The writer must have been created with
+ /// an explicit schema (via [`try_new`](Self::try_new)); a lazy schema cannot
+ /// be inferred here because individual calls need not cover every field.
+ ///
+ /// ```
+ /// # use arrow_array::{ArrayRef, Int32Array};
+ /// # use std::sync::Arc;
+ /// # use lance_file::writer::FileWriter;
+ /// # async fn example(writer: &mut FileWriter) -> lance_core::Result<()> {
+ /// // Field 0 gets three values, field 1 gets one — a non-rectangular file.
+ /// writer.write_column(0, Arc::new(Int32Array::from(vec![1, 2, 3]))).await?;
+ /// writer.write_column(1, Arc::new(Int32Array::from(vec![10]))).await?;
+ /// # Ok(())
+ /// # }
+ /// ```
+ pub async fn write_column(&mut self, column_index: usize, array: ArrayRef) -> Result<()> {
+ let schema = self.schema.as_ref().ok_or_else(|| {
+ Error::invalid_input_source(
+ "write_column requires the writer to be created with an explicit schema".into(),
+ )
+ })?;
+ let field = schema.fields.get(column_index).ok_or_else(|| {
+ Error::invalid_input_source(
+ format!(
+ "write_column: field index {} is out of bounds (schema has {} fields)",
+ column_index,
+ schema.fields.len()
+ )
+ .into(),
+ )
+ })?;
+ if array.len() as u64 > u32::MAX as u64 {
+ return Err(Error::invalid_input_source(
+ "cannot write Lance files with more than 2^32 rows".into(),
+ ));
+ }
+ Self::verify_field_nullability(&array.to_data(), field)?;
+
+ // A never-advanced field simply remains a zero-length column, which the
+ // encoders handle at `finish` time.
+ if array.is_empty() {
+ return Ok(());
+ }
+
+ let columns = [(column_index, array)];
+ let mut external_buffers =
+ OutOfLineBuffers::new(self.tell().await?, PAGE_BUFFER_ALIGNMENT as u64);
+ let encoding_tasks = self.encode_columns(&columns, &mut external_buffers)?;
+ for external_buffer in external_buffers.take_buffers() {
+ Self::do_write_buffer(&mut self.writer, &external_buffer).await?;
+ }
+ let encoding_tasks = encoding_tasks
+ .into_iter()
+ .flatten()
+ .collect::>();
+
+ self.advance_columns(&columns);
+ self.write_pages(encoding_tasks).await?;
+ Ok(())
+ }
+
async fn write_column_metadata(
&mut self,
metadata: pbfile::ColumnMetadata,
@@ -974,11 +1091,11 @@ mod tests {
use std::collections::HashMap;
use std::sync::Arc;
- use crate::reader::{FileReader, FileReaderOptions, describe_encoding};
+ use crate::reader::{FileReader, FileReaderOptions, ReaderProjection, describe_encoding};
use crate::testing::FsFixture;
use crate::writer::{ENV_LANCE_FILE_WRITER_MAX_PAGE_BYTES, FileWriter, FileWriterOptions};
use arrow_array::builder::{Float32Builder, Int32Builder};
- use arrow_array::{Int32Array, RecordBatch, UInt64Array};
+ use arrow_array::{ArrayRef, Int32Array, RecordBatch, UInt64Array};
use arrow_array::{RecordBatchReader, StringArray, types::Float64Type};
use arrow_schema::{DataType, Field, Field as ArrowField, Schema, Schema as ArrowSchema};
use lance_core::cache::LanceCache;
@@ -990,6 +1107,7 @@ mod tests {
use lance_encoding::version::LanceFileVersion;
use lance_io::object_store::ObjectStore;
use lance_io::utils::CachedFileSize;
+ use rstest::rstest;
#[tokio::test]
async fn test_basic_write() {
@@ -1040,6 +1158,326 @@ mod tests {
file_writer.finish().await.unwrap();
}
+ // Read a single column back at an explicit range/index set, returning its
+ // `Int32` values. Reading one column (or an equal-length group) at a time is
+ // how unequal-length files are consumed: a full scan across columns of
+ // differing lengths cannot form a single rectangular batch.
+ async fn read_int32_column(
+ reader: &FileReader,
+ schema: &LanceSchema,
+ version: LanceFileVersion,
+ name: &str,
+ params: lance_io::ReadBatchParams,
+ ) -> Vec> {
+ use futures::TryStreamExt;
+ use lance_encoding::decoder::FilterExpression;
+
+ let projection = ReaderProjection::from_column_names(version, schema, &[name]).unwrap();
+ let batches: Vec = reader
+ .read_stream_projected(params, 1024, 16, projection, FilterExpression::no_filter())
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ batches
+ .iter()
+ .flat_map(|b| {
+ b.column(0)
+ .as_any()
+ .downcast_ref::()
+ .unwrap()
+ .iter()
+ .collect::>()
+ })
+ .collect()
+ }
+
+ /// A single file may hold columns of differing item counts (no shared global
+ /// row counter). This is the physical layout used by sparse data overlay
+ /// files, where each field covers a different set of rows.
+ #[rstest]
+ #[tokio::test]
+ async fn test_write_columns_unequal_lengths(
+ #[values(LanceFileVersion::V2_0, LanceFileVersion::V2_1)] version: LanceFileVersion,
+ ) {
+ use lance_io::ReadBatchParams;
+
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("a", DataType::Int32, true),
+ ArrowField::new("b", DataType::Int32, true),
+ ArrowField::new("c", DataType::Int32, true),
+ ]));
+ let lance_schema = LanceSchema::try_from(arrow_schema.as_ref()).unwrap();
+
+ let fs = FsFixture::default();
+ let options = FileWriterOptions {
+ format_version: Some(version),
+ ..Default::default()
+ };
+ let mut writer = FileWriter::try_new(
+ fs.object_store.create(&fs.tmp_path).await.unwrap(),
+ lance_schema.clone(),
+ options,
+ )
+ .unwrap();
+
+ // Field "a" gets 5 values across two calls (appending), field "b" gets a
+ // single value, and field "c" is never written (a zero-length column).
+ let a1: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3]));
+ let b: ArrayRef = Arc::new(Int32Array::from(vec![10]));
+ writer.write_column(0, a1).await.unwrap();
+ writer.write_column(1, b).await.unwrap();
+ let a2: ArrayRef = Arc::new(Int32Array::from(vec![4, 5]));
+ writer.write_column(0, a2).await.unwrap();
+ // A zero-length array for an otherwise-unwritten field is a no-op.
+ let c_empty: ArrayRef = Arc::new(Int32Array::from(Vec::::new()));
+ writer.write_column(2, c_empty).await.unwrap();
+
+ let summary = writer.finish().await.unwrap();
+ // The file's logical length is the longest column.
+ assert_eq!(summary.num_rows, 5);
+
+ let file_scheduler = fs
+ .scheduler
+ .open_file(&fs.tmp_path, &CachedFileSize::unknown())
+ .await
+ .unwrap();
+ let reader = FileReader::try_open(
+ file_scheduler,
+ None,
+ Arc::::default(),
+ &LanceCache::no_cache(),
+ FileReaderOptions::default(),
+ )
+ .await
+ .unwrap();
+
+ // Per-column row counts are recorded in / derivable from file metadata.
+ assert_eq!(reader.num_rows(), 5);
+ assert_eq!(reader.column_num_rows(0), Some(5));
+ assert_eq!(reader.column_num_rows(1), Some(1));
+ assert_eq!(reader.column_num_rows(2), Some(0));
+ assert_eq!(reader.column_num_rows(3), None);
+
+ // Each column reads back independently at its own length.
+ assert_eq!(
+ read_int32_column(
+ &reader,
+ &lance_schema,
+ version,
+ "a",
+ ReadBatchParams::Range(0..5)
+ )
+ .await,
+ vec![Some(1), Some(2), Some(3), Some(4), Some(5)],
+ );
+ assert_eq!(
+ read_int32_column(
+ &reader,
+ &lance_schema,
+ version,
+ "b",
+ ReadBatchParams::Range(0..1)
+ )
+ .await,
+ vec![Some(10)],
+ );
+
+ // Random access by position within the longer column returns the right
+ // value even though other columns are shorter. (The take path requires
+ // strictly increasing indices.)
+ assert_eq!(
+ read_int32_column(
+ &reader,
+ &lance_schema,
+ version,
+ "a",
+ ReadBatchParams::Indices(arrow_array::UInt32Array::from(vec![0, 2, 4])),
+ )
+ .await,
+ vec![Some(1), Some(3), Some(5)],
+ );
+ }
+
+ /// Reading an unequal-length file:
+ /// - a projection whose columns are equal length full-scans normally;
+ /// - a full scan across columns of differing length is rejected up front,
+ /// before any batch is produced (even though a prefix would be rectangular);
+ /// - a bounded read is valid as long as every projected column covers it;
+ /// - a single-column `RangeFull` resolves to that column's own length, not
+ /// the file's (maximum) length.
+ #[rstest]
+ #[tokio::test]
+ async fn test_read_unequal_length_projection(
+ #[values(LanceFileVersion::V2_0, LanceFileVersion::V2_1)] version: LanceFileVersion,
+ ) {
+ use futures::TryStreamExt;
+ use lance_encoding::decoder::FilterExpression;
+ use lance_io::ReadBatchParams;
+
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("a", DataType::Int32, true),
+ ArrowField::new("b", DataType::Int32, true),
+ ArrowField::new("c", DataType::Int32, true),
+ ]));
+ let lance_schema = LanceSchema::try_from(arrow_schema.as_ref()).unwrap();
+ let fs = FsFixture::default();
+ let options = FileWriterOptions {
+ format_version: Some(version),
+ ..Default::default()
+ };
+ let mut writer = FileWriter::try_new(
+ fs.object_store.create(&fs.tmp_path).await.unwrap(),
+ lance_schema.clone(),
+ options,
+ )
+ .unwrap();
+ // "a" and "b" are equal length (5); "c" is shorter (1).
+ writer
+ .write_column(0, Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5])))
+ .await
+ .unwrap();
+ writer
+ .write_column(1, Arc::new(Int32Array::from(vec![6, 7, 8, 9, 10])))
+ .await
+ .unwrap();
+ writer
+ .write_column(2, Arc::new(Int32Array::from(vec![100])))
+ .await
+ .unwrap();
+ writer.finish().await.unwrap();
+
+ let file_scheduler = fs
+ .scheduler
+ .open_file(&fs.tmp_path, &CachedFileSize::unknown())
+ .await
+ .unwrap();
+ let reader = FileReader::try_open(
+ file_scheduler,
+ None,
+ Arc::::default(),
+ &LanceCache::no_cache(),
+ FileReaderOptions::default(),
+ )
+ .await
+ .unwrap();
+
+ let read = |names: &'static [&'static str], params: ReadBatchParams| {
+ let projection =
+ ReaderProjection::from_column_names(version, &lance_schema, names).unwrap();
+ async {
+ match reader
+ .read_stream_projected(
+ params,
+ 1024,
+ 16,
+ projection,
+ FilterExpression::no_filter(),
+ )
+ .await
+ {
+ Ok(stream) => stream.try_collect::>().await,
+ Err(e) => Err(e),
+ }
+ }
+ };
+ let col_values = |batches: &[RecordBatch], idx: usize| -> Vec> {
+ batches
+ .iter()
+ .flat_map(|b| {
+ b.column(idx)
+ .as_any()
+ .downcast_ref::()
+ .unwrap()
+ .iter()
+ .collect::>()
+ })
+ .collect()
+ };
+
+ // Equal-length projection [a, b] full-scans into rectangular batches.
+ let batches = read(&["a", "b"], ReadBatchParams::RangeFull).await.unwrap();
+ assert_eq!(
+ col_values(&batches, 0),
+ vec![Some(1), Some(2), Some(3), Some(4), Some(5)]
+ );
+ assert_eq!(
+ col_values(&batches, 1),
+ vec![Some(6), Some(7), Some(8), Some(9), Some(10)]
+ );
+
+ // A mismatched-length projection [a, c] (5 vs 1) is rejected before any
+ // batch is yielded, regardless of the read params — its columns cannot
+ // be combined into rectangular batches.
+ assert!(
+ read(&["a", "c"], ReadBatchParams::RangeFull).await.is_err(),
+ "full scan across unequal-length columns must error"
+ );
+ assert!(
+ read(&["a", "c"], ReadBatchParams::Range(0..1))
+ .await
+ .is_err(),
+ "even a common-prefix read of unequal-length columns must error"
+ );
+
+ // A single-column RangeFull resolves to that column's own length.
+ let batches = read(&["c"], ReadBatchParams::RangeFull).await.unwrap();
+ assert_eq!(col_values(&batches, 0), vec![Some(100)]);
+ let batches = read(&["a"], ReadBatchParams::RangeFull).await.unwrap();
+ assert_eq!(
+ col_values(&batches, 0),
+ vec![Some(1), Some(2), Some(3), Some(4), Some(5)]
+ );
+ }
+
+ /// Files written the ordinary (rectangular) way keep equal column lengths,
+ /// so the unequal-length support is backwards compatible.
+ #[tokio::test]
+ async fn test_write_batch_keeps_equal_lengths() {
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("a", DataType::Int32, true),
+ ArrowField::new("b", DataType::Int32, true),
+ ]));
+ let lance_schema = LanceSchema::try_from(arrow_schema.as_ref()).unwrap();
+
+ let fs = FsFixture::default();
+ let mut writer = FileWriter::try_new(
+ fs.object_store.create(&fs.tmp_path).await.unwrap(),
+ lance_schema,
+ FileWriterOptions::default(),
+ )
+ .unwrap();
+ let batch = RecordBatch::try_new(
+ arrow_schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3])),
+ Arc::new(Int32Array::from(vec![4, 5, 6])),
+ ],
+ )
+ .unwrap();
+ writer.write_batch(&batch).await.unwrap();
+ let summary = writer.finish().await.unwrap();
+ assert_eq!(summary.num_rows, 3);
+
+ let file_scheduler = fs
+ .scheduler
+ .open_file(&fs.tmp_path, &CachedFileSize::unknown())
+ .await
+ .unwrap();
+ let reader = FileReader::try_open(
+ file_scheduler,
+ None,
+ Arc::::default(),
+ &LanceCache::no_cache(),
+ FileReaderOptions::default(),
+ )
+ .await
+ .unwrap();
+ assert_eq!(reader.column_num_rows(0), Some(3));
+ assert_eq!(reader.column_num_rows(1), Some(3));
+ }
+
#[tokio::test]
async fn test_max_page_bytes_enforced() {
let arrow_field = Field::new("data", DataType::UInt64, false);
diff --git a/rust/lance-table/benches/manifest_intern.rs b/rust/lance-table/benches/manifest_intern.rs
index 78b7e352207..81bd57c1a22 100644
--- a/rust/lance-table/benches/manifest_intern.rs
+++ b/rust/lance-table/benches/manifest_intern.rs
@@ -59,6 +59,7 @@ fn make_uniform_pb_fragments(n: u64, num_fields: usize) -> Vec
file_size_bytes: 0,
base_id: None,
}],
+ overlays: vec![],
deletion_file: None,
row_id_sequence: None,
physical_rows: 1000,
@@ -135,6 +136,7 @@ fn make_diverse_pb_fragments(
file_size_bytes: 0,
base_id: None,
}],
+ overlays: vec![],
deletion_file: None,
row_id_sequence: None,
physical_rows: 1000,
diff --git a/rust/lance-table/src/feature_flags.rs b/rust/lance-table/src/feature_flags.rs
index 096f0da79e5..369c880c062 100644
--- a/rust/lance-table/src/feature_flags.rs
+++ b/rust/lance-table/src/feature_flags.rs
@@ -20,8 +20,22 @@ pub const FLAG_TABLE_CONFIG: u64 = 8;
pub const FLAG_BASE_PATHS: u64 = 16;
/// Disable writing transaction file under _transaction/, this flag is set when we only want to write inline transaction in manifest
pub const FLAG_DISABLE_TRANSACTION_FILE: u64 = 32;
+/// Fragments contain data overlay files, which supply new values for a subset of
+/// cells without rewriting base data files. A reader that does not understand
+/// overlays must refuse the dataset, since ignoring an overlay would silently
+/// return stale base values.
+///
+/// Data overlay files are not yet a released feature: in release builds this flag
+/// is treated as unknown (so a release reader/writer refuses an overlay dataset)
+/// unless [`ENABLE_DATA_OVERLAY_FILES_ENV`] is set, which lets benchmarks opt in.
+/// Debug builds always understand it so tests exercise the path.
+pub const FLAG_DATA_OVERLAY_FILES: u64 = 64;
/// The first bit that is unknown as a feature flag
-pub const FLAG_UNKNOWN: u64 = 64;
+pub const FLAG_UNKNOWN: u64 = 128;
+
+/// Environment variable that opts a release build into reading and writing data
+/// overlay files before the feature is generally released.
+pub const ENABLE_DATA_OVERLAY_FILES_ENV: &str = "LANCE_ENABLE_DATA_OVERLAY_FILES";
/// Set the reader and writer feature flags in the manifest based on the contents of the manifest.
pub fn apply_feature_flags(
@@ -71,18 +85,51 @@ pub fn apply_feature_flags(
manifest.writer_feature_flags |= FLAG_BASE_PATHS;
}
+ // Overlay files change cell values on read, so a reader that ignores them
+ // would return stale base values. Both readers and writers must understand
+ // them.
+ let has_overlays = manifest
+ .fragments
+ .iter()
+ .any(|frag| !frag.overlays.is_empty());
+ if has_overlays {
+ manifest.reader_feature_flags |= FLAG_DATA_OVERLAY_FILES;
+ manifest.writer_feature_flags |= FLAG_DATA_OVERLAY_FILES;
+ }
+
if disable_transaction_file {
manifest.writer_feature_flags |= FLAG_DISABLE_TRANSACTION_FILE;
}
Ok(())
}
+/// Whether this build understands data overlay files: always in debug builds,
+/// and in release builds only when [`ENABLE_DATA_OVERLAY_FILES_ENV`] is set.
+fn data_overlay_files_enabled() -> bool {
+ cfg!(debug_assertions) || std::env::var_os(ENABLE_DATA_OVERLAY_FILES_ENV).is_some()
+}
+
+/// The feature-flag bits this build understands, given whether overlay support
+/// is enabled. Split out from [`supported_flags`] so the policy is testable
+/// without toggling the build profile or environment.
+fn supported_flags_when(overlay_enabled: bool) -> u64 {
+ let mut supported = FLAG_UNKNOWN - 1;
+ if !overlay_enabled {
+ supported &= !FLAG_DATA_OVERLAY_FILES;
+ }
+ supported
+}
+
+fn supported_flags() -> u64 {
+ supported_flags_when(data_overlay_files_enabled())
+}
+
pub fn can_read_dataset(reader_flags: u64) -> bool {
- reader_flags < FLAG_UNKNOWN
+ reader_flags & !supported_flags() == 0
}
pub fn can_write_dataset(writer_flags: u64) -> bool {
- writer_flags < FLAG_UNKNOWN
+ writer_flags & !supported_flags() == 0
}
pub fn has_deprecated_v2_feature_flag(writer_flags: u64) -> bool {
@@ -103,6 +150,7 @@ mod tests {
assert!(can_read_dataset(super::FLAG_TABLE_CONFIG));
assert!(can_read_dataset(super::FLAG_BASE_PATHS));
assert!(can_read_dataset(super::FLAG_DISABLE_TRANSACTION_FILE));
+ assert!(can_read_dataset(super::FLAG_DATA_OVERLAY_FILES));
assert!(can_read_dataset(
super::FLAG_DELETION_FILES
| super::FLAG_STABLE_ROW_IDS
@@ -111,6 +159,53 @@ mod tests {
assert!(!can_read_dataset(super::FLAG_UNKNOWN));
}
+ #[test]
+ fn test_data_overlay_flag_release_gating() {
+ // Release default (overlays disabled): the overlay flag is treated as
+ // unknown so the dataset is refused, while other known flags still pass.
+ let supported = supported_flags_when(false);
+ assert_eq!(supported & FLAG_DATA_OVERLAY_FILES, 0);
+ assert_eq!(FLAG_DELETION_FILES & !supported, 0);
+ assert_ne!(FLAG_DATA_OVERLAY_FILES & !supported, 0);
+ // Enabled (debug or env opt-in): the overlay flag is understood.
+ let supported = supported_flags_when(true);
+ assert_eq!(FLAG_DATA_OVERLAY_FILES & !supported, 0);
+ }
+
+ #[test]
+ fn test_apply_feature_flags_sets_overlay_flag() {
+ use crate::format::{
+ DataFile, DataOverlayFile, DataStorageFormat, Fragment, OverlayCoverage,
+ };
+ use arrow_schema::{Field as ArrowField, Schema as ArrowSchema};
+ use lance_core::datatypes::Schema;
+ use roaring::RoaringBitmap;
+ use std::collections::HashMap;
+ use std::sync::Arc;
+
+ let arrow_schema = ArrowSchema::new(vec![ArrowField::new(
+ "id",
+ arrow_schema::DataType::Int64,
+ false,
+ )]);
+ let schema = Schema::try_from(&arrow_schema).unwrap();
+ let mut fragment = Fragment::new(0);
+ fragment.overlays = vec![DataOverlayFile {
+ data_file: DataFile::new_legacy_from_fields("o.lance", vec![0], None),
+ coverage: OverlayCoverage::dense(RoaringBitmap::from_iter([0u32])),
+ committed_version: 1,
+ }];
+ let mut manifest = Manifest::new(
+ schema,
+ Arc::new(vec![fragment]),
+ DataStorageFormat::default(),
+ HashMap::new(),
+ );
+ apply_feature_flags(&mut manifest, false, false).unwrap();
+ assert_ne!(manifest.reader_feature_flags & FLAG_DATA_OVERLAY_FILES, 0);
+ assert_ne!(manifest.writer_feature_flags & FLAG_DATA_OVERLAY_FILES, 0);
+ }
+
#[test]
fn test_write_check() {
assert!(can_write_dataset(0));
@@ -120,12 +215,14 @@ mod tests {
assert!(can_write_dataset(super::FLAG_TABLE_CONFIG));
assert!(can_write_dataset(super::FLAG_BASE_PATHS));
assert!(can_write_dataset(super::FLAG_DISABLE_TRANSACTION_FILE));
+ assert!(can_write_dataset(super::FLAG_DATA_OVERLAY_FILES));
assert!(can_write_dataset(
super::FLAG_DELETION_FILES
| super::FLAG_STABLE_ROW_IDS
| super::FLAG_USE_V2_FORMAT_DEPRECATED
| super::FLAG_TABLE_CONFIG
| super::FLAG_BASE_PATHS
+ | super::FLAG_DATA_OVERLAY_FILES
));
assert!(!can_write_dataset(super::FLAG_UNKNOWN));
}
diff --git a/rust/lance-table/src/format/fragment.rs b/rust/lance-table/src/format/fragment.rs
index 431e466dbd4..ef5166728ab 100644
--- a/rust/lance-table/src/format/fragment.rs
+++ b/rust/lance-table/src/format/fragment.rs
@@ -11,6 +11,7 @@ use lance_file::format::{MAJOR_VERSION, MINOR_VERSION};
use lance_file::version::LanceFileVersion;
use lance_io::utils::CachedFileSize;
use object_store::path::Path;
+use roaring::RoaringBitmap;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use crate::format::pb;
@@ -233,6 +234,210 @@ impl TryFrom for DataFile {
}
}
+/// Which `(physical offset, field)` cells a [`DataOverlayFile`] provides values
+/// for.
+///
+/// The coverage bitmaps index **physical** row offsets (positions in the base
+/// data files, counting deleted rows), so they are stable across deletions, like
+/// deletion vectors. Bitmaps are parsed from their 32-bit Roaring encoding once
+/// when the fragment is loaded and held behind an `Arc` so cloning a fragment is
+/// cheap; use [`DataOverlayFile::coverage_for_field`] to obtain the one that
+/// applies to a given field.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(into = "OverlayCoverageBytes", try_from = "OverlayCoverageBytes")]
+pub enum OverlayCoverage {
+ /// A single bitmap that applies to every field in the overlay's
+ /// `data_file.fields` (a dense / rectangular overlay): every covered offset
+ /// has a value for every field.
+ Shared(Arc),
+ /// One bitmap per field, in the same order as the overlay's
+ /// `data_file.fields` (a sparse overlay): different fields may cover
+ /// different offset sets.
+ PerField(Vec>),
+}
+
+/// Serialized form of [`OverlayCoverage`] — each bitmap as its 32-bit Roaring
+/// byte encoding. The in-memory form parses these once at load.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+enum OverlayCoverageBytes {
+ Shared(Vec),
+ PerField(Vec>),
+}
+
+fn deserialize_roaring(bytes: &[u8]) -> Result {
+ RoaringBitmap::deserialize_from(bytes).map_err(|e| {
+ Error::invalid_input(format!(
+ "failed to deserialize overlay coverage bitmap: {e}"
+ ))
+ })
+}
+
+fn serialize_roaring(bitmap: &RoaringBitmap) -> Vec {
+ let mut bytes = Vec::with_capacity(bitmap.serialized_size());
+ // Writing to a Vec is infallible.
+ bitmap.serialize_into(&mut bytes).unwrap();
+ bytes
+}
+
+impl From for OverlayCoverageBytes {
+ fn from(coverage: OverlayCoverage) -> Self {
+ match coverage {
+ OverlayCoverage::Shared(bitmap) => Self::Shared(serialize_roaring(&bitmap)),
+ OverlayCoverage::PerField(bitmaps) => {
+ Self::PerField(bitmaps.iter().map(|b| serialize_roaring(b)).collect())
+ }
+ }
+ }
+}
+
+impl TryFrom for OverlayCoverage {
+ type Error = Error;
+
+ fn try_from(bytes: OverlayCoverageBytes) -> Result {
+ Ok(match bytes {
+ OverlayCoverageBytes::Shared(b) => Self::Shared(Arc::new(deserialize_roaring(&b)?)),
+ OverlayCoverageBytes::PerField(bs) => Self::PerField(
+ bs.iter()
+ .map(|b| deserialize_roaring(b).map(Arc::new))
+ .collect::>()?,
+ ),
+ })
+ }
+}
+
+impl DeepSizeOf for OverlayCoverage {
+ fn deep_size_of_children(&self, _context: &mut lance_core::deepsize::Context) -> usize {
+ // RoaringBitmap does not expose its allocation size; its serialized size
+ // is a cheap, close proxy for the heap it holds.
+ let bitmap_heap = |bitmap: &RoaringBitmap| {
+ std::mem::size_of::() + bitmap.serialized_size()
+ };
+ match self {
+ Self::Shared(bitmap) => bitmap_heap(bitmap),
+ Self::PerField(bitmaps) => {
+ bitmaps.capacity() * std::mem::size_of::>()
+ + bitmaps.iter().map(|b| bitmap_heap(b)).sum::()
+ }
+ }
+ }
+}
+
+impl OverlayCoverage {
+ /// Build a dense coverage from a single bitmap shared across every field.
+ pub fn dense(bitmap: RoaringBitmap) -> Self {
+ Self::Shared(Arc::new(bitmap))
+ }
+
+ /// Build a sparse coverage from one bitmap per field.
+ pub fn sparse(bitmaps: Vec) -> Self {
+ Self::PerField(bitmaps.into_iter().map(Arc::new).collect())
+ }
+}
+
+/// An overlay file supplies new values for a subset of `(physical offset, field)`
+/// cells within a fragment, without rewriting the fragment's base data files. See
+/// the Data Overlay Files specification for the full resolution, coverage, and
+/// versioning rules.
+///
+/// The overlay's `data_file` stores one value column per field in
+/// `data_file.fields`, with **no** row-offset key column. Within a value column,
+/// the position of a covered offset's value is the **rank** (0-based count of set
+/// bits below it) of that offset in the field's coverage bitmap.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, DeepSizeOf)]
+pub struct DataOverlayFile {
+ /// The data file storing the overlay's new cell values.
+ pub data_file: DataFile,
+ /// Which cells this overlay provides values for.
+ pub coverage: OverlayCoverage,
+ /// The dataset version at which this overlay became effective (the version of
+ /// the commit that introduced it, stamped at commit time and re-stamped on
+ /// retry). Higher wins when two overlays cover the same `(offset, field)`.
+ pub committed_version: u64,
+}
+
+impl DataOverlayFile {
+ /// The parsed coverage bitmap that applies to the field stored at
+ /// `field_pos` within `data_file.fields`.
+ ///
+ /// For a dense overlay the same shared bitmap is returned for every field;
+ /// for a sparse overlay the per-field bitmap at `field_pos` is returned. The
+ /// bitmap is already parsed, so this is a cheap `Arc` clone.
+ pub fn coverage_for_field(&self, field_pos: usize) -> Result> {
+ match &self.coverage {
+ OverlayCoverage::Shared(bitmap) => Ok(bitmap.clone()),
+ OverlayCoverage::PerField(bitmaps) => {
+ bitmaps.get(field_pos).cloned().ok_or_else(|| {
+ Error::invalid_input(format!(
+ "overlay field_coverage has {} bitmaps but field position {} was requested",
+ bitmaps.len(),
+ field_pos
+ ))
+ })
+ }
+ }
+ }
+}
+
+/// Overlays are stored newest-last: a later list position is newer, and ties in
+/// `committed_version` are broken by position. Loading stable-sorts by
+/// `committed_version` so resolution can rely on the ordering without
+/// re-checking; the stable sort preserves the position tiebreak for equal
+/// versions.
+fn sort_overlays_newest_last(overlays: &mut [DataOverlayFile]) {
+ overlays.sort_by_key(|overlay| overlay.committed_version);
+}
+
+impl From<&DataOverlayFile> for pb::DataOverlayFile {
+ fn from(overlay: &DataOverlayFile) -> Self {
+ let coverage = match &overlay.coverage {
+ OverlayCoverage::Shared(bitmap) => {
+ pb::data_overlay_file::Coverage::SharedOffsetBitmap(serialize_roaring(bitmap))
+ }
+ OverlayCoverage::PerField(bitmaps) => {
+ pb::data_overlay_file::Coverage::FieldCoverage(pb::FieldCoverage {
+ offset_bitmaps: bitmaps.iter().map(|b| serialize_roaring(b)).collect(),
+ })
+ }
+ };
+ Self {
+ data_file: Some(pb::DataFile::from(&overlay.data_file)),
+ coverage: Some(coverage),
+ committed_version: overlay.committed_version,
+ }
+ }
+}
+
+impl TryFrom for DataOverlayFile {
+ type Error = Error;
+
+ fn try_from(proto: pb::DataOverlayFile) -> Result {
+ let data_file = proto
+ .data_file
+ .ok_or_else(|| Error::invalid_input("DataOverlayFile is missing its data_file"))?;
+ let coverage = match proto.coverage {
+ Some(pb::data_overlay_file::Coverage::SharedOffsetBitmap(bytes)) => {
+ OverlayCoverage::Shared(Arc::new(deserialize_roaring(&bytes)?))
+ }
+ Some(pb::data_overlay_file::Coverage::FieldCoverage(fc)) => OverlayCoverage::PerField(
+ fc.offset_bitmaps
+ .iter()
+ .map(|b| deserialize_roaring(b).map(Arc::new))
+ .collect::>()?,
+ ),
+ None => {
+ return Err(Error::invalid_input(
+ "DataOverlayFile is missing its coverage",
+ ));
+ }
+ };
+ Ok(Self {
+ data_file: DataFile::try_from(data_file)?,
+ coverage,
+ committed_version: proto.committed_version,
+ })
+ }
+}
+
/// Interns repeated data so that fragments with identical content share a
/// single heap allocation via `Arc`.
///
@@ -375,6 +580,15 @@ impl DataFileFieldInterner {
.into_iter()
.map(|f| self.intern_data_file(f))
.collect::>()?,
+ overlays: {
+ let mut overlays = p
+ .overlays
+ .into_iter()
+ .map(DataOverlayFile::try_from)
+ .collect::>>()?;
+ sort_overlays_newest_last(&mut overlays);
+ overlays
+ },
deletion_file: p.deletion_file.map(DeletionFile::try_from).transpose()?,
row_id_meta: p.row_id_sequence.map(RowIdMeta::try_from).transpose()?,
physical_rows,
@@ -483,6 +697,12 @@ pub struct Fragment {
/// Files within the fragment.
pub files: Vec,
+ /// Overlay files supplying new values for a subset of cells without
+ /// rewriting the base data files. Order is significant: a later entry is
+ /// newer than an earlier one. See [`DataOverlayFile`] for resolution rules.
+ #[serde(default, skip_serializing_if = "Vec::is_empty")]
+ pub overlays: Vec,
+
/// Optional file with deleted local row offsets.
#[serde(skip_serializing_if = "Option::is_none")]
pub deletion_file: Option,
@@ -510,6 +730,7 @@ impl Fragment {
Self {
id,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
physical_rows: None,
@@ -549,6 +770,7 @@ impl Fragment {
Self {
id,
files: vec![DataFile::new_legacy(path, schema, None, None)],
+ overlays: vec![],
deletion_file: None,
physical_rows,
row_id_meta: None,
@@ -669,6 +891,15 @@ impl TryFrom for Fragment {
.into_iter()
.map(DataFile::try_from)
.collect::>()?,
+ overlays: {
+ let mut overlays = p
+ .overlays
+ .into_iter()
+ .map(DataOverlayFile::try_from)
+ .collect::>>()?;
+ sort_overlays_newest_last(&mut overlays);
+ overlays
+ },
deletion_file: p.deletion_file.map(DeletionFile::try_from).transpose()?,
row_id_meta: p.row_id_sequence.map(RowIdMeta::try_from).transpose()?,
physical_rows,
@@ -716,6 +947,7 @@ impl From<&Fragment> for pb::DataFragment {
Self {
id: f.id,
files: f.files.iter().map(pb::DataFile::from).collect(),
+ overlays: f.overlays.iter().map(pb::DataOverlayFile::from).collect(),
deletion_file,
row_id_sequence,
physical_rows: f.physical_rows.unwrap_or_default() as u64,
@@ -734,6 +966,161 @@ mod tests {
use object_store::path::Path;
use serde_json::{Value, json};
+ #[test]
+ fn test_data_overlay_roundtrip() {
+ // A fragment carrying a dense overlay round-trips through protobuf and
+ // back, and the parsed coverage bitmap is recovered per field.
+ let mut bitmap = RoaringBitmap::new();
+ bitmap.insert(1);
+ bitmap.insert(3);
+
+ let overlay = DataOverlayFile {
+ data_file: DataFile::new_legacy_from_fields("overlay-0.lance", vec![3], None),
+ coverage: OverlayCoverage::dense(bitmap.clone()),
+ committed_version: 7,
+ };
+ let mut fragment = Fragment::new(0);
+ fragment.files = vec![DataFile::new_legacy_from_fields(
+ "base.lance",
+ vec![1, 3],
+ None,
+ )];
+ fragment.overlays = vec![overlay];
+
+ let proto = pb::DataFragment::from(&fragment);
+ assert_eq!(proto.overlays.len(), 1);
+ let round_tripped = Fragment::try_from(proto).unwrap();
+ assert_eq!(round_tripped, fragment);
+
+ // Dense coverage applies to every field.
+ let recovered = round_tripped.overlays[0].coverage_for_field(0).unwrap();
+ assert_eq!(*recovered, bitmap);
+ assert_eq!(
+ *round_tripped.overlays[0].coverage_for_field(5).unwrap(),
+ bitmap
+ );
+ }
+
+ #[test]
+ fn test_data_overlay_sparse_per_field_coverage() {
+ // A sparse overlay carries one bitmap per field, recovered by position.
+ let name_coverage = RoaringBitmap::from_iter([2u32, 3]);
+ let embedding_coverage = RoaringBitmap::from_iter([1u32]);
+ let overlay = DataOverlayFile {
+ data_file: DataFile::new_legacy_from_fields("overlay-1.lance", vec![2, 4], None),
+ coverage: OverlayCoverage::sparse(vec![
+ name_coverage.clone(),
+ embedding_coverage.clone(),
+ ]),
+ committed_version: 3,
+ };
+ let mut fragment = Fragment::new(1);
+ fragment.overlays = vec![overlay];
+
+ let round_tripped = Fragment::try_from(pb::DataFragment::from(&fragment)).unwrap();
+ assert_eq!(
+ *round_tripped.overlays[0].coverage_for_field(0).unwrap(),
+ name_coverage
+ );
+ assert_eq!(
+ *round_tripped.overlays[0].coverage_for_field(1).unwrap(),
+ embedding_coverage
+ );
+ }
+
+ #[test]
+ fn test_data_overlay_missing_fields_error() {
+ // A DataOverlayFile proto missing its coverage or data_file is rejected.
+ let no_coverage = pb::DataOverlayFile {
+ data_file: Some(pb::DataFile::from(&DataFile::new_legacy_from_fields(
+ "overlay.lance",
+ vec![3],
+ None,
+ ))),
+ coverage: None,
+ committed_version: 1,
+ };
+ let err = DataOverlayFile::try_from(no_coverage).unwrap_err();
+ assert!(err.to_string().contains("missing its coverage"), "{err}");
+
+ let no_data_file = pb::DataOverlayFile {
+ data_file: None,
+ coverage: Some(pb::data_overlay_file::Coverage::SharedOffsetBitmap(
+ serialize_roaring(&RoaringBitmap::from_iter([0u32])),
+ )),
+ committed_version: 1,
+ };
+ let err = DataOverlayFile::try_from(no_data_file).unwrap_err();
+ assert!(err.to_string().contains("missing its data_file"), "{err}");
+ }
+
+ #[test]
+ fn test_overlays_sorted_newest_last_on_load() {
+ // Overlays load stable-sorted by committed_version (newest last), with
+ // list position preserved as the tiebreak for equal versions.
+ let mk = |version: u64, field: i32| DataOverlayFile {
+ data_file: DataFile::new_legacy_from_fields("o.lance", vec![field], None),
+ coverage: OverlayCoverage::dense(RoaringBitmap::from_iter([0u32])),
+ committed_version: version,
+ };
+ let mut fragment = Fragment::new(0);
+ // Written out of order: v5, v2, v2 (second), v3.
+ fragment.overlays = vec![mk(5, 1), mk(2, 2), mk(2, 3), mk(3, 4)];
+
+ let loaded = Fragment::try_from(pb::DataFragment::from(&fragment)).unwrap();
+ let versions: Vec = loaded
+ .overlays
+ .iter()
+ .map(|o| o.committed_version)
+ .collect();
+ assert_eq!(versions, vec![2, 2, 3, 5]);
+ // Stable: the two v2 overlays keep their original relative order (field 2
+ // before field 3).
+ assert_eq!(
+ loaded.overlays[0].data_file.fields.as_ref(),
+ [2i32].as_slice()
+ );
+ assert_eq!(
+ loaded.overlays[1].data_file.fields.as_ref(),
+ [3i32].as_slice()
+ );
+ }
+
+ #[test]
+ fn test_overlay_coverage_serde_json_roundtrip() {
+ // The custom serde impl round-trips through JSON for dense/sparse,
+ // including empty bitmaps and a zero-bitmap sparse coverage.
+ for coverage in [
+ OverlayCoverage::dense(RoaringBitmap::from_iter([1u32, 5, 100])),
+ OverlayCoverage::dense(RoaringBitmap::new()),
+ OverlayCoverage::sparse(vec![
+ RoaringBitmap::from_iter([2u32, 3]),
+ RoaringBitmap::new(),
+ ]),
+ OverlayCoverage::sparse(vec![]),
+ ] {
+ let json = serde_json::to_string(&coverage).unwrap();
+ let back: OverlayCoverage = serde_json::from_str(&json).unwrap();
+ assert_eq!(back, coverage);
+ }
+ }
+
+ #[test]
+ fn test_coverage_for_field_out_of_bounds() {
+ let overlay = DataOverlayFile {
+ data_file: DataFile::new_legacy_from_fields("o.lance", vec![2, 4], None),
+ coverage: OverlayCoverage::sparse(vec![
+ RoaringBitmap::from_iter([1u32]),
+ RoaringBitmap::from_iter([2u32]),
+ ]),
+ committed_version: 1,
+ };
+ assert!(overlay.coverage_for_field(0).is_ok());
+ assert!(overlay.coverage_for_field(1).is_ok());
+ let err = overlay.coverage_for_field(5).unwrap_err();
+ assert!(err.to_string().contains("field position"), "{err}");
+ }
+
#[test]
fn test_new_fragment() {
let path = "foobar.lance";
diff --git a/rust/lance-table/src/format/manifest.rs b/rust/lance-table/src/format/manifest.rs
index 9845061b7e4..cd0a403621f 100644
--- a/rust/lance-table/src/format/manifest.rs
+++ b/rust/lance-table/src/format/manifest.rs
@@ -1316,6 +1316,7 @@ mod tests {
vec![0, 1, 2],
None,
)],
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
physical_rows: None,
@@ -1328,6 +1329,7 @@ mod tests {
DataFile::new_legacy_from_fields("path2", vec![0, 1, 43], None),
DataFile::new_legacy_from_fields("path3", vec![2], None),
],
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
physical_rows: None,
diff --git a/rust/lance/src/dataset/files.rs b/rust/lance/src/dataset/files.rs
index 848add7e4a8..2214822ed90 100644
--- a/rust/lance/src/dataset/files.rs
+++ b/rust/lance/src/dataset/files.rs
@@ -1036,6 +1036,7 @@ mod tests {
// No base_id -> falls back to the dataset base_uri.
mk_file("c.lance", None),
],
+ overlays: vec![],
// Deletion files also carry a base_id when they originate from a
// shallow clone, and must resolve against base_paths too.
deletion_file: Some(DeletionFile {
diff --git a/rust/lance/src/dataset/fragment.rs b/rust/lance/src/dataset/fragment.rs
index eb165e5f612..58a3ce0a9b6 100644
--- a/rust/lance/src/dataset/fragment.rs
+++ b/rust/lance/src/dataset/fragment.rs
@@ -900,6 +900,16 @@ impl FileFragment {
projection: &Schema,
read_config: FragReadConfig,
) -> Result {
+ // Overlay files supply newer cell values that must be merged on read.
+ // Until the scan/take merge path lands (the rest of OSS-1322 / OSS-1324),
+ // reading a fragment that has overlays would silently return stale base
+ // values, so we refuse rather than serve incorrect data.
+ if !self.metadata.overlays.is_empty() {
+ return Err(Error::not_supported(
+ "reading fragments with data overlay files is not yet supported \
+ (overlay merge is in progress)",
+ ));
+ }
let open_files = self.open_readers(projection, &read_config);
let deletion_vec_load = self.get_deletion_vector();
diff --git a/rust/lance/src/dataset/optimize.rs b/rust/lance/src/dataset/optimize.rs
index 87dda8e7e57..25168b54780 100644
--- a/rust/lance/src/dataset/optimize.rs
+++ b/rust/lance/src/dataset/optimize.rs
@@ -2125,6 +2125,7 @@ mod tests {
let fragment = Fragment {
id: 0,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
physical_rows: Some(0),
diff --git a/rust/lance/src/dataset/schema_evolution.rs b/rust/lance/src/dataset/schema_evolution.rs
index 5ef35a33ab7..8c1595c4daa 100644
--- a/rust/lance/src/dataset/schema_evolution.rs
+++ b/rust/lance/src/dataset/schema_evolution.rs
@@ -1941,6 +1941,7 @@ mod test {
Ok(Some(Fragment {
files: vec![],
id: 0,
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
physical_rows: Some(50),
diff --git a/rust/lance/src/dataset/transaction.rs b/rust/lance/src/dataset/transaction.rs
index 4555cd7ee6c..33248c2b68d 100644
--- a/rust/lance/src/dataset/transaction.rs
+++ b/rust/lance/src/dataset/transaction.rs
@@ -31,8 +31,9 @@ use lance_table::feature_flags::{FLAG_STABLE_ROW_IDS, apply_feature_flags};
use lance_table::rowids::read_row_ids;
use lance_table::{
format::{
- BasePath, DataFile, DataStorageFormat, Fragment, IndexFile, IndexMetadata, Manifest,
- RowDatasetVersionMeta, RowDatasetVersionRun, RowDatasetVersionSequence, RowIdMeta, pb,
+ BasePath, DataFile, DataOverlayFile, DataStorageFormat, Fragment, IndexFile, IndexMetadata,
+ Manifest, RowDatasetVersionMeta, RowDatasetVersionRun, RowDatasetVersionSequence,
+ RowIdMeta, pb,
},
io::{
commit::CommitHandler,
@@ -258,6 +259,17 @@ pub struct Transaction {
#[derive(Debug, Clone, DeepSizeOf, PartialEq)]
pub struct DataReplacementGroup(pub u64, pub DataFile);
+/// Overlay files to append to a single fragment, in order (the last entry is
+/// newest). The overlays are appended to the fragment's existing `overlays`
+/// list rather than replacing it, so overlays written by concurrent commits are
+/// preserved. Each overlay's `committed_version` is stamped to the new dataset
+/// version at commit time (re-stamped on retry).
+#[derive(Debug, Clone, DeepSizeOf, PartialEq)]
+pub struct DataOverlayGroup {
+ pub fragment_id: u64,
+ pub overlays: Vec,
+}
+
/// An entry for a map update. If value is None, the key will be removed from the map.
#[derive(Debug, Clone, DeepSizeOf, PartialEq)]
pub struct UpdateMapEntry {
@@ -367,6 +379,11 @@ pub enum Operation {
DataReplacement {
replacements: Vec,
},
+ /// Attach overlay files to fragments, supplying new values for a subset of
+ /// `(row offset, field)` cells without rewriting the fragments' base data
+ /// files. See [`DataOverlayFile`] and the Data Overlay Files specification
+ /// for resolution, coverage, and versioning rules.
+ DataOverlay { groups: Vec },
/// Merge a new column in
/// 'fragments' is the final fragments include all data files, the new fragments must align with old ones at rows.
/// 'schema' is not forced to include existed columns, which means we could use Merge to drop column data
@@ -499,6 +516,7 @@ impl std::fmt::Display for Operation {
Self::Project { .. } => write!(f, "Project"),
Self::UpdateConfig { .. } => write!(f, "UpdateConfig"),
Self::DataReplacement { .. } => write!(f, "DataReplacement"),
+ Self::DataOverlay { .. } => write!(f, "DataOverlay"),
Self::Clone { .. } => write!(f, "Clone"),
Self::UpdateMemWalState { .. } => write!(f, "UpdateMemWalState"),
Self::UpdateBases { .. } => write!(f, "UpdateBases"),
@@ -1345,6 +1363,8 @@ impl PartialEq for Operation {
(Self::Clone { .. }, Self::UpdateBases { .. }) => {
std::mem::discriminant(self) == std::mem::discriminant(other)
}
+ (Self::DataOverlay { groups: a }, Self::DataOverlay { groups: b }) => compare_vec(a, b),
+ (Self::DataOverlay { .. }, _) | (_, Self::DataOverlay { .. }) => false,
}
}
}
@@ -1521,6 +1541,7 @@ impl Operation {
Self::Project { .. } => "Project",
Self::UpdateConfig { .. } => "UpdateConfig",
Self::DataReplacement { .. } => "DataReplacement",
+ Self::DataOverlay { .. } => "DataOverlay",
Self::UpdateMemWalState { .. } => "UpdateMemWalState",
Self::Clone { .. } => "Clone",
Self::UpdateBases { .. } => "UpdateBases",
@@ -2300,6 +2321,49 @@ impl Transaction {
&replaced_fields,
);
}
+ Operation::DataOverlay { groups } => {
+ // Stamp each overlay with the version this commit is producing.
+ // build_manifest re-runs on every retry with an updated
+ // current_manifest, so this is naturally re-stamped on retry.
+ let new_version = current_manifest.map_or(1, |m| m.version + 1);
+
+ let existing_fragments = maybe_existing_fragments?;
+ // Multiple groups may target the same fragment; merge them in
+ // order rather than letting a HashMap collapse drop all but the
+ // last group's overlays.
+ let mut overlays_by_fragment: HashMap> = HashMap::new();
+ for group in groups {
+ overlays_by_fragment
+ .entry(group.fragment_id)
+ .or_default()
+ .extend(group.overlays.iter());
+ }
+
+ // Every group must target an existing fragment.
+ for fragment_id in overlays_by_fragment.keys() {
+ if !existing_fragments.iter().any(|f| f.id == *fragment_id) {
+ return Err(Error::invalid_input(format!(
+ "DataOverlay targets fragment {fragment_id}, which does not exist"
+ )));
+ }
+ }
+
+ for fragment in existing_fragments {
+ let mut fragment = fragment.clone();
+ if let Some(new_overlays) = overlays_by_fragment.get(&fragment.id) {
+ // Appended (not replaced) so concurrently-written overlays
+ // survive; later entries are newer.
+ fragment
+ .overlays
+ .extend(new_overlays.iter().map(|&overlay| {
+ let mut overlay = overlay.clone();
+ overlay.committed_version = new_version;
+ overlay
+ }));
+ }
+ final_fragments.push(fragment);
+ }
+ }
Operation::UpdateMemWalState { merged_generations } => {
update_mem_wal_index_merged_generations(
&mut final_indices,
@@ -3007,6 +3071,34 @@ impl TryFrom for DataReplacementGroup {
}
}
+impl From<&DataOverlayGroup> for pb::transaction::DataOverlayGroup {
+ fn from(group: &DataOverlayGroup) -> Self {
+ Self {
+ fragment_id: group.fragment_id,
+ overlays: group
+ .overlays
+ .iter()
+ .map(pb::DataOverlayFile::from)
+ .collect(),
+ }
+ }
+}
+
+impl TryFrom for DataOverlayGroup {
+ type Error = Error;
+
+ fn try_from(message: pb::transaction::DataOverlayGroup) -> Result {
+ Ok(Self {
+ fragment_id: message.fragment_id,
+ overlays: message
+ .overlays
+ .into_iter()
+ .map(DataOverlayFile::try_from)
+ .collect::>>()?,
+ })
+ }
+}
+
impl TryFrom for Transaction {
type Error = Error;
@@ -3289,6 +3381,14 @@ impl TryFrom for Transaction {
})) => Operation::UpdateBases {
new_bases: new_bases.into_iter().map(BasePath::from).collect(),
},
+ Some(pb::transaction::Operation::DataOverlay(pb::transaction::DataOverlay {
+ groups,
+ })) => Operation::DataOverlay {
+ groups: groups
+ .into_iter()
+ .map(DataOverlayGroup::try_from)
+ .collect::>>()?,
+ },
None => {
return Err(Error::internal(
"Transaction message did not contain an operation".to_string(),
@@ -3559,6 +3659,14 @@ impl From<&Transaction> for pb::Transaction {
.collect(),
})
}
+ Operation::DataOverlay { groups } => {
+ pb::transaction::Operation::DataOverlay(pb::transaction::DataOverlay {
+ groups: groups
+ .iter()
+ .map(pb::transaction::DataOverlayGroup::from)
+ .collect(),
+ })
+ }
Operation::UpdateMemWalState { merged_generations } => {
pb::transaction::Operation::UpdateMemWalState(pb::transaction::UpdateMemWalState {
merged_generations: merged_generations
@@ -3838,7 +3946,8 @@ mod tests {
use lance_file::version::LanceFileVersion;
use lance_io::utils::CachedFileSize;
use lance_table::format::{
- RowDatasetVersionMeta, RowDatasetVersionRun, RowDatasetVersionSequence, RowIdMeta,
+ OverlayCoverage, RowDatasetVersionMeta, RowDatasetVersionRun, RowDatasetVersionSequence,
+ RowIdMeta,
};
use lance_table::rowids::segment::U64Segment;
use lance_table::rowids::write_row_ids;
@@ -4148,6 +4257,7 @@ mod tests {
physical_rows: Some(100),
row_id_meta: None,
files: vec![],
+ overlays: vec![],
deletion_file: None,
last_updated_at_version_meta: None,
created_at_version_meta: None,
@@ -4180,6 +4290,7 @@ mod tests {
physical_rows: Some(50),
row_id_meta: Some(RowIdMeta::Inline(serialized)),
files: vec![],
+ overlays: vec![],
deletion_file: None,
last_updated_at_version_meta: None,
created_at_version_meta: None,
@@ -4212,6 +4323,7 @@ mod tests {
physical_rows: Some(50), // More physical rows than existing row IDs
row_id_meta: Some(RowIdMeta::Inline(serialized)),
files: vec![],
+ overlays: vec![],
deletion_file: None,
last_updated_at_version_meta: None,
created_at_version_meta: None,
@@ -4247,6 +4359,7 @@ mod tests {
physical_rows: Some(50), // Less physical rows than existing row IDs
row_id_meta: Some(RowIdMeta::Inline(serialized)),
files: vec![],
+ overlays: vec![],
deletion_file: None,
last_updated_at_version_meta: None,
created_at_version_meta: None,
@@ -4275,6 +4388,7 @@ mod tests {
physical_rows: Some(30), // No existing row IDs
row_id_meta: None,
files: vec![],
+ overlays: vec![],
deletion_file: None,
last_updated_at_version_meta: None,
created_at_version_meta: None,
@@ -4284,6 +4398,7 @@ mod tests {
physical_rows: Some(25), // Partial existing row IDs
row_id_meta: Some(RowIdMeta::Inline(serialized)),
files: vec![],
+ overlays: vec![],
deletion_file: None,
last_updated_at_version_meta: None,
created_at_version_meta: None,
@@ -4328,6 +4443,7 @@ mod tests {
physical_rows: None,
row_id_meta: None,
files: vec![],
+ overlays: vec![],
deletion_file: None,
last_updated_at_version_meta: None,
created_at_version_meta: None,
@@ -4835,6 +4951,7 @@ mod tests {
let fragment = Fragment {
id: 1,
files: vec![data_file],
+ overlays: vec![],
deletion_file: None,
row_id_meta,
physical_rows: Some(5),
@@ -5104,6 +5221,7 @@ mod tests {
None,
)],
physical_rows: Some(10),
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
last_updated_at_version_meta: None,
@@ -5195,6 +5313,7 @@ mod tests {
let prev_fragment = Fragment {
id: 0,
files: vec![mk_file("before.lance")],
+ overlays: vec![],
deletion_file: None,
row_id_meta,
physical_rows: Some(5),
@@ -5267,6 +5386,7 @@ mod tests {
let prev_fragment = Fragment {
id: 0,
files: vec![data_file.clone()],
+ overlays: vec![],
deletion_file: None,
row_id_meta: row_id_meta.clone(),
physical_rows: Some(5),
@@ -5286,6 +5406,7 @@ mod tests {
let merged_fragment = Fragment {
id: 0,
files: vec![data_file],
+ overlays: vec![],
deletion_file: None,
row_id_meta,
physical_rows: Some(5),
@@ -5335,6 +5456,7 @@ mod tests {
let prev_fragment = Fragment {
id: 0,
files: vec![mk_file("before.lance")],
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
physical_rows: Some(5),
@@ -5399,6 +5521,7 @@ mod tests {
let existing_fragment = Fragment {
id: 0,
files: vec![mk_file("existing.lance")],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&row_ids_0))),
physical_rows: Some(3),
@@ -5421,6 +5544,7 @@ mod tests {
let new_fragment = Fragment {
id: 1,
files: vec![mk_file("new.lance")],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&row_ids_1))),
physical_rows: Some(4),
@@ -5483,6 +5607,7 @@ mod tests {
let existing_fragment = Fragment {
id: 1,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&existing_seq))),
physical_rows: Some(3),
@@ -5496,6 +5621,7 @@ mod tests {
let new_fragment = Fragment {
id: 10,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&new_seq))),
physical_rows: Some(2),
@@ -5538,6 +5664,7 @@ mod tests {
Fragment {
id: 1,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&frag_a_seq))),
physical_rows: Some(2),
@@ -5549,6 +5676,7 @@ mod tests {
Fragment {
id: 2,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&frag_b_seq))),
physical_rows: Some(3),
@@ -5564,6 +5692,7 @@ mod tests {
let new_fragment = Fragment {
id: 10,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&new_seq))),
physical_rows: Some(2),
@@ -5605,6 +5734,7 @@ mod tests {
let existing_fragment = Fragment {
id: 1,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&existing_seq))),
physical_rows: Some(2),
@@ -5619,6 +5749,7 @@ mod tests {
let new_fragment = Fragment {
id: 10,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&new_seq))),
physical_rows: Some(2),
@@ -5663,6 +5794,7 @@ mod tests {
let existing_fragment = Fragment {
id: 1,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&existing_seq))),
physical_rows: Some(2),
@@ -5676,6 +5808,7 @@ mod tests {
let new_fragment = Fragment {
id: 20,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&new_seq))),
physical_rows: Some(4),
@@ -5709,6 +5842,7 @@ mod tests {
let existing_fragment = Fragment {
id: 1,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&existing_seq))),
physical_rows: Some(2),
@@ -5720,6 +5854,7 @@ mod tests {
let new_fragment = Fragment {
id: 10,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&new_seq))),
physical_rows: Some(1),
@@ -5748,6 +5883,7 @@ mod tests {
let existing_fragment = Fragment {
id: 1,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&existing_seq))),
physical_rows: Some(2),
@@ -5758,6 +5894,7 @@ mod tests {
let new_fragment = Fragment {
id: 10,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
physical_rows: Some(3),
@@ -5788,6 +5925,7 @@ mod tests {
let existing_fragment = Fragment {
id: 1,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&existing_seq))),
physical_rows: Some(2),
@@ -5801,6 +5939,7 @@ mod tests {
let new_fragment = Fragment {
id: 10,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&new_seq))),
physical_rows: Some(1),
@@ -5842,6 +5981,7 @@ mod tests {
let in_range_frag = Fragment {
id: 1,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&in_range_seq))),
physical_rows: Some(2),
@@ -5862,6 +6002,7 @@ mod tests {
let out_of_range_frag = Fragment {
id: 2,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&out_of_range_seq))),
physical_rows: Some(2),
@@ -5876,6 +6017,7 @@ mod tests {
let new_frag = Fragment {
id: 10,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&new_seq))),
physical_rows: Some(2),
@@ -5914,6 +6056,7 @@ mod tests {
let existing = Fragment {
id: 1,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&seq))),
physical_rows: Some(3),
@@ -5926,6 +6069,7 @@ mod tests {
let new_frag = Fragment {
id: 10,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&new_seq))),
physical_rows: Some(2),
@@ -5974,6 +6118,7 @@ mod tests {
let src_frag = Fragment {
id: 1,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&src_seq))),
physical_rows: Some(100),
@@ -5988,6 +6133,7 @@ mod tests {
let new_frag = Fragment {
id: 10,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&new_seq))),
physical_rows: Some(100),
@@ -6035,6 +6181,7 @@ mod tests {
Fragment {
id: 1,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&seq_a))),
physical_rows: Some(3),
@@ -6046,6 +6193,7 @@ mod tests {
Fragment {
id: 2,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&seq_b))),
physical_rows: Some(3),
@@ -6061,6 +6209,7 @@ mod tests {
let new_frag = Fragment {
id: 10,
files: vec![],
+ overlays: vec![],
deletion_file: None,
row_id_meta: Some(RowIdMeta::Inline(write_row_ids(&new_seq))),
physical_rows: Some(2),
@@ -6127,4 +6276,181 @@ mod tests {
assert!(!left.modifies_same_metadata(&different_key));
assert!(left.modifies_same_metadata(&replace));
}
+
+ #[test]
+ fn test_data_overlay_operation_roundtrips() {
+ // A DataOverlay operation survives the protobuf round-trip, preserving
+ // the target fragment, the overlay's coverage, and its committed_version.
+ use lance_table::format::{DataOverlayFile, OverlayCoverage};
+
+ let mut bitmap = roaring::RoaringBitmap::new();
+ bitmap.insert(1);
+ bitmap.insert(4);
+ let overlay = DataOverlayFile {
+ data_file: DataFile::new_legacy_from_fields("overlay-0.lance", vec![3], None),
+ coverage: OverlayCoverage::dense(bitmap.clone()),
+ committed_version: 6,
+ };
+ let pb_overlay = pb::DataOverlayFile::from(&overlay);
+
+ let message = pb::Transaction {
+ read_version: 1,
+ uuid: Uuid::new_v4().to_string(),
+ operation: Some(pb::transaction::Operation::DataOverlay(
+ pb::transaction::DataOverlay {
+ groups: vec![pb::transaction::DataOverlayGroup {
+ fragment_id: 7,
+ overlays: vec![pb_overlay],
+ }],
+ },
+ )),
+ ..Default::default()
+ };
+
+ let txn = Transaction::try_from(message).unwrap();
+ match txn.operation {
+ Operation::DataOverlay { groups } => {
+ assert_eq!(groups.len(), 1);
+ assert_eq!(groups[0].fragment_id, 7);
+ assert_eq!(groups[0].overlays.len(), 1);
+ assert_eq!(groups[0].overlays[0].committed_version, 6);
+ assert_eq!(
+ *groups[0].overlays[0].coverage_for_field(0).unwrap(),
+ bitmap
+ );
+ }
+ other => panic!("expected DataOverlay, got {other:?}"),
+ }
+ }
+
+ fn overlay_with_field(field: i32, committed_version: u64) -> DataOverlayFile {
+ DataOverlayFile {
+ data_file: DataFile::new_legacy_from_fields("o.lance", vec![field], None),
+ coverage: OverlayCoverage::dense(roaring::RoaringBitmap::from_iter([0u32])),
+ committed_version,
+ }
+ }
+
+ #[test]
+ fn test_data_overlay_build_manifest_appends_and_stamps() {
+ // A fragment already carrying an overlay (committed at v3) gets a new
+ // overlay appended and stamped to the new dataset version; the existing
+ // overlay is preserved with its version.
+ let mut fragment = Fragment::new(0);
+ fragment.overlays = vec![overlay_with_field(1, 3)];
+ let schema = ArrowSchema::new(vec![ArrowField::new("id", DataType::Int32, false)]);
+ let manifest = Manifest::new(
+ LanceSchema::try_from(&schema).unwrap(),
+ Arc::new(vec![fragment]),
+ lance_table::format::DataStorageFormat::new(LanceFileVersion::V2_0),
+ HashMap::new(),
+ );
+
+ let txn = Transaction::new(
+ manifest.version,
+ Operation::DataOverlay {
+ groups: vec![DataOverlayGroup {
+ fragment_id: 0,
+ overlays: vec![overlay_with_field(2, 0)],
+ }],
+ },
+ None,
+ );
+
+ let (result, _) = txn
+ .build_manifest(
+ Some(&manifest),
+ vec![],
+ "txn",
+ &ManifestWriteConfig::default(),
+ )
+ .unwrap();
+
+ let frag = &result.fragments[0];
+ assert_eq!(frag.overlays.len(), 2);
+ assert_eq!(frag.overlays[0].committed_version, 3);
+ assert_eq!(frag.overlays[1].committed_version, result.version);
+ assert!(result.version > manifest.version);
+ }
+
+ #[test]
+ fn test_data_overlay_build_manifest_merges_duplicate_groups() {
+ // Two groups targeting the same fragment must both survive (a HashMap
+ // collapse would have dropped the first).
+ let manifest = sample_manifest();
+ let txn = Transaction::new(
+ manifest.version,
+ Operation::DataOverlay {
+ groups: vec![
+ DataOverlayGroup {
+ fragment_id: 0,
+ overlays: vec![overlay_with_field(1, 0)],
+ },
+ DataOverlayGroup {
+ fragment_id: 0,
+ overlays: vec![overlay_with_field(2, 0)],
+ },
+ ],
+ },
+ None,
+ );
+
+ let (result, _) = txn
+ .build_manifest(
+ Some(&manifest),
+ vec![],
+ "txn",
+ &ManifestWriteConfig::default(),
+ )
+ .unwrap();
+
+ let overlays = &result.fragments[0].overlays;
+ assert_eq!(overlays.len(), 2);
+ assert_eq!(overlays[0].data_file.fields.as_ref(), [1i32].as_slice());
+ assert_eq!(overlays[1].data_file.fields.as_ref(), [2i32].as_slice());
+ }
+
+ #[test]
+ fn test_data_overlay_build_manifest_rejects_unknown_fragment() {
+ let manifest = sample_manifest();
+ let txn = Transaction::new(
+ manifest.version,
+ Operation::DataOverlay {
+ groups: vec![DataOverlayGroup {
+ fragment_id: 99,
+ overlays: vec![overlay_with_field(1, 0)],
+ }],
+ },
+ None,
+ );
+ let err = txn
+ .build_manifest(
+ Some(&manifest),
+ vec![],
+ "txn",
+ &ManifestWriteConfig::default(),
+ )
+ .unwrap_err();
+ assert!(err.to_string().contains("does not exist"), "{err}");
+ }
+
+ #[test]
+ fn test_data_overlay_operation_eq() {
+ let overlay = |field: i32| Operation::DataOverlay {
+ groups: vec![DataOverlayGroup {
+ fragment_id: 0,
+ overlays: vec![overlay_with_field(field, 1)],
+ }],
+ };
+ // Reflexive and value-based (the arm previously returned false for self).
+ assert_eq!(overlay(1), overlay(1));
+ assert_ne!(overlay(1), overlay(2));
+ // Not equal to a different operation kind (previously returned true vs Rewrite).
+ let rewrite = Operation::Rewrite {
+ groups: vec![],
+ rewritten_indices: vec![],
+ frag_reuse_index: None,
+ };
+ assert_ne!(overlay(1), rewrite);
+ }
}
diff --git a/rust/lance/src/dataset/write.rs b/rust/lance/src/dataset/write.rs
index ff0a119158c..47c4add18bd 100644
--- a/rust/lance/src/dataset/write.rs
+++ b/rust/lance/src/dataset/write.rs
@@ -3562,6 +3562,7 @@ mod tests {
let fragments = vec![Fragment {
id: 0,
files: vec![external_file, local_file],
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
physical_rows: Some(0),
diff --git a/rust/lance/src/dataset/write/commit.rs b/rust/lance/src/dataset/write/commit.rs
index baad71b3e39..7aed72a2d71 100644
--- a/rust/lance/src/dataset/write/commit.rs
+++ b/rust/lance/src/dataset/write/commit.rs
@@ -551,6 +551,7 @@ mod tests {
file_size_bytes: CachedFileSize::new(100),
base_id: None,
}],
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
physical_rows: Some(10),
diff --git a/rust/lance/src/io/commit.rs b/rust/lance/src/io/commit.rs
index ce0d29d550b..8980dd4f1ed 100644
--- a/rust/lance/src/io/commit.rs
+++ b/rust/lance/src/io/commit.rs
@@ -1687,6 +1687,7 @@ mod tests {
DataFile::new_legacy_from_fields("path1", vec![0, 1, 2], None),
DataFile::new_legacy_from_fields("unused", vec![9], None),
],
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
physical_rows: None,
@@ -1699,6 +1700,7 @@ mod tests {
DataFile::new_legacy_from_fields("path2", vec![0, 1, 2], None),
DataFile::new_legacy_from_fields("path3", vec![2], None),
],
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
physical_rows: None,
@@ -1736,6 +1738,7 @@ mod tests {
vec![0, 1, 10],
None,
)],
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
physical_rows: None,
@@ -1748,6 +1751,7 @@ mod tests {
DataFile::new_legacy_from_fields("path2", vec![0, 1, 2], None),
DataFile::new_legacy_from_fields("path3", vec![10], None),
],
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
physical_rows: None,
@@ -1838,6 +1842,7 @@ mod tests {
let fragment = Fragment {
id: 0,
files: vec![data_file],
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
physical_rows: Some(100),
diff --git a/rust/lance/src/io/commit/conflict_resolver.rs b/rust/lance/src/io/commit/conflict_resolver.rs
index dc898534c89..1e46c4acffa 100644
--- a/rust/lance/src/io/commit/conflict_resolver.rs
+++ b/rust/lance/src/io/commit/conflict_resolver.rs
@@ -137,6 +137,21 @@ impl<'a> TransactionRebase<'a> {
conflicting_mem_wal_merged_gens: Vec::new(),
})
}
+ Operation::DataOverlay { groups } => {
+ let modified_fragment_ids =
+ groups.iter().map(|g| g.fragment_id).collect::>();
+ let initial_fragments =
+ initial_fragments_for_rebase(dataset, &transaction, &modified_fragment_ids)
+ .await;
+ Ok(Self {
+ transaction,
+ affected_rows,
+ initial_fragments,
+ modified_fragment_ids,
+ conflicting_frag_reuse_indices: Vec::new(),
+ conflicting_mem_wal_merged_gens: Vec::new(),
+ })
+ }
Operation::Merge { fragments, .. } => {
let modified_fragment_ids = fragments.iter().map(|f| f.id).collect::>();
let initial_fragments =
@@ -203,6 +218,9 @@ impl<'a> TransactionRebase<'a> {
Operation::DataReplacement { .. } => {
self.check_data_replacement_txn(other_transaction, other_version)
}
+ Operation::DataOverlay { .. } => {
+ self.check_data_overlay_txn(other_transaction, other_version)
+ }
Operation::Merge { .. } => self.check_merge_txn(other_transaction, other_version),
Operation::Restore { .. } => self.check_restore_txn(other_transaction, other_version),
Operation::ReserveFragments { .. } => {
@@ -235,6 +253,10 @@ impl<'a> TransactionRebase<'a> {
| Operation::Project { .. }
| Operation::Append { .. }
| Operation::UpdateConfig { .. }
+ // A concurrent overlay is inert against the rows we delete
+ // (deletions take precedence over overlays) and otherwise
+ // preserves physical offsets, so it never conflicts.
+ | Operation::DataOverlay { .. }
| Operation::UpdateBases { .. } => Ok(()),
Operation::Rewrite { groups, .. } => {
if groups
@@ -382,6 +404,9 @@ impl<'a> TransactionRebase<'a> {
| Operation::Project { .. }
| Operation::Clone { .. }
| Operation::UpdateConfig { .. }
+ // A concurrent overlay preserves physical offsets and is newer
+ // than this update, so it wins its covered cells without conflict.
+ | Operation::DataOverlay { .. }
| Operation::UpdateBases { .. } => Ok(()),
Operation::Append { .. } => {
// If current transaction has primary key conflict detection,
@@ -498,6 +523,10 @@ impl<'a> TransactionRebase<'a> {
match &other_transaction.operation {
Operation::Append { .. }
| Operation::Clone { .. }
+ // An overlay committed after this index's version is newer than
+ // the index; the query path excludes its covered cells via the
+ // version gate, so the build does not conflict.
+ | Operation::DataOverlay { .. }
| Operation::UpdateBases { .. } => Ok(()),
Operation::CreateIndex {
new_indices: created_indices,
@@ -679,6 +708,20 @@ impl<'a> TransactionRebase<'a> {
Ok(())
}
}
+ Operation::DataOverlay { groups } => {
+ // Rewriting a fragment changes its physical row addresses, so
+ // an overlay addressed by physical offset on that fragment is
+ // invalidated and must be re-applied against the new base.
+ if groups
+ .iter()
+ .map(|g| g.fragment_id)
+ .any(|id| self.modified_fragment_ids.contains(&id))
+ {
+ Err(self.retryable_conflict_err(other_transaction, other_version))
+ } else {
+ Ok(())
+ }
+ }
Operation::Rewrite {
groups,
frag_reuse_index: committed_fri,
@@ -858,6 +901,7 @@ impl<'a> TransactionRebase<'a> {
| Operation::CreateIndex { .. }
| Operation::Rewrite { .. }
| Operation::DataReplacement { .. }
+ | Operation::DataOverlay { .. }
| Operation::Merge { .. }
| Operation::Restore { .. }
| Operation::ReserveFragments { .. }
@@ -891,7 +935,8 @@ impl<'a> TransactionRebase<'a> {
| Operation::Merge { .. }
| Operation::UpdateConfig { .. }
| Operation::Clone { .. }
- | Operation::DataReplacement { .. } => Ok(()),
+ | Operation::DataReplacement { .. }
+ | Operation::DataOverlay { .. } => Ok(()),
}
}
@@ -907,6 +952,9 @@ impl<'a> TransactionRebase<'a> {
| Operation::UpdateConfig { .. }
| Operation::ReserveFragments { .. }
| Operation::Project { .. }
+ // Both a column replacement and an overlay preserve physical row
+ // addresses; the overlay is newer and wins its covered cells.
+ | Operation::DataOverlay { .. }
| Operation::UpdateBases { .. } => Ok(()),
Operation::Merge { .. } => {
// Merge rewrites the whole fragment list; always conflict
@@ -1009,6 +1057,78 @@ impl<'a> TransactionRebase<'a> {
}
}
+ /// Conflict checks for our DataOverlay transaction against a concurrent one.
+ ///
+ /// Overlays are intentionally permissive (see the Data Overlay Files spec):
+ /// they stack with other overlays and tolerate appends, deletes, column
+ /// rewrites, and index builds, because overlay coverage is addressed by
+ /// physical offset and the version gate keeps indexes correct. The only
+ /// concurrent operations that invalidate an overlay are those that rewrite
+ /// rows or consume the overlays on one of our fragments (Rewrite / Merge),
+ /// and the whole-dataset replacements (Overwrite / Restore).
+ fn check_data_overlay_txn(
+ &mut self,
+ other_transaction: &Transaction,
+ other_version: u64,
+ ) -> Result<()> {
+ match &other_transaction.operation {
+ Operation::Append { .. }
+ | Operation::CreateIndex { .. }
+ | Operation::ReserveFragments { .. }
+ | Operation::Project { .. }
+ | Operation::UpdateConfig { .. }
+ | Operation::UpdateBases { .. }
+ | Operation::Clone { .. }
+ | Operation::UpdateMemWalState { .. }
+ | Operation::DataReplacement { .. }
+ | Operation::DataOverlay { .. } => Ok(()),
+ // A concurrent Update or Delete that *removes* one of our overlaid
+ // fragments leaves the overlay orphaned — it is addressed by physical
+ // offset into a fragment that would no longer exist — so conflict and
+ // retry against the new fragment list. Fragments merely updated in
+ // place (deletion vectors, column rewrites) preserve physical offsets,
+ // so the overlay stays valid and does not conflict.
+ Operation::Update {
+ removed_fragment_ids,
+ ..
+ }
+ | Operation::Delete {
+ deleted_fragment_ids: removed_fragment_ids,
+ ..
+ } => {
+ if removed_fragment_ids
+ .iter()
+ .any(|id| self.modified_fragment_ids.contains(id))
+ {
+ Err(self.retryable_conflict_err(other_transaction, other_version))
+ } else {
+ Ok(())
+ }
+ }
+ Operation::Rewrite { groups, .. } => {
+ // A rewrite (compaction / fold) of a fragment we are overlaying
+ // changes its physical row addresses, so our offsets would be
+ // invalid. Conflict only if it touches one of our fragments.
+ let touches_our_fragment = groups
+ .iter()
+ .flat_map(|g| g.old_fragments.iter())
+ .any(|f| self.modified_fragment_ids.contains(&f.id));
+ if touches_our_fragment {
+ Err(self.retryable_conflict_err(other_transaction, other_version))
+ } else {
+ Ok(())
+ }
+ }
+ Operation::Merge { .. } => {
+ // Merge rewrites the whole fragment list; always conflict.
+ Err(self.retryable_conflict_err(other_transaction, other_version))
+ }
+ Operation::Overwrite { .. } | Operation::Restore { .. } => {
+ Err(self.incompatible_conflict_err(other_transaction, other_version))
+ }
+ }
+ }
+
fn check_merge_txn(
&mut self,
other_transaction: &Transaction,
@@ -1026,7 +1146,8 @@ impl<'a> TransactionRebase<'a> {
| Operation::Delete { .. }
| Operation::Rewrite { .. }
| Operation::Merge { .. }
- | Operation::DataReplacement { .. } => {
+ | Operation::DataReplacement { .. }
+ | Operation::DataOverlay { .. } => {
Err(self.retryable_conflict_err(other_transaction, other_version))
}
Operation::Overwrite { .. }
@@ -1050,6 +1171,7 @@ impl<'a> TransactionRebase<'a> {
| Operation::CreateIndex { .. }
| Operation::Rewrite { .. }
| Operation::DataReplacement { .. }
+ | Operation::DataOverlay { .. }
| Operation::Merge { .. }
| Operation::Restore { .. }
| Operation::ReserveFragments { .. }
@@ -1078,6 +1200,7 @@ impl<'a> TransactionRebase<'a> {
| Operation::CreateIndex { .. }
| Operation::Rewrite { .. }
| Operation::DataReplacement { .. }
+ | Operation::DataOverlay { .. }
| Operation::Merge { .. }
| Operation::ReserveFragments { .. }
| Operation::Update { .. }
@@ -1102,6 +1225,7 @@ impl<'a> TransactionRebase<'a> {
| Operation::UpdateConfig { .. }
| Operation::CreateIndex { .. }
| Operation::DataReplacement { .. }
+ | Operation::DataOverlay { .. }
| Operation::Rewrite { .. }
| Operation::Clone { .. }
| Operation::ReserveFragments { .. }
@@ -1166,6 +1290,7 @@ impl<'a> TransactionRebase<'a> {
| Operation::CreateIndex { .. }
| Operation::Rewrite { .. }
| Operation::DataReplacement { .. }
+ | Operation::DataOverlay { .. }
| Operation::Merge { .. }
| Operation::Restore { .. }
| Operation::ReserveFragments { .. }
@@ -1238,6 +1363,7 @@ impl<'a> TransactionRebase<'a> {
| Operation::Overwrite { .. }
| Operation::Delete { .. }
| Operation::DataReplacement { .. }
+ | Operation::DataOverlay { .. }
| Operation::Merge { .. }
| Operation::Restore { .. }
| Operation::Clone { .. }
@@ -1331,6 +1457,7 @@ impl<'a> TransactionRebase<'a> {
Operation::Append { .. }
| Operation::Overwrite { .. }
| Operation::DataReplacement { .. }
+ | Operation::DataOverlay { .. }
| Operation::Merge { .. }
| Operation::Restore { .. }
| Operation::ReserveFragments { .. }
@@ -2734,6 +2861,129 @@ mod tests {
}
}
+ #[test]
+ fn test_data_overlay_conflicts() {
+ use crate::dataset::transaction::DataOverlayGroup;
+ use ConflictResult::*;
+ use lance_table::format::{DataOverlayFile, OverlayCoverage};
+ use roaring::RoaringBitmap;
+
+ // Our transaction overlays fragment 1.
+ let overlay_op = |fragment_id: u64| Operation::DataOverlay {
+ groups: vec![DataOverlayGroup {
+ fragment_id,
+ overlays: vec![DataOverlayFile {
+ data_file: DataFile::new_legacy_from_fields("overlay.lance", vec![0], None),
+ coverage: OverlayCoverage::dense(RoaringBitmap::from_iter([0u32])),
+ committed_version: 0,
+ }],
+ }],
+ };
+ let update_removing = |removed_fragment_ids: Vec| Operation::Update {
+ removed_fragment_ids,
+ updated_fragments: vec![],
+ new_fragments: vec![],
+ fields_modified: vec![],
+ merged_generations: Vec::new(),
+ fields_for_preserving_frag_bitmap: vec![],
+ update_mode: None,
+ inserted_rows_filter: None,
+ updated_fragment_offsets: None,
+ };
+ let delete = |updated: Vec, deleted: Vec| Operation::Delete {
+ updated_fragments: updated,
+ deleted_fragment_ids: deleted,
+ predicate: "x > 2".to_string(),
+ };
+ let rewrite_of = |old: &Fragment| Operation::Rewrite {
+ groups: vec![RewriteGroup {
+ old_fragments: vec![old.clone()],
+ new_fragments: vec![],
+ }],
+ rewritten_indices: vec![],
+ frag_reuse_index: None,
+ };
+
+ let fragment0 = Fragment::new(0);
+ let fragment1 = Fragment::new(1);
+
+ // Each case is checked against our overlay on fragment 1.
+ let cases: Vec<(Operation, ConflictResult)> = vec![
+ // Permissive: preserves physical offsets / leaves fragment 1 in place.
+ (
+ Operation::Append {
+ fragments: vec![fragment0.clone()],
+ },
+ Compatible,
+ ),
+ (
+ Operation::CreateIndex {
+ new_indices: vec![],
+ removed_indices: vec![],
+ },
+ Compatible,
+ ),
+ (
+ Operation::DataReplacement {
+ replacements: vec![DataReplacementGroup(
+ 1,
+ DataFile::new_legacy_from_fields("r.lance", vec![0], None),
+ )],
+ },
+ Compatible,
+ ),
+ // Another overlay on the same fragment stacks rather than conflicts.
+ (overlay_op(1), Compatible),
+ // Delete/Update that only updates fragment 1 in place is compatible.
+ (delete(vec![fragment1.clone()], vec![]), Compatible),
+ (update_removing(vec![2]), Compatible),
+ // ...but removing our overlaid fragment 1 orphans the overlay -> conflict.
+ (delete(vec![], vec![1]), Retryable),
+ (update_removing(vec![1]), Retryable),
+ // Rewriting fragment 1 invalidates its physical offsets -> conflict;
+ // a rewrite of a different fragment does not.
+ (rewrite_of(&fragment1), Retryable),
+ (rewrite_of(&fragment0), Compatible),
+ // Merge rewrites the whole fragment list; Restore replaces the dataset.
+ (
+ Operation::Merge {
+ fragments: vec![fragment1.clone()],
+ schema: lance_core::datatypes::Schema::default(),
+ },
+ Retryable,
+ ),
+ (Operation::Restore { version: 1 }, NotCompatible),
+ ];
+
+ for (other, expected) in cases {
+ let mut rebase = TransactionRebase {
+ transaction: Transaction::new(0, overlay_op(1), None),
+ initial_fragments: HashMap::new(),
+ modified_fragment_ids: modified_fragment_ids(&overlay_op(1))
+ .collect::>(),
+ affected_rows: None,
+ conflicting_frag_reuse_indices: Vec::new(),
+ conflicting_mem_wal_merged_gens: Vec::new(),
+ };
+ let other_txn = Transaction::new(0, other.clone(), None);
+ let result = rebase.check_txn(&other_txn, 1);
+ match expected {
+ Compatible => assert!(
+ result.is_ok(),
+ "overlay should be compatible with {other:?}, got {result:?}"
+ ),
+ Retryable => assert!(
+ matches!(result, Err(Error::RetryableCommitConflict { .. })),
+ "overlay should retryably conflict with {other:?}, got {result:?}"
+ ),
+ NotCompatible => assert!(
+ matches!(result, Err(Error::IncompatibleTransaction { .. })),
+ "overlay should be incompatible with {other:?}, got {result:?}"
+ ),
+ }
+ }
+ }
+
#[test]
fn test_create_index_conflicts_only_on_same_name() {
let index0 = IndexMetadata {
@@ -3208,6 +3458,7 @@ mod tests {
Operation::DataReplacement { replacements } => {
Box::new(replacements.iter().map(|r| r.0))
}
+ Operation::DataOverlay { groups } => Box::new(groups.iter().map(|g| g.fragment_id)),
}
}
diff --git a/rust/lance/src/utils/test.rs b/rust/lance/src/utils/test.rs
index 3338eee07a8..f804a7cc38a 100644
--- a/rust/lance/src/utils/test.rs
+++ b/rust/lance/src/utils/test.rs
@@ -243,6 +243,7 @@ impl TestDatasetGenerator {
Fragment {
id: 0,
files,
+ overlays: vec![],
deletion_file: None,
row_id_meta: None,
physical_rows: Some(batch.num_rows()),