[core] Add primary-key vector read kernel

[JunRuiLee]

Purpose

Linked issue: part of #514 — this PR implements the crate-private PK-vector read kernel / orchestration layer. It does not close the issue.

This is the Rust read-kernel counterpart of Apache Paimon Java apache/paimon#8576 (PrimaryKeyVectorPositionReader / PrimaryKeyIndexedSplitRead) plus the read-kernel subset of apache/paimon#8579 (PrimaryKeyVectorRead + PrimaryKeyVectorResult.splits()). It is intentionally not the full user-facing PK-vector search integration.

Given synthetic per-bucket search inputs, this PR can:

• materialize selected physical rows from one data file while preserving physical positions and aligned scores;
• validate and read one-file indexed splits over physical-position ranges;
• run bucket-local search via the already-merged bucket kernel ([core] Add primary-key vector bucket search kernel #516), merge deterministic global Top-K candidates across buckets, group survivors per data file, and build indexed splits for materialization.

#516 has already absorbed the current Java-master behavior for inactive ANN sources (file == null -> continue) and bounded heap merging, so this PR builds on that behavior. The inactive-source handling is therefore intentional and aligned with current Java master, not a fail-loud mismatch.

Scope

In scope:

1. pk_vector_position_read.rs — Rust equivalent of Java PrimaryKeyVectorPositionReader.
2. pk_vector_indexed_split_read.rs — Rust equivalent of Java PrimaryKeyIndexedSplitRead.
3. pk_vector_orchestrator.rs — synthetic-input orchestration mirroring the read-kernel part of Java PrimaryKeyVectorRead / PrimaryKeyVectorResult.splits().
4. VectorSearchMetric::distance_to_score, used when converting final distances into Java-compatible higher-is-better scores.

Out of scope for this PR, handled by the follow-up integration PR:

• PrimaryKeyVectorScan / snapshot and index-manifest planning;
• VectorSearchBuilder / table routing and public API wiring;
• construction of real ANN and exact-reader dependencies;
• real vindex ANN segment bytes and end-to-end public-path validation;
• GlobalIndexSearchMode::Fast routing behavior;
• refine/rerank from newer Java master.

Because there is no crate-level production caller for these new reader/orchestrator modules yet, they intentionally keep narrow #[allow(dead_code)] annotations. The follow-up integration PR wires the production caller and removes the allowances for the wired search path.

Brief Change Log

pk_vector_position_read.rs

Rust equivalent of Java PrimaryKeyVectorPositionReader.

• Reads selected 0-based physical positions from a single data file.
• Appends internal metadata columns:
  • _PKEY_VECTOR_POSITION (Int64, always);
  • _PKEY_VECTOR_SCORE (Float32, only when scores are supplied).
• Derives returned physical positions from the existing _ROW_ID channel (position = _ROW_ID - first_row_id) instead of guessing from batch offsets.
• Injects _ROW_ID internally and strips it from final output.
• Preserves null/deletion-vector physical-position alignment by using the actual rows returned by read_single_file_stream.
• Rejects reserved metadata-column name conflicts, requested _ROW_ID, missing first_row_id, predicate-bearing readers, empty positions, out-of-range positions, and score-key mismatches.

pk_vector_indexed_split_read.rs

Rust equivalent of Java PrimaryKeyIndexedSplitRead.

• Adds PkVectorIndexedSplit: one data file + inclusive physical-position ranges + optional scores aligned to expanded position order.
• Validates single-file input, range bounds, range ordering/non-overlap, non-empty selection, and score length.
• Expands ranges to physical positions and delegates materialization to PkVectorPositionRead.
• Rust intentionally fails loud on descending/overlapping physical ranges to protect score alignment. Java normalizes through a bitmap; this is a stricter validation, not a capability regression.

pk_vector_orchestrator.rs

Rust equivalent of the synthetic-input read-kernel part of Java PrimaryKeyVectorRead and PrimaryKeyVectorResult.splits().

• Calls [core] Add primary-key vector bucket search kernel #516 bucket_search for each synthetic bucket split.
• Merges candidates with a deterministic 5-level BEST_FIRST key:
  1. distance ascending;
  2. partition bytes unsigned lexicographic;
  3. bucket ascending;
  4. data file name ascending;
  5. row position ascending.
• Converts final distances to Java-compatible scores via VectorSearchMetric::distance_to_score.
• Groups survivors by data file into PkVectorIndexedSplits for materialization.
• Fails loud on cross-split ambiguity, duplicate (file, position), or a hit referencing a file absent from its bucket split.
• The materialized stream is file/position ordered, matching the indexed-split materialization contract; ranking order is preserved in the candidate list before grouping.

Validation

Synthetic unit/e2e tests cover:

• position/score alignment using real _ROW_ID values;
• null rows and deletion-vector interaction;
• indexed split validation and score alignment;
• global Top-K tie-breaks across buckets/files/positions;
• distance-to-score conversion;
• inactive ANN-source behavior inherited from [core] Add primary-key vector bucket search kernel #516;
• error paths for malformed inputs.

Validated locally:

• cargo fmt --all --check
• cargo clippy -p paimon --lib --tests -- -D warnings
• cargo test -p paimon pk_vector --lib

Full public-path validation with snapshot/index-manifest planning and real ANN/exact dependencies is intentionally deferred to the follow-up integration PR.

API and Format

No storage-format change. All additions are crate-private (pub(crate) / pub(super)); no public API is added or changed in this PR.

Documentation

No user-facing documentation changes.

[JingsongLi]

pk_vector_orchestrator.rs:86 uses f32::total_cmp to simulate Java sorting, but negative NaNs are sorted before finite values; after taking the negative of the distance to NaN in the inner product, an anomalous row may be selected as Top-1, whereas Java’s Float.compare places NaNs at the end. It is recommended to implement a Java-style NaN comparator and add signed-NaN tests. The existing pk_vector passes 68 out of 68 tests, but this scenario is not covered.

[JunRuiLee]

@JingsongLi good catch. I fixed this with a shared Java-compatible float comparator (java_float_compare: NaN sorts after all numeric values, non-NaN keeps total_cmp so -0.0 < +0.0) instead of using f32::total_cmp directly. The fix is applied across the full pk-vector Top-K path — ANN result ordering, bucket heap ordering, exact-search heap/final ordering, and orchestrator merge — because fixing only the orchestrator cannot recover candidates already dropped by the bucket/exact heaps. Tests now cover negative-NaN ordering, and both exact- and bucket-heap Top-1 behavior (a NaN-distance candidate never beats a finite one).