Supports range-based distributed BTree index

[steFaiz]

Introduction

The BTree index is a crucial scalar index in Lance. Many thanks to @westonpace for the implementation! Its main structure is shown below:

In summary, all <value, rowAddress> pairs are stored in a single page_data.lance in object storage. A B-tree structure from page_lookup.lance is kept in memory to enable fast access to a relatively small data block, and the final results are obtained through sequential scanning within that block.
To accelerate index building, @xloya contributed the distributed version (as illustrated below). The key points are:

1. Fragments are horizentally divided. Each subtask only build BTree index for a subset of the full-list fragments.
2. A dedicated merger is responsible for k-ways merging all page files into one final page file as well as the lookup file. Since all page files are already sorted, the merge process can be relativly faster.

However, inevitably, a single point (the Merger) has to process all the data, which becomes unacceptable at very large scales (tens of billions of records or more).

Resolution

Existing big data computing engines (such as Spark and Flink) provide mature range-shuffle capabilities, enabling efficient global sorting of datasets at the scale of hundreds of billions of records. Leveraging this, we have implemented a range-based BTree index using Lance-Spark and Lance, with the following key features:

1. Minimal changes to the existing Lance BTree index — only a few hundred lines of core code were added.
2. Full backward compatibility— completely compatible with the current implementation.
3. Dramatic performance improvement — for datasets with hundreds of millions of records, the merge phase is reduced from tens of minutes to under a few seconds, and end-to-end index construction completes in just a few minutes.

The range-based BTree index has a structure nearly identical to the current implementation, with the only addition being multiple distinct page files—each covering a non-overlapping data range:

Workflow

The overall workflow is as follows:

1. Spark/Flink reads the column to be indexed and its corresponding RowAddresses from Lance, performs a global sort, and applies range-based shuffling.
2. For each data range, it invokes Lance dataset.create_index with the corresponding rangeId passed in IndexParams. This create_index call is identical to the standard BTree index construction process.
3. The Merger combines all lookup files. Since this step only requires sequentially processing and merging the lookup files, the merge phase is extremely fast.

Partial Index Creation

This part of the implementation remains fully consistent with the current approach, with only two minor modifications:

1. A new field preprocessed_data is added to IndexCreateBuilder. When this data is present, the load_train_data step is skipped.
2. The generated page_file name now includes the rangeId as part of its filename. The rangeId is already sorted by its range, which is guaranteed by Spark/Flink.

Lookup File Merging

During the merge process, only the lookup files need to be merged. The purpose of this step is to:

1. Reassign a global PageIndex number for each page across all ranges.
2. Record the offset corresponding to each PageIndex in the consolidated pages file.

For example, consider these two lookup files:

In the merged lookup file, the page indices from lookup file 2 are reassigned to 3 and 4, which correctly reflect their global page indices. Additionally, the number of pages in each original lookup file is recorded, enabling the recovery of per-page-file internal page indices during read operations, as illustrated below:

Update

During updates, the distributed processing pipeline cannot be effectively leveraged for acceleration. Therefore, both update and remapping operations fall back to the standard BTree index behavior: all PageFiles are merged into a single file, resulting in only one PageFile being generated.
We will introduce distributed update mechanism in future issues.

Tests

We tested two tables with data volumes of 100 million and 10 billion rows respectively, and indexed an integer type column. The results are as follows:

num of rows	num of ranges	execution time	merge time
130 million	3	23 min	1 s
130 million	50	3 min	3 s
10 billion	1000	15 min	46 s

Even on the 10-billion-row dataset, we were able to complete the merge in just 46 seconds, the end to end latency is reduced to about 15 min.

[steFaiz]

@westonpace @wjones127 @jackye1995 Do you have any suggestions? I'm wondering whether the community would be open to allowing externally provided data during index creation—currently, Lance retrieves all required data by itself when building an index. However, since we’re integrating this functionality into Lance-Spark, the community can help ensure the correctness of this process (e.g., validating the legitimacy of the input data).
Also cc @xloya

[wjones127]

I'm wondering whether the community would be open to allowing externally provided data during index creation—currently, Lance retrieves all required data by itself when building an index. However, since we’re integrating this functionality into Lance-Spark, the community can help ensure the correctness of this process (e.g., validating the legitimacy of the input data).

Once we are getting down to this detail, it makes me wonder if IndexCreateBuilder is the right API. That's mean to be an API that's generic over the different index types.

IIUC, the distributed processing here is specific to BTree structure, so I think we might want instead a specific BTreeIndexBuilder for handling this. Could you create a proposal for that?

During updates, the distributed processing pipeline cannot be effectively leveraged for acceleration. Therefore, both update and remapping operations fall back to the standard BTree index behavior: all PageFiles are merged into a single file, resulting in only one PageFile being generated.

I'd like to see a better update story. Two cases you will want to handle:

1. Creating a new delta index: can be the same process as creating the initial index but just focusing on un-indexed fragments
2. Merging index deltas: I think this could be easily distributed as well.

[steFaiz]

@wjones127 Thanks for your reply！

I'd like to see a better update story. Two cases you will want to handle:

1. Creating a new delta index: can be the same process as creating the initial index but just focusing on un-indexed fragments
2. Merging index deltas: I think this could be easily distributed as well.

