feat: support create btree index distributedly

[xloya]

Closed #4665.

Overall Steps:

1. Create ordered Btree sub-page files / sub-lookup files at the fragment level based on Ray / Daft.
2. Sort and merge the sub-page files using a k-way merge sort algorithm, supporting prefetch data of sub-page files.
3. Output the final lookup file.
4. Commit the final index to dataset.

Production Test Results:
In a production scenario, using Ray and 50 workers on a string ID field in a dataset of 700 million records, we achieved the following:

1. Btree index build time was reduced from 190 minutes to 19 minutes, a 10x increase in build speed.
2. Peak memory usage on the Ray head node when creating the index was reduced from 90+ GB to 4+ GB, a 95%+ reduction.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 81.82927% with 149 lines in your changes missing coverage. Please review.
✅ Project coverage is 80.73%. Comparing base (76a710e) to head (65ad622).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
rust/lance-index/src/scalar/btree.rs	85.13%	89 Missing and 28 partials ⚠️
rust/lance-index/src/scalar/inverted/builder.rs	0.00%	32 Missing ⚠️

Additional details and impacted files

@@           Coverage Diff            @@
##             main    #4667    +/-   ##
========================================
  Coverage   80.72%   80.73%            
========================================
  Files         321      321            
  Lines      124043   124847   +804     
  Branches   124043   124847   +804     
========================================
+ Hits       100131   100792   +661     
- Misses      20340    20457   +117     
- Partials     3572     3598    +26     

Flag	Coverage Δ
unittests	80.73% <81.82%> (+<0.01%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[xloya]

@jackye1995 @westonpace @BubbleCal @chenghao-guo Please take a look when you have time, thx!

[chenghao-guo]

Thank you for the refactor. My previous design wasn’t thoroughly considered; yours is a significant improvement

[jackye1995]

this is amazing feature! I added some initial comments, will look more into the details tomorrow

[xloya]

@jackye1995 Thanks for your review, address all comments.

[westonpace]

This is great. I love the overall approach of adding a train method that takes in some fragments and creates a partial index and then a merge method that will complete all the fragments.

I think we can simplify the prefetch and just use the file reader's prefetch (I have some comments here).

Also, does this merge the batches themselves? For example, if I sort fragment 1 and I get a batch with values 10-50 and then I sort fragment 2 and I get a batch with values 30-80 is there some code that is merging these two batches (maybe the output is a batch with 10-50 and 50-80)? I might be missing it but maybe it is just emitting 10-50 and then 30-80 in sequence?

Also, if we want to simplify some of the merge code, Datafusion has a SortPreservingMerge operator that can do this for us too.

[xloya]

Also, does this merge the batches themselves? For example, if I sort fragment 1 and I get a batch with values 10-50 and then I sort fragment 2 and I get a batch with values 30-80 is there some code that is merging these two batches (maybe the output is a batch with 10-50 and 50-80)? I might be missing it but maybe it is just emitting 10-50 and then 30-80 in sequence?

I might not fully understand what's going on here. Let me explain the logic using these two fragments:

1. Use train_btree_index to construct sub-indexes for fragment_1 and fragment_2. If they are grouped together, I believe this code will ensure strict ordering: https://github.com/lancedb/lance/blob/main/rust/lance/src/index/scalar.rs#L131-L135, which generates the sequence [10, 80]. If they are not grouped together, the sequences [10, 50] and [30, 80] are generated, respectively.
2. Use the k-way merge algorithm to merge them, ensuring the final order.

[westonpace]

I might not fully understand what's going on here. Let me explain the logic using these two fragments:

1. Use train_btree_index to construct sub-indexes for fragment_1 and fragment_2. If they are grouped together, I believe this code will ensure strict ordering: https://github.com/lancedb/lance/blob/main/rust/lance/src/index/scalar.rs#L131-L135, which generates the sequence [10, 80]. If they are not grouped together, the sequences [10, 50] and [30, 80] are generated, respectively.
2. Use the k-way merge algorithm to merge them, ensuring the final order.

Ah, I see where I was confused. I thought the partition iterator was yielding batches. Instead it is yielding rows. So the heap is built one row at a time and batches are merged that way. You can ignore that particular comment.

[xloya]

@jackye1995 @westonpace @BubbleCal Hi, I've refactored the code based on your feedback. Please review it again when you have time, thanks! I've also noticed that using Datafusion's SortPreservingMerge has slowed down index creation compared to my previous version(the speed can be improved by adjusting the prefetch nums and the number of sub-indexes), but it's still significantly faster than the current single-node creation. Furthermore, the code has been significantly simplified. I think we can further optimize performance in future PRs.

[xloya]

@jackye1995 @westonpace @BubbleCal Gentle pin for this

[westonpace]

Thanks for working through the reviews, this looks good to me now.