feat(table): plan a PK-vector search from a decoded bucket split

[JunRuiLee]

Third step of #755: run a primary-key vector search over bucket splits an engine planned elsewhere, not only over a plan read from this table's index manifest.

Independent of #752 — the diff is against main alone.

PkVectorScan::plan_for_bucket_vector_splits takes the splits as authoritative: their payloads, row ranges and pinned snapshot are used as given, no manifest is read, and only the partition conjuncts are re-applied, since a caller may narrow further than the planner that produced them. Bucket grouping and segment selection reuse the manifest route.

Four decisions worth checking:

• Path resolution. _EXTERNAL_PATH wins; otherwise the bucket directory the split serialized when index-file-in-data-file-dir is set, else <table>/index. Java records an external path only for an index stored outside the table — PkVectorAnnSegmentFile writes null otherwise — so an ordinary bucket-local payload carries none. This duplicates only the small path decision fix(table): resolve index files by external path and bucket layout #752 centralizes; whichever lands second should consolidate this call site, and the option accessor added here is byte-identical to fix(table): resolve index files by external path and bucket layout #752's.
• The read is limited to the rows the split allows. The plan carries the selection as normalized Vec<RowRange>, and both paths that touch a data file read through those ranges: the residual predicate, and the exact fallback — which needs no predicate to be handed a narrow split. Positions are recovered by walking the selection in step with the emitted rows, and the two are checked against each other, so an over-long read fails loudly instead of returning a filtered answer.
• Absence means opposite things in the two formats. Java omits files its pre-filter did not narrow, while the search kernel reads a missing entry as "no rows". An omitted file therefore becomes an explicit full-file range; an explicitly empty list still excludes its file, and such a file is registered empty without being opened.
• Rejected rather than worked around: no splits, splits pinning different snapshots (checked before pruning, so an inconsistent input cannot hide behind an empty plan), two splits for one bucket, a nested data split carrying its own row ranges, and an unlisted file whose row count is unknown — reading ROW_COUNT_UNKNOWN as "no rows" would drop the file from the search silently.

A payload's deletion_vectors_ranges is ignored: Java reserves that field for deletion-vector index files and takes a read's deletion vectors from the data split.

What the range read saves is format-dependent, and the reader's contract says so: mosaic skips a row group before touching its column data, parquet skips pages through the offset index, .row prunes blocks. Avro is the exception — it loads the whole file and deserializes every record before applying the selection, so there a narrow selection saves only what comes after decoding. Making Avro prune physically needs a block-aware reader and is not part of this PR.

Deliberately not here:

• plan_for_bucket_vector_splits has no in-tree caller yet — the C entry point is the next step, and it carries #[allow(dead_code)] meanwhile. Happy to fold the caller in if you would rather not merge an uncalled entry point.
• An unrestricted split still materializes a whole-file allow-list for the search kernel, where Java returns no mask at all, which costs Lumina its batch search path. Saying "no mask" instead changes the plan's own type, so it belongs with the C entry point that first makes this route reachable.

[JingsongLi]

[P1] Apply the supplied physical ranges before scanning residuals

For an engine-supplied bucket split, plan.physical_row_ranges_by_split can restrict a large data file to a very small range. When a data predicate is also present, the code calls residual_positions_by_file first; that function invokes read_single_file_stream(..., None, None) and scans every physical row in each active file. Only here, afterward, are positions outside the supplied range discarded. Thus a 10-row split over a billion-row file turns into a full-file read for every residual-filtered vector query, defeating the distributed split and creating an extreme I/O/latency regression. Java evaluates the residual through an IndexedSplit built from the candidate ranges. Please pass the physical allow-list into residual evaluation, skip empty files, and read/evaluate only those ranges; a counting-reader test should assert that rows outside the engine range are never read.

[JunRuiLee]

Fixed in 3b9ba6f. The plan now carries the selection as normalized Vec<RowRange> rather than materialized positions, and both paths that touch a data file read through those ranges — the residual, and the exact fallback, which you did not mention but has the same defect and does not even need a predicate to hit it. Positions are recovered by walking the selection in step with the emitted rows and the two are checked against each other, so a read that overshoots fails loudly instead of quietly returning a filtered answer. A file the plan lists no rows for is registered empty without being opened.

DataFileReader already had a file-local ranges path — it coalesced positions into ranges internally — so it grew a ranges entry point rather than a new one. I did not use the generic row_ranges argument: that one translates through first_row_id, which is not the file-local space these ranges are in.

On the counting test: it is at the format layer, where the reading happens. A selection inside one row group requests strictly fewer byte ranges and fewer bytes than a full read. One caveat I should state rather than let you find — what the selection saves depends on the format. Mosaic skips a row group before touching its column data, parquet skips pages through the offset index, .row prunes blocks, but Avro loads the whole file and deserializes every record before applying the selection, so for an Avro data file the full read remains. Java allows file.format=avro for a vector column, so that is reachable. Making Avro prune physically needs a block-aware reader; I left it stated rather than half-done.

The intersection you pointed at stays. It cannot remove anything once the residual was evaluated over the same ranges, which is the invariant it now records.

[JingsongLi]

+1