feat: implement distributed VECTOR_SEARCH with parallel fragment/index scanning

[summaryzb]

Summary

Implements distributed execution for the VECTOR_SEARCH table function, enabling Spark-parallel vector similarity search across Lance datasets. When enabled via spark.sql.lance.search.distributed.enabled=true, the driver plans one Spark task per execution unit
(indexed segment or fallback fragment), and each worker runs a local ANN scan or fallback to KNN scan without indexed segment. Results are merged with a global sort on _distance.

This provides horizontal scalability for vector search workloads on large datasets without requiring a centralized vector index server.

Behavior

Condition	Execution Path
distributed.enabled=false	Single-partition namespace.queryTable()
distributed.enabled=true, has vector index	One task per index segment, plus fallback tasks for unindexed fragments (unless fastSearch=true)
distributed.enabled=true, no index	One task per fragment (flat KNN)

Notice

• Indexed-unit path with IVF_PQ is implemented but not yet covered by E2E tests (pending CREATE INDEX ... USING IVF_PQ SQL support in lance-spark extensions) which is rely on feat: distributed vector index creation #605.
• The prefilter=true flag is forced for fallback units due to a lance-core JNI limitation where Scanner::nearest rejects fragment-restricted scans without a prefilter expression.
• Specify indexSegments in ScanOptions is pre required which is rely on feat(java): support segment-based distributed vector search lance#7169, other wise compile fails

Testing

• BaseSparkDistributedVectorSearchTest exercises fallback-only scenarios
• Spark 3.4 and 3.5 modules have thin test subclasses

[summaryzb]

CI RED is expected since it rely on lance-format/lance#7169

[summaryzb]

@jackye1995 @Xuanwo @LuciferYang PTAL

[sezruby]

@summaryzb thanks for working on this. Quick data point in case it's useful:

Measured driver-side single-machine Dataset.newScan(...nearest(Query)), k=10, nprobes=16, IVF-PQ, ABFSS storage, single JVM, 100 timed queries on one open Lance handle:

Dataset	|R|	dim	p50	p90	p99
Cohere wikipedia-2023-11-embed-multilingual-v3	10M	1024	56 ms	182 ms	514 ms
Synthetic uniform-random (note: ~worst case for IVF clustering)	100M	128	157 ms	776 ms	1213 ms
Synthetic uniform-random	250M	128	1084 ms	1528 ms	2323 ms

Wondering whether the distributed path can beat sub-100ms single-machine once you account for Spark task scheduling overhead — would you be able to share end-to-end latency numbers from your benchmark setup at a similar scale? Would help inform docs around when to enable spark.sql.lance.search.distributed.enabled=true.

Thanks!

[FANNG1]

Thanks for driving this work.

This PR is very important for our Spark + Gravitino + Lance query path. Gravitino serves as the catalog/control plane, but it does not execute the queryTable data-plane API, so scalable direct Lance Dataset execution from Spark is essential for vector search in this setup.

Now that the prerequisite support for selecting vector index segments through the Lance Java/JNI scan API has been merged, would you be interested in continuing this PR?

Thanks again!

[FANNG1]

Filter semantics differ between indexed units and fallback units

While testing this PR (merged onto #819, which supplies the lance-core 12 dependency indexSegments needs), I hit a case where the same data, same query and same filter return different rows depending on whether a vector index exists.

LanceMergedSearchColumnarPartitionReader.buildScanOptions forces prefilter(true) only for fallback units:

boolean fallbackUnit = p.getIndexSegments().isEmpty();
boolean userRequestedPrefilter = Boolean.TRUE.equals(base.getPrefilter());
if (userRequestedPrefilter || fallbackUnit) {
  b.prefilter(true);
}

Indexed units therefore post-filter: each unit takes its own top-k first, then applies the filter, and the global merge combines those already-truncated per-unit results.

Repro

4 fragments x 64 rows, vector = [id, id, ...], query vector at the origin, k => 5, filter => 'id % 2 = 0' (true answer is [0, 2, 4, 6, 8]):

setup	result
distributed, no index (all fallback units)	[0, 2, 4, 6, 8] — correct
distributed, every fragment indexed	[0, 2, 4, 64, 66]
distributed, every fragment indexed, prefilter => true	[0, 2, 4, 6, 8] — correct
non-distributed reference (spark.sql.lance.search.distributed.enabled=false)	[0, 2, 4] — only 3 rows for k => 5

[0, 2, 4, 64, 66] is exactly what you get when fragment 0 returns its top-5 (ids 0-4) post-filtered to 0/2/4, fragment 1 returns 64/66/68, fragment 2 returns 128/130/132, and the global sort takes the first 5.

Reproduced both against a local dir namespace (JUnit) and against a real S3/MinIO dataset behind a REST namespace server, so it is not storage- or namespace-specific.

Note this also interacts with the default: shouldUseDistributed reads spark.sql.lance.search.distributed.enabled with a default of "true", so existing filtered VECTOR_SEARCH queries change behaviour silently on upgrade. The PR description says the feature is opt-in via that flag, which does not match the code.

Suggestion

Apply the same prefilter decision to both unit kinds whenever a filter is present. Fallback units already have to prefilter (the Scanner::nearest restriction called out in the code comment), so doing the same for indexed units makes the distributed result both self-consistent and equal to the true top-k.

A regression test that builds an indexed and an unindexed copy of the same table and asserts both return the same filtered rows would catch this — that is what surfaced it here.

[nyl3532016]

readOptions need to pinOpenedRef
We can refer to the logic of LanceScan.

[nyl3532016]

we can add distance ranger support after this pr merge