[python] Share raw vector scans across batch queries

[TheR1sing3un]

Purpose

Batch vector search currently calls _read_raw_search once per query for uncovered row ranges. With 32 queries, the same raw rows are planned, read and converted to Python lists 32 times, and each scan materializes the entire raw table.

Read the filtered, snapshot-pinned ranges once through the Arrow batch reader. Convert each batch once, update a separate top-k heap for every query, and merge each raw result with its corresponding indexed result. The scalar distance calculation, metric selection, dimension validation and tie-breaking helpers stay unchanged. Close both the Arrow reader and its underlying iterator on success or failure.

Single-query search and indexed reranking keep their existing paths.

Tests

• 82 passed: python -m pytest pypaimon/tests/batch_vector_raw_scan_test.py pypaimon/tests/vector_search_filter_test.py -q
• Coverage includes L2/cosine/inner product, NULL/zero vectors, ties, overlapping ranges, scalar/partition/prefilters, empty input, mixed indexed/raw results, planned snapshots, batch-by-batch scoring and iterator closure on dimension/read errors.
• Repository-configured flake8 and git diff --check passed.

Benchmark and ablation

Added dev/benchmark_batch_vector_raw_scan.py. It calls the public execute_batch_local() API on real local Paimon/Parquet tables with no vector index and vector-index.search-mode=full.

• repeated: reproduces the previous per-query raw scan loop.
• shared-table: scans once but materializes the full table/Python lists, isolating the shared-scan benefit.
• shared-stream: this change, sharing the scan and consuming one batch at a time.

All three use identical scalar distance and top-k helpers. Float32 vectors, 128 dimensions, L2, top-k 10, batch size 1,024, read parallelism 1. Each configuration runs three times in fresh processes, sequentially with shuffled configuration order. Timing includes planning, reading, conversion, scoring and merging; excludes data generation/process startup. RSS is total process peak including imports and Arrow buffers. Environment: macOS 26.4.1 arm64, Python 3.9.6, PyArrow 19.0.1, NumPy 2.0.2; OMP/OPENBLAS thread counts 1. Filesystem cache is not flushed.

Rows	Queries	Mode	Median seconds (min–max)	Median peak RSS, MiB	Raw scans
16,384	1	repeated	0.410 (0.409–0.427)	255.7	1
16,384	1	shared-table	0.414 (0.403–0.433)	255.6	1
16,384	1	shared-stream	0.395 (0.390–0.406)	170.5	1
16,384	8	repeated	3.199 (3.124–3.253)	256.1	8
16,384	8	shared-table	1.369 (1.369–1.413)	255.8	1
16,384	8	shared-stream	1.357 (1.355–1.377)	170.6	1
16,384	32	repeated	12.599 (12.472–12.713)	256.5	32
16,384	32	shared-table	4.635 (4.624–4.637)	256.7	1
16,384	32	shared-stream	4.646 (4.530–4.677)	171.0	1
65,536	8	repeated	12.589 (12.282–12.817)	581.0	8
65,536	8	shared-table	5.521 (5.428–5.785)	564.8	1
65,536	8	shared-stream	5.357 (5.353–5.375)	192.9	1

The 8-query cases improve 2.35–2.36x; 32 queries improve 2.71x. The shared-table ablation shows that shared scanning accounts for the timing improvement, while streaming lowers peak RSS. For 65,536 rows and 8 queries, peak RSS falls from 581.0 to 192.9 MiB (66.8%). Single-query timing is similar.

Batch-size sensitivity at 16,384 rows / 8 queries:

Batch rows	Median seconds	Median peak RSS, MiB
256	1.395	164.2
1024	1.357	170.6
4096	1.386	196.8

All 42 runs produce exactly matching row IDs and floating-point scores within each dataset/query-count group, including across batch sizes. Instrumentation verifies Q raw plans/reader passes and N×Q delivered rows for repeated, versus one plan/pass and N delivered rows for both shared modes. These are logical reader counters, not physical disk reads.

Scoring remains O(NQD); the fallback keeps one batch and Q top-k heaps, with no Q×N score matrix. Storage reader buffers and scan metadata are outside that bound. The existing batch-reader path is serial; eager to_arrow() can read splits concurrently. These measurements fix read parallelism to 1 and do not establish remote-storage or multi-split parallel throughput. No GPU/distributed benchmark or bounded parallel prefetch is included.

Reproduce from paimon-python with project dependencies installed:

export PYTHONPATH=.
export OMP_NUM_THREADS=1
export OPENBLAS_NUM_THREADS=1
python dev/benchmark_batch_vector_raw_scan.py prepare --warehouse /tmp/raw-scan-bench --rows 16384 --dimension 128
python dev/benchmark_batch_vector_raw_scan.py run --warehouse /tmp/raw-scan-bench --mode repeated --queries 8 --output /tmp/repeated.json
python dev/benchmark_batch_vector_raw_scan.py run --warehouse /tmp/raw-scan-bench --mode shared-table --queries 8 --output /tmp/shared-table.json
python dev/benchmark_batch_vector_raw_scan.py run --warehouse /tmp/raw-scan-bench --mode shared-stream --queries 8 --output /tmp/shared-stream.json

Repeat each run three times in a fresh process and compare result_sha256 for identical datasets/query counts. Repeat with queries 1/32; prepare a new warehouse with 65,536 rows for the larger case. Use --batch-size 256/4096 for the batch-size ablation.

[JingsongLi]

Reviewed c718d0b. Requirement fit: supported; one performance regression needs attention.

Sharing the raw scan across queries removes real repeated work, and the focused tests passed (82 tests). The new streaming path, however, serializes split I/O that the previous path ran concurrently; the existing benchmark fixes read.parallelism to 1 and therefore cannot reveal this regression.

[JingsongLi]

[P2] Preserve split-read parallelism in the shared scan

_new_arrow_batch_reader iterates splits serially and does not apply read.parallelism. The previous _read_raw_search used TableRead.to_arrow, which honors that option and automatically parallelizes multiple splits. Thus batches with one or a few queries on remote multi-split tables become slower even when users configured parallel reads. Through execute_batch_local() on a real four-partition Parquet table with read.parallelism=4 and 150 ms injected split-open latency, one query changed from 160 ms / 4 concurrent opens to 626 ms / 1; two queries changed from 321 ms to 634 ms, with identical IDs and scores. Please preserve bounded split concurrency (and merge per-query Top-K state) while sharing the scan, with a multi-split regression that checks concurrency and result equivalence. The acknowledged serial-reader tradeoff currently discards an effective read configuration in this public API.