Lance-spark support in Gluten

[FelixYBW]

Description

To add Lance read/write support, we have two ways:

• Use lance connector in Velox, offload the whole read/write operator to velox plan [WIP]Add Lance connector facebookincubator/velox#16556
• Use lance-spark datasource. To read, gluten gets the arrow data from spark-spark, fwd to velox pipeline. To write, gluten gets the velox data from Velox, converts it as arrow, fwd to lance-spark. No C2R/R2C conversion in the whole process

We are going to try lance-spark as the preferred way.

Gluten version

None

[FelixYBW]

FYI @sezruby

[FelixYBW]

Lance-spark memory management:
#12226 (comment)

[zhztheplayer]

Lance reader seems relatively easier to integrate, as it reads data into a arrow-compatible format, ref.

Lance writer (ref) is a bit more complicated, the writer API is row-based which is constrained by the Spark standard. Very similar to Iceberg.

For the time being we can optimize the read-side efficiency in Gluten, by inserting arrow-to-velox columnar transition between lance scan and Gluten to eliminate r2c.

@sezruby let me know if you have any comments on this.

cc @malinjawi if you'd like to offer some help as well.

[sezruby]

Sounds good — happy to help review or work on the Lance-spark integration as it's relatively close to my current work.

[FelixYBW]

The best way is to add API to Spark and support columnar-based write. Before that, we either need to add it to lance-spark or copy the code from lance-spark to Gluten and modify it.

@JkSelf, your thoughts?

[jackylee-ch]

Before that, we either need to add it to lance-spark or copy the code from lance-spark to Gluten and modify it.

I'd prefer to introduce Lance support as a separate module, following the same pattern we used for Iceberg, and pull in lance-spark as a dependency from there.

A few reasons this is the cleanest path:

• It's consistent with how Gluten already integrates external data sources (Iceberg being the precedent), which lowers the maintenance and review burden and keeps the project's structure predictable for contributors.

• The integration framework is engine-agnostic. Once the Velox path lands, we should be able to reuse most of this scaffolding to wire Lance up to Velox with only minor changes, rather than reworking the integration.

cc @LuciferYang Any thoughts from the lance-spark side?

[FelixYBW]

With fix #12226, Gluten can read/write lance data with C2R

config

jar download from https://repo1.maven.org/maven2/com/lancedb/lance-spark-bundle-3.5_2.12/0.0.15/lance-spark-bundle-3.5_2.12-0.0.15.jar

conf.set('spark.sql.catalog.lance','com.lancedb.lance.spark.LanceNamespaceSparkCatalog')
conf.set('spark.sql.catalog.lance.impl','dir')
conf.set('spark.sql.catalog.lance.root','/opt/spark/work-dir')

write, C2R in Gluten + R2C in lance

CREATE TABLE lance.lineitem
USING lance
AS SELECT *
FROM spark_catalog.tpch_sf100_parquet_snappy.lineitem;

read, C2R in Spark + R2C in Gluten.

I remember we have a PR to convert Spark's columnar to Velox directly. @zhztheplayer is it disabled or doesn't work here?

select max(tax) from lance.lineitem group by shipmode

[FelixYBW]

Before that, we either need to add it to lance-spark or copy the code from lance-spark to Gluten and modify it.

I'd prefer to introduce Lance support as a separate module, following the same pattern we used for Iceberg, and pull in lance-spark as a dependency from there.

Yes, we need a separate module. Two things I'm thinking:

1. To support iceberg or delta, we copied several files from their data source and hacked in Gluten. See if we can avoid doing this in lance-spark. Create APIs in lance-spark and call from Gluten
2. General arrow input/output support. See if we can submit new APIs in Spark to support arrow format input/output. It's something we plan to do when we support bigquery connector. In future, if data source is Arrow format, Gluten can automatically support.

[JkSelf]

Lance-spark's write path is row-based primarily because Spark's DSv2 sink APIs are still bound to InternalRow. Internally, lance-spark converts and buffers these rows into Arrow columnar batches before flushing them to Lance.

Therefore, since Gluten already operates on Arrow columnar data, we can bypass the standard row-based lance-spark sink path entirely. A more possible direction would be:

Velox columnar
-> Arrow C Data Interface / ArrowArrayStream
-> lance-core (via Dataset.write().stream(...).execute())

[jackylee-ch]

2. General arrow input/output support. See if we can submit new APIs in Spark to support arrow format input/output. It's something we plan to do when we support bigquery connector. In future, if data source is Arrow format, Gluten can automatically support.

This seems to be more friendly to gluten and other native engines, but it increases the complexity of lance-spark. @LuciferYang PTAL

[FelixYBW]

This seems to be more friendly to gluten and other native engines, but it increases the complexity of lance-spark. @LuciferYang PTAL

Without API in spark, we either need to add the interface in lance-spark for Gluten, or Gluten needs to hack lance-spark code as iceberg/delta today.

[malinjawi]

I remember we have a PR to convert Spark's columnar to Velox directly

@FelixYBW that's #7392 (ArrowColumnarToVeloxColumnarExec). It's not disabled, it just never triggers here: the transition only fires for plan nodes that declare Gluten's Arrow batch types, and a vanilla BatchScanExec over lance-spark declares the vanilla batch type. There's no vanilla->Velox transition registered since Gluten can't assume an arbitrary ColumnarBatch is Arrow-backed. Hence C2R in Spark + R2C in Gluten.

