This is an automated email from the ASF dual-hosted git repository. taiyang-li pushed a commit to branch fake_add_bolt_backend in repository https://gitbox.apache.org/repos/asf/gluten.git
commit 3bc62436837e0a2d8b8b6aa57526d46853ee53e0 Author: liyang.127 <[email protected]> AuthorDate: Wed Jul 1 16:02:51 2026 +0800 [GLUTEN][CORE] Share ColumnarCollectTailExec under backends-core Move the common CollectTail implementation into backends-core so the velox backend keeps only the backend-specific slicing hook. This reduces duplicated JVM execution code and keeps future backend sync work smaller. --- .../execution/SharedColumnarCollectTailExec.scala | 18 ++++---- .../gluten/execution/ColumnarCollectTailExec.scala | 53 ++-------------------- 2 files changed, 12 insertions(+), 59 deletions(-) diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarCollectTailExec.scala b/backends-core/src/main/scala/org/apache/gluten/execution/SharedColumnarCollectTailExec.scala similarity index 82% copy from backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarCollectTailExec.scala copy to backends-core/src/main/scala/org/apache/gluten/execution/SharedColumnarCollectTailExec.scala index 7977c9d610..6c4c6437cb 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarCollectTailExec.scala +++ b/backends-core/src/main/scala/org/apache/gluten/execution/SharedColumnarCollectTailExec.scala @@ -17,7 +17,6 @@ package org.apache.gluten.execution import org.apache.gluten.columnarbatch.ColumnarBatches -import org.apache.gluten.columnarbatch.VeloxColumnarBatches import org.apache.spark.sql.execution.SparkPlan import org.apache.spark.sql.vectorized.ColumnarBatch @@ -25,10 +24,14 @@ import org.apache.spark.sql.vectorized.ColumnarBatch import scala.collection.mutable import scala.util.control.Breaks._ -case class ColumnarCollectTailExec( - limit: Int, - child: SparkPlan -) extends ColumnarCollectTailBaseExec(limit, child) { +/** + * Shared collect-tail implementation for backends that can retain incoming columnar batches and + * slice the first overflow batch using a backend-specific primitive. + */ +abstract class SharedColumnarCollectTailExec(limit: Int, child: SparkPlan) + extends ColumnarCollectTailBaseExec(limit, child) { + + protected def sliceBatch(batch: ColumnarBatch, offset: Int, length: Int): ColumnarBatch override protected def collectTailRows( partitionIter: Iterator[ColumnarBatch], @@ -68,14 +71,11 @@ case class ColumnarCollectTailExec( if (overflow > 0) { val first = tailQueue.remove(0) val keep = first.numRows() - overflow - val sliced = VeloxColumnarBatches.slice(first, overflow.toInt, keep.toInt) + val sliced = sliceBatch(first, overflow.toInt, keep.toInt) tailQueue.prepend(sliced) first.close() } tailQueue.iterator } - - override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan = - copy(child = newChild) } diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarCollectTailExec.scala b/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarCollectTailExec.scala index 7977c9d610..cbc3c42a07 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarCollectTailExec.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarCollectTailExec.scala @@ -16,65 +16,18 @@ */ package org.apache.gluten.execution -import org.apache.gluten.columnarbatch.ColumnarBatches import org.apache.gluten.columnarbatch.VeloxColumnarBatches import org.apache.spark.sql.execution.SparkPlan import org.apache.spark.sql.vectorized.ColumnarBatch -import scala.collection.mutable -import scala.util.control.Breaks._ - case class ColumnarCollectTailExec( limit: Int, child: SparkPlan -) extends ColumnarCollectTailBaseExec(limit, child) { - - override protected def collectTailRows( - partitionIter: Iterator[ColumnarBatch], - limit: Int - ): Iterator[ColumnarBatch] = { - if (!partitionIter.hasNext || limit <= 0) { - return Iterator.empty - } - - val tailQueue = new mutable.ListBuffer[ColumnarBatch]() - var totalRowsInTail = 0L - - while (partitionIter.hasNext) { - val batch = partitionIter.next() - val batchRows = batch.numRows() - ColumnarBatches.retain(batch) - tailQueue += batch - totalRowsInTail += batchRows - - breakable { - while (tailQueue.nonEmpty) { - val front = tailQueue.head - val frontRows = front.numRows() - - if (totalRowsInTail - frontRows >= limit) { - val dropped = tailQueue.remove(0) - dropped.close() - totalRowsInTail -= frontRows - } else { - break - } - } - } - } - - val overflow = totalRowsInTail - limit - if (overflow > 0) { - val first = tailQueue.remove(0) - val keep = first.numRows() - overflow - val sliced = VeloxColumnarBatches.slice(first, overflow.toInt, keep.toInt) - tailQueue.prepend(sliced) - first.close() - } +) extends SharedColumnarCollectTailExec(limit, child) { - tailQueue.iterator - } + override protected def sliceBatch(batch: ColumnarBatch, offset: Int, length: Int): ColumnarBatch = + VeloxColumnarBatches.slice(batch, offset, length) override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan = copy(child = newChild) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
