[python] Share raw vector scans across batch queries - #9753

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.

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:

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 ×ばつ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 ×ばつ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.