You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by ke...@apache.org on 2026/03/24 16:14:44 UTC
(gluten) branch main updated: [GLUTEN-11766][VL] Implement the executeCollect() method in ColumnarCollectLimitExec (#11802)
This is an automated email from the ASF dual-hosted git repository.
kejia 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 71e0db193d [GLUTEN-11766][VL] Implement the executeCollect() method in ColumnarCollectLimitExec (#11802)
71e0db193d is described below
commit 71e0db193df5e0516a3850835bd694577ebf59af
Author: JiaKe <ke...@ibm.com>
AuthorDate: Tue Mar 24 16:14:37 2026 +0000
[GLUTEN-11766][VL] Implement the executeCollect() method in ColumnarCollectLimitExec (#11802)
---
.../gluten/execution/ColumnarCollectLimitExec.scala | 19 +++++++++++++++++++
.../gluten/execution/VeloxColumnarToRowExec.scala | 9 +++++++++
2 files changed, 28 insertions(+)
diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarCollectLimitExec.scala b/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarCollectLimitExec.scala
index a3ee421e59..56b595b48b 100644
--- a/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarCollectLimitExec.scala
+++ b/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarCollectLimitExec.scala
@@ -19,6 +19,7 @@ package org.apache.gluten.execution
import org.apache.gluten.columnarbatch.ColumnarBatches
import org.apache.gluten.columnarbatch.VeloxColumnarBatches
+import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.vectorized.ColumnarBatch
@@ -96,6 +97,24 @@ case class ColumnarCollectLimitExec(
}
}
+ override def executeCollect(): Array[InternalRow] = {
+ val rowsRdd = child.executeColumnar().mapPartitions {
+ it =>
+ val rows = VeloxColumnarToRowExec.toRowIterator(it)
+ rows.map(_.copy())
+ }
+
+ if (limit >= 0) {
+ if (offset > 0) {
+ rowsRdd.take(limit).drop(offset)
+ } else {
+ rowsRdd.take(limit)
+ }
+ } else {
+ rowsRdd.collect().drop(offset)
+ }
+ }
+
override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan =
copy(child = newChild)
}
diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxColumnarToRowExec.scala b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxColumnarToRowExec.scala
index 097c7d4897..2354ebf39f 100644
--- a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxColumnarToRowExec.scala
+++ b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxColumnarToRowExec.scala
@@ -92,6 +92,15 @@ case class VeloxColumnarToRowExec(child: SparkPlan) extends ColumnarToRowExecBas
VeloxColumnarToRowExec.toRowIterator(_, numOutputRows, numInputBatches, convertTime))
}
+ override def executeCollect(): Array[InternalRow] = {
+ child match {
+ case l: ColumnarCollectLimitExec =>
+ l.executeCollect()
+ case _ =>
+ super.executeCollect()
+ }
+ }
+
protected def withNewChildInternal(newChild: SparkPlan): VeloxColumnarToRowExec =
copy(child = newChild)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org