This is an automated email from the ASF dual-hosted git repository.
zhztheplayer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 0ef7c9325a [MINOR][DOC] Add developer doc on Arrow columnar conversion
(#13374)
0ef7c9325a is described below
commit 0ef7c9325a5dd8e7785146d9ca45980b0d941b50
Author: BInwei Yang <[email protected]>
AuthorDate: Tue Oct 20 02:15:44 2026 -0700
[MINOR][DOC] Add developer doc on Arrow columnar conversion (#13374)
* [MINOR][DOC] Add developer doc on Arrow columnar conversion
Describe how Gluten converts between rows, Spark batches, Java Arrow,
native Arrow and Velox batches today (conventions, transition graph,
what each transition does, copies, batch lifecycle), and how to add an
Arrow-based columnar operator, using the Lance scan offload as example.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* [MINOR][DOC] Add Arrow UDF buffer lifecycle example and diagrams
- Add section 4: a Velox BIGINT column through LoadArrowData, an Arrow
UDF and back to Velox, with the reference counts at each step and the
release callbacks back to Velox's memory (PlantUML sequence diagram).
- Explain how an ArrowBuf is released depending on whether it was
allocated in Java or imported through C Data, and the memory
accounting of imported buffers (second PlantUML diagram).
- Replace the mermaid transition graph with rendered images.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
docs/developers/ArrowColumnarConversion.md | 487 ++++++++++++++++++++++++++++
docs/image/ArrowBufReleaseDispatch.png | Bin 0 -> 96431 bytes
docs/image/GlutenBatchTypeTransition.png | Bin 0 -> 154325 bytes
docs/image/GlutenConversionImages.png | Bin 0 -> 389243 bytes
docs/image/VeloxColumnArrowUdfLifecycle.png | Bin 0 -> 144971 bytes
mkdocs.yml | 1 +
6 files changed, 488 insertions(+)
diff --git a/docs/developers/ArrowColumnarConversion.md
b/docs/developers/ArrowColumnarConversion.md
new file mode 100644
index 0000000000..1901ffb4b5
--- /dev/null
+++ b/docs/developers/ArrowColumnarConversion.md
@@ -0,0 +1,487 @@
+---
+layout: page
+title: Arrow Columnar Conversion
+nav_order: 21
+parent: /developer-overview/
+---
+
+# Arrow Columnar Conversion in Gluten
+
+This document describes how Gluten converts data between rows, Spark columnar
batches, Java Arrow
+batches, native Arrow batches and Velox batches today, and how to add a new
operator that consumes
+or produces Arrow batches, using the Lance scan offload
+([#12958](https://github.com/apache/gluten/pull/12958)) as the example.
+
+## 1. Conventions
+
+Every physical plan node in a Gluten query has a *convention*: the row type
and the batch type it
+outputs. Gluten never asks a node to convert its output itself. Instead, each
type registers the
+transitions to and from other types, and the planner inserts the transition
operators between a
+child and a parent whose conventions don't match.
+
+
+
+
+| Concept | Code | Meaning |
+|---|---|---|
+| `Convention.RowType` | `gluten-core/.../transition/Convention.scala` | A row
format. `VanillaRowType` is Spark's `InternalRow`; `None` means the node
outputs no rows |
+| `Convention.BatchType` | same | A columnar batch format. `VanillaBatchType`
is Spark's `ColumnarBatch`; `None` means no batches |
+| `ConventionReq` | `ConventionReq.scala` | What a parent requires from each
child: `Any` or `Is(type)`, for rows and for batches |
+| `Transition` | `Transition.scala`, `TransitionGraph.scala` | An edge of the
graph: a function `SparkPlan => SparkPlan` that wraps the child with a
converting operator, or `Transition.empty` when the data is already compatible |
+| `ConventionFunc` | `ConventionFunc.scala` | Derives the convention of any
plan, including vanilla Spark plans and plans through `QueryStageExec` /
`ReusedExchangeExec` |
+
+A type registers its transitions in `registerTransitions()` with `fromRow`,
`toRow`, `fromBatch` and
+`toBatch`. All edges form one graph, and the planner uses the shortest path
between two types
+(`FloydWarshallGraph`), so a conversion over several steps is assembled
automatically.
+
+How `ConventionFunc` derives a convention:
+
+- A `GlutenPlan` declares it: `batchType()` and `rowType()`.
`supportsColumnar` and
+ `supportsRowBased` are derived from them and are `final`.
+- A vanilla Spark plan gets `VanillaBatchType` if `supportsColumnar`, and
`VanillaRowType` if it
+ supports rows.
+- Components can override both through `Component.convFuncOverride()`.
+
+How the planner requires a convention from children (`conventionReqOf`):
+
+- A `GlutenPlan` declares `requiredChildConvention()`. By default, children
must have the parent's
+ own convention; a node that supports both rows and batches requires only its
batch type.
+- `ColumnarToRowExec` requires `VanillaBatchType`; `RowToColumnarExec` and
`UnionExec` require
+ `VanillaRowType`.
+
+`InsertTransitions` removes existing transitions, then for each node compares
+`conventionReqOf(node)` with `conventionOf(child)` and applies the transition
path found in the
+graph.
+
+## 2. Types and how batches look
+
+### Types registered with the Velox backend
+
+| Type | Defined in | Physical form | Notes |
+|---|---|---|---|
+| `VanillaRowType` | `gluten-core` | Spark `InternalRow` | |
+| `VanillaBatchType` | `gluten-core` | Spark `ColumnarBatch` of any
`ColumnVector` | Transitions are Spark's `RowToColumnarExec` /
`ColumnarToRowExec` |
+| `ArrowJavaBatchType` | `gluten-arrow` `ArrowBatchTypes` | *Heavy* batch:
every column is an `ArrowWritableColumnVector`, whose data is in Java Arrow
vectors | Readable from the JVM through the `ColumnVector` getters |
+| `ArrowNativeBatchType` | `gluten-arrow` `ArrowBatchTypes` | *Light* batch:
column 0 is an `IndicatorVector` holding the handle of a native `ColumnarBatch`
(e.g. `ArrowCStructColumnarBatch`), the other columns are `PlaceholderVector`s
| Not readable from the JVM |
+| `VeloxBatchType` | `backends-velox` `VeloxBatchType` | Light batch whose
native object is a `VeloxColumnarBatch` (a Velox `RowVector`) | |
+| `VeloxCarrierRowType` | `backends-velox` `VeloxCarrierRowType` | Rows that
carry a Velox batch | Used for row-based outputs that keep Velox data, e.g. for
writes |
+
+`ColumnarBatches` (`gluten-arrow/.../columnarbatch/ColumnarBatches.java`)
tells heavy and light
+batches apart, and holds the reference counts of both: `IndicatorVector` for
light batches and
+`ArrowWritableColumnVector` for heavy ones.
+
+### Transition graph
+
+
+
+- **`ArrowJavaBatchType` → `VanillaBatchType` is empty** because
`ArrowWritableColumnVector` is a
+ Spark `ColumnVector`. A vanilla consumer such as `ColumnarToRowExec` reads
Gluten's Java Arrow
+ batches directly.
+- **`VeloxBatchType` → `ArrowNativeBatchType` is empty** because a
`VeloxColumnarBatch` is a native
+ `ColumnarBatch` that can export itself as Arrow C data, which is all that
`ArrowNativeBatchType`
+ promises.
+
+Composite paths that the planner builds from these edges:
+
+| From → To | Operators inserted |
+|---|---|
+| Velox → Java Arrow | `LoadArrowDataExec` (Velox → Arrow native is empty) |
+| Java Arrow → Velox | `OffloadArrowDataExec` →
`ArrowColumnarToVeloxColumnarExec` |
+| Java Arrow → rows | `ColumnarToRowExec` (Java Arrow → vanilla batch is
empty) |
+| Rows → Java Arrow | `RowToVeloxColumnarExec` → `LoadArrowDataExec` |
+| Velox → rows | `VeloxColumnarToRowExec` |
+
+## 3. What each transition does
+
+| Transition | Implementation | What happens to the data |
+|---|---|---|
+| `LoadArrowDataExec` (native → Java Arrow) | `ColumnarBatches.load` | The
native batch exports Arrow C structs (`ColumnarBatchJniWrapper.exportToArrow`).
For a Velox batch, `VeloxColumnarBatch::exportArrowArray` first flattens
dictionary and constant vectors (`ensureFlattened`), then calls
`velox::exportToArrow`. Java imports the arrays into
`ArrowWritableColumnVector`s (`ArrowAbiUtil.importToSparkColumnarBatch`) |
+| `OffloadArrowDataExec` (Java Arrow → native) | `ColumnarBatches.offload` |
Java exports the vectors as Arrow C structs
(`ArrowAbiUtil.exportFromSparkColumnarBatch`), and native wraps them as an
`ArrowCStructColumnarBatch` (`createWithArrowArray`). No copy |
+| `ArrowColumnarToVeloxColumnarExec` (native Arrow → Velox) |
`VeloxColumnarBatches.toVeloxBatch` → `VeloxColumnarBatch::from` |
`velox::importFromArrowAsOwner` builds a Velox `RowVector` that owns the Arrow
buffers |
+| `RowToVeloxColumnarExec` / `VeloxColumnarToRowExec` | `backends-velox` |
Converts between rows and Velox vectors (always copies) |
+| `ColumnarToRowExec` over Java Arrow | Spark | Reads rows through
`ArrowWritableColumnVector`'s getters |
+
+### Copies on the Arrow path
+
+| Step | Fixed-width columns | Strings | Dictionary / constant vectors |
+|---|---|---|---|
+| Velox → Java Arrow (`load`) | No copy | Copied: Velox's `StringView` layout
is exported as an Arrow string array, since Arrow string views aren't enabled |
Flattened (copied) |
+| Java Arrow → native Arrow (`offload`) | No copy | No copy | Not used |
+| Native Arrow → Velox | No copy | A new `StringView` buffer is built (strings
of 12 bytes or less are inlined) | Not used |
+
+### Batch lifecycle
+
+- `load`, `offload` and `toVeloxBatch` convert **in place**: the new vectors
are moved into the
+ input `ColumnarBatch` object (`SparkColumnarBatchUtil.transferVectors`), and
the output keeps the
+ input's reference count. Code that holds the input object sees the converted
batch.
+- A batch is released with `ColumnarBatches.release` or
`ColumnarBatch.close()`, which decrements
+ the reference count of its vectors. An operator that needs a batch after its
child has moved on
+ must retain it (`ColumnarBatches.retain`) and release it later, as
`ColumnarArrowEvalPythonExec`
+ does.
+- `ColumnarToColumnarExec` closes the batches it emits on the next call only
when
+ `needRecyclePayload` is true (e.g. `VeloxResizeBatchesExec`). The Arrow
transitions keep the
+ default, `false`.
+- Java Arrow memory comes from `ArrowBufferAllocators.contextInstance()`,
Gluten's per-task
+ allocator, whose listener reports to Spark's memory manager.
+
+## 4. Example: a Velox column through an Arrow UDF
+
+This follows one fixed-width column, a `BIGINT` column `x`, through
+`Velox op → LoadArrowDataExec → ColumnarArrowEvalPythonExec →
OffloadArrowDataExec →
+ArrowColumnarToVeloxColumnarExec → Velox op`, for a query such as `SELECT x,
f(x)` with a pandas or
+Arrow UDF `f`. .
+
+
+
+Three counts are involved, and none of them crosses JNI:
+
+- **Velox buffer references** (`BufferPtr`), held by Velox vectors and by an
exported `ArrowArray`.
+- **Arrow Java `ArrowBuf` references**. An imported array is released through
its C `release`
+ callback once its last `ArrowBuf` is released.
+- **Gluten's `refCnt`** on `ArrowWritableColumnVector`. It only decides when
Gluten closes the Java
+ vector.
+
+The Arrow C Data interface has no reference count. Ownership of an
`ArrowArray` moves once, and the
+holder calls `release` exactly once.
+
+| Step | Velox buffer refs | `x.refCnt` | What happens |
+|---|---|---|---|
+| 1. Export | 2 | – | `VeloxColumnarBatch::exportArrowArray` exports
`ArrowArray` A, which keeps a reference to the buffer. No copy for a flat
fixed-width vector |
+| 2. Import | 2 | 1 | `Data.importIntoVectorSchemaRoot` wraps the Velox memory
as `ArrowBuf`s, and `loadColumns` creates `x` |
+| 3. Close the Velox batch | 1 | 1 | `load` closes the input batch:
`ObjectStore::release` frees the `RowVector`. A keeps the buffer alive |
+| 4. Retain for the UDF | 1 | 2 | `ColumnarArrowEvalPythonExec` retains the
input batch until the result comes back |
+| 5. Send to Python | 1 | 2 | `ArrowStreamWriter` serializes the UDF columns
into the socket. The Python worker reads its own copy; when pyarrow releases
it, only Python's memory is freed |
+| 6. Join the result | 1 | 1 | The output batch is `[x, r]`. `adjustRefCnt`
sets every column to 1, because `offload` requires one count for the whole
batch |
+| 7. Export to native | 1 | 0 | `offload` exports `ArrowArray` B, which
retains the `ArrowBuf`s, then closes the input batch. `x` is closed, and B
keeps the memory alive |
+| 8. Import into Velox | 1 | – | `importFromArrowAsOwner(B)` builds a Velox
vector over the same memory. No copy |
+| 9. Release | 0 | – | When the downstream Velox operator frees the vector,
B's `release` (Arrow Java's export callback) drops the last `ArrowBuf`, which
calls A's `release` (Velox's export callback). The buffer goes back to Velox's
memory pool |
+
+So the column's memory is freed by the last Velox operator that uses it,
through two `release`
+callbacks: Velox → Arrow Java → Velox. If the UDF output goes to a vanilla
Spark operator instead,
+step 7 doesn't happen, and the memory is freed when
`ColumnarArrowEvalPythonExec` closes its output
+batch on the next call (`recyclePayload`).
+
+<details>
+<summary>PlantUML source: Velox column through an Arrow UDF</summary>
+
+```plantuml
+@startuml VeloxColumnArrowUdfLifecycle
+skinparam shadowing false
+skinparam sequenceMessageAlign left
+skinparam responseMessageBelowArrow true
+hide footbox
+
+title A Velox BIGINT column through an Arrow UDF and back (Gluten main)
+
+box "Native (C++)" #EEF7EE
+participant "Velox buffer\n(values of x)" as Buf
+participant "VeloxColumnarBatch\n(upstream op)" as VB
+participant "Velox\n(downstream op)" as VD
+end box
+box "JVM" #E8F0FB
+participant "LoadArrowDataExec\nColumnarBatches.load" as Load
+participant "ArrowWritableColumnVector x\n(Arrow BigIntVector)" as X
+participant "ColumnarArrowEvalPythonExec" as Py
+participant "OffloadArrowDataExec\n+ ArrowColumnarToVelox" as Off
+end box
+participant "Python worker\n(pyarrow UDF)" as W
+
+== 1. Velox -> Arrow C -> Java (no copy) ==
+Load -> VB : JNI exportToArrow(handle)
+VB -> VB : ensureFlattened()\nvelox::exportToArrow()
+VB -> Buf : ArrowArray A keeps a buffer ref\n<b>Velox refs: 2</b> (vector + A)
+VB --> Load : ArrowArrayMove(A)
+Load -> X : Data.importIntoVectorSchemaRoot(A)\nArrowBuf wraps Velox
memory\nloadColumns() <b>x.refCnt = 1</b>
+Load -> VB : input.close() -> ObjectStore::release\nRowVector freed
+VB -> Buf : <b>Velox refs: 1</b> (only A)
+
+== 2. To Python and back (IPC copy) ==
+Py -> X : ColumnarBatches.retain <b>x.refCnt = 2</b>\ncache input batch
+Py -> W : ArrowStreamWriter.writeBatch\n(serialize x into the socket)
+W -> W : pyarrow reads its own copy,\nruns the UDF, frees its copy\n(no effect
on JVM memory)
+W --> Py : result batch r (IPC)
+Py -> X : join [x, r]\nadjustRefCnt(x, 1) <b>x.refCnt = 1</b>
+
+== 3. Java -> Arrow C -> Velox (no copy) ==
+Off -> X : Data.exportVectorSchemaRoot -> ArrowArray B\nB retains the ArrowBufs
+Off -> X : offload closes the input batch\n<b>x.refCnt = 0</b> ->
vector.close()\n(ArrowBufs still held by B)
+Off -> VD : createWithArrowArray(B)\nimportFromArrowAsOwner(B)
+note right of VD : Velox reads the same memory
+
+== 4. Release chain ==
+VD -> VD : downstream op done with the vector
+VD -> X : B.release() (Arrow Java export callback)\nlast ArrowBuf released
+X -> Buf : A.release() (Velox export callback)\n<b>Velox refs: 0</b>, buffer
returned to the MemoryPool
+@enduml
+```
+
+</details>
+
+### Buffer release
+
+Neither `ArrowWritableColumnVector.close()` nor `B.release()` checks where a
buffer came from.
+Both drop one reference on each `ArrowBuf`, and when the last one goes, the
buffer's
+`AllocationManager`, chosen when the buffer was created, does the rest:
+
+- **Allocated in Java** (`allocator.buffer(...)`, e.g. by `ColumnarRangeExec`
or the Python result
+ reader): the memory is freed in Java.
+- **Imported through C Data** (`Data.importIntoVectorSchemaRoot`): the buffer
has a
+ `ForeignAllocationManager`, whose `release0()` calls
`ReferenceCountedArrowArray.release()`. When
+ the last buffer of the imported array is released, it calls the array's C
`release` callback,
+ here Velox's. It's the `release` member function of Arrow C_data interface:
+
+```c++
+struct ArrowArray {
+ // Array data description
+ int64_t length;
+ int64_t null_count;
+ int64_t offset;
+ int64_t n_buffers;
+ int64_t n_children;
+ const void** buffers;
+ struct ArrowArray** children;
+ struct ArrowArray* dictionary;
+
+ // Release callback
+ void (*release)(struct ArrowArray*);
+ // Opaque producer-specific data
+ void* private_data;
+};
+
+```
+
+In both cases the allocator then calls `listener.onRelease(size)`, and Gluten's
+`ManagedAllocationListener` returns the memory to Spark.
+
+
+
+<details>
+<summary>PlantUML source: release</summary>
+
+```plantuml
+@startuml ArrowBufReleaseDispatch
+skinparam shadowing false
+skinparam sequenceMessageAlign left
+skinparam responseMessageBelowArrow true
+hide footbox
+
+title How an ArrowBuf is released (Arrow Java 15)
+
+participant "Caller" as C
+participant "ArrowBuf /\nBufferLedger" as L
+participant "AllocationManager" as M
+participant "Gluten allocator\n+ ManagedAllocationListener" as A
+participant "ReferenceCountedArrowArray\n(imported ArrowArray A)" as R
+participant "Native producer\n(Velox export)" as N
+
+note over C
+ Two callers, same path:
+ - ArrowWritableColumnVector.close(), when refCnt reaches 0
+ -> FieldVector.close()
+ - B.release() -> ExportedArrayPrivateData.close()
+ (drops the reference the export retained)
+end note
+
+C -> L : close() -> release()
+L -> L : bufRefCnt - 1
+alt bufRefCnt > 0
+ L --> C : still in use, nothing else happens
+else bufRefCnt == 0
+ L -> M : release(ledger)
+ M -> A : releaseBytes(size)
+ alt allocated in Java (allocator.buffer)
+ M -> M : release0(): free the memory
+ else imported through C Data (ForeignAllocationManager)
+ M -> R : release0() -> ForeignAllocation.release0()\n->
ReferenceCountedArrowArray.release()
+ R -> R : count - 1 (one per imported buffer)
+ opt count == 0 (last buffer of the array)
+ R -> N : ArrowArray.release() (C callback)
+ N -> N : drop BufferPtr\nVelox frees the buffer
+ end
+ end
+ M -> A : listener.onRelease(size)\nrepay Spark memory
+end
+
+== When the buffer was created ==
+note over A
+ allocator.buffer(size): onPreAllocation borrows from Spark, then allocates.
+ wrapForeignAllocation(size): onPreAllocation also borrows from Spark,
+ but the memory is Velox's, already counted by Velox's pool.
+end note
+@enduml
+```
+
+</details>
+
+## 5. Arrow-based operators today
+
+| Operator | Batch type in → out | How it's planned |
+|---|---|---|
+| `ColumnarArrowEvalPythonExec` (Arrow / pandas UDFs) | Java Arrow → Java
Arrow | Gluten's own operator and Python runner, from `OffloadSingleNodeRules` |
+| `ColumnarPartialProjectExec`, `ColumnarPartialGenerateExec` | Velox → Velox
| Load to Java Arrow inside the operator, evaluate the Spark expressions row by
row, and write the results with `ArrowProjection` |
+| `ColumnarRangeExec` | – → Java Arrow | Builds `ArrowWritableColumnVector`s
in the JVM |
+| `VeloxRDDScanTransformer`, `VeloxLocalTableScanTransformer` | – → Velox |
Write rows into `ArrowWritableColumnVector`s, then convert them to Velox inside
the operator |
+| `LanceScanTransformer` (`-Plance`) | – → Java Arrow | Imports an Arrow C
stream from lance-spark |
+
+Any other Spark operator that wants Arrow, such as `MapInArrowExec` or
`FlatMapGroupsInArrowExec`,
+is a vanilla row-based plan to Gluten, so Gluten plans `VeloxColumnarToRow →
Spark operator →
+RowToVeloxColumnar` around it.
+
+## 6. Adding an Arrow-based columnar operator
+
+The Lance scan offload ([#12958](https://github.com/apache/gluten/pull/12958))
is a complete example: a leaf operator that produces Java Arrow batches from an
external Arrow C stream,
+and leaves every conversion to the planner. Its files:
+
+| File | Role |
+|---|---|
+|
`backends-velox/src-lance/main/scala/.../execution/LanceScanTransformer.scala`
| The operator and its RDD |
+| `backends-velox/src-lance/main/scala/.../extension/OffloadLanceScan.scala` |
The rule that replaces `BatchScanExec` |
+|
`backends-velox/src-lance/main/scala/.../component/VeloxLanceComponent.scala` |
Registers the rule when lance-spark is on the classpath |
+|
`backends-velox/src-lance/main/resources/META-INF/gluten-components/...VeloxLanceComponent`
| Makes the component discoverable |
+| `backends-velox/src-lance/test/scala/.../VeloxLanceSuite.scala` | Plan
shape, results and fallback tests |
+| `pom.xml`, `backends-velox/pom.xml` | `lance` / `lance-test` profiles that
add `src-lance` |
+
+### Step 1: Declare the convention
+
+Implement a `GlutenPlan` (usually through `ValidatablePlan`) and declare what
it outputs and
+requires:
+
+```scala
+case class MyArrowExec(child: SparkPlan) extends UnaryExecNode with
ValidatablePlan {
+ override def batchType(): Convention.BatchType = ArrowJavaBatchType
+ override def rowType(): Convention.RowType = Convention.RowType.None
+ // The default requires the children to have this node's convention, here
ArrowJavaBatchType.
+ // Override it only if the operator accepts something else.
+ override def requiredChildConvention(): Seq[ConventionReq] =
+ Seq(ConventionReq.ofBatch(ConventionReq.BatchType.Is(ArrowJavaBatchType)))
+
+ override protected def doExecute(): RDD[InternalRow] =
+ throw new UnsupportedOperationException(s"$nodeName does not support
row-based execution.")
+ override protected def doExecuteColumnar(): RDD[ColumnarBatch] = ...
+}
+```
+
+Don't insert `LoadArrowDataExec`, `OffloadArrowDataExec` or
`ArrowColumnarToVeloxColumnarExec`
+yourself. With these declarations `InsertTransitions` produces, for example:
+
+- Velox child → your operator: `LoadArrowDataExec` is added below it.
+- Your operator → Velox parent: `OffloadArrowDataExec` →
`ArrowColumnarToVeloxColumnarExec` is
+ added above it.
+- Your operator → vanilla row parent: only `ColumnarToRowExec`, with no Velox
round trip.
+
+### Step 2: Produce batches that Gluten can own
+
+The batches you emit are offloaded and closed by the transitions downstream,
so each batch must
+own its data:
+
+- **Build the columns as `ArrowWritableColumnVector`s**
(`ArrowWritableColumnVector.loadColumns`
+ over `FieldVector`s). `ColumnarBatches` rejects other vectors in a Java
Arrow batch.
+- **Allocate from `ArrowBufferAllocators.contextInstance()`.** This keeps the
memory accounted by
+ Spark, and keeps the vectors in the allocator tree that `offload` expects.
+- **Don't emit vectors that a reader reuses.** An `ArrowReader` (e.g. from
`Data.importArrayStream`)
+ loads every batch into the same `VectorSchemaRoot`. Lance transfers each
vector out of the root
+ before wrapping it (`getTransferPair(allocator).transfer()`), which moves
the buffers without a
+ copy and lets the batch outlive the reader.
+- **Close what you emit.** Lance wraps its iterator with
+ `Iterators.wrap(...).recyclePayload(_.close())`, like `ColumnarRangeExec`.
+- **Release native resources early and on task end.** Close a reader as soon
as it's exhausted,
+ and add a `TaskContext` completion listener as a safety net for `LIMIT` and
failures. Make
+ `close()` idempotent.
+- **Cross library boundaries with C structs only.** If the source has its own
Arrow Java build,
+ pass the address of the `ArrowArrayStream` / `ArrowArray` and wrap it with
Gluten's Arrow
+ (`ArrowArrayStream.wrap(address)`). Lance does this so that lance-spark and
Gluten can use
+ different Arrow versions.
+
+For an operator with a child (e.g. a UDF operator), the input batch is valid
only until the next
+call on the child iterator. If you keep it longer, for example until a result
comes back, call
+`ColumnarBatches.retain` on it and `ColumnarBatches.release` once you're done.
+
+### Step 3: Validate and fall back
+
+Only offload what the operator can run. Check unsupported cases at planning
time, not at run time:
+once a node is offloaded, an exception fails the query instead of falling
back. Lance keeps a scan
+on vanilla Spark when it has a pushed aggregation or a full-text query
+(`LanceScanTransformer.isExportable`). Put further checks in
`doValidateInternal()`.
+
+### Step 4: Write the offload rule
+
+Implement `OffloadSingleNode` and inject it with a validator, so that the rule
only fires for
+plans that pass validation:
+
+```scala
+case class OffloadMyOperator() extends OffloadSingleNode {
+ override def offload(plan: SparkPlan): SparkPlan = plan match {
+ case p: SomeSparkExec if MyArrowExec.isSupported(p) => MyArrowExec(p.child)
+ case other => other
+ }
+}
+
+object OffloadMyOperator {
+ def inject(injector: Injector): Unit = {
+ injector.gluten.legacy.injectTransform {
+ c =>
+ val offload = Seq(OffloadMyOperator())
+ HeuristicTransform.Simple(
+ Validators.newValidator(new GlutenConfig(c.sqlConf), offload),
+ offload)
+ }
+ }
+}
+```
+
+### Step 5: Register a component (for optional dependencies)
+
+If the operator depends on a library that isn't always present, register it as
a `Component`:
+
+- `dependencies()`: e.g. `classOf[VeloxBackend] :: Nil`.
+- `isRuntimeCompatible`: check that the library is on the classpath (Lance
checks for
+ `org.lance.spark.read.LanceScan`), so the default bundle can ship the
component.
+- `injectRules(injector)`: call the rule's `inject`.
+- List the class in `src-xxx/main/resources/META-INF/gluten-components/<class
name>`.
+- Put the sources under `backends-velox/src-xxx` behind a Maven profile, like
`src-lance` and
+ `src-iceberg`. Exclude the library's transitive Arrow so Gluten's Arrow
version is kept.
+
+### Step 6: Test
+
+Cover at least the following, as `VeloxLanceSuite` does:
+
+- The node is offloaded, and the plan has the expected transitions: none to a
vanilla row consumer,
+ and `OffloadArrowDataExec` → `ArrowColumnarToVeloxColumnarExec` to a Velox
consumer.
+- Results match vanilla Spark over several batches and several partitions,
with nulls.
+- `LIMIT` and other early stops release native resources.
+- Unsupported cases fall back to vanilla Spark.
+
+Add the profile to the CI job that runs the suite, and a page under
+`docs/get-started/` if users have to enable anything.
+
+### Checklist
+
+- [ ] `batchType()` / `rowType()` / `requiredChildConvention()` declared; no
transitions inserted
+ by hand
+- [ ] Columns are `ArrowWritableColumnVector`s from
`ArrowBufferAllocators.contextInstance()`
+- [ ] Emitted batches own their buffers; reused reader vectors are transferred
first
+- [ ] Input batches kept beyond the next call are retained and released
+- [ ] Resources released on exhaustion and on task completion
+- [ ] Unsupported cases rejected at planning time
+- [ ] Rule injected with a validator; component gated on the classpath
+- [ ] Tests for plan shape, results, early stop and fallback
+
+## Ongoing work
+
+These change parts of the above once merged:
+
+- [#13280](https://github.com/apache/gluten/pull/13280) replaces
`ArrowWritableColumnVector` with
+ `RefCountedArrowColumnVector`, a subclass of Spark's `ArrowColumnVector`,
and replaces
+ `ArrowProjection` with Spark's `ArrowWriter`. New operators then build
columns with
+ `ArrowColumnVectors.toBatch(root)` / `ArrowColumnVectors.fromRows(...)`.
`offload` also accepts
+ batches of plain Spark `ArrowColumnVector`s, which it moves (same allocator
root) or copies into
+ Gluten's allocator.
+- SPARK-57468
([apache/spark#59154](https://github.com/apache/spark/pull/59154)) adds
conventions to
+ Spark itself, with an `ArrowBatchType`. With the prototype bridge in
+ [#13140](https://github.com/apache/gluten/issues/13140),
`ArrowJavaBatchType` is bound to Spark's
+ `ArrowBatchType`, so a Spark operator that declares `ArrowBatchType` gets
Gluten's transitions
+ without a Gluten operator. Gluten's own `ColumnarArrowEvalPythonExec` and
Python runner are then
+ replaced by Spark's `ArrowEvalPythonExec`.
+- Spark's batch ownership rules for Arrow vectors are still being settled
upstream (SPARK-60294,
+ SPARK-60306, SPARK-60325). Operators that keep Arrow batches across calls
should follow them once
+ merged.
diff --git a/docs/image/ArrowBufReleaseDispatch.png
b/docs/image/ArrowBufReleaseDispatch.png
new file mode 100644
index 0000000000..bfd326b73d
Binary files /dev/null and b/docs/image/ArrowBufReleaseDispatch.png differ
diff --git a/docs/image/GlutenBatchTypeTransition.png
b/docs/image/GlutenBatchTypeTransition.png
new file mode 100644
index 0000000000..7193957905
Binary files /dev/null and b/docs/image/GlutenBatchTypeTransition.png differ
diff --git a/docs/image/GlutenConversionImages.png
b/docs/image/GlutenConversionImages.png
new file mode 100644
index 0000000000..ac9ffcfbe7
Binary files /dev/null and b/docs/image/GlutenConversionImages.png differ
diff --git a/docs/image/VeloxColumnArrowUdfLifecycle.png
b/docs/image/VeloxColumnArrowUdfLifecycle.png
new file mode 100644
index 0000000000..54a7343efb
Binary files /dev/null and b/docs/image/VeloxColumnArrowUdfLifecycle.png differ
diff --git a/mkdocs.yml b/mkdocs.yml
index 1aa7e16e3a..4d4f355d1c 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -67,6 +67,7 @@ nav:
- Memory Profiling: developers/ProfileMemoryOfGlutenWithVelox.md
- Dynamic Off-Heap Sizing: developers/VeloxDynamicSizingOffheap.md
- Partial Project: developers/PartialProject.md
+ - Arrow Columnar Conversion: developers/ArrowColumnarConversion.md
- Bloop Integration: developers/bloop-integration.md
- Debug CH Backend: developers/clickhouse-backend-debug.md
- Jemalloc with CH: developers/UsingJemallocWithCH.md
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
