feat(gluten): export a Lance fragment scan as an Arrow C stream - #12
Closed
sezruby wants to merge 8 commits into
Closed
feat(gluten): export a Lance fragment scan as an Arrow C stream#12sezruby wants to merge 8 commits into
sezruby wants to merge 8 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.
The Arrow C stream export path added in the following commit calls LanceScanner#exportArrowStream(long), which is first available in lance-core 11.0.0-beta.21 (lance-format/lance#7259). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Add LanceArrowStreamScanner, which plans a fragment scan via the existing LanceFragmentScanner and exports it as an Arrow C Data Interface stream (ArrowArrayStream) for native consumers such as Apache Gluten / Velox. Only the ArrowArrayStream C-struct address crosses the JVM/native boundary, so the consumer's Arrow build and classloader do not need to match lance-spark's (Gluten builds Arrow 15 to match Spark 3.5 / Velox, while the Lance Java SDK is on Arrow 18). The Lance native core populates the caller-owned stream directly through LanceScanner#exportArrowStream(long), so no Arrow data is materialized on the JVM heap on this path. All scan planning (column projection, filter pushdown, limit/offset, row-id / row-address, batch size) is delegated to LanceFragmentScanner, so the exported stream yields exactly the same rows in the same order as the Spark columnar reader. LanceArrowStream owns the exported stream and the scan behind it; closing it releases the native scan (via the stream's release callback) and then the scanner and dataset handles. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
sezruby
force-pushed
the
feat/gluten-arrow-stream-forwarding
branch
from
August 23, 2026 16:09
2a7f762 to
d94b5c6
Compare
Owner
Author
|
Superseded by the upstream PR lance-format#778, which uses the direct LanceScanner#exportArrowStream(long) (lance-core 11.0.0-beta.21) instead of the double-hop staged here. |
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
4 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.
What
Adds
LanceArrowStreamScannerto the base module: it plans a Lance fragment scan through the existingLanceFragmentScannerand re-exports the scan'sArrowReaderas an Arrow C Data Interface stream (ArrowArrayStream).Only the stream's C-struct address (
LanceArrowStream#streamAddress()) crosses the JVM/native boundary, so a native consumer such as Apache Gluten / Velox does not need to match lance-spark's Arrow build or classloader.Why
This is step 2 of the JVM Arrow-in/Arrow-out read path for Gluten support (apache/gluten#12263, Approach 2): lance-spark hands a native engine an Arrow C stream over a planned fragment scan. A gluten-side scan-offload rule that consumes
streamAddress()is the remaining piece.Details
LanceFragmentScanner, so the exported stream yields the same rows in the same order as the Spark columnar reader.LanceArrowStreamhandle owns the exported stream and the scanner behind it.exportArrayStreamtransfers ownership of the reader to the stream, soclose()releases the stream (which closes the reader via the C release callback) and then the scanner and dataset — it never closes the reader itself.TODO(lance#7259)marks a one-line swap toLanceScanner#exportArrowStream(long)once that lands upstream: the Rust core would populate the caller's stream directly and skip the JVM-side Arrow materialization. TheLanceArrowStreamcontract is unchanged, so no consumer needs to be touched.Test
LanceArrowStreamScannerTestexports each fragment of the bundled test table, re-imports it viaData.importArrayStream(standing in for a native consumer), drains it, and asserts the rows match the Spark columnar reader. Closing the imported reader and then the handle under the leak-checkingRootAllocatoralso verifies the export/reader/scanner lifecycle releases cleanly.🤖 Generated with Claude Code