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 26ec3497b84f9a010882f4f9aa4af38d35486670 Author: liyang.127 <[email protected]> AuthorDate: Tue Jun 30 21:56:13 2026 +0800 [GLUTEN][CORE] Share tier-2 function test suites under backends-core --- backends-core/pom.xml | 16 +++++- .../gluten/execution/SharedRowToColumnarExec.scala | 14 ++++- .../metrics/NestedLoopJoinMetricsUpdater.scala | 2 +- .../gluten/metrics/SharedJoinMetricsUpdater.scala | 60 ++++++++++++++++++++++ .../benchmarks/NativeBenchmarkPlanGenerator.scala | 4 +- .../functions/ArithmeticAnsiValidateSuite.scala | 0 .../functions/MathFunctionsValidateSuite.scala | 0 .../functions/WindowFunctionsValidateSuite.scala | 0 .../org/apache/gluten/metrics/MetricsUtil.scala | 11 ++++ 9 files changed, 100 insertions(+), 7 deletions(-) diff --git a/backends-core/pom.xml b/backends-core/pom.xml index 64dcfbe29e..68aa022d77 100644 --- a/backends-core/pom.xml +++ b/backends-core/pom.xml @@ -62,6 +62,12 @@ <version>${scala.version}</version> <scope>provided</scope> </dependency> + <dependency> + <groupId>com.google.jimfs</groupId> + <artifactId>jimfs</artifactId> + <version>1.3.0</version> + <scope>compile</scope> + </dependency> <dependency> <groupId>org.apache.gluten</groupId> <artifactId>gluten-substrait</artifactId> @@ -112,7 +118,7 @@ <groupId>org.apache.arrow</groupId> <artifactId>arrow-vector</artifactId> <version>${arrow.version}</version> - <scope>test</scope> + <scope>compile</scope> </dependency> </dependencies> @@ -208,7 +214,13 @@ </dependency> <dependency> <groupId>org.apache.iceberg</groupId> - <artifactId>iceberg-spark-runtime-${sparkbundle.version}_${scala.binary.version}</artifactId> + <artifactId>iceberg-core</artifactId> + <version>${iceberg.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.iceberg</groupId> + <artifactId>iceberg-api</artifactId> <version>${iceberg.version}</version> <scope>provided</scope> </dependency> diff --git a/backends-core/src/main/scala/org/apache/gluten/execution/SharedRowToColumnarExec.scala b/backends-core/src/main/scala/org/apache/gluten/execution/SharedRowToColumnarExec.scala index 5f375f53ae..22fd7a59ef 100644 --- a/backends-core/src/main/scala/org/apache/gluten/execution/SharedRowToColumnarExec.scala +++ b/backends-core/src/main/scala/org/apache/gluten/execution/SharedRowToColumnarExec.scala @@ -29,7 +29,8 @@ import org.apache.spark.broadcast.Broadcast import org.apache.spark.rdd.RDD import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.{UnsafeProjection, UnsafeRow} -import org.apache.spark.sql.execution.{BroadcastUtils, SparkPlan} +import org.apache.spark.sql.catalyst.plans.physical.{BroadcastMode, BroadcastPartitioning, Partitioning} +import org.apache.spark.sql.execution.SparkPlan import org.apache.spark.sql.execution.metric.SQLMetric import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.StructType @@ -92,7 +93,7 @@ abstract class SharedRowToColumnarExec(child: SparkPlan) extends RowToColumnarEx val convertTime = longMetric("convertTime") val numRows = GlutenConfig.get.maxBatchSize val numBytes = preferredBatchBytes - val mode = BroadcastUtils.getBroadcastMode(outputPartitioning) + val mode = SharedRowToColumnarExec.getBroadcastMode(outputPartitioning) val relation = child.executeBroadcast() sparkToBackendUnsafe( sparkContext, @@ -114,6 +115,15 @@ abstract class SharedRowToColumnarExec(child: SparkPlan) extends RowToColumnarEx object SharedRowToColumnarExec { + private def getBroadcastMode(partitioning: Partitioning): BroadcastMode = { + partitioning match { + case BroadcastPartitioning(mode) => + mode + case _ => + throw new IllegalArgumentException("Unexpected partitioning: " + partitioning.toString) + } + } + def toColumnarBatchIterator( it: Iterator[InternalRow], schema: StructType, diff --git a/backends-core/src/main/scala/org/apache/gluten/metrics/NestedLoopJoinMetricsUpdater.scala b/backends-core/src/main/scala/org/apache/gluten/metrics/NestedLoopJoinMetricsUpdater.scala index f2a4dfff36..56b57bc61e 100644 --- a/backends-core/src/main/scala/org/apache/gluten/metrics/NestedLoopJoinMetricsUpdater.scala +++ b/backends-core/src/main/scala/org/apache/gluten/metrics/NestedLoopJoinMetricsUpdater.scala @@ -25,7 +25,7 @@ import java.util import scala.collection.JavaConverters._ class NestedLoopJoinMetricsUpdater(override val metrics: Map[String, SQLMetric]) - extends JoinMetricsUpdaterBase(metrics) { + extends SharedJoinMetricsUpdaterBase(metrics) { val nestedLoopJoinBuildInputRows: SQLMetric = metrics("nestedLoopJoinBuildInputRows") val nestedLoopJoinBuildOutputRows: SQLMetric = metrics("nestedLoopJoinBuildOutputRows") diff --git a/backends-core/src/main/scala/org/apache/gluten/metrics/SharedJoinMetricsUpdater.scala b/backends-core/src/main/scala/org/apache/gluten/metrics/SharedJoinMetricsUpdater.scala new file mode 100644 index 0000000000..a393ff2562 --- /dev/null +++ b/backends-core/src/main/scala/org/apache/gluten/metrics/SharedJoinMetricsUpdater.scala @@ -0,0 +1,60 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.gluten.metrics + +import org.apache.gluten.metrics.Metrics.SingleMetric +import org.apache.gluten.substrait.JoinParams + +import org.apache.spark.sql.execution.metric.SQLMetric + +import java.util + +trait SharedJoinMetricsUpdater extends MetricsUpdater { + def updateJoinMetrics( + joinMetrics: util.ArrayList[OperatorMetrics], + singleMetrics: SingleMetric, + joinParams: JoinParams): Unit +} + +abstract class SharedJoinMetricsUpdaterBase(val metrics: Map[String, SQLMetric]) + extends SharedJoinMetricsUpdater { + val postProjectionCpuCount: SQLMetric = metrics("postProjectionCpuCount") + val postProjectionWallNanos: SQLMetric = metrics("postProjectionWallNanos") + val numOutputRows: SQLMetric = metrics("numOutputRows") + val numOutputVectors: SQLMetric = metrics("numOutputVectors") + val numOutputBytes: SQLMetric = metrics("numOutputBytes") + + final override def updateJoinMetrics( + joinMetrics: util.ArrayList[OperatorMetrics], + singleMetrics: SingleMetric, + joinParams: JoinParams): Unit = { + if (joinParams != null && joinParams.postProjectionNeeded) { + val postProjectMetrics = joinMetrics.remove(0) + postProjectionCpuCount += postProjectMetrics.cpuCount + postProjectionWallNanos += postProjectMetrics.wallNanos + numOutputRows += postProjectMetrics.outputRows + numOutputVectors += postProjectMetrics.outputVectors + numOutputBytes += postProjectMetrics.outputBytes + } + + updateJoinMetricsInternal(joinMetrics, joinParams) + } + + protected def updateJoinMetricsInternal( + joinMetrics: util.ArrayList[OperatorMetrics], + joinParams: JoinParams): Unit +} diff --git a/backends-velox/src/test/scala/org/apache/gluten/benchmarks/NativeBenchmarkPlanGenerator.scala b/backends-core/src/test/scala/org/apache/gluten/benchmarks/NativeBenchmarkPlanGenerator.scala similarity index 96% rename from backends-velox/src/test/scala/org/apache/gluten/benchmarks/NativeBenchmarkPlanGenerator.scala rename to backends-core/src/test/scala/org/apache/gluten/benchmarks/NativeBenchmarkPlanGenerator.scala index 6886adbd26..257f24c5ac 100644 --- a/backends-velox/src/test/scala/org/apache/gluten/benchmarks/NativeBenchmarkPlanGenerator.scala +++ b/backends-core/src/test/scala/org/apache/gluten/benchmarks/NativeBenchmarkPlanGenerator.scala @@ -17,7 +17,7 @@ package org.apache.gluten.benchmarks import org.apache.gluten.config.GlutenConfig -import org.apache.gluten.execution.{VeloxWholeStageTransformerSuite, WholeStageTransformer} +import org.apache.gluten.execution.{SharedWholeStageTransformerSuite, WholeStageTransformer} import org.apache.spark.SparkConf import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec @@ -30,7 +30,7 @@ import java.io.File object GenerateExample extends Tag("org.apache.gluten.tags.GenerateExample") -class NativeBenchmarkPlanGenerator extends VeloxWholeStageTransformerSuite { +class NativeBenchmarkPlanGenerator extends SharedWholeStageTransformerSuite { override protected val resourcePath: String = "/tpch-data-parquet" override protected val fileFormat: String = "parquet" val generatedPlanDir = getClass.getResource("/").getPath + "../../../generated-native-benchmark/" diff --git a/backends-velox/src/test/scala/org/apache/gluten/functions/ArithmeticAnsiValidateSuite.scala b/backends-core/src/test/scala/org/apache/gluten/functions/ArithmeticAnsiValidateSuite.scala similarity index 100% rename from backends-velox/src/test/scala/org/apache/gluten/functions/ArithmeticAnsiValidateSuite.scala rename to backends-core/src/test/scala/org/apache/gluten/functions/ArithmeticAnsiValidateSuite.scala diff --git a/backends-velox/src/test/scala/org/apache/gluten/functions/MathFunctionsValidateSuite.scala b/backends-core/src/test/scala/org/apache/gluten/functions/MathFunctionsValidateSuite.scala similarity index 100% rename from backends-velox/src/test/scala/org/apache/gluten/functions/MathFunctionsValidateSuite.scala rename to backends-core/src/test/scala/org/apache/gluten/functions/MathFunctionsValidateSuite.scala diff --git a/backends-velox/src/test/scala/org/apache/gluten/functions/WindowFunctionsValidateSuite.scala b/backends-core/src/test/scala/org/apache/gluten/functions/WindowFunctionsValidateSuite.scala similarity index 100% rename from backends-velox/src/test/scala/org/apache/gluten/functions/WindowFunctionsValidateSuite.scala rename to backends-core/src/test/scala/org/apache/gluten/functions/WindowFunctionsValidateSuite.scala diff --git a/backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala b/backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala index 878eb2dcf1..5c04675126 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala @@ -276,6 +276,17 @@ object MetricsUtil extends Logging { p } ju.updateJoinMetrics(operatorMetrics, metrics.getSingleMetrics, joinParams) + case ju: SharedJoinMetricsUpdaterBase => + // JoinRel and CrossRel output two suites of metrics respectively for build and probe. + // Therefore, fetch one more suite of metrics here. + operatorMetrics.add(metrics.getOperatorMetrics(curMetricsIdx)) + curMetricsIdx -= 1 + val joinParams = Option(joinParamsMap.get(operatorIdx)).getOrElse { + val p = JoinParams() + p.postProjectionNeeded = false + p + } + ju.updateJoinMetrics(operatorMetrics, metrics.getSingleMetrics, joinParams) case u: UnionMetricsUpdater => // Union outputs two suites of metrics respectively. // Therefore, fetch one more suite of metrics here. --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
