FelixYBW commented on issue #13140:
URL: https://github.com/apache/gluten/issues/13140#issuecomment-6281254885
# Arrow UDF support in Gluten: today vs. target (Spark 4.2 + SPARK-57468)
**Goal:** Gluten supports Spark's Arrow-based columnar operators
(`ArrowEvalPythonExec`,
`ArrowEvalPythonUDTFExec`, ...) in one of two ways.
- Keeps Spark's operator and only inserts Arrow <-> Velox transitions
- It adds a thin columnar wrapper. Gluten stops re-implementing Spark's
operators and Python runners.
## 0. Sources reviewed
Spark commits on `SPARK-57468-convention-4.2`:
| Commit | Change |
|---|---|
| `98c9eac81ee` | [SPARK-57468] Convention API: `RowType` / `BatchType` /
`ArrowBatchType`, `TransitionGraph`, `SparkPlan.convention` /
`requiredChildConventions`, graph-based transition insertion (+593/−53) |
| `effa8e8cd3d` | Follow-up: `ArrowEvalPythonExec` consumes and produces
`ArrowBatchType` |
| `8341e63bf92` | Follow-up: ownership contract of `ArrowBatchType` batches,
and `ArrowBatchOwner` |
| `5c4190677bb` | Follow-up: buffers of a producer that has an allocation
listener stay in the producer's allocator |
Gluten commits on `spark-convention-arrow-udf-4.2`:
| Commit | Change |
|---|---|
| `4a5a7a46dd` | `SparkConventions` bridge, `SparkArrowBatchType`, rewrite
of `ColumnarArrowEvalPythonExec` on Spark APIs only, `BasePythonRunnerShim`
removed |
| `bc293a5c77` | Replace `ArrowWritableColumnVector` with Spark's
`ArrowColumnVector` + `ArrowWriter` ("Arrow write"), `ArrowJavaBatchType` bound
to Spark's `ArrowBatchType` (+378/−4420) |
| `65db047126` | Port to Spark 4.2: use Spark's columnar Arrow runner
(SPARK-56350) |
| `8a430f38e4` | Remove Gluten's `ColumnarArrowEvalPythonExec` and use
Spark's `ArrowEvalPythonExec` directly |
| `ec3f05360d` | Gluten's per-task Arrow allocator becomes a child of
`ArrowUtils.rootAllocator`, so UDF results move into it without a copy |
Upstream status (2026-10-13):
- SPARK-56350 / apache/spark#55120 (columnar input for Arrow UDFs) is in
branch-4.2.
- SPARK-57468 / apache/spark#59154 is open with no reviews yet. It targets
master, so the 4.2 line
needs a backport or our own patch.
- SPARK-59792 / apache/spark#59062 (UDTF columnar input) is open.
- The three SPARK-57468 follow-ups are only on the branch; no upstream PR
exists for them yet.
- Gluten #12276 (Arrow UDTF) is open.
---
## 1. What Gluten does today (Gluten main)
### Plan
```mermaid
flowchart TD
V1["VeloxOp<br/>(VeloxBatchType)"]
L["[LoadArrowDataExec]<br/>native Arrow -> Java Arrow<br/>(C Data;
copies strings and dictionaries)"]
subgraph E["ColumnarArrowEvalPythonExec — Gluten's own operator
(ArrowJavaBatchType in and out)"]
R["ColumnarArrowPythonRunner<br/>Gluten's own runner<br/>built on
BasePythonRunnerShim"]
end
O["[OffloadArrowDataExec]<br/>Java Arrow -> native Arrow"]
A["[ArrowColumnarToVeloxColumnarExec]"]
V2["VeloxOp"]
V1 --> L --> E --> O --> A --> V2
classDef gluten fill:#fde7e7,stroke:#c0392b,color:#1f2328;
classDef inserted fill:#e8f0fb,stroke:#3b82d4,color:#1f2328;
classDef velox fill:#eef7ee,stroke:#2e7d32,color:#1f2328;
class R gluten;
style E fill:#fdf3f3,stroke:#c0392b,color:#1f2328;
class L,O,A inserted;
class V1,V2 velox;
```
Gluten's `ConventionFunc` sees every other Spark Arrow operator as a vanilla
row or vanilla
columnar plan. Gluten then plans `VeloxColumnarToRow -> Spark op ->
RowToVeloxColumnar`.
### Components Gluten owns today
| Area | Gluten code | Size |
|---|---|---|
| Python exec |
`backends-velox/.../spark/api/python/ColumnarArrowEvalPythonExec.scala` (exec +
`ColumnarArrowPythonRunner`, `adjustRefCnt`, input batch cache with retain /
release) | 579 lines |
| Worker protocol shims |
`shims/spark3x,40,41/.../BasePythonRunnerShim.scala` | 4 files, about 66 lines
each |
| Arrow vector | `gluten-arrow/.../ArrowWritableColumnVector.java` (read,
write, ref count) | 2,309 lines |
| Vector API shims | `shims/*/WritableColumnVectorShim.java` | 4 × about 223
lines |
| Arrow write helpers | `ArrowProjection`, `GenerateArrowProjection`,
`InterpretedArrowProjection`, `ArrowColumnarBatch`, `ArrowColumnarRow`,
`ArrowColumnVectorUtils` | about 690 lines |
| UDTF | #12276 (open): `ColumnarArrowEvalPythonUDTFExec` + per-version
`ArrowEvalPythonUDTFShim`, then a copy of the Python output into Gluten's
allocator | – |
| Pre-project | `PullOutArrowEvalPythonPreProjectHelper` (complex UDF args
-> columns) | – |
| Allocator | Per-task `RootAllocator` with `ManagedAllocationListener`,
separate from Spark's root | – |
### Pain points
1. **Duplicated Spark logic.** Gluten must track every change to Spark's
Python runner and worker
protocol (e.g. SPARK-44705, SPARK-51384 `runnerConf` / `evalConf`). Each
new Spark Arrow
operator needs its own Gluten re-implementation.
2. **No Arrow type in Spark's planner.** A vanilla plan with
`supportsColumnar = true` is always
`VanillaBatchType`, so "Velox -> Arrow -> Spark operator -> Arrow ->
Velox" needs special cases
for each operator.
3. **Unsafe output hand-off.** `BasicPythonArrowOutput` allocates from a
separate child of Spark's
root and closes it at end of stream. Gluten has its own root, so it must
copy the UDF output
(#12276), and Spark's side of that memory isn't tracked.
4. **Ref-count lifecycle.** `ColumnarBatches` loads and offloads batches in
place, with reference
counts kept by `ArrowWritableColumnVector` and `IndicatorVector`. This
leads to workarounds such
as `adjustRefCnt`.
5. **Everything else falls back to rows**: UDTF (until #12276), `MapInArrow`
/ `MapInPandas`,
`FlatMapGroupsIn*`, `FlatMapCoGroupsIn*`, `ArrowAggregatePython`,
`ArrowWindowPython`.
---
## 2. What Gluten will do (Spark 4.2 + SPARK-57468 and follow-ups, Gluten
Arrow write)
### Plan
The plan below is produced by `ArrowEvalPythonExecSuite` on the 4.2 branch.
Every bracketed node
is inserted automatically.
```mermaid
flowchart TD
V1["VeloxOp<br/>(VeloxBatchType)"]
L["[LoadArrowDataExec]<br/>reports Spark's
ArrowBatchType<br/>(ArrowJavaBatchType is bound to it)"]
subgraph E["ArrowEvalPythonExec"]
direction TB
T["Spark's own operator, unchanged<br/>convention = (VanillaRow,
ArrowBatchType)"]
R["Spark's Python
runner<br/>(ColumnarArrowPythonWithNamedArgumentRunner)"]
B["ArrowBatchOwner<br/>takes ownership of input vectors<br/>by
transfer (no copy)"]
end
O["[OffloadArrowDataExec]<br/>adopt(): moves vectors, same root
allocator,<br/>no copy for UDF results"]
A["[ArrowColumnarToVeloxColumnarExec]"]
V2["VeloxOp"]
V1 --> L --> E --> O --> A --> V2
classDef spark fill:#fff4e0,stroke:#d98e04,color:#1f2328;
classDef inserted fill:#e8f0fb,stroke:#3b82d4,color:#1f2328;
classDef velox fill:#eef7ee,stroke:#2e7d32,color:#1f2328;
class T,R,B spark;
style E fill:#fffaf0,stroke:#d98e04,color:#1f2328;
class L,O,A inserted;
class V1,V2 velox;
```
### How it works
- **Spark** (`SparkPlan.convention`, `requiredChildConventions`,
`TransitionGraph`): every plan
declares its row and batch types. `ArrowBatchType` means "a batch of
`ArrowColumnVector`", and
its transition to `VanillaBatchType` is free.
- **Spark `ArrowEvalPythonExec`**: if `arrowBatchInput` holds, the operator
declares
`(VanillaRowType, ArrowBatchType)`, requires Arrow batches from its child,
always takes the Arrow
columnar path, and outputs the child's columns followed by the UDF results
as Arrow. The
condition is: the child's batch type is not vanilla, a path to
`ArrowBatchType` exists, and every
UDF input is a child column without UDTs.
- **Ownership contract**: an `ArrowBatchType` batch is valid until the next
`hasNext` / `next`. To
keep a batch longer, a consumer transfers its vectors. The vectors move to
the consumer's
allocator only when it shares the root and the producer's allocator has no
listener. Otherwise
they stay in the producer's allocator.
- **Gluten bridge** (`SparkConventions` + 7 lines in `ConventionFunc` +
`GlutenPlan.convention`):
- A Spark plan that declares a non-vanilla Spark type is planned as the
Gluten type bound to it.
- A Gluten plan whose type is bound to a Spark type reports that type to
Spark. For example,
`LoadArrowDataExec` reports `ArrowBatchType`.
- Gluten plans declare `Any` on their children to Spark, because Gluten
plans its own transitions.
- **Gluten "Arrow write"**:
- `RefCountedArrowColumnVector` (a subclass of Spark's
`ArrowColumnVector`) replaces
`ArrowWritableColumnVector`.
- Rows are written to Arrow with Spark's `ArrowWriter`
(`ArrowColumnVectors.fromRows`) in
`ColumnarPartialProjectExec`, `ColumnarPartialGenerateExec`,
`ColumnarRangeExec` and
`ExecUtil`.
- `ArrowJavaBatchType` is bound to Spark's `ArrowBatchType`.
- **Allocator**: Gluten's task allocator is a child of
`ArrowUtils.rootAllocator`. `adopt()`
transfers Spark's vectors and moves their accounting between the two
allocation listeners.
### Copies left on the path (from the "final plan" comment in #13140)
| Step | Copy? |
|---|---|
| Velox -> Java Arrow | Fixed-width: none. Dictionary or constant vectors:
flattened (copy). Strings: copied, because Arrow string views aren't enabled |
| Spark `ArrowEvalPythonExec` input | No conversion. The UDF columns are
serialized to IPC directly from Gluten's buffers |
| Spark -> Gluten (offload) | None for pass-through columns or UDF results
(same root since `ec3f05360d`) |
| Java Arrow -> Velox | Fixed-width: none. Strings: a new string-view buffer
(strings of 12 bytes or less are inlined) |
---
## 3. Side-by-side comparison
| Aspect | Today (Gluten main) | Target (Spark 4.2 + SPARK-57468, Gluten
branch) |
|---|---|---|
| Operator executing Arrow UDFs | Gluten `ColumnarArrowEvalPythonExec` (a
`GlutenPlan`) | Spark `ArrowEvalPythonExec`, unchanged and not offloaded |
| Python runner / worker protocol | Gluten `ColumnarArrowPythonRunner` +
`BasePythonRunnerShim` × 4 | Spark's runners; Gluten has no Python runner code |
| How Gluten learns the operator wants Arrow | Hard-coded offload rule
(`createColumnarArrowEvalPythonExec`) | `SparkPlan.convention` =
`ArrowBatchType`, mapped through `SparkConventions` |
| Transition insertion | Gluten's own operator declares `ArrowJavaBatchType`
| Generic: any Spark plan declaring a bound type gets Gluten's transitions |
| Operator-specific Gluten code | Exec + runner + shims (about 850 lines) |
`InsertArrowInputForArrowEvalPython` (53 lines), pre-project helper |
| Java Arrow vector | `ArrowWritableColumnVector` +
`WritableColumnVectorShim` × 4 | Spark `ArrowColumnVector`
(`RefCountedArrowColumnVector`, 57 lines) |
| Writing rows into Arrow | `ArrowWritableColumnVector.put*`, Gluten
`ArrowProjection` codegen | Spark `ArrowWriter` + `MutableProjection` |
| Gluten Arrow batch type vs. Spark | Unrelated; Spark sees vanilla |
`ArrowJavaBatchType` ≡ Spark `ArrowBatchType` |
| Input batch lifetime in UDF exec | Retain / release + `adjustRefCnt` |
`ArrowBatchOwner` transfers vectors, releases after the output is consumed |
| UDF result columns | Read by Gluten's runner into Gluten's allocator |
Read by Spark's runner, moved into Gluten by `adopt()` (same root, no copy) |
| Arrow allocator | Separate Gluten `RootAllocator` | Child of Spark's
`ArrowUtils.rootAllocator`, with listener accounting moved on transfer |
| Memory accounting of UDF output | Spark side untracked | Moved to Gluten's
`ManagedAllocationListener` on adopt |
| UDTF | #12276: Gluten exec + shim + output copy | Spark
`ArrowEvalPythonUDTFExec` after SPARK-59792 + an `ArrowBatchType` declaration
(still to do) |
| MapInArrow / FlatMapGroups / Window / Agg | Row fallback (C2R + R2C) |
Still row fallback until Spark operators declare `ArrowBatchType` (still to do)
|
| New Spark Arrow operator | New Gluten re-implementation | Zero Gluten code
once it declares `ArrowBatchType` |
| Supported Spark versions | 3.4 – 4.1 (4.2 via #13126) | Spark with
SPARK-57468 only; older versions need the old path kept (still to do) |
| Code delta | – | Spark +939/−79; Gluten +1,370/−6,025 (`bc293a5c77` alone
is +378/−4,420) |
| Performance (5M rows, 4 pandas UDFs) | 2.14–2.22 s | 2.24–2.35 s (within
noise, measured before `ec3f05360d` removed the result copy) |
---
## 4. Prototype status
Done and tested on the branches:
- Spark: convention API, `ConventionSuite`; `ArrowEvalPythonExec` on
`ArrowBatchType`; ownership
contract; listener-aware transfer. All are covered by
`ArrowColumnarPythonUDFSuite`.
- Gluten: bridge; `ArrowJavaBatchType` bound to Spark; switch to
`ArrowColumnVector` /
`ArrowWriter`; Gluten's Python exec and shims removed; shared root
allocator.
`ArrowEvalPythonExecSuite`, `ColumnarBatchTest`, `ArrowColumnVectorTest`
and the transition and
fallback suites pass.
Gaps in the prototype:
1. `InsertArrowInputForArrowEvalPython` is still a rule specific to
`ArrowEvalPythonExec`. Spark
picks the Arrow path from `child.convention`, and a Velox child reports
`VanillaBatchType`, so
Gluten has to insert `LoadArrowDataExec` first and then check what Spark
decided.
2. `gluten-core` / `gluten-arrow` reference
`org.apache.spark.sql.execution.convention` directly.
They no longer compile against Spark versions without SPARK-57468, so the
bridge needs a shim
(or a no-op).
3. `backends-bolt` still uses the removed `BasePythonRunnerShim` /
`ArrowWritableColumnVector` and
doesn't compile.
4. `RefCountedArrowColumnVector` keeps Gluten's reference count. The
in-place load / offload
lifecycle in `ColumnarBatches` hasn't been moved to ownership transfer
yet.
5. The fallback reporter lists Spark Arrow operators planned through a
convention as "not
supported" (log noise only).
6. The write-path performance of `ArrowWriter` vs. the old codegen
projection hasn't been
benchmarked.
7. Only `ArrowEvalPythonExec` is covered.
---
## 5. Features to add
### 5.1 Spark (target: our 4.2 line, and upstream master)
| # | Feature | Why | Status |
|---|---|---|---|
| S1 | **SPARK-57468 convention API**: `Convention`, `RowType`, `BatchType`
(`Vanilla*`, `ArrowBatchType`), `TransitionGraph` with transition costs,
`SparkPlan.convention` / `requiredChildConventions`, graph-based
`ApplyColumnarRulesAndInsertTransitions`, convention propagation through
`QueryStageExec` / `Exchange` / `DataWritingCommandExec` | The foundation: lets
a plan declare the Arrow layout and lets plugins bind their types | Branch
`98c9eac81ee`; upstream PR apache/spark#59154 open |
| S2 | **`ArrowEvalPythonExec` declares `ArrowBatchType`**
(`arrowBatchInput`, `inputsAreChildColumns`, always takes the Arrow path,
outputs Arrow) | Spark's operator is used as is, with no Gluten wrapper |
Branch `effa8e8cd3d`; no upstream PR yet |
| S3 | **`ArrowBatchType` ownership contract** + `ArrowBatchOwner` (transfer
on read, release after the output is consumed, close at end of output or task)
| Makes zero-copy hand-off safe for producers that allocate per batch or reuse
vectors (Velox -> Arrow, DSv2 readers) | Branch `8341e63bf92`; no upstream PR
yet |
| S4 | **Listener-aware transfer**: keep a producer's buffers in its
allocator when it has an `AllocationListener` or a different root | Without it,
Gluten's memory reservation grows until the task ends | Branch `5c4190677bb`;
no upstream PR yet |
| S5 | **SPARK-59792 UDTF columnar input** + make `ArrowEvalPythonUDTFExec`
declare `ArrowBatchType` the same way as S2 / S3 | Replaces Gluten #12276 |
apache/spark#59062 open; the `ArrowBatchType` declaration is not started |
| S6 | **A planner-level "prefer Arrow" request** for operators that benefit
from Arrow input, e.g. let `ArrowEvalPythonExec` require `ArrowBatchType`
whenever any registered path from the child exists, not only when the child
already declares a non-vanilla type. Alternative: Gluten makes Velox visible to
Spark as a Spark `BatchType` (see G5) | Removes Gluten's
`InsertArrowInputForArrowEvalPython`, so support is fully automatic | Not
started; needs design |
| S7 | **Other Arrow operators declare `ArrowBatchType` input and output**:
`MapInArrowExec` / `MapInPandasExec`, `FlatMapGroupsInArrowExec` / `Pandas`,
`FlatMapCoGroupsInArrowExec` / `Pandas`, `ArrowAggregatePythonExec`,
`ArrowWindowPythonExec` | Today they take rows and output rows, so Gluten does
C2R + R2C around them. The grouped and window operators need a group-aware
Arrow input, which is harder | Not started |
| S8 | **`RowToArrowColumnarExec`** on `ArrowWriter`, registered as
`ArrowBatchType.fromRow` (revive SPARK-37124) | Planner path rows -> Arrow, so
consumers can drop their row-by-row fallbacks. Also covers local table scan,
RDD scan and range | Not started |
| S9 | **Public Arrow batch utilities**: `isArrowBacked` + layout check
(`ArrowUtils.isCompatibleWithDeclaredField`), unwrap a batch into a
`VectorSchemaRoot` / record batch, take ownership by transfer | Gluten, the
Arrow cache (SPARK-57268) and the UDTF path stop duplicating private helpers |
Not started |
| S10 | *(Optional)* Direct vanilla columnar -> `ArrowBatchType` transition
(on-heap / off-heap vectors -> Arrow) | Parquet / ORC vectorized reads feeding
UDFs without a row round trip | Not started |
| S11 | *(Optional)* Python reader allocator closed at task end rather than
at end of stream | No longer needed for Gluten once G6 is in (same root plus
adopt-on-consume). Still useful for other consumers that keep output batches |
Not needed for Gluten |
| S12 | *(Optional)* Arrow string view (`Utf8View`) support in
`ArrowColumnVector` / the IPC path to pyarrow | Removes the string copy on
Velox -> Arrow export (see G10) | Not started |
| S13 | *(Optional)* Spark-side rule that pulls complex UDF arguments into a
`ProjectExec` (the TODO in `ColumnarArrowEvalPythonEvaluatorFactory`) |
Satisfies `inputsAreChildColumns` in plain Spark too; Gluten could then drop
`PullOutArrowEvalPythonPreProjectHelper` | Not started |
### 5.2 Gluten
| # | Feature | Why | Status |
|---|---|---|---|
| G1 | **`SparkConventions` bridge**: bind Spark and Gluten types;
`ConventionFunc` maps a Spark plan's declared type and child requirements;
`GlutenPlan.convention` reports bound types to Spark, with `Any` child
requirements | Generic transition insertion for any Spark operator that
declares a bound type | Done (`4a5a7a46dd`, `8a430f38e4`) |
| G2 | **Gluten Arrow batches on Spark's `ArrowColumnVector`**:
`RefCountedArrowColumnVector`; `ArrowJavaBatchType` bound to Spark
`ArrowBatchType`; delete `ArrowWritableColumnVector`,
`WritableColumnVectorShim` × 4, `ArrowColumnarBatch`, `ArrowColumnarRow`,
`ArrowColumnVectorUtils` | One Java Arrow batch type shared with Spark; no
Spark-to-Gluten Arrow transitions; no shims | Done (`bc293a5c77`) |
| G3 | **Arrow write via Spark's `ArrowWriter`**:
`ColumnarPartialProjectExec` (`MutableProjection` + `ArrowWriter`),
`ColumnarPartialGenerateExec`, `ColumnarRangeExec`, `ExecUtil`,
`ArrowColumnVectors.fromRows`; drop the `ArrowProjection` codegen | Removes
Gluten's own Arrow writer and codegen | Done (`bc293a5c77`); **benchmark
pending** |
| G4 | **Remove Gluten's Python exec**: `ColumnarArrowEvalPythonExec`,
`ColumnarArrowPythonRunner`, `BasePythonRunnerShim` × 5;
`createColumnarArrowEvalPythonExec` returns Spark's exec | Gluten no longer
owns any Python runner or worker protocol code | Done for Velox (`8a430f38e4`) |
| G5 | **Remove `InsertArrowInputForArrowEvalPython`**: once S6 lands, or by
making Velox visible to Spark (bind `VeloxBatchType` to a Spark `BatchType`
with a registered path to `ArrowBatchType`) | Fully automatic support for any
Arrow operator, with no rule per operator | Interim rule done (53 lines);
removal not started |
| G6 | **Shared root allocator**: the task allocator is a child of
`ArrowUtils.rootAllocator`; `ArrowColumnVectors.adopt` transfers vectors and
moves listener accounting | Zero-copy UDF results; memory accounted by Gluten |
Done (`ec3f05360d`) |
| G7 | **Version gating / shim for the bridge**: keep the today path (Gluten
exec + runner + `ArrowWritableColumnVector`) for Spark 3.4 – 4.1 (and 4.2
without SPARK-57468); put `SparkConventions`, the `GlutenPlan.convention`
overrides and the `ArrowBatchType` binding behind a shim | The branch only
compiles against patched Spark | Not started |
| G8 | **Lifecycle rewrite**: replace reference counts and in-place load /
offload in `ColumnarBatches` with "the producer owns a batch until the next
one; a consumer that keeps it transfers the vectors". The `IndicatorVector`
reference count stays for native batches shared by limit, tail and broadcast |
Removes the reference count from `RefCountedArrowColumnVector`; matches Spark's
`ArrowBatchType` contract | Not started (largest and riskiest part) |
| G9 | **Fallback reporter**: skip Spark operators planned through a Spark
convention | Spurious "not supported" entries | Not started |
| G10 | *(Optional)* Velox -> Arrow export: enable Arrow string views and
avoid flattening where the consumer accepts it (needs S12) | Removes the
remaining string and dictionary copies | Not started |
| G11 | **UDTF**: drop #12276 in favor of S5; Gluten only needs the existing
transitions | No Gluten UDTF exec, shim or output copy | Waiting for S5 |
| G12 | **Other Arrow operators** (S7): no Gluten code beyond tests once
Spark declares `ArrowBatchType`. Before S7, a one-line convention declaration
(rule) per operator is enough for the input side | Removes the C2R + R2C around
them | Waiting for S7 |
| G13 | **Pre-project**: keep `PullOutArrowEvalPythonPreProjectHelper` (in
its own file now) so complex UDF arguments become columns and
`inputsAreChildColumns` holds; drop it if S13 lands | Otherwise Spark takes the
row path | Done (kept) |
| G14 | **`backends-bolt` migration** to the same model | It doesn't compile
after G2 / G4 | Not started |
| G15 | **Tests**: `gluten-ut` coverage for the Spark Arrow operators on
Velox input (multi-batch, nulls, chained UDFs, ANSI on and off, AQE, config
`columnarInput.enabled` off), memory leak checks, and an end-to-end pyspark
benchmark after G6 / G3 | Required before upstreaming | Partly done
(`ArrowEvalPythonExecSuite` with ANSI off) |
---
## 6. Suggested order
1. **Upstream S1** (apache/spark#59154), then S2 – S4 as SPARK-57468
follow-ups. Backport them to
our 4.2 line.
2. **Gluten G7** (shim and version gating), so G1 – G6 can merge into Gluten
main without breaking
Spark 3.4 – 4.1.
3. **Merge G1, G2, G3, G4, G6 and G13** for Spark builds with SPARK-57468,
after the G3 benchmark
and G15 tests.
4. **S5 + G11** (UDTF), replacing #12276.
5. **S6 + G5**, so the remaining operator-specific rule goes away.
6. **G8** (lifecycle rewrite), then G9 and G14.
7. **S7 / G12** (other Arrow operators), then S8 – S10 and S12 / G10 as
optimizations.
## 7. Risks and open points
- **Upstream acceptance of S1 – S4.** Spark has to accept a plugin-facing
plan API and an ownership
contract. The UDF-specific follow-ups may need to be generalized (S6)
before review.
- **Planning interplay.** Spark's `insertTransitions` and Gluten's
`InsertTransitions` both run. It
must be verified that they never insert competing transitions. Gluten
plans report `Any` child
requirements to Spark for this reason.
- **Memory accounting.** Arrow doesn't notify listeners of transfers. Both
S4 and `adopt()`
compensate for this manually, which is fragile if Arrow changes its
behavior.
- **Grouped and window operators (S7)** need grouped Arrow input. That's a
bigger Spark change than
for map-style operators.
- **Older Spark versions** keep today's code until they're dropped, so
Gluten carries both paths
for a while.
# Arrow UDF support in Gluten: today vs. target (Spark 4.2 + SPARK-57468)
Tracking issue:
[apache/gluten#13140](https://github.com/apache/gluten/issues/13140)
**Goal:** Gluten supports Spark's Arrow-based columnar operators
(`ArrowEvalPythonExec`,
`ArrowEvalPythonUDTFExec`, ...) in one of two ways. It either keeps Spark's
operator and only
inserts Arrow <-> Velox transitions, or it adds a thin columnar wrapper.
Gluten stops
re-implementing Spark's operators and Python runners.
## 0. Sources reviewed
| Repo | Location | Branch / base |
|---|---|---|
| Gluten (today) | local `main` (`8fa03c0fe`) | apache/gluten main, Spark
3.4 – 4.1 shims |
| Spark (target) | `[email protected]:2202:/spark` |
`SPARK-57468-convention-4.2` on top of `v4.2.1` (`4a1cd45b891`) |
| Gluten (target) | `[email protected]:2202:/incubator-gluten` |
`spark-convention-arrow-udf-4.2` on top of the Spark 4.2 support (#13126) |
Spark commits on `SPARK-57468-convention-4.2`:
| Commit | Change |
|---|---|
| `98c9eac81ee` | [SPARK-57468] Convention API: `RowType` / `BatchType` /
`ArrowBatchType`, `TransitionGraph`, `SparkPlan.convention` /
`requiredChildConventions`, graph-based transition insertion (+593/−53) |
| `effa8e8cd3d` | Follow-up: `ArrowEvalPythonExec` consumes and produces
`ArrowBatchType` |
| `8341e63bf92` | Follow-up: ownership contract of `ArrowBatchType` batches,
and `ArrowBatchOwner` |
| `5c4190677bb` | Follow-up: buffers of a producer that has an allocation
listener stay in the producer's allocator |
Gluten commits on `spark-convention-arrow-udf-4.2`:
| Commit | Change |
|---|---|
| `4a5a7a46dd` | `SparkConventions` bridge, `SparkArrowBatchType`, rewrite
of `ColumnarArrowEvalPythonExec` on Spark APIs only, `BasePythonRunnerShim`
removed |
| `bc293a5c77` | Replace `ArrowWritableColumnVector` with Spark's
`ArrowColumnVector` + `ArrowWriter` ("Arrow write"), `ArrowJavaBatchType` bound
to Spark's `ArrowBatchType` (+378/−4420) |
| `65db047126` | Port to Spark 4.2: use Spark's columnar Arrow runner
(SPARK-56350) |
| `8a430f38e4` | Remove Gluten's `ColumnarArrowEvalPythonExec` and use
Spark's `ArrowEvalPythonExec` directly |
| `ec3f05360d` | Gluten's per-task Arrow allocator becomes a child of
`ArrowUtils.rootAllocator`, so UDF results move into it without a copy |
Upstream status (2026-10-13):
- SPARK-56350 / apache/spark#55120 (columnar input for Arrow UDFs) is in
branch-4.2.
- SPARK-57468 / apache/spark#59154 is open with no reviews yet. It targets
master, so the 4.2 line
needs a backport or our own patch.
- SPARK-59792 / apache/spark#59062 (UDTF columnar input) is open.
- The three SPARK-57468 follow-ups are only on the branch; no upstream PR
exists for them yet.
- Gluten #12276 (Arrow UDTF) is open.
---
## 1. What Gluten does today (Gluten main)
### Plan
```mermaid
flowchart TD
V1["VeloxOp<br/>(VeloxBatchType)"]
L["[LoadArrowDataExec]<br/>native Arrow -> Java Arrow<br/>(C Data;
copies strings and dictionaries)"]
subgraph E["ColumnarArrowEvalPythonExec — Gluten's own operator
(ArrowJavaBatchType in and out)"]
R["ColumnarArrowPythonRunner<br/>Gluten's own runner<br/>built on
BasePythonRunnerShim"]
end
O["[OffloadArrowDataExec]<br/>Java Arrow -> native Arrow"]
A["[ArrowColumnarToVeloxColumnarExec]"]
V2["VeloxOp"]
V1 --> L --> E --> O --> A --> V2
classDef gluten fill:#fde7e7,stroke:#c0392b,color:#1f2328;
classDef inserted fill:#e8f0fb,stroke:#3b82d4,color:#1f2328;
classDef velox fill:#eef7ee,stroke:#2e7d32,color:#1f2328;
class R gluten;
style E fill:#fdf3f3,stroke:#c0392b,color:#1f2328;
class L,O,A inserted;
class V1,V2 velox;
```
Gluten's `ConventionFunc` sees every other Spark Arrow operator as a vanilla
row or vanilla
columnar plan. Gluten then plans `VeloxColumnarToRow -> Spark op ->
RowToVeloxColumnar`.
### Components Gluten owns today
| Area | Gluten code | Size |
|---|---|---|
| Python exec |
`backends-velox/.../spark/api/python/ColumnarArrowEvalPythonExec.scala` (exec +
`ColumnarArrowPythonRunner`, `adjustRefCnt`, input batch cache with retain /
release) | 579 lines |
| Worker protocol shims |
`shims/spark3x,40,41/.../BasePythonRunnerShim.scala` | 4 files, about 66 lines
each |
| Arrow vector | `gluten-arrow/.../ArrowWritableColumnVector.java` (read,
write, ref count) | 2,309 lines |
| Vector API shims | `shims/*/WritableColumnVectorShim.java` | 4 × about 223
lines |
| Arrow write helpers | `ArrowProjection`, `GenerateArrowProjection`,
`InterpretedArrowProjection`, `ArrowColumnarBatch`, `ArrowColumnarRow`,
`ArrowColumnVectorUtils` | about 690 lines |
| UDTF | #12276 (open): `ColumnarArrowEvalPythonUDTFExec` + per-version
`ArrowEvalPythonUDTFShim`, then a copy of the Python output into Gluten's
allocator | – |
| Pre-project | `PullOutArrowEvalPythonPreProjectHelper` (complex UDF args
-> columns) | – |
| Allocator | Per-task `RootAllocator` with `ManagedAllocationListener`,
separate from Spark's root | – |
### Pain points
1. **Duplicated Spark logic.** Gluten must track every change to Spark's
Python runner and worker
protocol (e.g. SPARK-44705, SPARK-51384 `runnerConf` / `evalConf`). Each
new Spark Arrow
operator needs its own Gluten re-implementation.
2. **No Arrow type in Spark's planner.** A vanilla plan with
`supportsColumnar = true` is always
`VanillaBatchType`, so "Velox -> Arrow -> Spark operator -> Arrow ->
Velox" needs special cases
for each operator.
3. **Unsafe output hand-off.** `BasicPythonArrowOutput` allocates from a
separate child of Spark's
root and closes it at end of stream. Gluten has its own root, so it must
copy the UDF output
(#12276), and Spark's side of that memory isn't tracked.
4. **Ref-count lifecycle.** `ColumnarBatches` loads and offloads batches in
place, with reference
counts kept by `ArrowWritableColumnVector` and `IndicatorVector`. This
leads to workarounds such
as `adjustRefCnt`.
5. **Everything else falls back to rows**: UDTF (until #12276), `MapInArrow`
/ `MapInPandas`,
`FlatMapGroupsIn*`, `FlatMapCoGroupsIn*`, `ArrowAggregatePython`,
`ArrowWindowPython`.
---
## 2. What Gluten will do (Spark 4.2 + SPARK-57468 and follow-ups, Gluten
Arrow write)
### Plan
The plan below is produced by `ArrowEvalPythonExecSuite` on the 4.2 branch.
Every bracketed node
is inserted automatically.
```mermaid
flowchart TD
V1["VeloxOp<br/>(VeloxBatchType)"]
L["[LoadArrowDataExec]<br/>reports Spark's
ArrowBatchType<br/>(ArrowJavaBatchType is bound to it)"]
subgraph E["ArrowEvalPythonExec"]
direction TB
T["Spark's own operator, unchanged<br/>convention = (VanillaRow,
ArrowBatchType)"]
R["Spark's Python
runner<br/>(ColumnarArrowPythonWithNamedArgumentRunner)"]
B["ArrowBatchOwner<br/>takes ownership of input vectors<br/>by
transfer (no copy)"]
end
O["[OffloadArrowDataExec]<br/>adopt(): moves vectors, same root
allocator,<br/>no copy for UDF results"]
A["[ArrowColumnarToVeloxColumnarExec]"]
V2["VeloxOp"]
V1 --> L --> E --> O --> A --> V2
classDef spark fill:#fff4e0,stroke:#d98e04,color:#1f2328;
classDef inserted fill:#e8f0fb,stroke:#3b82d4,color:#1f2328;
classDef velox fill:#eef7ee,stroke:#2e7d32,color:#1f2328;
class T,R,B spark;
style E fill:#fffaf0,stroke:#d98e04,color:#1f2328;
class L,O,A inserted;
class V1,V2 velox;
```
### How it works
- **Spark** (`SparkPlan.convention`, `requiredChildConventions`,
`TransitionGraph`): every plan
declares its row and batch types. `ArrowBatchType` means "a batch of
`ArrowColumnVector`", and
its transition to `VanillaBatchType` is free.
- **Spark `ArrowEvalPythonExec`**: if `arrowBatchInput` holds, the operator
declares
`(VanillaRowType, ArrowBatchType)`, requires Arrow batches from its child,
always takes the Arrow
columnar path, and outputs the child's columns followed by the UDF results
as Arrow. The
condition is: the child's batch type is not vanilla, a path to
`ArrowBatchType` exists, and every
UDF input is a child column without UDTs.
- **Ownership contract**: an `ArrowBatchType` batch is valid until the next
`hasNext` / `next`. To
keep a batch longer, a consumer transfers its vectors. The vectors move to
the consumer's
allocator only when it shares the root and the producer's allocator has no
listener. Otherwise
they stay in the producer's allocator.
- **Gluten bridge** (`SparkConventions` + 7 lines in `ConventionFunc` +
`GlutenPlan.convention`):
- A Spark plan that declares a non-vanilla Spark type is planned as the
Gluten type bound to it.
- A Gluten plan whose type is bound to a Spark type reports that type to
Spark. For example,
`LoadArrowDataExec` reports `ArrowBatchType`.
- Gluten plans declare `Any` on their children to Spark, because Gluten
plans its own transitions.
- **Gluten "Arrow write"**:
- `RefCountedArrowColumnVector` (a subclass of Spark's
`ArrowColumnVector`) replaces
`ArrowWritableColumnVector`.
- Rows are written to Arrow with Spark's `ArrowWriter`
(`ArrowColumnVectors.fromRows`) in
`ColumnarPartialProjectExec`, `ColumnarPartialGenerateExec`,
`ColumnarRangeExec` and
`ExecUtil`.
- `ArrowJavaBatchType` is bound to Spark's `ArrowBatchType`.
- **Allocator**: Gluten's task allocator is a child of
`ArrowUtils.rootAllocator`. `adopt()`
transfers Spark's vectors and moves their accounting between the two
allocation listeners.
### Copies left on the path (from the "final plan" comment in #13140)
| Step | Copy? |
|---|---|
| Velox -> Java Arrow | Fixed-width: none. Dictionary or constant vectors:
flattened (copy). Strings: copied, because Arrow string views aren't enabled |
| Spark `ArrowEvalPythonExec` input | No conversion. The UDF columns are
serialized to IPC directly from Gluten's buffers |
| Spark -> Gluten (offload) | None for pass-through columns or UDF results
(same root since `ec3f05360d`) |
| Java Arrow -> Velox | Fixed-width: none. Strings: a new string-view buffer
(strings of 12 bytes or less are inlined) |
---
## 3. Side-by-side comparison
| Aspect | Today (Gluten main) | Target (Spark 4.2 + SPARK-57468, Gluten
branch) |
|---|---|---|
| Operator executing Arrow UDFs | Gluten `ColumnarArrowEvalPythonExec` (a
`GlutenPlan`) | Spark `ArrowEvalPythonExec`, unchanged and not offloaded |
| Python runner / worker protocol | Gluten `ColumnarArrowPythonRunner` +
`BasePythonRunnerShim` × 4 | Spark's runners; Gluten has no Python runner code |
| How Gluten learns the operator wants Arrow | Hard-coded offload rule
(`createColumnarArrowEvalPythonExec`) | `SparkPlan.convention` =
`ArrowBatchType`, mapped through `SparkConventions` |
| Transition insertion | Gluten's own operator declares `ArrowJavaBatchType`
| Generic: any Spark plan declaring a bound type gets Gluten's transitions |
| Operator-specific Gluten code | Exec + runner + shims (about 850 lines) |
`InsertArrowInputForArrowEvalPython` (53 lines), pre-project helper |
| Java Arrow vector | `ArrowWritableColumnVector` +
`WritableColumnVectorShim` × 4 | Spark `ArrowColumnVector`
(`RefCountedArrowColumnVector`, 57 lines) |
| Writing rows into Arrow | `ArrowWritableColumnVector.put*`, Gluten
`ArrowProjection` codegen | Spark `ArrowWriter` + `MutableProjection` |
| Gluten Arrow batch type vs. Spark | Unrelated; Spark sees vanilla |
`ArrowJavaBatchType` ≡ Spark `ArrowBatchType` |
| Input batch lifetime in UDF exec | Retain / release + `adjustRefCnt` |
`ArrowBatchOwner` transfers vectors, releases after the output is consumed |
| UDF result columns | Read by Gluten's runner into Gluten's allocator |
Read by Spark's runner, moved into Gluten by `adopt()` (same root, no copy) |
| Arrow allocator | Separate Gluten `RootAllocator` | Child of Spark's
`ArrowUtils.rootAllocator`, with listener accounting moved on transfer |
| Memory accounting of UDF output | Spark side untracked | Moved to Gluten's
`ManagedAllocationListener` on adopt |
| UDTF | #12276: Gluten exec + shim + output copy | Spark
`ArrowEvalPythonUDTFExec` after SPARK-59792 + an `ArrowBatchType` declaration
(still to do) |
| MapInArrow / FlatMapGroups / Window / Agg | Row fallback (C2R + R2C) |
Still row fallback until Spark operators declare `ArrowBatchType` (still to do)
|
| New Spark Arrow operator | New Gluten re-implementation | Zero Gluten code
once it declares `ArrowBatchType` |
| Supported Spark versions | 3.4 – 4.1 (4.2 via #13126) | Spark with
SPARK-57468 only; older versions need the old path kept (still to do) |
| Code delta | – | Spark +939/−79; Gluten +1,370/−6,025 (`bc293a5c77` alone
is +378/−4,420) |
| Performance (5M rows, 4 pandas UDFs) | 2.14–2.22 s | 2.24–2.35 s (within
noise, measured before `ec3f05360d` removed the result copy) |
---
## 4. Prototype status
Done and tested on the branches:
- Spark: convention API, `ConventionSuite`; `ArrowEvalPythonExec` on
`ArrowBatchType`; ownership
contract; listener-aware transfer. All are covered by
`ArrowColumnarPythonUDFSuite`.
- Gluten: bridge; `ArrowJavaBatchType` bound to Spark; switch to
`ArrowColumnVector` /
`ArrowWriter`; Gluten's Python exec and shims removed; shared root
allocator.
`ArrowEvalPythonExecSuite`, `ColumnarBatchTest`, `ArrowColumnVectorTest`
and the transition and
fallback suites pass.
Gaps in the prototype:
1. `InsertArrowInputForArrowEvalPython` is still a rule specific to
`ArrowEvalPythonExec`. Spark
picks the Arrow path from `child.convention`, and a Velox child reports
`VanillaBatchType`, so
Gluten has to insert `LoadArrowDataExec` first and then check what Spark
decided.
2. `gluten-core` / `gluten-arrow` reference
`org.apache.spark.sql.execution.convention` directly.
They no longer compile against Spark versions without SPARK-57468, so the
bridge needs a shim
(or a no-op).
3. `backends-bolt` still uses the removed `BasePythonRunnerShim` /
`ArrowWritableColumnVector` and
doesn't compile.
4. `RefCountedArrowColumnVector` keeps Gluten's reference count. The
in-place load / offload
lifecycle in `ColumnarBatches` hasn't been moved to ownership transfer
yet.
5. The fallback reporter lists Spark Arrow operators planned through a
convention as "not
supported" (log noise only).
6. The write-path performance of `ArrowWriter` vs. the old codegen
projection hasn't been
benchmarked.
7. Only `ArrowEvalPythonExec` is covered.
---
## 5. Features to add
### 5.1 Spark (target: our 4.2 line, and upstream master)
| # | Feature | Why | Status |
|---|---|---|---|
| S1 | **SPARK-57468 convention API**: `Convention`, `RowType`, `BatchType`
(`Vanilla*`, `ArrowBatchType`), `TransitionGraph` with transition costs,
`SparkPlan.convention` / `requiredChildConventions`, graph-based
`ApplyColumnarRulesAndInsertTransitions`, convention propagation through
`QueryStageExec` / `Exchange` / `DataWritingCommandExec` | The foundation: lets
a plan declare the Arrow layout and lets plugins bind their types | Branch
`98c9eac81ee`; upstream PR apache/spark#59154 open |
| S2 | **`ArrowEvalPythonExec` declares `ArrowBatchType`**
(`arrowBatchInput`, `inputsAreChildColumns`, always takes the Arrow path,
outputs Arrow) | Spark's operator is used as is, with no Gluten wrapper |
Branch `effa8e8cd3d`; no upstream PR yet |
| S3 | **`ArrowBatchType` ownership contract** + `ArrowBatchOwner` (transfer
on read, release after the output is consumed, close at end of output or task)
| Makes zero-copy hand-off safe for producers that allocate per batch or reuse
vectors (Velox -> Arrow, DSv2 readers) | Branch `8341e63bf92`; no upstream PR
yet |
| S4 | **Listener-aware transfer**: keep a producer's buffers in its
allocator when it has an `AllocationListener` or a different root | Without it,
Gluten's memory reservation grows until the task ends | Branch `5c4190677bb`;
no upstream PR yet |
| S5 | **SPARK-59792 UDTF columnar input** + make `ArrowEvalPythonUDTFExec`
declare `ArrowBatchType` the same way as S2 / S3 | Replaces Gluten #12276 |
apache/spark#59062 open; the `ArrowBatchType` declaration is not started |
| S6 | **A planner-level "prefer Arrow" request** for operators that benefit
from Arrow input, e.g. let `ArrowEvalPythonExec` require `ArrowBatchType`
whenever any registered path from the child exists, not only when the child
already declares a non-vanilla type. Alternative: Gluten makes Velox visible to
Spark as a Spark `BatchType` (see G5) | Removes Gluten's
`InsertArrowInputForArrowEvalPython`, so support is fully automatic | Not
started; needs design |
| S7 | **Other Arrow operators declare `ArrowBatchType` input and output**:
`MapInArrowExec` / `MapInPandasExec`, `FlatMapGroupsInArrowExec` / `Pandas`,
`FlatMapCoGroupsInArrowExec` / `Pandas`, `ArrowAggregatePythonExec`,
`ArrowWindowPythonExec` | Today they take rows and output rows, so Gluten does
C2R + R2C around them. The grouped and window operators need a group-aware
Arrow input, which is harder | Not started |
| S8 | **`RowToArrowColumnarExec`** on `ArrowWriter`, registered as
`ArrowBatchType.fromRow` (revive SPARK-37124) | Planner path rows -> Arrow, so
consumers can drop their row-by-row fallbacks. Also covers local table scan,
RDD scan and range | Not started |
| S9 | **Public Arrow batch utilities**: `isArrowBacked` + layout check
(`ArrowUtils.isCompatibleWithDeclaredField`), unwrap a batch into a
`VectorSchemaRoot` / record batch, take ownership by transfer | Gluten, the
Arrow cache (SPARK-57268) and the UDTF path stop duplicating private helpers |
Not started |
| S10 | *(Optional)* Direct vanilla columnar -> `ArrowBatchType` transition
(on-heap / off-heap vectors -> Arrow) | Parquet / ORC vectorized reads feeding
UDFs without a row round trip | Not started |
| S11 | *(Optional)* Python reader allocator closed at task end rather than
at end of stream | No longer needed for Gluten once G6 is in (same root plus
adopt-on-consume). Still useful for other consumers that keep output batches |
Not needed for Gluten |
| S12 | *(Optional)* Arrow string view (`Utf8View`) support in
`ArrowColumnVector` / the IPC path to pyarrow | Removes the string copy on
Velox -> Arrow export (see G10) | Not started |
| S13 | *(Optional)* Spark-side rule that pulls complex UDF arguments into a
`ProjectExec` (the TODO in `ColumnarArrowEvalPythonEvaluatorFactory`) |
Satisfies `inputsAreChildColumns` in plain Spark too; Gluten could then drop
`PullOutArrowEvalPythonPreProjectHelper` | Not started |
### 5.2 Gluten
| # | Feature | Why | Status |
|---|---|---|---|
| G1 | **`SparkConventions` bridge**: bind Spark and Gluten types;
`ConventionFunc` maps a Spark plan's declared type and child requirements;
`GlutenPlan.convention` reports bound types to Spark, with `Any` child
requirements | Generic transition insertion for any Spark operator that
declares a bound type | Done (`4a5a7a46dd`, `8a430f38e4`) |
| G2 | **Gluten Arrow batches on Spark's `ArrowColumnVector`**:
`RefCountedArrowColumnVector`; `ArrowJavaBatchType` bound to Spark
`ArrowBatchType`; delete `ArrowWritableColumnVector`,
`WritableColumnVectorShim` × 4, `ArrowColumnarBatch`, `ArrowColumnarRow`,
`ArrowColumnVectorUtils` | One Java Arrow batch type shared with Spark; no
Spark-to-Gluten Arrow transitions; no shims | Done (`bc293a5c77`) |
| G3 | **Arrow write via Spark's `ArrowWriter`**:
`ColumnarPartialProjectExec` (`MutableProjection` + `ArrowWriter`),
`ColumnarPartialGenerateExec`, `ColumnarRangeExec`, `ExecUtil`,
`ArrowColumnVectors.fromRows`; drop the `ArrowProjection` codegen | Removes
Gluten's own Arrow writer and codegen | Done (`bc293a5c77`); **benchmark
pending** |
| G4 | **Remove Gluten's Python exec**: `ColumnarArrowEvalPythonExec`,
`ColumnarArrowPythonRunner`, `BasePythonRunnerShim` × 5;
`createColumnarArrowEvalPythonExec` returns Spark's exec | Gluten no longer
owns any Python runner or worker protocol code | Done for Velox (`8a430f38e4`) |
| G5 | **Remove `InsertArrowInputForArrowEvalPython`**: once S6 lands, or by
making Velox visible to Spark (bind `VeloxBatchType` to a Spark `BatchType`
with a registered path to `ArrowBatchType`) | Fully automatic support for any
Arrow operator, with no rule per operator | Interim rule done (53 lines);
removal not started |
| G6 | **Shared root allocator**: the task allocator is a child of
`ArrowUtils.rootAllocator`; `ArrowColumnVectors.adopt` transfers vectors and
moves listener accounting | Zero-copy UDF results; memory accounted by Gluten |
Done (`ec3f05360d`) |
| G7 | **Version gating / shim for the bridge**: keep the today path (Gluten
exec + runner + `ArrowWritableColumnVector`) for Spark 3.4 – 4.1 (and 4.2
without SPARK-57468); put `SparkConventions`, the `GlutenPlan.convention`
overrides and the `ArrowBatchType` binding behind a shim | The branch only
compiles against patched Spark | Not started |
| G8 | **Lifecycle rewrite**: replace reference counts and in-place load /
offload in `ColumnarBatches` with "the producer owns a batch until the next
one; a consumer that keeps it transfers the vectors". The `IndicatorVector`
reference count stays for native batches shared by limit, tail and broadcast |
Removes the reference count from `RefCountedArrowColumnVector`; matches Spark's
`ArrowBatchType` contract | Not started (largest and riskiest part) |
| G9 | **Fallback reporter**: skip Spark operators planned through a Spark
convention | Spurious "not supported" entries | Not started |
| G10 | *(Optional)* Velox -> Arrow export: enable Arrow string views and
avoid flattening where the consumer accepts it (needs S12) | Removes the
remaining string and dictionary copies | Not started |
| G11 | **UDTF**: drop #12276 in favor of S5; Gluten only needs the existing
transitions | No Gluten UDTF exec, shim or output copy | Waiting for S5 |
| G12 | **Other Arrow operators** (S7): no Gluten code beyond tests once
Spark declares `ArrowBatchType`. Before S7, a one-line convention declaration
(rule) per operator is enough for the input side | Removes the C2R + R2C around
them | Waiting for S7 |
| G13 | **Pre-project**: keep `PullOutArrowEvalPythonPreProjectHelper` (in
its own file now) so complex UDF arguments become columns and
`inputsAreChildColumns` holds; drop it if S13 lands | Otherwise Spark takes the
row path | Done (kept) |
| G14 | **`backends-bolt` migration** to the same model | It doesn't compile
after G2 / G4 | Not started |
| G15 | **Tests**: `gluten-ut` coverage for the Spark Arrow operators on
Velox input (multi-batch, nulls, chained UDFs, ANSI on and off, AQE, config
`columnarInput.enabled` off), memory leak checks, and an end-to-end pyspark
benchmark after G6 / G3 | Required before upstreaming | Partly done
(`ArrowEvalPythonExecSuite` with ANSI off) |
---
## 6. Suggested order
1. **Upstream S1** (apache/spark#59154), then S2 – S4 as SPARK-57468
follow-ups. Backport them to
our 4.2 line.
2. **Gluten G7** (shim and version gating), so G1 – G6 can merge into Gluten
main without breaking
Spark 3.4 – 4.1.
3. **Merge G1, G2, G3, G4, G6 and G13** for Spark builds with SPARK-57468,
after the G3 benchmark
and G15 tests.
4. **S5 + G11** (UDTF), replacing #12276.
5. **S6 + G5**, so the remaining operator-specific rule goes away.
6. **G8** (lifecycle rewrite), then G9 and G14.
7. **S7 / G12** (other Arrow operators), then S8 – S10 and S12 / G10 as
optimizations.
## 7. Risks and open points
- **Upstream acceptance of S1 – S4.** Spark has to accept a plugin-facing
plan API and an ownership
contract. The UDF-specific follow-ups may need to be generalized (S6)
before review.
- **Planning interplay.** Spark's `insertTransitions` and Gluten's
`InsertTransitions` both run. It
must be verified that they never insert competing transitions. Gluten
plans report `Any` child
requirements to Spark for this reason.
- **Memory accounting.** Arrow doesn't notify listeners of transfers. Both
S4 and `adopt()`
compensate for this manually, which is fragile if Arrow changes its
behavior.
- **Grouped and window operators (S7)** need grouped Arrow input. That's a
bigger Spark change than
for map-style operators.
- **Older Spark versions** keep today's code until they're dropped, so
Gluten carries both paths
for a while.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
