Stable Row ID

[westonpace]

Lance currently has no stable row ids / primary keys. This is a UX challenge: once a row is inserted, it’s not obvious how to get it back, update it, or delete it. It also creates challenges for keeping secondary indices up-to-date. If a the “row id” the secondary index points to changes, we need to update and re-write the secondary index.

In practice, these two challenges are somewhat separate:

• For secondary indices, the main challenge is what happens when a row id moved. When a row id updated, we don’t want to keep it in the index anyways.
• For primary keys, we care about when the row is moved and updated.

Therefore, to solve this incrementally, we can first solve the secondary index problem. Then later solve the primary key problem.

How Lance works now

Right now, Lance has the concept of the _rowid . It is a 64-bit values that is the combination of two 32-bit numbers: the fragment_id the row is in and the offset the row is in that file. The _rowid provides easy constant-time access to any row. But as soon as that row moves, whether due to compaction or updates, this id changes. Because the _rowid field represents the physical location of the row, for the rest of this document we will call it the “row address” so we can differentiate it from the stable row id.

Goals

What are the outcomes that will result from these changes? How will we evaluate success for the proposed changes?

• Introduce a row id that is stable when moved. The id of a row should not change if it is compacted. However, it may change if it is updated.
• Accessing by row id should require near constant time complexity.
• Make sure we never run out of row ids.

Non-Goals

To narrow the scope of what we're working on, outline what this proposal will not accomplish.

• Introduce a public primary key that is stable when rows are updated. This will be handled in a future feature.

Proposed Solution

Describe the solution to the problems outlined above. Include enough detail to allow for productive discussion and comments from readers.

TLDR:

• Add row id as auto-incrementing u64 id.
• Each fragment metadata will contain its row id index
• Deletion files data will be superseded by tombstones in the row id index
• We need new rules for how indices can cover fragments

Auto-incrementing u64 id

Row ids will be incrementing u64 id. Assignment of the row ids will be similar to how fragment ids are assigned: We will keep a high water mark in the manifest. During the commit loop, we will save the used row ids in the fragment metadata (more detail on how this works soon).

The high water mark will be a new field max_row_id , which will work similar to max_fragment_id.

Row id indices inside fragment metadata

To make random access fast, we will need a secondary index on row id. This will map from row id to row address. We will update this index with every write.

The row id index will have separate code paths from other indices. Each fragment will have a separate piece of the index and these pieces will be combined in memory. Because the row addresses within a fragment are a simple sequence, we can serialize the index as an array of row ids. In other words, they values of the mapping are always 0..physical_rows, so there’s no need to serialize that part.

The row id index could be encoded to be very small and written inline in the manifest. For example, a fragment that was just inserted would just contain a contiguous sequence of row ids, so it just needs to record the start and end of the range. If the index ever got large enough, it could be written to another file. The encoding scheme is an implementation detail, but ideally it should be encoded both on-disk and in-memory.

message DataFragment {
  // Unique ID of each DataFragment
  uint64 id = 1;
  RowIdIndex row_id_index = 5;
  ...
}
message RowIdIndex {
  oneof location {
    // If the index is < 100KB, it should be stored here
    bytes inline = 1;
    // Larger indices should be in separate files
    // It's in a separate file so it doesn't have to be rewritten in each
    // manifest.
    string file_path = 2;
  }
}

Replace deletion files with tombstones in row id index

To reduce the number of files needed, deletion files will be eliminated, and instead deletions will be marked in the row id index. This eliminates extra IOs, since whenever the index is read the deletion file would have also need to be read. We can encode the deleted rows as a sentinel value.

Example operations

Let’s review some operations and how they would write row ids:

1. Insert 100 rows. This would set row ids to 0..99. max_row_id is now 99.
2. Insert 50 rows. This would set row ids 100..149. max_row_id is now 149.
3. Delete rowid=120. Since row id 120 is in fragment 2, we would modify the row id index to put a tombstone where id 120 was. (In the diagrams, the tombstone is -1, but it could be a different sentinel value.)
4. Compact. We merge fragments 1 and 2 into fragment 3, and remove the deleted row in frag 2. The new row ids are 0..119, 121.149 . Notice how because we use an incrementing range and kept it sorted by row id, we could combine the ranges 0..99 and 100..119 into a single range.
5. Update rowid=35. When we update, we still need to delete the old row and add a new one. This is because we don’t have a way to invalidate the value in the index. So we modify the row id index for frag 3 to put a tombstone for row id 35. Then we add a new fragment with the updated row.

Reserving row ids

