feat: indexed nearest-by join over Lance — Spark 4.2 SQL Catalyst - #14
Closed
sezruby wants to merge 11 commits into
Closed
feat: indexed nearest-by join over Lance — Spark 4.2 SQL Catalyst#14sezruby wants to merge 11 commits into
sezruby wants to merge 11 commits into
Conversation
…nce-format#528) ## Lance dependency — required to compile * Depends on lance-format/lance#6748, which adds `Update.Builder.updatedFragmentOffsets(...)` / `Update.updatedFragmentOffsets()` and JNI FromJava + IntoJava so the driver commit passes matched offsets into Rust. Bump **`lance.version`** in `pom.xml` to a release that includes lance#6748 before building or merging this PR; otherwise build fails on the new API calls. ## Summary * Fixes lance-format#418. * `UpdateColumnsWriter.processFragment`: after each `fragment.updateColumns()` call, reads `result.getUpdatedRowOffsets()` and accumulates a `Map<Long, long[]>` of fragment id → matched physical row offsets. * `TaskCommit`: carries the per-fragment offset map alongside the existing `updatedFragments` and `fieldsModified` fields. * `UpdateColumnsBackfillBatchWrite.commit()`: merges offset maps from all task commits and passes them to `Update.builder().updatedFragmentOffsets(...)`. Lance's `build_manifest` then calls the partial `_row_last_updated_at_version` refresh only for the matched rows, leaving unmatched rows and untouched fragments unchanged. * `BaseUpdateColumnsBackfillTest`: flips `testUpdateColumnsPreservesCreatedAtAndAdvancesLastUpdatedWithStableRowIds` from the "known gap" pin (assertEquals, no change) to the correct assertion (`assertTrue(after > before)`); updates Javadoc accordingly. ## Background `UPDATE COLUMNS FROM` rewrites column data in place via Lance's `Operation::Update` with `RewriteColumns` mode. Lance's `build_manifest` can partially refresh `_row_last_updated_at_version` for only the matched rows — but only when the `updated_fragment_offsets` map is non-empty on the commit. Previously `UpdateColumnsBackfillBatchWrite` never populated this map, so the partial refresh never activated and `_row_last_updated_at_version` stayed stale after every UPDATE COLUMNS commit, breaking CDF consumers. The matched row offsets are already computed inside Lance during `fragment.updateColumns()` and surfaced via `FragmentUpdateResult.getUpdatedRowOffsets()` (lance#6650). The missing piece was wiring those offsets from the executor task result through `TaskCommit` to the driver commit, and then setting them on the `Update` operation — which this PR does. ## Test plan * `BaseUpdateColumnsBackfillTest#testUpdateColumnsPreservesCreatedAtAndAdvancesLastUpdatedWithStableRowIds` — creates a stable-row-id table, runs UPDATE COLUMNS over all rows, and asserts `_row_last_updated_at_version` strictly increases for each row while `_row_created_at_version` is unchanged. --------- Co-authored-by: Jing chen He <jingh@adobe.com> Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com> Co-authored-by: Daniel Rammer <hamersaw@protonmail.com>
… raw version number (lance-format#770) Related lance-format#713 ### Description This PR refactors Spark read and write options to use LanceRef instead of passing around raw version numbers directly. The goal is to make reference handling more explicit and extensible, especially for future branch and tag support. The change updates the Spark option model and related call sites to carry reference information through the connector, while preserving the ability to resolve snapshot versions where needed. This lays the groundwork for consistent handling of main, versioned, branch, and tag references across read, scan, blob/search, and write paths. ### Summary 1. Replace raw version-based Spark options with LanceRef. 2. Update read/write option APIs and affected Spark call sites. 3. Align test utilities with the new reference model. 4. Prepare the connector for branch/tag-aware reference handling. --------- Co-authored-by: fangbo <fangbo.0511@bytedance.com>
Related lance-format#713 # Motivation In some use cases, users need to run DQL against a specific tag rather than relying on the default resolution behavior. Tags are read-only. Tags could just stay on time travel. # How to query ```SQL SELECT * FROM lance.users VERSION AS OF 'specific_tag'; ``` # Changes * Add support for executing DQL against a specific tag. * Enable more explicit tag-scoped reads in query workflows. * Improve flexibility for scenarios where users need to access data from a designated tag. # Testing * Added/updated tests for DQL queries with a specific tag. * Verified that existing query behavior remains unchanged when no tag is specified. * Validated the new tag-specific query path works as expected. --------- Co-authored-by: fangbo <fangbo.0511@bytedance.com>
…t#777) Branch identifiers were failing on Glue because we first tried to resolve the full branch shaped identifier as a literal table. Glue only accepts database.table, so that lookup failed before we could fall back to the parent table and branch. This changes branch identifier resolution so the literal table lookup is just a probe. If it does not resolve as a real table, we load the parent table and apply the branch instead. Literal tables still win on catalogs that support them. The collision test is skipped on Glue since Glue cannot represent that table shape.
## Summary - Balance scalar index fragment batches using fragment row counts. - Keep assignments deterministic and document the non-contiguous batching behavior. ## Testing - Spark 3.5 / Scala 2.12: AddIndexTest (48 tests) - Scala 2.12 and 2.13: IndexUtilsTest (26 tests each) Closes lance-format#757
…ormat#766) Closes lance-format#765. ## What `LanceScanBuilder.pushAggregation` decided `COUNT(*)` could be answered from `ManifestSummary.getTotalRows()` whenever no predicate was pushed, without checking `readOptions.getFullTextQuery()`. Because the FTS rule moves the predicate out of the `Filter` and into the relation options, a query like ```sql SELECT count(*) FROM t WHERE lance_match(body, 'hello') ``` has no `Filter` and no pushed predicate, so it returned the table's total row count instead of the number of FTS matches. The same omission in `LanceCountStarPartitionReader.computeCount` made the scan-based count path (FTS combined with a pushable scalar filter) ignore the FTS query too. `LanceScan.pruneByLimit` already guards on `getFullTextQuery() != null` for exactly this reason; both count paths now do the same. The count scan also opts out of Lance's `_score` autoprojection, which an empty column list would otherwise trigger once per task; the row-scan path deliberately keeps that autoprojection, since it is how `_score` is delivered. ## Scope This fixes the `COUNT(*)` that pushes down. `build()`'s namespace FTS branch requires no pushed aggregation, so a pushed-down `COUNT(*)` is never routed to a namespace's `queryTable` and is now answered by the local per-fragment scan on every namespace. On catalog-only namespaces (Glue/Hive/Iceberg) the row query already applied the predicate through that same scan, so the count now agrees with it. Two cases stay wrong and are out of scope, both recorded under Known Limitations in `docs/src/operations/dql/fts.md`: - on namespaces served by `queryTable` — `dir`, and REST implementations that skip a structured full-text query — the row query still returns every row, so it can disagree with the count; - a `COUNT(*)` whose other filters cannot push down leaves `pushedAggregation` empty, which routes it through `queryTable` like a row query and leaves it unfiltered there as well. ## Tests Three cases in `BaseFtsCatalogOnlyNamespaceTest` cover the metadata shortcut (matching and non-matching terms) and the scan-based path; all three fail on the unfixed tree (20/20/10 instead of 15/0/5). The first also asserts the executed plan's scan output list is `[count#N]`, which ties the count to the scan rather than to re-aggregated rows — matching only the word `count` does not, since the un-pushed plan contains `count(1)` as well. A unit test in `LanceScanBuilderTest` covers the other half: with a full-text query set, `build()` must return `LanceScan`, not `LanceLocalScan`. The fixture's javadoc claimed "Two fragments" while each `INSERT` writes one fragment per write task (four under `local[4]`, so eight in total, 2-3 rows each). That comment misled two rounds of analysis during review, so it is corrected here to state the real layout and warn that per-fragment pushdown reasoning must use eight. ## Follow-up from review Routing FTS-only `COUNT(*)` through `LanceCountStarPartitionReader` exposed a pre-existing gap: unlike `LanceFragmentScanner.create()`, `computeCount()` opened the dataset with only `initialStorageOptions` and never rebuilt the namespace, so on namespaces that vend short-lived credentials (Iceberg REST, Polaris, Unity) a long FTS count could fail after the driver-fetched credentials expired. `f1086cc` applies the same enabled/disabled refresh gate before the open, keeping the `executor_credential_refresh=false` opt-out that Kerberized HMS catalogs rely on, with `LanceCountStarPartitionReaderTest` covering both flag values. Verified on `lance-spark-3.4_2.12`: full suite 1196 run, 0 failures, 113 skipped (the skipped set includes three FTS suites disabled via `assumeTrue(false)`).
## Summary
- apply the target table's persisted column properties and actual Lance file format version to `ADD COLUMNS ... FROM` backfill schemas
- support adding and backfilling a blob v2 `BINARY` column on file format 2.2+ tables
- validate configured blob columns are `BINARY` before append
- document the supported workflow and add integration coverage for success and error paths
## Usage
Persist the future column's blob encoding before running the backfill:
```sql
CREATE TABLE users (
id INT,
name STRING
) USING lance
TBLPROPERTIES ('file_format_version' = '2.2');
ALTER TABLE users
SET TBLPROPERTIES ('content.lance.encoding' = 'blob');
CREATE TEMPORARY VIEW content_backfill AS
SELECT _rowaddr, _fragid, CAST(name AS BINARY) AS content
FROM users;
ALTER TABLE users ADD COLUMNS content FROM content_backfill;
```
After the operation, reads expose `content` as the existing blob v2 descriptor struct, including fields such as `size` and `kind`.
## Why
The add-columns backfill path previously used the source view's raw Spark schema. As a result, a new `BINARY` column did not receive the `lance.blob.v2` Arrow extension metadata from the target table's persisted properties and was written as plain binary.
Using `ALTER TABLE ... SET TBLPROPERTIES` ensures the future column's encoding is persisted for both managed and unmanaged catalogs before `ADD COLUMNS` reloads the target table.
## Validation
- GitHub Actions: all checks passed
- integration coverage verifies blob v2 descriptor values after a successful `BINARY` backfill
- integration coverage verifies non-`BINARY` inputs fail before the invalid column is added
- existing non-blob add-columns scenarios remain covered
Local Maven, pytest, lint, and format checks were intentionally not run per the Lance repository workflow; GitHub Actions was used as the validation environment.
Accelerate Spark 4.2's `APPROX NEAREST k BY DISTANCE ...` join
(SPARK-56395) over Lance tables by intercepting the `NearestByJoin`
operator and running a per-row native ANN/exact probe against the
Lance vector index instead of the default O(|L|x|R|) cross-product +
MaxMinByK lowering.
New opt-in module `lance-spark-knn-4.2_2.13`:
- Catalyst postHocResolutionRule (`IndexedNearestByJoinRule`) that
recognizes `NearestByJoin` and rewrites it to a single no-shuffle
logical node; must run in postHoc (not the optimizer) because
Spark's `RewriteNearestByJoin` fires first in `FinishAnalysis`.
- Planner strategy + exec (`LanceKnnJoinStrategy`/`LanceKnnJoinExec`)
lowering to one `mapPartitions` -> per-left-row native probe ->
bounded top-K heap -> late materialize by row address. No shuffle,
no broadcast.
- Probe core (`LanceProbe`/`LanceKnnJoinStage`/`TopKHeap`/`Metric`)
supporting L2, cosine, and dot metrics with `nprobes`/`refineFactor`
tuning.
- Right-side WHERE pushdown as a Lance prefilter, including nested
struct-field predicates rendered as dotted paths (`col.field`).
Conservative: if any sub-expression is unsupported the whole
rewrite is refused and Spark's brute-force lowering runs unchanged.
Off by default (`spark.lance.knn.indexedNearestByJoin.enabled=false`);
wired via `LanceKnnSparkSessionExtensions`. Part of the effort tracked
in lance-format#541.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Owner
Author
|
Superseded by the upstream PR lance-format#796 (same branch, rebased onto upstream main and squashed). Continuing review there. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
7 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Adds a single module,
lance-spark-knn-4.2_2.13, that transparently accelerates Spark 4.2's nativeAPPROX NEAREST k BY DISTANCE ...join (theNearestByJoinoperator from SPARK-56395) over a Lance scan.What it does
Spark's default rewrite lowers
NearestByJointo an O(|L|×|R|) cross-product +MaxMinByK. This module rewrites it onto a no-shuffle, per-partition native probe against a Lance vector index instead.IndexedNearestByJoinRule— a post-hoc resolution rule (the only injection point that sees the unrewrittenNearestByJoin; Spark's ownRewriteNearestByJoinruns first, in theFinishAnalysisbatch) rewrites the operator into a singleLanceKnnJoinLogicalPlan.LanceKnnJoinStrategylowers that toLanceKnnJoinExec, which drives onemapPartitions→ per-rowLanceProbe(native ANN/exact search) → bounded top-K → late materialize by row address. No shuffle, no broadcast.spark.lance.knn.indexedNearestByJoin.enabled(off by default). When disabled, the query falls through to Spark's built-in rewrite.WHEREis translated to a Lance prefilter, or the rewrite is refused when the predicate can't be pushed in full — never a silently dropped residual.nprobes/refineFactorare tunable viaspark.lance.knn.*configs.The module carries its own probe core (
LanceKnnJoinStage/LanceProbe/TopKHeap/Metric); it depends onlance-spark-baseforLanceRuntimeand on the Spark 4.2 connector at test scope.Tests
WHEREprefilter pushdown and the opt-in gating.refineFactorhelps).Notes
spark.ymlrunsmvn test -pl lance-spark-<spark>_<scala> -am) builds the connector modules and their upstream deps only; this module is downstream, so it is not built by that matrix. Verifying it needs a reactor build (e.g../mvnw test -pl lance-spark-knn-4.2_2.13 -am).df.kNearestJoinDataFrame API is deferred — it can be added later as a Spark-version-agnostic base module when there's demand.🤖 Generated with Claude Code