Repository navigation
[GLUTEN-13140][VL] Replace ArrowWritableColumnVector with Spark's ArrowColumnVector and ArrowWriter - #13280
Open
FelixYBW wants to merge 5 commits into
Open
[GLUTEN-13140][VL] Replace ArrowWritableColumnVector with Spark's ArrowColumnVector and ArrowWriter#13280FelixYBW wants to merge 5 commits into
FelixYBW wants to merge 5 commits into
Conversation
|
Run Gluten Clickhouse CI on x86 |
…owColumnVector and ArrowWriter
Gluten's Java Arrow batches now consist of RefCountedArrowColumnVector, a
subclass of Spark's ArrowColumnVector that keeps Gluten's reference count.
The batch lifecycle is unchanged: ColumnarBatches still retains and releases
batches by reference count, and loads and offloads them in place.
- Reading goes through Spark's ArrowColumnVector accessors, which check for
nulls on all supported Spark versions, so the ArrowColumnarBatch /
ArrowColumnarRow workaround for Spark's ColumnarRow before 4.0 is no
longer needed.
- Writing fills the Arrow vectors directly or with Spark's ArrowWriter:
- ColumnarPartialProjectExec: Spark's MutableProjection + ArrowWriter
(replaces ArrowProjection / GenerateArrowProjection /
InterpretedArrowProjection).
- ColumnarPartialGenerateExec: ArrowWriter over the generated rows.
- ColumnarRangeExec, ExecUtil: set the Arrow vector directly.
- ArrowColumnVectors: allocate a root from Gluten's allocator, wrap
vectors, build a batch from rows with ArrowWriter, and adopt (move or
copy) a vector into Gluten's allocator. ColumnarBatches.offload adopts
batches of plain Spark ArrowColumnVectors.
- ColumnarArrowEvalPythonExec wraps the Python output vectors with
RefCountedArrowColumnVector.
- VeloxColumnarWriteFilesExec reads the VARBINARY fragments column with
getBinary (the old accessor allowed getUTF8String on binary).
- Removed: ArrowWritableColumnVector, WritableColumnVectorShim (spark34 -
spark42), ArrowColumnarBatch, ArrowColumnarRow, ArrowColumnVectorUtils,
ArrowProjection and its code generators.
ArrowColumnVector, ArrowWriter.create(VectorSchemaRoot) and MutableProjection
are available with the same API in Spark 3.4 - 4.2, so no shim is needed.
Tests: ColumnarBatchTest ported to ArrowWriter-built batches; new
ArrowColumnVectorTest cases (ref count, decimal round trip, offloading a
Spark Arrow batch from another allocator root); new ArrowEvalPythonExecSuite
case with chained UDFs over many small batches with nulls.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…wEvalPythonExecSuite Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
FelixYBW
force-pushed
the
arrow-column-vector
branch
from
October 20, 2026 20:44
c6a63d5 to
1b6ee72
Compare
Copilot stopped reviewing on behalf of
FelixYBW due to an error
October 20, 2026 22:19
|
Run Gluten Clickhouse CI on x86 |
1 similar comment
|
Run Gluten Clickhouse CI on x86 |
Copilot stopped reviewing on behalf of
FelixYBW due to an error
October 20, 2026 23:26
…rrowColumnVector Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Copilot stopped reviewing on behalf of
FelixYBW due to an error
October 21, 2026 00:43
|
Run Gluten Clickhouse CI on x86 |
…ic ref count - ColumnarBatches.adopt copies instead of moving a vector when moving would empty it for someone else: a Gluten RefCountedArrowColumnVector passed through in a mixed batch, or a vector used by more than one column. - ArrowColumnVectors.fromRows closes the root if writing fails. - RefCountedArrowColumnVector updates its count with compare-and-set, never goes below 0, and rejects retain() after release. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Contributor
Author
|
I went through all 18 Copilot comments. They come down to 7 distinct points. e10fa72 fixes 3 of them; the rest are false positives or optional. Fixed in e10fa72
2 and 3 are also bugs in the current code on main, not regressions from this PR:
They're fixed here because the new code puts both in one place. New tests in
The two False positives
Optional, not changed
|
Contributor
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
22 open issues
root.setRowCount(numRows)but never setsvector.setValueCount(numRows). If… · NewColumnarRangeExec, this fills anIntVectorbut doesn’t set… · NewtoBatchrelies onroot.getRowCount, but it does not ensure each underlyingFieldVectorhas… · Newclose()is not refcount-safe under concurrent callers: two threads can both observe `refCnt.get()…inputis a \"foreign\" Spark Arrow batch, calling `ArrowColumnVectors.adopt(vector,…ColumnarRangeExec, this code writes intopidVecbut only sets…root.setRowCount(numRows)but never setsvector.setValueCount(numRows). For…toBatchcurrently trusts that each underlyingFieldVectoralready has the correctvalueCount.…ArrowColumnVectorbatches,ArrowColumnVectors.adoptmay choose a zero-copy…ColumnarRangeExec:pidVec.set(i, pid)doesn’t updatevalueCount, and…VectorSchemaRoot.setRowCountdoes not guarantee the underlying Arrow vectors’valueCountis…close()calls can both pass the…adopt(...)may do a zero-copy… · Newtarget.allocateNew()uses default sizing, which can cause repeated…ArrowWriteris created but never closed. Even if most memory is owned by the vectors,…adopt()can…target.allocateNew()allocates using defaults, which can cause…0boundary:close()does a…rows.foreach(writer.write)throws (bad input, expression bug, etc.),root(and its allocated…close()implementation is not atomic with respect to concurrent callers: two threads can both…
And 2 more that still need to be addressed.
Review effort: Lite
Comment on lines
+103
to
+113
| val root = ArrowColumnVectors.allocateRoot(schema) | ||
| val vector = root.getVector(0).asInstanceOf[BigIntVector] | ||
| vector.allocateNew(numRows) | ||
| for (i <- 0 until numRows) { | ||
| val value = current + i * step | ||
| vectors(0).putLong(i, getSafeMargin(value)) | ||
| vector.set(i, getSafeMargin(value)) | ||
| } | ||
| vectors.foreach(_.setValueCount(numRows)) | ||
| root.setRowCount(numRows) | ||
| current += numRows * step | ||
|
|
||
| val batch = new ColumnarBatch(vectors.asInstanceOf[Array[ColumnVector]], numRows) | ||
| batch | ||
| ArrowColumnVectors.toBatch(root) |
Comment on lines
+147
to
+160
| val pidRoot = | ||
| ArrowColumnVectors.allocateRoot(new StructType().add("pid", IntegerType)) | ||
| val pidVec = pidRoot.getVector(0).asInstanceOf[IntVector] | ||
| pidVec.allocateNew(cb.numRows) | ||
| convertColumnarToRow(cb).zipWithIndex.foreach { | ||
| case (row, i) => | ||
| val pid = rangePartitioner.get.getPartition(partitionKeyExtractor(row)) | ||
| pidVec.putInt(i, pid) | ||
| pidVec.set(i, pid) | ||
| } | ||
| pidRoot.setRowCount(cb.numRows) | ||
| val pidBatch = VeloxColumnarBatches.toVeloxBatch( | ||
| ColumnarBatches.offload( | ||
| ArrowBufferAllocators.contextInstance(), | ||
| new ColumnarBatch(Array[ColumnVector](pidVec), cb.numRows))) | ||
| ArrowColumnVectors.toBatch(pidRoot))) |
Comment on lines
+50
to
+54
| def toBatch(root: VectorSchemaRoot): ColumnarBatch = { | ||
| new ColumnarBatch( | ||
| wrap(root.getFieldVectors).map(_.asInstanceOf[ColumnVector]), | ||
| root.getRowCount) | ||
| } |
Comment on lines
+269
to
+274
| if (isForeignArrowBatch(input)) { | ||
| // A batch of Spark ArrowColumnVectors (e.g. the output of a Spark Arrow operator) is owned | ||
| // by its producer and only valid until the next one is requested: take ownership of its | ||
| // vectors in a Gluten batch and offload that one, leaving the input to its producer. | ||
| return offload(allocator, adopt(allocator, input)); | ||
| } |
|
Run Gluten Clickhouse CI on x86 |
Copilot stopped reviewing on behalf of
FelixYBW due to an error
October 21, 2026 06:52
jinchengchenghh
approved these changes
Oct 21, 2026
zhztheplayer
approved these changes
Oct 21, 2026
| * <p>Reading goes through Spark's accessors. To write, fill the Arrow vectors (e.g. with Spark's | ||
| * {@code ArrowWriter}) and wrap them, see {@link ArrowColumnVectors}. | ||
| */ | ||
| public final class RefCountedArrowColumnVector extends ArrowColumnVector { |
Member
There was a problem hiding this comment.
Since we have dropped Spark 3.2, now reference-counting can be done in columnar batch wise by inheriting Spark API ColumnarBatch.
We'd plan it as another future refactor to simplify code further. Many batch utilities can be removed then.
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes are proposed in this pull request?
ArrowWritableColumnVector was introduced by Gazelle back in 2019, and then Gluten reused it. We tried to upstream it in apache/spark#34396, while @cloud-fan @viirya @sunchao suggested to use Spark's ArrowColumnVector and ArrowWriter to implement the functionality of ArrowWritableColumnVector, which makes sense to us but the work is paused.
Recently, a couple of PRs in upstream Spark are going to enable more Arrow-based operators. Meanwhile the 3rd party libraries like lance-spark also introduce Arrow based operators. Our goal is to reuse those operators or have a thin wrapper around those operators. But one issue we meet is the compatibility of the ArrowColumnVector. The PR is going to use Spark's ArrowColumnVector and ArrowWriter to replace Gluten's old ArrowWritableColumnVector. So Gluten and Spark can transfer the ArrowColumnVector each other.
Lance. The Lance scan offload (#12958) currently needs its own
LanceScanTransformerand RDD, because Gluten only accepted batches ofArrowWritableColumnVector. With this PR,ColumnarBatches.offloadaccepts batches of Spark'sArrowColumnVector. That's the Gluten side of the longer-term plan in #12958 (comment): if lance-spark's columnar reader gets an Arrow output mode that emits SparkArrowColumnVectors, the vanillaBatchScanExeccan be offloaded with only a batch-type rule. It would then keep Lance's metrics, runtime filtering and partitioning. That follow-up needs a change in lance-spark and isn't part of this PR.It's part of #13140 (items G2 + G3). No Spark change is needed:
ArrowColumnVector(ValueVector),getValueVector,ArrowWriter.create(VectorSchemaRoot)andMutableProjectionhave the same API in Spark 3.4 – 4.2.RefCountedArrowColumnVector: a subclass of Spark'sArrowColumnVectorthat keeps Gluten's reference count. The batch lifecycle is unchanged:ColumnarBatchesstill retains and releases batches by reference count, and loads and offloads them in place.ArrowColumnarBatch/ArrowColumnarRowworkaround for Spark'sColumnarRowbefore 4.0 is removed.ColumnarPartialProjectExec: Spark'sMutableProjection+ArrowWriter, replacingArrowProjection/GenerateArrowProjection/InterpretedArrowProjection.ColumnarPartialGenerateExec:ArrowWriterover the generated rows. The per-column size estimation is no longer needed.ColumnarRangeExec,ExecUtil(range partition ids): set the Arrow vector directly.ArrowColumnVectors(new): allocate a root from Gluten's allocator, wrap vectors, build a batch from rows withArrowWriter, and adopt (move, or copy across allocator roots) a vector into Gluten's allocator.ColumnarBatches.offloadadopts batches of plain SparkArrowColumnVectors.ColumnarArrowEvalPythonExecwraps the Python output vectors withRefCountedArrowColumnVector. Gluten's runner and its shims are unchanged.LanceScanTransformer(-Plance): wraps the vectors it transfers out of the imported Arrow stream withArrowColumnVectors.wrap, instead ofArrowWritableColumnVector.loadColumns. Its behavior is unchanged.docs/developers/ArrowColumnarConversion.mdis updated to describeRefCountedArrowColumnVectorandArrowColumnVectors.VeloxColumnarWriteFilesExecreads the VARBINARY fragments column withgetBinary. The old accessor allowedgetUTF8Stringon binary columns; Spark's doesn't.ArrowWritableColumnVector,WritableColumnVectorShim(spark34 – spark42),ArrowColumnarBatch,ArrowColumnarRow,ArrowColumnVectorUtils, andArrowProjectionwith its code generators.The new solution actually simplified the implementation with a net: +406 / −4,521 lines.
Not included:
backends-boltuses the same removed classes and is not migrated in this PR.How was this patch tested?
ColumnarBatchTest: ported to batches built withArrowWriter.ArrowColumnVectorTest: new cases for the reference count, a decimal round trip, and offloading a Spark Arrow batch allocated from another allocator root.ArrowEvalPythonExecSuite: a new case with two chained Arrow UDFs over many small batches with nulls, checking the results against vanilla Spark and theLoadArrowData/OffloadArrowDatatransitions around the UDFs.The write paths now use
ArrowWriter, which writes one row at a time like the code it replaces. I haven't benchmarked them yet.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (claude-opus-5-5)
🤖 Generated with Claude Code