feat(table): let a vector search return its splits and an ordinary read consume them

[JunRuiLee]

Purpose

Follow-up to #771, now merged. This PR contains the staged two-step read API extracted from earlier revisions of #771 so its contract can be reviewed separately.

Reading the diff: this branch is rebased onto current main, including #771. GitHub now shows this PR's own change: 13 files / +2,176−30 in one commit.

Linked issue: #755

#771 reads engine-planned bucket splits through one terminal: split bytes in, Arrow rows out. That terminal cannot answer what the search found until the rows are already materialized, so a caller cannot decide whether the read is worth doing, and cannot read one search twice under different projections.

This adds the two-step form alongside it, following Java's layering:

let vector_read = search.new_vector_read()?;
let selected = vector_read.read(bucket_splits).await?;             // WHICH rows
// ... inspect: how many files matched, which positions, at what scores
let rows = read_builder.new_read()?.to_arrow_indexed(&selected)?;  // WHICH columns

This is the API that earlier revisions of #771 carried and that the review asked to move out. #771 now contains only the one-shot boundary; this PR keeps the staged form in a separate review with its own contract discussion.

Brief change log

• PkVectorIndexedSplit becomes public with getters and crate-private construction. It carries the same three payloads as Java's globalindex.IndexedSplit: the data split, the selected physical ranges, the aligned scores. Java's is also SERIALIZABLE (its own MAGIC/VERSION frame plus SplitSerializer); this one is not, because both halves run in one address space.
• VectorSearchBuilder::new_vector_read() → VectorRead::read(Vec<BucketVectorSearchSplit>) returns the selected splits without materializing user columns.
• TableRead::to_arrow_indexed(&[PkVectorIndexedSplit]) materializes them under the read builder's own projection. TableRead::indexed_read_type() reports that output schema, because read_type() describes to_arrow and omits the score column, and a search that matched nothing yields no batch to learn the schema from. The score field takes id i32::MAX and the name Java's VectorSearchProcedure.SEARCH_SCORE_FIELD uses, declared NON-NULL because that is what the read emits.
• paimon_vector_search_splits_free reads inner before reclaiming the outer pointer. Reclaiming assumes the search allocated it, so a caller declaring paimon_vector_search_splits s = {0}; and freeing &s would otherwise hand stack storage to the allocator. The sibling terminals already check inner for the same reason; this one has to check it first. The zero-handle test now frees one, and fails with SIGABRT against the old order.
• C ABI: four new symbols beside the existing one-shot terminal — paimon_vector_search_builder_search_for_bucket_splits returning an opaque handle, paimon_vector_search_splits_count, paimon_vector_search_splits_free, and paimon_table_read_to_arrow_indexed, which BORROWS the handle so one search can be read again under a different projection. Nothing is serialized. Every new terminal checks its handle's inner, not only the outer pointer, since a #[repr(C)] wrapper can arrive zero-initialized.

The one-shot terminal from #771 is unchanged and remains public. They are not two spellings of one thing: one call per bucket with ranked rows is a different contract from a reusable selection a caller inspects, reads under its own projection, and may read more than once. Keeping both makes this diff additive (+2,176/−30) rather than a removal plus a replacement.

What the read refuses rather than ignores

A filter, a with_limit, explicit with_row_ranges, and a row_filter_factory.

The middle two matter because neither ever reaches a TableRead — the read builder keeps both for TableScan, which this read does not run — so accepting them would return MORE rows than asked for, and with_row_ranges(vec![]) documents "selects no rows". Java ignores a limit here rather than refusing it: ReadBuilderImpl.newRead forwards it, but PrimaryKeyIndexedSplitRead does not override withLimit and inherits SplitRead's no-op, so withLimit(1) over a split selecting three rows still returns three. Refusing is a deliberate divergence, on the grounds that a silently dropped limit is exactly the failure this read refuses everywhere else.

A filter is rejected rather than forwarded because Rust recovers physical positions by zipping returned batches against the requested selection, so a predicate that drops rows desyncs the position and score cursor. Java can forward one because its reader reports each row's own returnedPosition().

The search likewise refuses a with_projection, which belongs to the read.

Not mirrored, deliberately: no TableScan stage, so no read-protection tag — Java's is opt-in on scan.plan-auto-tag-for-read.time-retained and Rust has none on any route.

Tests

Rust (15 in the bucket-split suite, 9 new here):

• a_search_is_inspectable_before_its_rows_are_read — asserts what the search selected BEFORE any data file is opened: how many files, which rows, at what scores, plus the pinned snapshot and that the derived split is not raw-convertible. This is the property the route exists for.
• the_two_step_route_agrees_with_the_one_shot_terminal — the two terminals select the same rows.
• rows_come_back_in_physical_order_carrying_their_scores — queries nearest [4,0], whose rank order is the reverse of physical order, so this is the test that would catch the read starting to rank. Every other test queries [0,0], where the two coincide.
• indexed_read_type_matches_the_rows_that_come_back — against both real batches and an empty result.
• One test per refusal: reserved projection, filter on the read builder, projection on the search.
• Unit tests in table_read.rs and read_builder.rs for the fan-in guard, per-split validation before streaming, the score-presence requirement, and the partition-only filter case (which is what makes the filter_set bit load-bearing — the data half of such a filter is empty).

C: the happy path over the Java fixture plus handle and pointer safety (zero-initialized handles, null arguments, count/free on null). The projection and filter contracts are asserted on the Rust side, where a failure names the semantic that broke.

Post-rebase gates on current main: cargo test -p paimon --test pk_vector_bucket_split_read_test 15 passed; cargo test -p paimon-c 79 passed; cargo fmt --all -- --check clean.

API and Format

New public Rust API (VectorRead, PkVectorIndexedSplit, TableRead::{to_arrow_indexed, indexed_read_type}) and four new C ABI symbols. No existing symbol changes signature; the ABI signature guards in bindings/c/src/vector_search.rs pin that.

No storage-format change, and deliberately no new wire format: the indexed splits are an in-process handle, not bytes.

Documentation

None added. The staged distributed form #755 describes — candidate-only search, global merge/rerank, deferred materialization — is a further step and would be where a documented wire format belongs.

[JunRuiLee]

Closing this PR since it has been superseded by #822.