perf(vindex): parallelize Parquet reads and multipart index uploads

[jerry-024]

Purpose

Speed up vector-index builds by removing two object-storage bottlenecks:

1. full-file scans were unnecessarily excluded from parallel Parquet row-group reads;
2. serialized indexes uploaded multipart chunks one at a time.

Changes

• Treat full-coverage row_ranges without a deletion vector as an unfiltered scan, allowing the existing parallel row-group path to run. Partial ranges, deletion vectors, and empty ranges keep their previous semantics.
• Keep oversized row groups charged against the full configured Parquet read budget, preserving max_inflight_bytes as a strict projected-byte admission limit, and warn once when a row group exceeds it.
• Upload vector-index multipart chunks with four parts in flight. Other writers retain their existing serial behavior.
• Add opt-in aggregate diagnostics for vector-index build phases and Parquet row-group reads.
• Add a production-path IVF-PQ build benchmark example.

Reproduce the IVF-PQ build benchmark

The benchmark is in ivfpq_build_benchmark.rs. It builds an IVF-PQ index with dimension 768, cosine distance, nlist=4096, and pq.m=192 on an existing Paimon table:

PAIMON_CATALOG_OPTIONS='<catalog-options-json>' \
PAIMON_LOG_VECTOR_INDEX_BUILD_TIMING=1 \
cargo run --release -p paimon --example ivfpq_build_benchmark -- \
  <database> <table> <vector-column> --drop-existing

Benchmark

Cohere 10M × 768 dimensions, cosine IVF-PQ (nlist=4096, pq.m=192, 8 bits), 10 Parquet files / 100 row groups, direct OSS, and no local cache.

Parallel Parquet reads

Both baseline and this PR use an explicit 512 MiB Parquet read budget. The metrics below cover only the Parquet-read work changed by this PR.

metric	baseline	this PR	change
source batch wait	105.346s	51.148s	−51.45%
peak in-flight row groups	1	4	+3

The current-PR values are from the latest single validation run. The 512 MiB budget is a workload-specific benchmark setting and must be configured explicitly. The default remains 256 MiB.

Concurrent multipart upload

With all other benchmark inputs held constant, four in-flight multipart uploads reduced the serialize/upload phase from 30.3s to 4.6s (−84.8%). The vector-index writer alone uses this concurrency; other writers keep their existing behavior.

Tests

• Full, stitched, partial, empty, and deletion-vector row selections, including _ROW_ID parity.
• Shared Parquet budget accounting and peak in-flight diagnostics.
• Oversized row groups consume the full budget and emit a one-time warning; ordinary and tiny row-group accounting remains strict.
• Vector-index build timing aggregation and existing builder tests.

API and format

No public API or file/index format changes. The concurrent writer helper is crate-private.

[shyjsarah]

Thanks for the optimization work. I found one issue that I think should be addressed before merge:

Concurrent multipart uploads are not bounded to 32 MiB

async_writer_with_concurrency(4) documents the buffering cost as 4 * 8 MiB, but that is not the behavior of the pinned OpenDAL 0.58 implementation.

MultipartWriter creates ConcurrentTasks with prefetch = 8192. With concurrency greater than one, completed tasks are not continuously reaped during write; each completed task still owns its WriteInput, including the 8 MiB part buffer, until tasks are drained during close(). If uploads complete faster than index serialization, a ~2 GiB index can therefore retain close to the whole index size in completed upload buffers, rather than ~32 MiB. The previous concurrency-1 path executes inline and does not accumulate these buffers.

Please either keep concurrency at 1 for now, use/patch an OpenDAL implementation with strictly bounded completed-task buffering, or introduce a bounded uploader that continuously reaps completed parts. A fast fake multipart backend test asserting peak retained bytes would also help prevent regression.

Two non-blocking diagnostics/benchmark notes:

• For the target 768-dimensional workload, a ~294 MiB row group consumes the entire default 256 MiB read budget, so the parallel path can still run effectively serially unless the option is tuned. It would be useful for the benchmark to print the effective budget and observed peak concurrency, and warn when the peak is 1.
• Parquet diagnostics are collected only after read_budget is filtered for parallel-path eligibility. Partial row selections can perform Parquet I/O while reporting zero row-group/projected-byte diagnostics. Consider decoupling diagnostics collection from fast-path eligibility, or naming the fields as parallel-path-only metrics.

[JingsongLi]

Because read_budget has already been filtered by row_selection.is_none(), any partial-row-range read reaches this block with read_budget == None. Vector-index shard boundaries can cut through a file, so these reads still perform Parquet I/O while reporting parquet_row_group_count and all projected-byte diagnostics as zero. Please compute and record the diagnostic sizes from self.read_budget independently of parallel-path eligibility, or rename these fields to make their parallel-path-only scope explicit.