feat: support build FTS index distributedly

[chenghao-guo]

Close #4514
Related with lance-format/lance-ray#12

This PR introduces distributed fts index capabilities enabling parallel index creation across multiple fragments.

New added methods

• execute_uncommitted(): Creates index metadata without dataset commitment, returning IndexMetadata for distributed coordination. Chainable method specifying target fragment IDs for selective indexing as suggested by Will Jones in discussion feat: add create_fragment_inverted_index for distributed inverted index building #4514 (comment)

              let partial_index=CreateIndexBuilder::new(&mut dataset, &["text"], IndexType::Inverted, &params)
                    .name("distributed_index".to_string())
                    .fragments(vec![fragment_id])
                    .fragment_uuid(shared_uuid.clone())
                    .execute_uncommitted()
                    .await?;

• merge_index_metadata(): Merges distributed index metadata from multiple workers into consolidated final metadata

Distributed Workflow

1. Split and Parallel Phase: Ray header distributes the fragments to different workers. (We may distribute the fragments evenly to different workers by fragment statistics, which will be implemented in lance-ray connector)
   Ray Workers call execute_uncommitted() on specific fragments using fragments() method. Shared UUID via fragment_uuid() ensures consistent index identity.
2. Merge Phase: merge_index_metadata() consolidates partition metadata files (part_*_metadata.lance)
3. Commit Phase: Final index commitment with unified metadata

Example
The workflow example can be found in test_distribute_fts_index_build in test_scalar_index.py.

Following work after this PR:
The distributed index building workflow PR will be proposed to lance-ray connector.
lance-ray draft PR lance-format/lance-ray#45

Other implementation details on fragment_mask
Optional mask with fragment_id in high 32 bits. When provided,
only partitions whose partition id matches this fragment will be included.
The fragment mask is constructed as (fragment_id as u64) << 32.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 61.73184% with 137 lines in your changes missing coverage. Please review.
✅ Project coverage is 80.59%. Comparing base (103006f) to head (56cbff1).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
rust/lance-index/src/scalar/inverted/builder.rs	20.76%	95 Missing and 8 partials ⚠️
rust/lance/src/index/scalar.rs	78.78%	9 Missing and 5 partials ⚠️
rust/lance-index/src/scalar/inverted/index.rs	57.69%	11 Missing ⚠️
rust/lance/src/index/create.rs	95.86%	5 Missing ⚠️
rust/lance-index/src/scalar/inverted.rs	77.77%	2 Missing ⚠️
rust/lance/src/index.rs	0.00%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #4578      +/-   ##
==========================================
- Coverage   80.62%   80.59%   -0.03%     
==========================================
  Files         317      317              
  Lines      118785   119097     +312     
  Branches   118785   119097     +312     
==========================================
+ Hits        95769    95985     +216     
- Misses      19568    19660      +92     
- Partials     3448     3452       +4     

Flag	Coverage Δ
unittests	80.59% <61.73%> (-0.03%)	⬇️

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.

[BubbleCal]

This is a great work!
Just some questions

[BubbleCal]

I think each indexing worker/fragment would have partition IDs starting from 0, we need to map each new partition id to all_partitions.len() + part_id?

[chenghao-guo]

Thanks for the suggestion — that’s convincing. I’ll adjust the partition ID assignment logic and make the corresponding changes.

[chenghao-guo]

I think each indexing worker/fragment would have partition IDs starting from 0, we need to map each new partition id to all_partitions.len() + part_id?

Hi @BubbleCal , thanks for taking a look. I want to confirm that I’m interpreting the implementation correctly.

In the previous design, each partition ID was assigned using the mask fragment_id << 32 to ensure isolation across fragments and avoid conflicts. As a result, partition IDs are not sequential (i.e., not 0, 1, 2, 3, …).

Based on the suggestions, in the final stage, I reorder and rename all fragment partitions and then reassign partition IDs starting from 0. While this entails some rename operations, I’m not sure this is the best approach. I can’t assign sequential partition IDs 0,1,2... at fragment-index creation time because that work happens on different machines (e.g., Ray workers). Therefore, the only practical option I see is to sort the IDs and remap them in the last stage in a coordinator (e.g. Ray header)

I’ve made some changes to the function accordingly—please see the diff here:

• https://github.com/lancedb/lance/pull/4578/files#diff-613f7dde20f16c5577e9f69e63d58c5fe3601f7c2053c831bc8d2b7562b56d3bR833-R908

I’d appreciate any feedback or alternative suggestions if you think there’s a better way to handle this.

[BubbleCal]

This makes sense to me, TBH I didn't realize that the partition IDs were encoded by frag_id << 32 + i, but I think the current reorder and re-assign design is still better, just in case we will probably build single index over multiple fragments (e.g. there are many small fragments?)

[chenghao-guo]

@BubbleCal Yes, I completely agree with your idea. Using “reorder” and “reassign” is definitely clearer, since the generated FTS index is almost the same as the single-node index.

My only concern is that the renaming and overall code structure may not be very elegant, which could make the code harder for others to understand. To mitigate this, I’ve added inline comments for clarity. I’ll likely refactor the code structure in the near future.

BTW, kindly let me know if any other suggestions, or if I could get your approval to merge the PR.

[BubbleCal]

it would be nice if we can add a helper function that builds the FTS index in distributed way, then we can add it into all existing FTS tests, to make sure the 2 ways will always get the same result

[chenghao-guo]

Got it, thank you for your great suggestions. I will add them later.

[chenghao-guo]

Done. Added helper function assert_distributed_fts_consistency and applied it to all existing FTS tests. Validated that distributed execution produces results consistent with single-node FTS index searches.

[chenghao-guo]

Rebased to resolve conflict.

[BubbleCal]

This work is amazing, thanks!

[BubbleCal]

NIT, we've changed the default value to False for with_position, so no need to specify it here. we can remove this in the future PR

[chenghao-guo]

Got it , thanks !