We have considered and drafted an initial design for this functionality—for example, how to split ranges when merging two ranged B-trees to enable concurrent rewriting. However, implementing distributed updates is a considerably complex process—it would require planning, distributed execution, and then commit phases, leading to significant changes in the existing codebase. For instance, it might necessitate introducing a new distributed update interface that works across all index types, which would require extensive design discussions and community alignment. So we are planning to gradually introduce this part in future issues.

IIUC, the distributed processing here is specific to BTree structure, so I think we might want instead a specific BTreeIndexBuilder for handling this. Could you create a proposal for that?

I agree with you! We’ll submit the initial commit in the next couple of days, and we can discuss the implementation details in the PR!

[steFaiz]

added some test results

[westonpace]

Given the scales we are using here and the training times I wonder if there is much need for a full distributed update. For example, let's say we create two "shards". One shard is the "stable shard" for the old data, which follows this range-based approach. One shard is the "new shard" for the new data, which follows the simple style. As new data arrives we use the simple update and update the new data shard. Then, once a week, just retrain the entire dataset and replace the "old shard".

As for the approach, this makes sense at high scales to avoid the very expensive merge phase. It seems we are trading off some search time penalty in exchange for a much faster build time. Also, if the indexed column is unique, then there is no penalty to search time, since the matching row is only going to be in a single page data file.

[westonpace]

Once we are getting down to this detail, it makes me wonder if IndexCreateBuilder is the right API. That's mean to be an API that's generic over the different index types.

IIUC, the distributed processing here is specific to BTree structure, so I think we might want instead a specific BTreeIndexBuilder for handling this. Could you create a proposal for that?

I think we have two APIs today. First, the IndexCreateBuilder, which is the "simplified Lance will do everything for you" API. Second, is the Dataset.commit API which is "you do everything and just record the index in Lance" API.

I do think individual indexes might have unique APIs and I also share this concern that trying to make the IndexCreateBuilder API do everything will be too challenging. I think, for btree, I'd like to see what we think a btree-specific API might look like at a high level.

[westonpace]

Also, the missing piece here that spark and flink seem to be providing, is just the ability to come up with good partition bounds right? If we had something like that it would be really nice to have the DF-based engine support it as well.

[steFaiz]

@westonpace Thanks for your insightful comments!

For example, let's say we create two "shards". One shard is the "stable shard" for the old data, which follows this range-based approach. One shard is the "new shard" for the new data, which follows the simple style. As new data arrives we use the simple update and update the new data shard. Then, once a week, just retrain the entire dataset and replace the "old shard".

I strongly agree with this idea—it makes our BTree index design more LSM-tree-like, which would be especially beneficial for scenarios involving small amounts of data, such as minor data insertions or updates. However, this would introduce some management complexity. For example, we’d need to expose concepts like “index health,” so that callers (LanceDB, Spark, Flink, and end users) can decide whether to perform an update or rebuild the index from scratch. We can discuss these design considerations in detail in a follow-up issue.

I think, for btree, I'd like to see what we think a btree-specific API might look like at a high level.

Thanks for this! We will modify our initial commit in PR

Also, the missing piece here that spark and flink seem to be providing, is just the ability to come up with good partition bounds right? If we had something like that it would be really nice to have the DF-based engine support it as well.

I think the answer is: "We can partially achieve this."

Currently, Spark determines partition boundaries through data sampling, and Lance can implement this even more efficiently—thanks to its take API—potentially outperforming Spark in this aspect.

However, another major challenge is shuffle. Spark can read the entire table concurrently and then shuffle the data according to the computed partitions, ensuring that all records within the same range end up on a single node. This shuffle operation itself is quite heavy—we, for example, rely on Apache Celeborn to handle it.

I’m not sure whether DataFusion (if that’s what you meant by “DF”) is capable of performing such a distributed shuffle and range-based data co-location effectively.

[xloya]

I'm considering whether we could exposing Lance file writers and file readers to clients instead of making special modifications to the existing API to accommodate distributed processing.

The advantage of this approach is that Lance's own API remains stable, while exposing the file writers and readers allows the operational logic of the Lance API to remain consistent with the distributed engine connectors in such as lance-ray or lance-spark, thus fully leveraging distributed capabilities and maintaining considerable flexibility on the engine side. Perhaps distributed operations can be applied beyond just index building.

As for the example of distributed creation of Btree indexes, exposing the writers and readers allows the corresponding logic to be implemented in the distributed engine—such as sorting by range, merging the final page data and lookup files, and committing it to Lance—maintaining consistency with the current index structure and fully utilizing distributed capabilities.

What do you think?

[steFaiz]

@xloya Thanks for your advise!
I think this is a great idea. My main concern is that the current Java SDK is still experimental (see comment) —it lags behind the Rust implementation in both stability and feature completeness. Additionally, I worry that the Java side may still be missing some key capabilities; for example, during merge operations we need to list files, which would require introducing Java dependencies related to ObjectStore.

On the other hand, if we move all this logic outside of Rust, we’d end up duplicating the same logic in both Python and Java, which isn’t ideal. Therefore, I believe it might be better to first design a dedicated BTreeIndexBuilder on the Rust side. If we later find stronger requirements for distributed scalar indexing, we could then design a unified distributed index interface at the Rust layer.