Repository navigation
[VL] Add Velox offload for Lance scans via Arrow C stream - #12958
Conversation
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
|
@sezruby the coding style issue can be fixed by running this tool: |
0a7569f to
5080d2f
Compare
|
Thanks @zhouyuan! Ran Heads up on the rest of CI: this is still a draft because the lance-spark release carrying lance-format/lance-spark#778 isn't out yet, so the |
|
Run Gluten Clickhouse CI on x86 |
|
just link, no followup needed This is the scan-side version of the same problem. With the Spark changes proposed in #13140, a connector-specific scan wouldn't be needed:
|
5080d2f to
5b47a2b
Compare
|
Run Gluten Clickhouse CI on x86 |
5b47a2b to
10d5502
Compare
|
Run Gluten Clickhouse CI on x86 |
| TAGS_EXCLUDE="org.apache.spark.tags.ExtendedSQLTest,org.apache.spark.tags.SlowHiveTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.CudfTest,org.apache.gluten.tags.SkipTest" | ||
| if [ "${{ matrix.group }}" = "1" ]; then | ||
| $MVN_CMD clean test -Pspark-3.5 -Pjava-17 -Pbackends-velox -Piceberg -Pdelta -Phudi -Ppaimon -Pspark-ut \ | ||
| $MVN_CMD clean test -Pspark-3.5 -Pjava-17 -Pbackends-velox -Piceberg -Pdelta -Phudi -Ppaimon -Plance -Plance-test -Pspark-ut \ |
There was a problem hiding this comment.
could also enable the 4.0 test for lance if it's available?
There was a problem hiding this comment.
Thanks! lance-spark 0.8.0 publishes lance-spark-4.0_2.13, so I enabled -Plance -Plance-test in the Spark 4.0 job (group 1, which runs org.apache.gluten.*). VeloxLanceSuite passes 5/5 locally on Spark 3.5 and 4.0. I also pinned spark.sql.ansi.enabled=false in the suite, since Spark 4 defaults to ANSI. Docs updated.
There was a problem hiding this comment.
Warning
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Copilot review overview
4 open findings
wrap(...)or… · Newreader/handleis only closed on task completion. If a Spark partition contains… · New-Planceis enabled.… · New
What changed in this PR
Adds Velox-side offload support for read-only Lance BatchScanExec by importing Lance-exported Arrow C streams into Gluten’s Arrow-Java columnar batches, with opt-in build/test profiles and CI validation.
Changes:
- Introduces a new Velox “lance” component + rule to rewrite exportable Lance scans to
LanceScanTransformer. - Implements
LanceScanTransformerthat imports per-fragment Arrow C streams and emitsArrowJavaBatchTypecolumnar batches. - Adds Maven profiles, documentation, CI wiring, and a new
VeloxLanceSuitecovering offload vs fallback cases.
| File | Description |
|---|---|
| pom.xml | Adds lance / lance-test profiles to include Lance-specific sources/resources. |
| backends-velox/pom.xml | Adds Lance dependencies (provided/test) with Arrow/Jackson exclusions and Arrow dataset API. |
| backends-velox/src-lance/main/.../OffloadLanceScan.scala | Adds heuristic rule to rewrite exportable Lance scans to LanceScanTransformer. |
| backends-velox/src-lance/main/.../LanceScanTransformer.scala | Implements Arrow C stream import path into Arrow-Java columnar batches. |
| backends-velox/src-lance/main/.../VeloxLanceComponent.scala | Registers the Lance offload component, gated by runtime class presence. |
| backends-velox/src-lance/test/.../VeloxLanceSuite.scala | Adds execution-plan and correctness tests for offload/fallback behavior. |
| docs/get-started/VeloxLance.md | Documents requirements, build flags, behavior, and limitations. |
| .github/workflows/velox_backend_x86.yml | Enables Lance profiles in CI group 1 so new suite runs. |
🧠 Review effort: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| val stream = ArrowArrayStream.wrap(handle.streamAddress()) | ||
| val reader = Data.importArrayStream(allocator, stream) | ||
| context.addTaskCompletionListener[Unit] { | ||
| _ => | ||
| try { | ||
| // Drains + runs the C release callback, tearing down the native scan. | ||
| reader.close() | ||
| } finally { | ||
| // Frees the caller-owned stream struct and closes the scanner + dataset. | ||
| handle.close() | ||
| } |
There was a problem hiding this comment.
Fixed in d017739. Per-fragment reading is now in LanceFragmentIterator. If wrap/importArrayStream throws, the handle is closed before rethrowing, and any close failure is added as suppressed.
| val handle = LanceArrowStreamScanner.export(fragId.intValue(), lancePartition) | ||
| val stream = ArrowArrayStream.wrap(handle.streamAddress()) | ||
| val reader = Data.importArrayStream(allocator, stream) | ||
| context.addTaskCompletionListener[Unit] { | ||
| _ => | ||
| try { | ||
| // Drains + runs the C release callback, tearing down the native scan. | ||
| reader.close() | ||
| } finally { | ||
| // Frees the caller-owned stream struct and closes the scanner + dataset. | ||
| handle.close() | ||
| } | ||
| } | ||
|
|
||
| new Iterator[ColumnarBatch] { | ||
| private var advanced = false | ||
| private var hasMore = false | ||
|
|
||
| private def advance(): Unit = | ||
| if (!advanced) { | ||
| hasMore = reader.loadNextBatch() | ||
| advanced = true | ||
| } | ||
|
|
||
| override def hasNext: Boolean = { | ||
| advance() | ||
| hasMore | ||
| } |
There was a problem hiding this comment.
Done in d017739. Reader + handle are closed as soon as loadNextBatch() returns false, so a multi-fragment partition holds at most one open scan. close() is idempotent, and the task-completion listener stays as a safety net for early termination. Emitted batches own their transferred buffers, so closing early doesn't affect them. Added a multi-fragment test (including a LIMIT query that stops early and relies on the listener).
| <build> | ||
| <plugins> | ||
| <plugin> | ||
| <groupId>org.codehaus.mojo</groupId> | ||
| <artifactId>build-helper-maven-plugin</artifactId> | ||
| <executions> | ||
| <execution> | ||
| <id>add-lance-sources</id> | ||
| <goals> | ||
| <goal>add-source</goal> | ||
| </goals> | ||
| <phase>generate-sources</phase> | ||
| <configuration> | ||
| <sources> | ||
| <source>${project.basedir}/src-lance/main/scala</source> | ||
| </sources> | ||
| </configuration> | ||
| </execution> | ||
| <execution> | ||
| <id>add-lance-resources</id> | ||
| <goals> | ||
| <goal>add-resource</goal> | ||
| </goals> | ||
| <phase>generate-resources</phase> | ||
| <configuration> | ||
| <resources> | ||
| <resource> | ||
| <directory>${project.basedir}/src-lance/main/resources</directory> | ||
| </resource> | ||
| </resources> | ||
| </configuration> | ||
| </execution> | ||
| </executions> | ||
| </plugin> | ||
| </plugins> | ||
| </build> | ||
| </profile> | ||
| <profile> | ||
| <id>lance-test</id> | ||
| <activation> | ||
| <activeByDefault>false</activeByDefault> | ||
| </activation> | ||
| <build> | ||
| <plugins> | ||
| <plugin> | ||
| <groupId>org.codehaus.mojo</groupId> | ||
| <artifactId>build-helper-maven-plugin</artifactId> | ||
| <executions> | ||
| <execution> | ||
| <id>add-lance-test-sources</id> | ||
| <goals> | ||
| <goal>add-test-source</goal> | ||
| </goals> | ||
| <phase>generate-test-sources</phase> | ||
| <configuration> | ||
| <sources> | ||
| <source>${project.basedir}/src-lance/test/scala</source> | ||
| </sources> | ||
| </configuration> | ||
| </execution> | ||
| <execution> | ||
| <id>add-lance-test-resources</id> | ||
| <goals> | ||
| <goal>add-test-resource</goal> | ||
| </goals> | ||
| <phase>generate-test-resources</phase> | ||
| <configuration> | ||
| <resources> | ||
| <resource> | ||
| <directory>${project.basedir}/src-lance/test/resources</directory> | ||
| </resource> | ||
| </resources> | ||
| </configuration> | ||
| </execution> | ||
| </executions> | ||
| </plugin> | ||
| </plugins> | ||
| </build> |
There was a problem hiding this comment.
This is intentional and mirrors the existing delta/iceberg/hudi/paimon profiles in the root pom, which add src-<format> roots the same way. build-helper skips non-existent directories, so other modules are unaffected.
|
|
||
| ## Reading | ||
|
|
||
| Offload / Fallback. |
10d5502 to
d017739
Compare
|
Run Gluten Clickhouse CI on x86 |
Offloads read-only Lance BatchScanExec to the Velox backend through the Arrow C Data Interface, using lance-spark's LanceArrowStreamScanner (lance-spark#778). Only the Arrow C-struct address crosses the classloader boundary, so Gluten's Arrow version stays decoupled from lance-spark's. The offload is gated on being a plain columnar Lance scan (no pushed aggregation, no full-text query); anything else falls back to vanilla Spark. A velox-lance Component registers the rule and only engages when lance-spark is on the classpath. Code lives under backends-velox/src-lance behind a new `lance` Maven profile, mirroring the src-iceberg velox-integration layout. The read path is entirely velox-specific (ArrowWritableColumnVector, ArrowBufferAllocators, ValidatablePlan), so no top-level gluten-lance module is added yet; one can be introduced when an engine-agnostic write path lands. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Supportability follow-ups to the Lance Velox offload: - lance-test profile (root pom): add-test-source/add-test-resource for backends-velox/src-lance/test, activated alongside -Plance. - VeloxLanceSuite: asserts a columnar Lance scan offloads to LanceScanTransformer, that projection + pushed filter stay offloaded with correct results, and that a pushed-down aggregation falls back to a vanilla BatchScanExec. API-grounded against VeloxWholeStageTransformerSuite (checkGlutenPlan/checkSparkPlan/runQueryAndCompare, Iceberg-suite parity). - docs/get-started/VeloxLance.md: read-only scope, -Plance build, lance-spark version + linux-x86-64 platform requirements, offload/fallback conditions, and the Arrow C stream topology note. The suite requires the lance-spark native runtime, published for linux-x86-64 only, so it executes in that environment; the offload/fallback behavior it encodes was validated under linux-x86-64. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Ran ./dev/format-scala-code.sh (spotless:apply) so the Lance offload sources conform to Gluten's scalafmt config -- wraps javadoc lines over the 100-column limit and drops a stray import-group blank line. spotless:check is now clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
- Pin lance.version to the released lance-spark 0.8.0 (first release with
LanceArrowStreamScanner) in the root pom.
- lance-test: add the Spark-version-specific lance-spark artifact, which
registers the "lance" data source used by the suite.
- Keep Gluten's Arrow on the classpath: exclude lance-core's newer Arrow and
jackson-jaxrs, and add arrow-dataset at ${arrow.version}.
- Fix an executor-side NPE: LanceScanTransformer.output was read through the
transient batchScan after serialization (e.g. by DeserializeToObjectExec on
collect()); capture it eagerly.
- Enable -Plance -Plance-test in the Spark 3.5 x86 CI job.
- Docs: lance-spark 0.8.0 requirement, supported native platforms, test command.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Plain scan: the Lance scan's Arrow Java batches go straight to vanilla ColumnarToRowExec (no Velox round trip). - New test: with a Velox consumer (grouped aggregation), the scan feeds ArrowColumnarToVeloxColumnarExec(OffloadArrowDataExec(scan)) and is never converted to rows first. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
…ark 4.0 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
d017739 to
0f9b38a
Compare
|
Run Gluten Clickhouse CI on x86 |
|
For the record: longer term, Gluten should reuse lance-spark's own columnar scan, with at most a thin Proposal. lance-spark keeps the standard Three pieces are needed: 1. lance-spark: an Arrow output mode in the columnar reader Today
An Arrow mode would:
2. lance-spark: solve the Arrow version/classloader mismatch on its side This mismatch is why this PR passes a C-struct address. If lance-spark's batches carry lance's own (newer or relocated) Arrow classes, Gluten can't use them as Arrow. Two options:
The first is cleaner and matches apache/spark#59313. 3. Gluten: declare the batch type and accept Spark's
With these in place, the plan becomes |
What changes are proposed in this pull request?
Offloads read-only Lance
BatchScanExecto the Velox backend through the Arrow C Data Interface. This is the Gluten consumer side of the three-step Lance integration; both upstream pieces are released:LanceScanner#exportArrowStream(long)in lance-core (Arrow C stream export)LanceArrowStreamScannerforwarding in lance-spark (feat(gluten): export a Lance fragment scan as an Arrow C stream lance-format/lance-spark#778), first released in lance-spark 0.8.0How it works
OffloadLanceScanrule matches a LanceBatchScanExecand rewrites it toLanceScanTransformer, which produces anArrowJavaBatchTypecolumnar RDD.LanceArrowStreamScanner.export(...)hands back an Arrow C stream; Gluten wraps the stream address withArrowArrayStream.wrap(...)/Data.importArrayStream(...)and loads the batches intoArrowWritableColumnVector. Only the Arrow C-struct address crosses the classloader boundary, so Gluten's Arrow version stays decoupled from lance-spark's.velox-lanceComponentregisters the rule and activates only when lance-spark is on the classpath (org.lance.spark.read.LanceScanpresent).Relationship to the Iceberg scan transformer
LanceScanTransformeris intentionally different in shape fromIcebergScanTransformer. Iceberg extendsBatchScanExecTransformerBaseand is a native Velox scan: it emits Substrait, pushes filters into Velox, and reports scan-level SQL metrics. Lance instead extendsLeafExecNode with ValidatablePlanand bridges an already-produced Arrow stream into Velox: filters and projection are pushed into the Lance reader by lance-spark before export, not into Velox, and there is an Arrow-Java → Velox conversion at import. Consequences of this boundary:_rowid/_rowaddr/ blob columns) are not offloaded.These are follow-up opportunities, not blockers for the read-only offload. (See also #13140 for the more general "Spark Arrow batch → Velox" direction.)
Where the code lives / build
The read path is entirely Velox-specific (
ArrowWritableColumnVector,ArrowBufferAllocators,ValidatablePlan), so it is placed underbackends-velox/src-lancebehind a newlanceMaven profile — mirroring thesrc-icebergVelox-integration layout rather than adding a top-level module.lance.version(rootpom.xml) =0.8.0.-Plance: compiles againstlance-spark-base(provided). lance-core declares a newer Arrow than Gluten, so lance's transitive Arrow / jackson-jaxrs are excluded and Gluten's Arrow is kept (arrow-datasetis added at${arrow.version}). This matches the publishedlance-spark-bundle-3.5, which ships Arrow 15. With these exclusions-Planceonly adds jars to the classpath; it does not change any existing dependency version.-Plance-test: adds the Spark-version-specificlance-spark-${sparkbundle.version}runtime (registers thelancedata source) and the test sources.Build:
-Pbackends-velox -Pspark-3.5 -Plance.docs/get-started/VeloxLance.mddocuments scope, build, and requirements.Open question for reviewers: a top-level
gluten-lancemodule (likegluten-iceberg) makes sense once there is engine-agnostic code to host — e.g. a write path. The read-only offload has none today, so this PR keeps it Velox-local. Happy to restructure if you'd prefer the module up front.How was this patch tested?
VeloxLanceSuite(-Plance -Plance-test) asserts:LanceScanTransformer; with no Velox consumer its Arrow batches go straight to vanillaColumnarToRowExec(no Velox round trip),ArrowColumnarToVeloxColumnarExec(OffloadArrowDataExec(scan))into a Velox aggregate, with no row conversion off the scan,LIMITthat stops early (native scans are released per fragment on exhaustion, with a task-completion listener as the safety net),BatchScanExec.CI: the Spark 3.5 and Spark 4.0 x86 jobs (group 1, which runs the
org.apache.glutensuites) now build with-Plance -Plance-test, so the suite runs on every PR. The suite setsspark.sql.ansi.enabled=falsebecause Spark 4 defaults to ANSI mode.Locally, the suite passes (5/5) on linux-aarch64 (
apache/gluten:centos-8-jdk8, JDK 17, Spark 3.5 and Spark 4.0 / Scala 2.13) against lance-spark 0.8.0 and the current nightly Velox native libs:Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code; GitHub Copilot CLI 1.0.93 (Claude Opus 5.5)
Relates to #12263.