Some systems, like a write-head log, might want to reserve a set of row ids ahead of time. For example, they might reserve 1,000 row ids. When it receives an insert transaction, it assigns one of it’s reserved row ids to that row. When it receives an update transaction on a row that hasn’t been flushed, it can make that update reference the row id that’s been reserved and assigned. Once it is ready to flush a row, it can do so and immediately write the row ids safely, knowing they are already reserved.

We can do this with the same mechanism as we use for reserving fragment ids. You can commit a transaction requesting a certain number of rows, and then from the transaction you can get the range based on how much max_row_id changed.

Time and Space Complexity of the index

When the manifest is read, readers will combine these per-fragment indices into an in-memory index data structure. The space and time complexity will be sensitive to how many ranges there are stored in this rowid index. It can be kept very fast if the row ids are always full contiguous ranges. But once there are a lot of holes and jumps, it will degrade in performance. Space complexity will become O(num_rows) (up to 16GB for 1 billion rows, assuming a u64 for key and value and completely random ordering). Time complexity could still be pretty fast if we use [interpolation search](https://en.wikipedia.org/wiki/Interpolation_search) (average complexity O(log(log(n))) but worst case O(n)). There’s also hash tables for O(n) .

The upshot of using incrementing ids is sets of new rows can just be represented by a range. In addition, when we compact we can sort by id, potentially merging ranges. This will happen naturally in a LSM tree for an append only table where we keep the fragments clustered by row id.

When updates happen, the rowids will be moved to new fragments, creating holes in the original fragments. This will increase the size of the rowid index. We can have merge strategies prefer to merge together ranges of rowids, so that those holes will later be filled in upon compaction.

The most pathological is deletion. Deleting rows will create holes in the row ids, which can never be filled in. (If we are making a public row id, we shouldn’t re-use ids.)

ℹ️ **Other clustering schemes.** We’ve contemplated having other clustering schemes, including HNSW order (for fast vector search take) or based on filter columns (for fast filtering). These other clustering schemes will degrade the performance of the row id index by making it much larger. We could try to make an alternative row id assignment scheme that aligned well to a clustering. In order for it to work, it would need to have a mechanism to prevent duplicate ids and allow for reserving ids ahead of time (or some other affordance for a WAL).2

Backwards compatibility

Initially, stable row ids will be an opt-in feature. This will let us experiment with and test this on main. A new feature flag will be added to the format, which will prevent older readers and writers from trying to use the dataset.

If we wanted to, we could make a method to transition an existing table. It would use the existing row addresses as the row ids for all rows currently in use.

Indices

Can we re-use the current row address logic for row ids?

• Row ids are the same size / data type as row addresses
• They are stable, so re-mapping is unnecessary
• They do not decompose into two meaningful u32s, unlike row address.

Conclusion: as long as the logic doesn’t rely upon the decomposition of the row address, we should be able to re-use the logic. We can search for >> 32 and << 32 to help find these spots.

Fragment coverage bitmaps

Right now, we make certain assumptions about how indices cover fragments:

1. If a fragment has any rows in an index, all of its rows must be in the index. A fragment cannot be partially covered.
2. Two delta indices can’t cover the same fragment. Or, each fragment is covered by one index file per index.

This second one could now be violated if fragments are compacted:

This should be fine for now, but this will get more complex when we tackle primary keys

• What if bitmap is over rows? Could be something like max id indexed.
• It’s a many-to-many relationship
• Could we record bitmap at top-level index name? We do care about this for LSM-tree purposes.

Pre-filtering

Previously, when we saw tombstoned rows or missing fragments, we can easily compute the row ids that should be considered missing. With move-stable row ids, it’s safe to say the tombstoned rows are deleted (though this will change in the Primary keys / fully-stable row ids). But with missing fragments, it’s ambiguous whether the row ids were deleted or moved. It’s also unclear which row ids should be considered affected, since we can’t just mask the fragment id part of the row id anymore.

The pre-filter will need to be replaced by the row id index combined with the tombstone bitmap. To check whether a row should be filtered, we should check they exist in the index, but not in the tombstone bitmap. We could try translating this into a pre-filter to collapse masks if we need to. We could optimize conversion of row id sequences into the pre-filter.

Risks

Highlight risks so your reviewers can direct their attention here.

• This changes a lot of code paths, which will be a significant amount of work to test. The structure of row ids is built into the codebase, so there is risk we could accidentally introduce bugs.
• If we keep this opt-in, we could end up maintaining many duplicate and complicated code paths.

[westonpace]

CC @wjones127 who authored this proposal (I simply copied it over)

[jackye1995]

After thinking about this problem for some time, I feel there is an alternative we could evaluate here and compare the tradeoffs for solving the issue of secondary index out of date when rows have moved. I call this approach remap separation.

Remap Separation

Problem Analysis

Today, in order to compensate for the fact that secondary indexes become out of date when rows have moved, we remap all the old row addresses to new ones, and commit the newly remapped indexes as a part of the compaction. This has a few issues:

1. compaction is slow, because you have to remap all indexes before you can claim compaction completes.
2. because compaction is slow, it is more likely that something conflicting has happened in the middle, e.g. a merge
3. most importantly, because compaction now also overwrites indexes, it conflicts with the other asynchronous processes that is trying to optimize the index, e.g. reindexing after a merge

Solution

Since the root cause of this issue is the remap process during compaction, one alternative way we could solve the issue is to just separate the remap process out from the compaction process, so compaction only compact files.

There are 2 challenges we need to overcome in this solution:

1. If remap is not performed as a part of compaction, we have lost the mapping between old and new row addresses. This is not easy to recover because of (1) deletion files in old fragments, (2) one rewrite group might have produced multiple new files and the cut boundary is not always deterministic.
2. Assuming remap will run as a separated async process, then before that process is run, all compacted data will become unindexed. This will be bad for read performance and is not acceptable.

Old Fragment Reuse Index

To solve these 2 challenges, we will let compaction produce a new index, which we call Old Fragment Reuse (OFR) index. The idea is that after a compaction process have compacted some old fragments to new fragments, we record:

1. fragment replacements: the mapping of old fragments to new fragments. This is similar to Rewrite Group in transaction file.
2. row ID map: the detailed mapping (HashMap<u64, Option<u64>>) of old row ID to new row ID as a new index. This will be serialized as an Arrow file with 2 columns.

Note: we don't want to reuse the transaction file because:

1. it will be difficult to search for older transaction files later at read time if there are other things happened to the table after compaction.
2. we can be a bit more concise in the information we record, e.g. new fragment only need an ID not the full fragment since that is already committed in the manifest.

Because there might be multiple compactions before a remap happens, we will store one fragment map and row ID map for each version. This results in the following proto definitions:

message OFRIndexDetails {

  message NewFragmentSignature {
    uint64 fragment_id = 1;
    // if the number of rows have changed, the fragment has logically changed (e.g. new deletes)
    uint64 num_rows = 2;
  }

  message Replacement {
    repeated DataFragment old_fragments = 1;

    repeated NewFragmentSignature new_fragment_signatures = 2;
  }

  message VersionMapEntry {
      uint64 dataset_version = 1;

      repeated Replacement replacements = 2;

      // The path to the row ID mapping file for this version, relative to the index root
      string row_id_map_path = 3;
  }

  message VersionMapEntries {
    repeated VersionMapEntry entries = 1;
  }

  oneof version_map {
    // If small (< 200KB), store inline
    VersionMapEntries entries = 1;
    ExternalFile external_file = 2;
  }
}

Change to compaction process

We will introduce a flag in CompactionOptions called defer_index_remap. When true, instead of remapping all indexes, it will write out the OFR index:

• if there is no existing OFR index, a new one is created with the initial VersionMapEntry
• if there is an exiting OFR index, it appends a new VersionMapEntry to the version_map.entries

Remap Index Operation

We will introduce a new index operation in DatasetIndexExt:

/// Remap an index using the OFR index information
async fn remap_index(&mut self, index_id: &Uuid) -> Result<()>;

Specifically, it will:

1. load one specific index by UUID
2. load the OFR index
3. load all the row_id_map files in each versions, make a combined row ID map
4. remap the index based on the combined row ID map
5. replace the old index with the new index

Remap index as a part of reindexing

Because remap index will conflict with a reindexing process, we can do an optimization to run them together. To achieve that, we add an option remap_index in DatasetIndexExt.create_index. When this flag is true, we will also run remap_index if possible.

Trim OFR Index

The OFR index needs to be trimmed regularly. When all indexes have a IndexMetadata.dataset_version higher than a VersionMapEntry.dataset_version, the entry can be trimmed.We can introduce a trim_remap_index function to run this process.

A few considerations:

1. This will conflict with the compaction since they both touch the same OFR index. So ideally OFR index trim and compaction should run sequentially. But even if it conflicts, either compaction or trim can be rebased on top of the other one.
2. Similar to run remap as a part of reindexing, maybe we could run this as a part of cleanup, since this is essentially cleaning up the OFR index.

Using OFR Index at read time

Now comes the important part. We want to leverage the OFR index at read time, so that index is still able to work on newly compacted fragments although these fragments are not in the index. The idea is the following:

1. we load the OFR index fragment replacement, this is a cold start
2. if a plan determines a fragment is unindexed, check if it has replacement of old fragments. If so, use the old fragments that is covered by index.
3. if a plan requests row address (e.g. with_row_address=true, merge-insert which tries to write a deletion file at a row position of the new fragment), then load the row ID map (which requires a cold start), and use the row ID map to derive what is the new row address of a given selected old row address.

Compared to the stable row ID approach, this seems to be more efficient both from space and runtime perspective:

1. instead of having to construct and use an in memory index of stable row ID to fragment ID + row position which is O(num_rows) for every query, in this approach the row ID map is O(num_rows_compacted_not_trimmed), which is always much smaller. Also the row ID map is required only in cases described in 3 instead of all the time.
2. The runtime of searching the row ID map in OFR index should always be better than the stable row ID index. If they are using the same data structure, then row ID map in OFR index is faster due to smaller size. And because of the small size, we can afford to use a HashMap to achieve O(1).
3. Both solutions have a cold start, so it should be similar in that regard

Change to GC process

In this approach, we also need to not delete old fragments until it is removed from OFR index.

After thoughts on stable row ID

I looked into how systems like Postgres and Oracle (traditional database systems that are more on the OLTP side) solves such issue since they also have indexes that need to be updated when data is vacumed. Looks like most such database systems use physical row address instead of a globally stable row ID for indexes, and does something similar to the asynchronous remapping with a temp OFR index to serve traffic in the mean time. It is definitely going to make the query planning and execution phase more complex, but it seems like the right direction to go here.

I understand systems like Snowflake, Iceberg (and Delta if my memory is right, not super sure though) uses a global counter stable row ID approach, but I think that is more OLAP focused, with use cases like CDC incremental view. So overall I feel remap separation might be a better approach for solving the index coverage problem after compaction, but we should definitely explore if we could leverage a stable row ID for use cases like CDC incremental view.

What do you think? @westonpace @wjones127

[westonpace]

I agree with the analysis and splitting out remapping to be a separate process.

I'm still wrapping my head around the OFR index. If I understand correctly there is at most one OFR index per manifest right?

It seems that VersionMapEntry has a lot more data than I would expect. I think all you would need is the row_id_map_path right? How would the other information be helpful? Specifically, do we need to keep old_fragments? This seems like it could grow quite large.

I hadn't thought about using the OFR index at read time but I suppose that could work well. I'm not quite sure about the approach listed here though. It seems we have (at least) two options in the case where compaction has run but indices have not yet been remapped.

1. Use the old fragments (I think this is what you have proposed) and only materialize the addr->addr map if the scan requests the row addr column. (I guess this answers my question about why we need the old fragments)
2. Always materialize the addr->addr map and only scan the new fragments.

Approach 1 could be slow because it still needs to scan the uncompacted data. Approach 2 would be faster I think but have the burden of keeping the row id mapping in memory. Given compaction doesn't run all that often I don't think the cold start penalty of loading the row id map is too high (and we could prewarm the cache with the updated row id map when we run compaction).

I suppose there is also an option 3 which might be easier in the short term. Just drop the compacted fragment from the index and fallback to a runtime scan / filter of the data. This would probably be just as fast for filtered scans but maybe slower for embedding searches.

[jackye1995]

It seems that VersionMapEntry has a lot more data than I would expect. I think all you would need is the row_id_map_path right? How would the other information be helpful? Specifically, do we need to keep old_fragments? This seems like it could grow quite large.

Yeah I was thinking about if we need that, my current conclusion is we need it.

Consider the following setup, the table has fragments [1,2,3,4], and a compaction makes [1,2,3,4] -> [5,6]. At this time, all indexes are still indexed against [1,2,3,4]. When a query comes, we have logic to check if we need to generate a split plan, by looking at if the index fragment bitmap covers all current fragments. And in this case, it will determine [5,6] are not covered, and produce a scan plan. So we need the logic to say [5,6] has equivalent old fragments [1,2,3,4] so do not produce the scan plan.

I originally thought about using a bitmap for the check, but I ended up just use the original mapping structure because we have to check that the scan covers both 5 and 6, and also 5 and 6 has not changed (e.g. deletion file has updated), only when these 2 conditions are true we can swap to old fragments 1,2,3,4.

For old_fragments, if we don't keep the full information, then we don't know what data file to read in the end when we take rows. But I thought that is okay since these fragments were in the manifest, so it is basically just as bad as the previous uncompacted state of the table.

[jackye1995]

If I understand correctly there is at most one OFR index per manifest right?

Yes. The alternative is to have multiple delta indexes one per version. I actually haven't thought about the pros and cons of that (because I did not read into the delta indexes part until probably yesterday...)

Maybe the details structure can be simplified a bit if we use multiple delta indexes, every compaction produces a separated delta of the OFR index and when we read we combine them. The tradeoff is probably if there are too many old fragments, then we will use an external file, and with delta indexes there could be multiple external files increasing IO. The tradeoff is faster compaction completion time, but that process is already async so having a single index seems to be a bit more advantageous.

[westonpace]

Ah, yeah, that seems a little messy. I agree that one-per-manifest seems cleaner.

[jackye1995]

1. Use the old fragments (I think this is what you have proposed) and only materialize the addr->addr map if the scan requests the row addr column. (I guess this answers my question about why we need the old fragments)
2. Always materialize the addr->addr map and only scan the new fragments.

Actually can you explain a bit more about these 2 options. I don't quite get why there are 2 options. When I say "use old fragments", the goal is to leverage the index for random access because old fragments are likely indexed. So when you say "Approach 1 could be slow because it still needs to scan the uncompacted data.", it seems to be opposite to my expectation, shouldn't it be faster because the uncompacted data is indexed? (if it is not indexed, then we will just scan compacted data)

[westonpace]

Ah, by "only scan the new fragments" I meant, apply the index to the new fragments. You are right that in many cases (e.g. search) we are targeting only a few rows and so it doesn't matter much how many fragments we have.

In my head I was thinking about another case (filtered scan) where the filter might eliminate 95% of the data. In this case, on cloud storage, we are still mostly doing a scan, but we are using the index to speed up the compute cost of the filter. Plus, on NVMe we can still use the index to skip data loading. This sort of filtered scan would be faster on the compacted data..

[jackye1995]

I see. In general the gist is that whenever it makes sense to use the old fragments, we do that, and if it would impact perf we should just use new fragments. Currently the line is cut by if (1) it is a search or point query, and (2) if the uncovered fragments have old fragment mapping, (3) if the old fragment is indexed, we will use the old fragments.

[jackye1995]

I created this Github Issue for the tasks breakdown #3831, will push up my code first for doing the remap separation from compaction to run independently, and we can discuss more about the read side optimization in the specific tasks.

[jackye1995]

Just to provide an update on this, the feature I described above has been merged, by introducing the fragment reuse index (#3847) and audi-remap feature (#3971), so now compaction will not conflict with index update, and has minimal impact on read performance.

I think it would be great if we can now discuss about how we should push the original stable row ID effort forward. One great use case I can think of is to use it to track row lineage. This would be similar to the Iceberg row lineage feature (https://iceberg.apache.org/spec/#row-lineage), where we can know exactly how a row has been changing across the table versions, and have a CDC incremental view of the data when we want. This will require move and update stable row ID, so some additional work is needed to make it update stable.

[yanghua]

If the scenario is solid, to push this work, what is needed to be done next?

[jackye1995]

The work was divided to having a move-stable row ID, and then make it update-stable. You can check the existing work in #2307. We probably need to verify everything is still working end to end since this has been an experiemtnal feature. If everything still works, the next step is to preserve the row ID after an update. And to fulfill CDC use case, I think we also need to introduce the last_updated_version metadata field alongside the stable row ID field.

[yanghua]

Thanks for sharing this.

We probably need to verify everything is still working end to end since this has been an experiemtnal feature.

Let me catch the context and try to find a way to verify the move-stable row ID still works fine.

[yanghua]

We probably need to verify everything is still working end to end since this has been an experiemtnal feature.

@jackye1995 I found there are many test cases in dataset/rowids.rs and test_dataset.py that have covered many scenarios(via setting enable_move_stable_row_ids: true). Does that mean the move-stable row ids work fine and has been verified?

What's more, there is a TODO comment in dataset/rowids.rs's test block:

// TODO: query / scan / take after deletion, compaction, then deletion

Shall we realize these test cases first?

[jackye1995]

Yes that sounds like a good start. Also so far there are places we use row ID to mean row address. With this change, row ID and row address will start to mean 2 totally different things. We probably need to be super clear about which place is using row id and which place is using row address, and change the variable name accordingly in different places.