I went through lance-spark's code to figure out what the integration actually needs from their side. Turns out almost everything is already public, and the "add interface in lance-spark vs copy code like iceberg/delta" question comes down to three small accessors:

1. read: no public way to get the raw Arrow data. LanceFragmentColumnarBatchScanner doesn't expose its VectorSchemaRoot and LanceArrowColumnVector doesn't expose the wrapped ValueVector. Either one works for us.
2. write: LanceBatchWrite.TaskCommit's constructor is package-private, we need a public factory to produce commit messages their driver-side commit() accepts.
3. write: SparkWrite.getWriteOptions() is package-private (LanceSparkWriteOptions.toWriteParams() itself is public).

With those, the read side is what @zhztheplayer suggested: an OffloadLanceScan rule (same shape as OffloadIcebergScan) swapping the scan for an exec that declares ArrowJavaBatchType, rewrapping via ArrowWritableColumnVector.loadColumns() without copying, and the existing transitions handle the rest. Since this wraps the already-planned LanceScan, lance's pushdowns (filter, count-star, TopN, zonemap pruning) are preserved. One caveat: the scanner also emits non-Arrow virtual vectors (fragment-id constant, blob position/size), so the rule needs validation fallback for those projections.

For the write side, LanceDataWriter already ends in Fragment.create(datasetUri, arrowStream, params), which is @JkSelf's pipeline with one adjustment: per-task Fragment.create rather than Dataset.write(), with the single commit staying on the driver, which is what lance-spark already does. So a Gluten writer (via ColumnarV2TableWriteExec, same as iceberg native write) can export Velox batches through the C data interface into that same call and skip the row->Arrow buffering entirely. No Spark API change needed to ship this, though the Arrow I/O API in Spark is still worth pursuing long term.

Btw the repro above uses the com.lancedb 0.0.15 bundle, which is the last release under those coordinates. Releases moved to org.lance after the package rename (lance-format/lance-spark#133), currently 0.5.1, so even the catalog class name in that config differs now. Everything I checked above is against 0.5.1.

[FelixYBW]

Arrow UDTF:

from pyspark.sql.functions import lit, udtf
import math

@udtf(returnType="num: int, square: int, cube: int, factorial: int", useArrow=True)
class GenerateComplexNumbers:
    def eval(self, start: int, end: int):
        for num in range(start, end + 1):
            yield (num, num ** 2, num ** 3, math.factorial(num))
GenerateComplexNumbers(lit(1), lit(10)).show()

See if we can make a general solution. Let's submit PRs to upstream Spark if we need to. @zhztheplayer @JkSelf

[LuciferYang]

@jackylee-ch @FelixYBW Sorry for the late reply. I went through lance-spar, the lance Java SDK and gluten upmaster before answering, so this took a bit.

From the lance-spark side: my personal read is that what Gluten needs here is small and shouldn't add much maintenance burden. On the technical substance, @malinjawi's audit matches what I see in the code, but I'd cut the list down to two changes, and adjust the first one:

1. Read: expose the Arrow C stream instead of VectorSchemaRoot. Inside the lance SDK, LanceScanner.scanBatches() already gets the scan result as a native FFI_ArrowArrayStream and only then imports it into Java Arrow. If lance-spark exposes the stream at that level, Gluten can import it with its own Arrow copy, or directly into Velox, and the two projects never have to share Java Arrow objects. The reason this matters: Gluten on Spark 3.5 is on Arrow 15.0.0 (vanilla, after #12244) and keeps org.apache.arrow.{memory,vector,c,dataset} unrelocated in its bundle jar; lance-spark's bundle also ships unrelocated arrow-c-data/arrow-dataset, while lance-core is built against Arrow 18.3. A getVectorSchemaRoot()-style accessor plus ArrowWritableColumnVector.loadColumns(...) only works if both sides load the same org.apache.arrow.vector classes, which today depends on classloader order. Passing C-ABI pointers sidesteps the whole problem.

2. Write: a public factory for TaskCommit. It just wraps List<FragmentMetadata>, so this is a one-liner. SparkWrite.getWriteOptions() doesn't need to change — LanceSparkWriteOptions.toWriteParams() and all the individual getters are already public.

The write path is actually in better shape than the thread assumed. Fragment.write().data(ArrowArrayStream) hands the raw stream pointer straight to JNI without materializing Java Arrow vectors, and the driver-side commit (Transaction + Append + CommitBuilder) is all public — it's exactly what LanceBatchWrite.commit() does internally. Lance also recently added WriteFragmentBuilder.schema(...) (lance-format/lance#6919, already in the 8.0.0-beta.9 that lance-spark depends on) so executor-side fragment writes no longer need to open the dataset first. To be clear, lance-spark's own writer doesn't use these yet (it still goes through the deprecated Fragment.create without the schema override) — the point is that Gluten's columnar writer can call the SDK directly and only reuse lance-spark's driver-side commit. So @JkSelf's pipeline works today, no Spark API change required.

One smaller thing: lance-spark uses a single process-wide unbounded RootAllocator (LanceRuntime.allocator()), not tied into Spark's task memory. Per-batch usage is bounded by maxBatchBytes so it's acceptable for a connector, but if Gluten wants the read path accounted we could additionally accept a caller-supplied BufferAllocator. Optional, can come later.

If we converge on the C-ABI direction here, we can open an issue on the lance-spark side with the concrete API proposal and gather feedback from the other maintainers, and we can push this forward together from both ends. I'd suggest settling the Arrow question first since it decides the shape of the read API.