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 94eb1d5628116d542c928f771d1f370570b21185 Author: liyang.127 <[email protected]> AuthorDate: Wed Jul 1 16:26:32 2026 +0800 [GLUTEN][CORE] Share velox execution helpers in backends-core Move shared Arrow transition, collect-tail, and metrics updater logic into backends-core so velox can reuse common implementations and reduce duplicated backend glue. --- .../SharedArrowColumnarToBackendColumnarExec.scala | 19 +++--- .../SharedHashAggregateMetricsUpdater.scala | 16 ++--- .../SharedInputIteratorMetricsUpdater.scala | 20 +++--- .../ArrowColumnarToVeloxColumnarExec.scala | 14 +--- .../metrics/HashAggregateMetricsUpdater.scala | 76 ++-------------------- .../metrics/InputIteratorMetricsUpdater.scala | 34 +++------- 6 files changed, 46 insertions(+), 133 deletions(-) diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/ArrowColumnarToVeloxColumnarExec.scala b/backends-core/src/main/scala/org/apache/gluten/execution/SharedArrowColumnarToBackendColumnarExec.scala similarity index 74% copy from backends-velox/src/main/scala/org/apache/gluten/execution/ArrowColumnarToVeloxColumnarExec.scala copy to backends-core/src/main/scala/org/apache/gluten/execution/SharedArrowColumnarToBackendColumnarExec.scala index cca381ee84..0e77f15e07 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/execution/ArrowColumnarToVeloxColumnarExec.scala +++ b/backends-core/src/main/scala/org/apache/gluten/execution/SharedArrowColumnarToBackendColumnarExec.scala @@ -17,25 +17,28 @@ package org.apache.gluten.execution import org.apache.gluten.backendsapi.arrow.ArrowBatchTypes.ArrowNativeBatchType -import org.apache.gluten.backendsapi.velox.VeloxBatchType -import org.apache.gluten.columnarbatch.VeloxColumnarBatches import org.apache.gluten.extension.columnar.transition.Convention import org.apache.spark.sql.execution.SparkPlan import org.apache.spark.sql.vectorized.ColumnarBatch -case class ArrowColumnarToVeloxColumnarExec(override val child: SparkPlan) +/** + * Shared Arrow-native -> backend-native transition for backends whose only difference is the target + * batch type constant and the concrete `ColumnarBatch` conversion helper. + */ +abstract class SharedArrowColumnarToBackendColumnarExec( + child: SparkPlan, + backendBatchType: Convention.BatchType) extends ColumnarToColumnarExec(child) with GlutenColumnarToColumnarTransition { + protected def toBackendBatch(batch: ColumnarBatch): ColumnarBatch + override protected val from: Convention.BatchType = ArrowNativeBatchType - override protected val to: Convention.BatchType = VeloxBatchType + override protected val to: Convention.BatchType = backendBatchType override protected def mapIterator(in: Iterator[ColumnarBatch]): Iterator[ColumnarBatch] = { - in.map(b => VeloxColumnarBatches.toVeloxBatch(b)) + in.map(toBackendBatch) } - - override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan = - copy(child = newChild) } diff --git a/backends-velox/src/main/scala/org/apache/gluten/metrics/HashAggregateMetricsUpdater.scala b/backends-core/src/main/scala/org/apache/gluten/metrics/SharedHashAggregateMetricsUpdater.scala similarity index 88% copy from backends-velox/src/main/scala/org/apache/gluten/metrics/HashAggregateMetricsUpdater.scala copy to backends-core/src/main/scala/org/apache/gluten/metrics/SharedHashAggregateMetricsUpdater.scala index dcfa0d7cf3..c87d815f24 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/metrics/HashAggregateMetricsUpdater.scala +++ b/backends-core/src/main/scala/org/apache/gluten/metrics/SharedHashAggregateMetricsUpdater.scala @@ -24,14 +24,14 @@ import org.apache.spark.task.TaskResources import scala.collection.JavaConverters._ -trait HashAggregateMetricsUpdater extends MetricsUpdater { +trait SharedHashAggregateMetricsUpdater extends MetricsUpdater { def updateAggregationMetrics( aggregationMetrics: java.util.ArrayList[OperatorMetrics], aggParams: AggregationParams): Unit } -class HashAggregateMetricsUpdaterImpl(val metrics: Map[String, SQLMetric]) - extends HashAggregateMetricsUpdater { +abstract class SharedHashAggregateMetricsUpdaterBase(val metrics: Map[String, SQLMetric]) + extends SharedHashAggregateMetricsUpdater { val aggOutputRows: SQLMetric = metrics("aggOutputRows") val aggOutputVectors: SQLMetric = metrics("aggOutputVectors") val aggOutputBytes: SQLMetric = metrics("aggOutputBytes") @@ -44,7 +44,6 @@ class HashAggregateMetricsUpdaterImpl(val metrics: Map[String, SQLMetric]) val aggSpilledPartitions: SQLMetric = metrics("aggSpilledPartitions") val aggSpilledFiles: SQLMetric = metrics("aggSpilledFiles") val flushRowCount: SQLMetric = metrics("flushRowCount") - val abandonedPartialAggregationRows: SQLMetric = metrics("abandonedPartialAggregationRows") val loadedToValueHook: SQLMetric = metrics("loadedToValueHook") val rowConstructionCpuCount: SQLMetric = metrics("rowConstructionCpuCount") @@ -53,12 +52,9 @@ class HashAggregateMetricsUpdaterImpl(val metrics: Map[String, SQLMetric]) val extractionCpuCount: SQLMetric = metrics("extractionCpuCount") val extractionWallNanos: SQLMetric = metrics("extractionWallNanos") - val finalOutputRows: SQLMetric = metrics("finalOutputRows") - val finalOutputVectors: SQLMetric = metrics("finalOutputVectors") - val loadLazyVectorTime: SQLMetric = metrics("loadLazyVectorTime") - override def updateAggregationMetrics( + final override def updateAggregationMetrics( aggregationMetrics: java.util.ArrayList[OperatorMetrics], aggParams: AggregationParams): Unit = { var idx = 0 @@ -82,8 +78,8 @@ class HashAggregateMetricsUpdaterImpl(val metrics: Map[String, SQLMetric]) aggSpilledPartitions += aggMetrics.spilledPartitions aggSpilledFiles += aggMetrics.spilledFiles flushRowCount += aggMetrics.flushRowCount - abandonedPartialAggregationRows += aggMetrics.abandonedPartialAggregationRows loadedToValueHook += aggMetrics.loadedToValueHook + updateAggSpecificMetrics(aggMetrics) idx += 1 if (aggParams.rowConstructionNeeded) { @@ -103,4 +99,6 @@ class HashAggregateMetricsUpdaterImpl(val metrics: Map[String, SQLMetric]) aggMetrics.spilledBytes) } } + + protected def updateAggSpecificMetrics(aggMetrics: OperatorMetrics): Unit = {} } diff --git a/backends-velox/src/main/scala/org/apache/gluten/metrics/InputIteratorMetricsUpdater.scala b/backends-core/src/main/scala/org/apache/gluten/metrics/SharedInputIteratorMetricsUpdater.scala similarity index 73% copy from backends-velox/src/main/scala/org/apache/gluten/metrics/InputIteratorMetricsUpdater.scala copy to backends-core/src/main/scala/org/apache/gluten/metrics/SharedInputIteratorMetricsUpdater.scala index ed006a5434..82ec8030a7 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/metrics/InputIteratorMetricsUpdater.scala +++ b/backends-core/src/main/scala/org/apache/gluten/metrics/SharedInputIteratorMetricsUpdater.scala @@ -18,30 +18,30 @@ package org.apache.gluten.metrics import org.apache.spark.sql.execution.metric.SQLMetric -case class InputIteratorMetricsUpdater(metrics: Map[String, SQLMetric], forBroadcast: Boolean) +abstract class SharedInputIteratorMetricsUpdater( + val metrics: Map[String, SQLMetric], + val forBroadcast: Boolean) extends MetricsUpdater { - override def updateNativeMetrics(opMetrics: IOperatorMetrics): Unit = { + + final override def updateNativeMetrics(opMetrics: IOperatorMetrics): Unit = { if (opMetrics != null) { val operatorMetrics = opMetrics.asInstanceOf[OperatorMetrics] metrics("cpuCount") += operatorMetrics.cpuCount metrics("wallNanos") += operatorMetrics.wallNanos if (!forBroadcast) { if (operatorMetrics.outputRows == 0 && operatorMetrics.outputVectors == 0) { - // Sometimes, velox does not update metrics for intermediate operator, - // here we try to use the input metrics + // Sometimes, native backend does not update metrics for intermediate operator, + // here we try to use the input metrics. metrics("numOutputRows") += operatorMetrics.inputRows metrics("outputVectors") += operatorMetrics.inputVectors } else { metrics("numOutputRows") += operatorMetrics.outputRows metrics("outputVectors") += operatorMetrics.outputVectors } - metrics.get("valueStreamDynamicFiltersAccepted").foreach { - _ += operatorMetrics.numDynamicFiltersAccepted - } - metrics.get("valueStreamDynamicFilterInputRows").foreach { - _ += operatorMetrics.numDynamicFilterInputRows - } + updateExtraOutputMetrics(operatorMetrics) } } } + + protected def updateExtraOutputMetrics(operatorMetrics: OperatorMetrics): Unit = {} } diff --git a/backends-velox/src/main/scala/org/apache/gluten/execution/ArrowColumnarToVeloxColumnarExec.scala b/backends-velox/src/main/scala/org/apache/gluten/execution/ArrowColumnarToVeloxColumnarExec.scala index cca381ee84..bc205b5f3e 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/execution/ArrowColumnarToVeloxColumnarExec.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/execution/ArrowColumnarToVeloxColumnarExec.scala @@ -16,25 +16,17 @@ */ package org.apache.gluten.execution -import org.apache.gluten.backendsapi.arrow.ArrowBatchTypes.ArrowNativeBatchType import org.apache.gluten.backendsapi.velox.VeloxBatchType import org.apache.gluten.columnarbatch.VeloxColumnarBatches -import org.apache.gluten.extension.columnar.transition.Convention import org.apache.spark.sql.execution.SparkPlan import org.apache.spark.sql.vectorized.ColumnarBatch case class ArrowColumnarToVeloxColumnarExec(override val child: SparkPlan) - extends ColumnarToColumnarExec(child) - with GlutenColumnarToColumnarTransition { + extends SharedArrowColumnarToBackendColumnarExec(child, VeloxBatchType) { - override protected val from: Convention.BatchType = ArrowNativeBatchType - - override protected val to: Convention.BatchType = VeloxBatchType - - override protected def mapIterator(in: Iterator[ColumnarBatch]): Iterator[ColumnarBatch] = { - in.map(b => VeloxColumnarBatches.toVeloxBatch(b)) - } + override protected def toBackendBatch(batch: ColumnarBatch): ColumnarBatch = + VeloxColumnarBatches.toVeloxBatch(batch) override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan = copy(child = newChild) diff --git a/backends-velox/src/main/scala/org/apache/gluten/metrics/HashAggregateMetricsUpdater.scala b/backends-velox/src/main/scala/org/apache/gluten/metrics/HashAggregateMetricsUpdater.scala index dcfa0d7cf3..2837a35d54 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/metrics/HashAggregateMetricsUpdater.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/metrics/HashAggregateMetricsUpdater.scala @@ -19,88 +19,22 @@ package org.apache.gluten.metrics import org.apache.gluten.substrait.AggregationParams import org.apache.spark.sql.execution.metric.SQLMetric -import org.apache.spark.sql.utils.SparkMetricsUtil -import org.apache.spark.task.TaskResources -import scala.collection.JavaConverters._ - -trait HashAggregateMetricsUpdater extends MetricsUpdater { +trait HashAggregateMetricsUpdater extends SharedHashAggregateMetricsUpdater { def updateAggregationMetrics( aggregationMetrics: java.util.ArrayList[OperatorMetrics], aggParams: AggregationParams): Unit } -class HashAggregateMetricsUpdaterImpl(val metrics: Map[String, SQLMetric]) - extends HashAggregateMetricsUpdater { - val aggOutputRows: SQLMetric = metrics("aggOutputRows") - val aggOutputVectors: SQLMetric = metrics("aggOutputVectors") - val aggOutputBytes: SQLMetric = metrics("aggOutputBytes") - val aggCpuCount: SQLMetric = metrics("aggCpuCount") - val aggWallNanos: SQLMetric = metrics("aggWallNanos") - val aggPeakMemoryBytes: SQLMetric = metrics("aggPeakMemoryBytes") - val aggNumMemoryAllocations: SQLMetric = metrics("aggNumMemoryAllocations") - val aggSpilledBytes: SQLMetric = metrics("aggSpilledBytes") - val aggSpilledRows: SQLMetric = metrics("aggSpilledRows") - val aggSpilledPartitions: SQLMetric = metrics("aggSpilledPartitions") - val aggSpilledFiles: SQLMetric = metrics("aggSpilledFiles") - val flushRowCount: SQLMetric = metrics("flushRowCount") +class HashAggregateMetricsUpdaterImpl(override val metrics: Map[String, SQLMetric]) + extends SharedHashAggregateMetricsUpdaterBase(metrics) + with HashAggregateMetricsUpdater { val abandonedPartialAggregationRows: SQLMetric = metrics("abandonedPartialAggregationRows") - val loadedToValueHook: SQLMetric = metrics("loadedToValueHook") - - val rowConstructionCpuCount: SQLMetric = metrics("rowConstructionCpuCount") - val rowConstructionWallNanos: SQLMetric = metrics("rowConstructionWallNanos") - - val extractionCpuCount: SQLMetric = metrics("extractionCpuCount") - val extractionWallNanos: SQLMetric = metrics("extractionWallNanos") val finalOutputRows: SQLMetric = metrics("finalOutputRows") val finalOutputVectors: SQLMetric = metrics("finalOutputVectors") - val loadLazyVectorTime: SQLMetric = metrics("loadLazyVectorTime") - - override def updateAggregationMetrics( - aggregationMetrics: java.util.ArrayList[OperatorMetrics], - aggParams: AggregationParams): Unit = { - var idx = 0 - - if (aggParams.extractionNeeded) { - extractionCpuCount += aggregationMetrics.get(idx).cpuCount - extractionWallNanos += aggregationMetrics.get(idx).wallNanos - idx += 1 - } - - val aggMetrics = aggregationMetrics.get(idx) - aggOutputRows += aggMetrics.outputRows - aggOutputVectors += aggMetrics.outputVectors - aggOutputBytes += aggMetrics.outputBytes - aggCpuCount += aggMetrics.cpuCount - aggWallNanos += aggMetrics.wallNanos - aggPeakMemoryBytes += aggMetrics.peakMemoryBytes - aggNumMemoryAllocations += aggMetrics.numMemoryAllocations - aggSpilledBytes += aggMetrics.spilledBytes - aggSpilledRows += aggMetrics.spilledRows - aggSpilledPartitions += aggMetrics.spilledPartitions - aggSpilledFiles += aggMetrics.spilledFiles - flushRowCount += aggMetrics.flushRowCount + override protected def updateAggSpecificMetrics(aggMetrics: OperatorMetrics): Unit = { abandonedPartialAggregationRows += aggMetrics.abandonedPartialAggregationRows - loadedToValueHook += aggMetrics.loadedToValueHook - idx += 1 - - if (aggParams.rowConstructionNeeded) { - rowConstructionCpuCount += aggregationMetrics.get(idx).cpuCount - rowConstructionWallNanos += aggregationMetrics.get(idx).wallNanos - idx += 1 - } - - loadLazyVectorTime += aggregationMetrics.asScala.last.loadLazyVectorTime - - if (TaskResources.inSparkTask()) { - SparkMetricsUtil.incMemoryBytesSpilled( - TaskResources.getLocalTaskContext().taskMetrics(), - aggMetrics.spilledInputBytes) - SparkMetricsUtil.incDiskBytesSpilled( - TaskResources.getLocalTaskContext().taskMetrics(), - aggMetrics.spilledBytes) - } } } diff --git a/backends-velox/src/main/scala/org/apache/gluten/metrics/InputIteratorMetricsUpdater.scala b/backends-velox/src/main/scala/org/apache/gluten/metrics/InputIteratorMetricsUpdater.scala index ed006a5434..f722a8f2a2 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/metrics/InputIteratorMetricsUpdater.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/metrics/InputIteratorMetricsUpdater.scala @@ -18,30 +18,16 @@ package org.apache.gluten.metrics import org.apache.spark.sql.execution.metric.SQLMetric -case class InputIteratorMetricsUpdater(metrics: Map[String, SQLMetric], forBroadcast: Boolean) - extends MetricsUpdater { - override def updateNativeMetrics(opMetrics: IOperatorMetrics): Unit = { - if (opMetrics != null) { - val operatorMetrics = opMetrics.asInstanceOf[OperatorMetrics] - metrics("cpuCount") += operatorMetrics.cpuCount - metrics("wallNanos") += operatorMetrics.wallNanos - if (!forBroadcast) { - if (operatorMetrics.outputRows == 0 && operatorMetrics.outputVectors == 0) { - // Sometimes, velox does not update metrics for intermediate operator, - // here we try to use the input metrics - metrics("numOutputRows") += operatorMetrics.inputRows - metrics("outputVectors") += operatorMetrics.inputVectors - } else { - metrics("numOutputRows") += operatorMetrics.outputRows - metrics("outputVectors") += operatorMetrics.outputVectors - } - metrics.get("valueStreamDynamicFiltersAccepted").foreach { - _ += operatorMetrics.numDynamicFiltersAccepted - } - metrics.get("valueStreamDynamicFilterInputRows").foreach { - _ += operatorMetrics.numDynamicFilterInputRows - } - } +case class InputIteratorMetricsUpdater( + override val metrics: Map[String, SQLMetric], + override val forBroadcast: Boolean) + extends SharedInputIteratorMetricsUpdater(metrics, forBroadcast) { + override protected def updateExtraOutputMetrics(operatorMetrics: OperatorMetrics): Unit = { + metrics.get("valueStreamDynamicFiltersAccepted").foreach { + _ += operatorMetrics.numDynamicFiltersAccepted + } + metrics.get("valueStreamDynamicFilterInputRows").foreach { + _ += operatorMetrics.numDynamicFilterInputRows } } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
