[python] Search vector index shards concurrently

[TheR1sing3un]

Purpose

Vector index readers run their native search synchronously before returning a completed Future, so collecting those futures still opens and searches shards one at a time.

Add the Python query/table option vector.search.parallelism (positive integer, default 1) to schedule complete shard open/search/close operations with a bounded thread pool for both single and batch vector queries. Results are merged in split order, persisted metric checks are synchronized without serializing index loading, and pending tasks are cancelled while started tasks finish cleanup on failure. Single-shard and serial searches keep the direct path.

Tests

• python -m pytest -q pypaimon/tests/vector_index_parallel_search_test.py pypaimon/tests/vector_search_filter_test.py pypaimon/tests/vector_metric_consistency_test.py pypaimon/tests/vindex_vector_index_test.py pypaimon/tests/lumina_vector_index_test.py pypaimon/tests/batch_vector_raw_scan_test.py: 112 passed, 1 skipped.
• New coverage exercises concurrent index loading, worker limits, single/batch result ordering, offset pre-filters, metric mismatch and search-error cleanup, option validation, table/query precedence, and serial fast paths.
• Flake8, license-header checks, and git diff --check passed.

Benchmark

macOS 26.4.1 arm64, Python 3.9.6, paimon-vindex 0.4.0, NumPy 2.0.2. Each native IVF-flat shard contains 4,096 vectors of 64 dimensions with 16 clusters. Queries use Top-K=10 and nprobe=4; per-shard Vindex I/O parallelism is fixed at 1.

The temporary harness runs the production index read/merge path against local index files. Timings include index opening, native initialization/search, result conversion, merging, and closing; table planning and scalar pre-filter evaluation are excluded. Each process performs one warm-up and five measured iterations; values below are medians. A delay of 1 ms is injected per positional read to isolate storage-latency sensitivity; this is not a live object-store measurement.

Shards	Queries	Read delay	Master (ms)	Parallelism 1 (ms)	2 (ms)	4 (ms)	8 (ms)
8	1	1 ms	61.957	61.746	32.263	16.291	8.787
8	8	0 ms	2.827	2.883	2.998	4.005	4.218
8	8	1 ms	165.151	165.171	83.661	42.373	22.892
16	8	1 ms	331.570	331.942	167.545	86.012	44.061

At 16 shards / 8 queries, parallelism 4 improves latency by 3.85x and parallelism 8 by 7.53x versus master. Peak process RSS for parallelism 1/2/4/8 was 139.8/139.6/153.6/155.7 MiB. All result IDs and score bytes matched the serial baseline, and open stream counts never exceeded the configured parallelism.

Local tiny searches do not benefit from additional threads, which motivates retaining the default of 1. A separate 50-iteration local single-query check measured 1.075 ms on master versus 1.085 ms with parallelism 1. Real gains depend on shard size, storage, native thread settings, and concurrent query load.