feat(knn): probe core + no-shuffle join stage for indexed Lance vector join (Spark 4.2)

[sezruby]

What

The foundation of the indexed APPROX NEAREST join over Lance (SPARK-56395), as a new opt-in module lance-spark-knn-4.2_2.13: the JVM probe primitive plus the no-shuffle per-partition join stage that drives it. No Spark Catalyst / plan integration yet, so it reviews on its own.

Split out of the umbrella PR #796 for reviewability. The SQL Catalyst rewrite, physical operator, session extension, and end-to-end SQL/recall tests are the remaining follow-up (see below).

This was previously split further into [1/3] probe-core and [2/3] join-stage (#799); those are collapsed here so the probe primitive lands together with its only caller. #799 is closed in favor of this PR.

How it works

The indexed nearest-by join runs per Spark partition with no shuffle. Each task opens R's whole index (fragmentIds = None) and issues one native LanceProbe.probeRows(...) per left row; Lance does the cross-fragment IVF probe and heap-merge internally and returns the final top-k best-first, each hit already carrying its projected payload. There is no Exchange, no JVM-side cross-fragment merge, and no over-fetch / trim / second-materialize scan.

Contents

• LanceProbe — opens a Lance dataset once and serves per-query nearest searches against a fixed fragment set. probeRows folds the nearest search and payload projection into a single native scan, returning each hit's row-id, ranking score, and materialized payload (via the connector's canonical Arrow→Spark adapter). Handles 64-bit-unsigned row ids. A schema-level guard declines any table whose own schema owns a reserved column (_rowid / _distance / _score), which the injected metadata would otherwise shadow.
• LanceKnnJoinStage — the per-partition execution: runPartition opens one probe per partition, streams the join output lazily (no per-partition buffering), and closes the native handle on task completion (success or failure). resolveReadContext / mergeReadOptions pin the Lance snapshot once on the driver so every task probes one consistent version (mirroring the connector's incompatible-pinned-ref guard). coerceToSpark shapes payloads (structs, arrays, and the Arrow entry-list form a MapType cell arrives in) to what the join's encoder expects.
• Metric / MaterializedHit — small value types.

Tests

Run standalone. The probe validation suite writes a real Lance dataset with Spark and needs no vector index — Lance's brute-force scan is an exact recall=1.0 oracle, isolating probe correctness from index quality:

• LanceProbeValidationTest — probe result shape, brute-force-oracle equivalence, metric-direction correctness, unmapped-projection preservation, the schema-eligibility contract and reserved-column decline, dataset-handle reuse across probes, fragment-id restriction, and the executor namespace policy.
• LanceKnnJoinStageTest (backend-free) — lazy streaming output, read-option merge conflicts, and schema-aware payload coercion.

Follow-up

• [3/3] the Catalyst APPROX NEAREST interception rule, physical operator, session extension, and end-to-end SQL / recall tests + docs.

Note: this module is downstream of the connector modules, so the default -am connector build does not exercise it; build it with ./mvnw -pl lance-spark-knn-4.2_2.13 test (connector artifacts resolved from the local repo).

🤖 Generated with Claude Code

[github-actions]

ACTION NEEDED
Lance follows the Conventional Commits specification for release automation.

The PR title and description are used as the merge commit message. Please update your PR title and description to match the specification.

For details on the error please inspect the "PR Title Check" action.

[sezruby]

@geruh @yanghua @LuciferYang — could I get a maintainer review on this when you have a moment?

Why this exists: the current path to KNN entries in Spark is a JVM cross join + distance compute. That's only viable on small datasets, and even there it's very slow; at larger scale it doesn't complete at all — brute-force cross join isn't a feasible way to get nearest neighbors at any real size. This series replaces it with Lance's native indexed vector search: LanceProbe opens the dataset once and serves per-query nearest searches through the native API, and probeRows folds the search and the payload projection into a single native scan so materialization happens natively too (no overfetch, no JVM-side join). It's substantially faster.

The offload is gated on index availability, with a dataset-size gate planned as well.

This PR: slice 1 of 3 of the indexed APPROX NEAREST join (SPARK-56395), split out of #796 to be reviewable on its own. It's the JVM probe-core layer only — LanceProbe / TopKHeap / value types — with no Spark plan integration, so it stands alone.

Status: CI is green across Spark 3.4–4.2, and the Lance Gatekeeper's recommendation is approve with a non-blocking risk (the _score reserved-name guard is conservative vs. the current _distance path). Mergeable, no conflicts. Follow-ups [2/3] (LanceKnnJoinStage) and [3/3] (the Catalyst rule + e2e tests) build on top of this.

[LuciferYang]

I’m missing some context. Is this a feature that was already discussed and agreed to add?

[lance-gatekeeper]

🟡 Gate recommendation: maintainer decision required.

This revision folds the no-shuffle join stage from #799 into the foundation, but it does not resolve the open feature-agreement question. The scope record in #798 remains an author proposal rather than a verified maintainer decision.

Accepting the direction would land the probe with its only caller while Catalyst integration follows; deferring avoids maintaining an otherwise unused Spark 4.2 module. Maintainers should choose that scope explicitly. The prior probe-core _score fallback risk remains bounded, but it does not settle this product choice.