feat(gluten): export a Lance fragment scan as an Arrow C stream by sezruby · Pull Request #778 · lance-format/lance-spark

sezruby · 2026-08-23T16:09:57Z

What

Adds LanceArrowStreamScanner, which plans a Lance fragment scan and exports it as an Arrow C Data Interface stream (ArrowArrayStream) for native consumers such as Apache Gluten / Velox.

Why

This is the read-side building block for offloading Lance scans to a native engine. 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. Passing the raw struct address sidesteps that mismatch entirely.

How

LanceArrowStreamScanner.export(fragmentId, inputPartition) returns a LanceArrowStream handle exposing stream() / streamAddress().
All scan planning (column projection, filter pushdown, limit/offset, row-id / row-address, batch size) is delegated to the existing LanceFragmentScanner, so the exported stream yields exactly the same rows in the same order as the Spark columnar reader.
The Lance native core populates the caller-owned stream directly via LanceScanner#exportArrowStream(long) (feat(java): expose ArrowArrayStream export on LanceScanner lance#7259), so no Arrow data is materialized on the JVM heap on this path.
LanceArrowStream owns the exported stream and the scan behind it; closing it releases the native scan (through the stream's release callback) and then the scanner and dataset handles.

Dependency

exportArrowStream(long) first ships in lance-core 11.0.0-beta.21, so this bumps lance.version from 11.0.0-beta.10 (isolated in its own commit).

Testing

New LanceArrowStreamScannerTest: exports each fragment of the bundled test table, re-imports it on the JVM (standing in for a native consumer), and asserts the rows match the columnar reader — run under the leak-checking allocator to verify the export/scanner lifecycle releases cleanly.
Existing LanceFragmentColumnarBatchScannerTest still passes (regression on the touched LanceFragmentScanner).

Complements feat: add task commit factory for Apache Gluten writes #715 (Gluten write path).
Distinct mechanism from Expose native Lance scan descriptor for datafusion-comet integration #624 / Expose a stable native scan descriptor for Lance Spark reads and datafusion-comet integration #623 (serializable native scan descriptor for datafusion-comet): this exposes a live Arrow C stream handle rather than a re-plannable descriptor.

🤖 Generated with Claude Code

sezruby · 2026-08-23T17:06:45Z

Addressed all three findings in e914259.

Release callback (line 119). LanceArrowStream.close() now runs the C stream release callback via a guarded releaseStream() before freeing the struct, so an abandoned or partially-consumed stream drops the provider's private_data and record-batch stream. It is idempotent against a consumer that already imported the stream: Data.importArrayStream snapshots the callback and closes our struct, so releaseStream() skips when snapshot() reports the struct is freed or the release address is NULL per the C ABI. Added closeReleasesStreamThatWasNeverImported, which closes the handle with no JVM import — under the leak-checking allocator it fails if the callback or struct is left dangling.

Schema contract (line 60). export() now calls checkExportableSchema() first and rejects, with UnsupportedOperationException, the shapes the columnar reader fixes up on the JVM after import: the synthesized _fragid, an empty projection (which surfaces the internal _rowid), metadata columns (_rowid / _rowaddr / row-version / _score), and blob columns. Those cannot be reproduced on the zero-copy native path without materializing on the JVM heap, so the caller falls back to the columnar reader instead of receiving a wrong schema. Ordinary data projections still export a stream that matches the partition schema (Lance has no Hive-style partition columns). Added exportRejectsSchemasNeedingJvmPostProcessing.

Construction leak (line 54). ArrowArrayStream.allocateNew now runs inside the cleanup scope, so a throwing allocation (e.g. a bounded allocator) closes both the stream and the scanner/dataset opened by LanceFragmentScanner.create().

LanceArrowStreamScannerTest (3/3) and the LanceFragmentColumnarBatchScannerTest regression (1/1) pass; spotless:check is clean.

sezruby · 2026-08-23T19:09:04Z

Addressed the pushed-aggregation finding in f78423b.

checkExportableSchema is now checkExportablePartition, which rejects a partition with a present pushedAggregation (e.g. COUNT(*)) before the schema checks. Such a partition is served by a dedicated reader (LanceCountStarPartitionReader) that returns a single count column, whereas LanceFragmentScanner ignores pushedAggregation and scans data rows — so the raw export would emit the wrong output. Filter, limit, offset, and top-N ordering stay exportable because they are pushed faithfully into the native scan.

Added exportRejectsPushedAggregation (your reproducer's COUNT(*) shape). LanceArrowStreamScannerTest 4/4, LanceFragmentColumnarBatchScannerTest 1/1, spotless:check clean.

sezruby · 2026-08-23T19:26:56Z

Addressed the full-text finding in 33c2788, and generalized the admission check so this class of divergence can't recur.

Rather than enumerate more column-name cases, admission now verifies the actual native scan schema against the declared partition schema. After planning the scan, checkNativeSchemaMatchesPartition compares LanceScanner#schema() (surfaced via a new LanceFragmentScanner#schema()) to inputPartition.getSchema() by field name and order, and rejects on any mismatch. A full-text query's auto-projected _score therefore falls out as [id, body] vs [id, body, _score] → reject. The same guard covers the synthesized _fragid, the _rowaddr added for blobs, the _rowid an empty projection surfaces, and reordered row-version columns, since the zero-copy export cannot re-project or reorder the way the columnar reader does after import.

Pushed aggregation stays a separate pre-plan check: an aggregate partition's declared schema can equal the native data schema, so the schema comparison alone cannot catch it.

exportRejectsSchemasNeedingJvmPostProcessing now exercises both mismatch directions — native returns fewer columns (_fragid) and native returns an extra column (empty projection surfacing _rowid, the same shape as the full-text _score). A _rowid column the native scan produces in matching order is now legitimately exportable, so that assertion was dropped. LanceArrowStreamScannerTest 4/4, LanceFragmentColumnarBatchScannerTest 1/1, spotless:check clean.

sezruby · 2026-08-25T16:09:05Z

@yanghua could you take a look when you have a chance? This is the lance-spark side of the Gluten/Velox read path — it forwards a Lance fragment scan as an Arrow C stream via LanceScanner#exportArrowStream(long) (the API from #7259, now shipping in lance-core 11.0.0-beta.21, which this PR bumps to).

CI is green and it's MERGEABLE; unit tests pass (LanceArrowStreamScannerTest 4/4, LanceFragmentColumnarBatchScannerTest 1/1). cc @jackye1995

yanghua · 2026-08-26T03:47:15Z

CI is red?

sezruby · 2026-08-26T05:34:55Z

The red CI isn't from this PR's changes; it's the beta.10 → beta.21 lance-core bump exposing a pre-existing issue in the Kryo codec (LanceSerializeUtil), unrelated to the Arrow-stream forwarding.

LanceSerializeUtil configures Kryo with an Objenesis instantiator (StdInstantiatorStrategy), which constructs objects without calling their constructors. The JDK immutable collections (List.of/copyOf, Set.of, Map.of) have no no-arg constructor and keep their elements in a private java.base-internal field, so Kryo produces an instance whose backing array is never populated. beta.21's CompactionOptions.excludedFragmentIds is a List.copyOf(...), and OptimizeExec ships each CompactionTask through this codec to the executors — so it fails to deserialize there:

java.lang.NullPointerException: Cannot read the array length because "this.elements" is null

That's why OptimizeTest, VacuumTest, and ShowIndexesTest fail across every Spark/Scala combination — it's version-independent.

I've split the fix into #783: it registers scoped Kryo CollectionSerializer/MapSerializer implementations for the immutable-collection base classes (ported from Kryo 5's ImmutableCollectionsSerializers) that read each element through Kryo and rebuild the collection via its public factory (List/Set/Map.copyOf), so they roundtrip correctly without routing decode through ObjectInputStream. It adds a regression test, and bumps lance-core to beta.21 so the same OptimizeTest/VacuumTest/ShowIndexesTest run against the version that reproduces the failure. #783 is from a fork, so its Test / Integration Test workflows are waiting on approval — could you trigger them (Approve and run)? Once #783 merges I'll rebase this PR onto it (the bump here becomes a no-op) and CI will be green.

yanghua · 2026-08-26T06:29:55Z

The red CI isn't from this PR's changes; it's the beta.10 → beta.21 lance-core bump exposing a pre-existing issue in the Kryo codec (LanceSerializeUtil), unrelated to the Arrow-stream forwarding.

LanceSerializeUtil configures Kryo with an Objenesis instantiator (StdInstantiatorStrategy), which constructs objects without calling their constructors. The JDK immutable collections (List.of/copyOf, Set.of, Map.of) have no no-arg constructor and keep their elements in a private java.base-internal field, so Kryo produces an instance whose backing array is never populated. beta.21's CompactionOptions.excludedFragmentIds is a List.copyOf(...), and OptimizeExec ships each CompactionTask through this codec to the executors — so it fails to deserialize there:
java.lang.NullPointerException: Cannot read the array length because "this.elements" is null
That's why OptimizeTest, VacuumTest, and ShowIndexesTest fail across every Spark/Scala combination — it's version-independent.

I've split the fix into #783: it registers Kryo's JavaSerializer as the default serializer for the immutable-collection base classes so they roundtrip correctly, adds a regression test, and bumps lance-core to beta.21 so the same OptimizeTest/VacuumTest/ShowIndexesTest run against the version that reproduces the failure. #783 is from a fork, so its Test / Integration Test workflows are waiting on approval — could you trigger them (Approve and run)? Once #783 merges I'll rebase this PR onto it (the bump here becomes a no-op) and CI will be green.

Got it. Will take a look.

yanghua

Left two comments.

yanghua · 2026-08-26T12:02:10Z

+   * projection, reordering row-version columns — but this zero-copy export cannot, so any mismatch
+   * in field names or order must fall back to the columnar reader.
+   */
+  private static void checkNativeSchemaMatchesPartition(


Can we also check the arrow data type?

Done in 235791d. checkNativeSchemaMatchesPartition now compares the declared and native schemas by Arrow ArrowType and nullability (recursing into children), not only by field name and order. The declared Spark schema is converted through LanceArrowUtils — the same adapter the read path uses — so the Arrow-specific distinctions a Spark DataType alone cannot express (LargeUtf8 vs Utf8, LargeBinary, Date(MILLISECOND), FixedSizeBinary, Float16) survive the comparison, and a column whose native type differs from the declared one (e.g. a narrower/wider int or a different time-zone timestamp) is no longer streamed as if it matched.

yanghua · 2026-08-26T12:03:47Z

+   * verifies the export/reader/scanner lifecycle releases cleanly.
+   */
+  @Test
+  public void exportsFragmentAsArrowCStream() throws Exception {


Can we also check a case about an empty fragment?

Added exportsEmptyFragmentScanPreservingSchema in 235791d: a fragment scan whose filter matches no rows (x < 0, every x is 0..3) still exports its full x, y, b, c schema up front and drains cleanly with zero batches, releasing without a leak under the leak-checking allocator.

yanghua · 2026-08-27T03:04:42Z

@sezruby, do you have WeChat? We can contact each other more effectively.

sezruby · 2026-08-27T06:02:52Z

@sezruby, do you have WeChat? We can contact each other more effectively.

I don't have WeChat, but feel free to reach me on the ASF Slack (EJ Song) or LinkedIn. And I can join WeChat if there's a group chat for Lance.

yanghua · 2026-08-27T06:51:30Z

And I can join WeChat if there's a group chat for Lance.

yes, we have multiple WeChat groups for Chinese. But you do not have a WeChat account; how to join?

sezruby · 2026-08-30T22:06:19Z

I created one. @yanghua

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>

…e schemas, close on alloc failure Addresses the three gatekeeper review findings: - Release callback (line 119): ArrowArrayStream.close() only frees the struct buffer, not the native release callback, so an abandoned or partially-consumed stream leaked the provider's private_data and record batch stream. LanceArrowStream.close() now runs the release callback via a guarded releaseStream() before freeing the struct. It is idempotent against a consumer that already imported the stream: importArrayStream snapshots the callback and closes our struct, so we skip when snapshot() reports the struct is freed or the release address is NULL. - Schema contract (line 60): the raw native scan schema diverges from the Spark partition schema for shapes the columnar reader fixes up on the JVM (synthesized _fragid, empty projection surfacing _rowid, metadata _rowid/_rowaddr/version/_score, and blob columns). export() now rejects those via checkExportableSchema() with UnsupportedOperationException so the caller falls back to the columnar reader instead of getting a wrong schema. Lance has no Hive-style partition columns, so ordinary data projections still export a stream that matches the partition schema. - Construction leak (line 54): the dataset and scanner opened by LanceFragmentScanner.create() were acquired before the cleanup try, so a throwing ArrowArrayStream.allocateNew (e.g. a bounded allocator) leaked both. The allocation now runs inside the cleanup scope, which closes the stream and the scanner on any construction failure. Tests: LanceArrowStreamScannerTest adds closeReleasesStreamThatWasNeverImported (release-callback path with no JVM import) and exportRejectsSchemasNeedingJvmPostProcessing. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Second gatekeeper pass: the admission check inspected only the partition schema, so a partition with a pushed aggregation (e.g. COUNT(*)) still passed. LanceScan routes such a partition to a dedicated reader (LanceCountStarPartitionReader) that returns a single count column, while LanceFragmentScanner ignores pushedAggregation and scans data rows — so the export would emit the wrong output for a valid partition. Renamed checkExportableSchema to checkExportablePartition, which now rejects a present pushedAggregation before the schema checks. Filter, limit, offset, and top-N ordering remain exportable because they are pushed faithfully into the native scan. Adds LanceArrowStreamScannerTest#exportRejectsPushedAggregation. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Third gatekeeper pass: a full-text query is another partition mode that diverges. LanceFragmentScanner enables it on the native scan, which auto-projects a _score column, so an export of a [id, body] partition returned [id, body, _score] — more columns than the partition declares. The read-option-driven divergence is invisible to a schema-only check. Replace the enumerated column-name checks with a generic guard: after planning the scan, compare the schema the native scan actually produces (LanceScanner#schema, now surfaced via LanceFragmentScanner#schema) against the declared partition schema and reject on any field name/order mismatch. This subsumes the full-text _score case, the synthesized _fragid, the _rowaddr added for blobs, the _rowid an empty projection surfaces, and reordered row-version columns — and any future mode — because the zero-copy export cannot re-project or reorder the way the columnar reader does. The pushed-aggregation check stays a separate pre-plan guard: an aggregate partition's declared schema can equal the native data schema, so the schema comparison cannot catch it. Tests: exportRejectsSchemasNeedingJvmPostProcessing now covers both mismatch directions (native fewer columns via _fragid; native extra column via empty projection, the same shape as full-text _score). A _rowid column the native scan produces in matching order is now legitimately exportable, so that assertion is dropped. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Address review feedback on the fragment-scan Arrow C stream export: - checkNativeSchemaMatchesPartition now compares the declared and native schemas by Arrow ArrowType and nullability (recursing into children), not only field name and order. The declared Spark schema is converted through LanceArrowUtils — the same adapter the read path uses — so the Arrow-specific distinctions a Spark DataType alone cannot express (LargeUtf8 vs Utf8, LargeBinary, Date(MILLISECOND), FixedSizeBinary, Float16) survive the comparison and a column whose native type differs from the declared one (e.g. a narrower/wider int or a different time-zone timestamp) is no longer streamed as if it matched. - Add exportsEmptyFragmentScanPreservingSchema: a scan whose filter matches no rows still exports its full declared schema up front and drains cleanly with zero batches, releasing without a leak under the leak-checking allocator. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

sezruby · 2026-09-01T20:08:50Z

@yanghua Rebased onto main now that #783 has merged, so the earlier beta.21/Kryo blocker is resolved — the lance-core bump is already on main and dropped out of this PR as a no-op, leaving just the three Gluten files.

The Spark test workflow passed on the prior head across Spark 3.4–4.2 / Scala 2.12–2.13; I've re-pushed to refresh the checks. The lance-gatekeeper check went to action_required after the rebase and should re-evaluate on this push. Could you approve the pending workflow run when you get a chance? Thanks!

lance-gatekeeper

✅ Gate recommendation: approve.

The author’s rebase update is verified: #783’s immutable-collection Kryo fix is in the live base, the lance-core bump is no longer part of this PR, and the formerly failing Optimize path passes.

The remaining three-file export preserves the existing scan contract by rejecting pushed aggregates and native schemas that differ in field order, Arrow type, nullability, or nested shape. Its release and cleanup path is idempotent, and focused C-stream, empty-scan, and columnar-scanner tests pass.

yanghua

+1

sezruby mentioned this pull request Aug 23, 2026

feat(gluten): export a Lance fragment scan as an Arrow C stream sezruby/lance-spark#12

Closed

github-actions Bot added the enhancement New feature or request label Aug 23, 2026