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

...splits
Final step of apache#755: let paimon-rust execute a primary-key (bucket-local ANN) vector
search that an ENGINE planned. Paimon Java plans `BucketVectorSearchSplit`s on a
planner node and ships each one to a worker, which calls this entry point:
 let mut search = table.new_vector_search_builder();
 search.with_vector_column("v").with_query_vector(q).with_limit(k);
 let rows = search.execute_read_for_bucket_splits(&split_bytes).await?;
Split bytes in, Arrow record batches out, with `__paimon_search_score` attached.
Planning is done by the caller; this does not read the index manifest and does not
re-plan the scan. Over the C ABI the same terminal is
`paimon_vector_search_builder_execute_read_for_bucket_splits`. Callers distributing
work across buckets merge the per-bucket Top-K themselves.
This is the trust boundary that makes the change more than an entry point: the
supplied plan and the row counts inside it come from outside the process. What
follows is the consequence, and it is where most of this diff is.
A file the split does not list means "unrestricted", as Java's `rowRangesByFile`
specifies, and absence carries that state -- `FileRowSelection` is the per-file
three-state selection (absent unrestricted, empty excluded, non-empty restricted).
The kernel previously read a missing entry as "no rows allowed", the opposite of
Java, so planning normalized every omitted file into an explicit `[0, row_count - 1]`
range. That expansion is gone: ranges stay ranges from planning through the ANN mask,
the ANN result check and the exact fallback, and a split that narrowed nothing
(`rangeFileCount == 0`, which is both the committed fixture and the shape an engine
ships by default) reaches the ANN backend with no filter at all rather than an
all-permitting mask that Lumina turns into one `u64` per live row.
Row counts and range endpoints are validated where they enter, none of them a bound
invented here. A listed range is checked against its source file's row count, as Java
checks it. Row counts are checked non-negative at decode for every data file, not
only the files the message goes on to list ranges for. `MAX_LIVE_ROW_IDS` bounds a
segment's mask and is charged before each insertion, because a mask spans its sources'
wire-supplied row counts: without it, one restricted sibling file or one deletion
vector is enough to insert an unrestricted file wholesale at whatever size its count
claims, and the tests for that run for minutes or OOM-kill the process. The limit is
Java's own for this quantity (`LuminaVectorGlobalIndexReader.toScopedIds` refuses
above `Integer.MAX_VALUE`); our dense conversion had no guard, so `to_scoped_ids`
mirrors it, and `lumina.search.max-filter-bytes` (default 64 MiB) bounds the
allocation under concurrent searches.
The Java-generated fixture covers the external-planner split format end to end from
both Rust and C.
Not here, deliberately. The result-splitting read API (`VectorRead`,
`TableRead::to_arrow_indexed`, a public `PkVectorIndexedSplit`, the C split handle)
was staged in earlier revisions of this branch and is WITHDRAWN: it has no consumer
independently of this entry point, and freezing a materialize contract before one
exists is the thing apache#755 said to avoid. It is proposed separately, and so is the
non-finite-score check that belongs to it -- that route hands scores out as split
METADATA a caller ranks on, whereas this one emits them as a column, unvalidated,
exactly as the pre-existing `execute_read` does. Likewise three independent hardening
fixes that were riding along became their own PRs -- the deletion-vector position
bound (apache#802), Lumina's empty-index error precedence (apache#803) and the scalar batch arity
check (apache#801). All three are merged and this branch is rebased onto them, so it uses
apache#801's `take_only_result` for its own single-query unwrap rather than repeating the
check inline.
apache#802 turned out to be load-bearing for the trust boundary rather than incidental to
it: a bucket split names its own deletion files, so a forged one supplies deletion
positions, and without that bound a position past its file lands in the NEXT file's
ordinal range and drops one of ITS rows -- silently, since the mask stays well
formed. Now that it is on main this branch inherits it; noted because the split
looked like three unrelated hardening fixes and one of them was not.
The mask bound and apache#803's step order also meet on the batch path, and neither pins
the meeting alone: `an_empty_index_outranks_a_filter_too_large_to_densify` drives an
empty index with an include-set far above the byte budget and asserts it reports no
hits rather than an oversized filter. Without the ordering the budget rejects it;
without the budget there is nothing to order against.
The C terminal checks the builder's `inner`, not only the outer pointer: a
`#[repr(C)]` wrapper can arrive zero-initialized, which passes a null-POINTER check
while carrying a null `inner` that the terminal then dereferences. The sibling
terminals on this ABI predate that check; this one is new, so it starts with it.
Tests: the Rust integration suite drives the committed Java fixture -- the split's
row ranges as the read's authority, agreement with the manifest route, a narrower
limit, and an empty plan. The C suite covers argument marshalling, corrupt bytes, and
one happy path over the same Java bytes; the semantic contract is asserted on the
Rust side where a failure names what broke.
Gates: 2621 `cargo test -p paimon --lib` (2696 with `fulltext`), the bucket-split
integration suite, `-p paimon-c` 73 passed with the pre-existing
`vector_search_append_filter_returns_invalid_input` failure, which was re-run on
`origin/main` and fails there identically. `cargo fmt --all -- --check` clean;
workspace clippy `-D warnings` clean except `pypaimon_rust`, which cannot build on
this machine (pyo3 needs Python >= 3.10, it has 3.9) and is untouched here.

...ad consume them
Follow-up to apache#771, which 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, which positions, at what scores
 let rows = read_builder.new_read()?.to_arrow_indexed(&selected)?; // WHICH columns
`PkVectorIndexedSplit` carries the same three payloads as Java's
`globalindex.IndexedSplit` -- the data split, the selected physical ranges, the aligned
scores. The type already existed in the read kernel with no public producer; this
exposes it with getters and crate-private construction. Java's is also SERIALIZABLE
(its own MAGIC/VERSION frame plus `SplitSerializer`); this one is not, because both
halves run in one address space, and freezing a wire format before there is a consumer
for it is what apache#755 said to avoid.
The one-shot terminal is UNCHANGED and still public, and its rows stay best-first;
the two-step route returns rows in physical order carrying `__paimon_search_score`,
as Java does, sorting after the read. 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 is also what makes this diff additive. Once apache#771 is in, its Rust entry
point and its C ABI symbol are published surface, so removing them would be a
breaking change rather than the cleanup it would have been while the branch was
unmerged. `the_two_step_route_agrees_with_the_one_shot_terminal` pins that the two
terminals select the same rows.
`TableRead::indexed_read_type()` reports the two-step 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. Its 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.
The read refuses, rather than ignores, every input it cannot honour: 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 refusing is a deliberate divergence, on the grounds that a
silently dropped limit is the failure this read refuses everywhere else. A filter is
rejected
rather than forwarded because Rust recovers 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.
`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, and 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 --
`..._search_for_bucket_splits` returning an opaque handle, `..._splits_count`,
`..._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.
Tests: the Rust suite asserts what the search selected BEFORE any data file is opened
(the property this route exists for), physical ordering under a query whose rank order
differs from it, `indexed_read_type` against both real and empty results, and each
refusal. The C suite keeps to the happy path over the Java fixture plus handle and
pointer safety; the projection and filter contracts are asserted on the Rust side,
where a failure names the semantic that broke.
Gates: 2655 `cargo test -p paimon --lib`, 15 in the bucket-split suite, `-p paimon-c`
77 passed with the pre-existing `vector_search_append_filter_returns_invalid_input`
failure, which fails identically on `origin/main`. `cargo fmt --all -- --check` clean;
workspace clippy `-D warnings` clean except `pypaimon_rust`, which cannot build on
this machine (pyo3 needs Python >= 3.10, it has 3.9) and is untouched here.