malinjawi commented on code in PR #13358:
URL: https://github.com/apache/gluten/pull/13358#discussion_r4491631094
##########
backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala:
##########
@@ -359,6 +371,16 @@ object MetricsUtil extends Logging {
aggregated.bloomFilterTestedRows = bloomFilterTestedRows
aggregated.bloomFilterAcceptedRows = bloomFilterAcceptedRows
aggregated.bloomFilterBypassed = bloomFilterBypassed
+ val expressionStats = new java.util.HashMap[String, java.lang.Long]()
+ operatorMetrics.asScala.foreach {
+ metrics =>
+ metrics.expressionStats.asScala.foreach {
+ case (name, rows) =>
+ val sum =
Option(expressionStats.get(name)).map(_.longValue()).getOrElse(0L) + rows
+ expressionStats.put(name, java.lang.Long.valueOf(sum))
Review Comment:
Leaving this as is. The field is initialized to an empty `HashMap` in its
declaration, so every constructor sets it, and the only two assignments
(`operatorMetricFromJson` and `mergeMetrics`) store a fresh map. Nothing in the
repo or JNI writes null to it.
##########
backends-velox/src/main/scala/org/apache/gluten/metrics/ProjectMetricsUpdater.scala:
##########
@@ -38,6 +40,9 @@ class ProjectMetricsUpdater(
case (name, metric) =>
name match {
case "increment_metric" => metric += operatorMetrics.outputRows
+ case slot if
slot.startsWith(IncrementMetricCall.functionNamePrefix) =>
+ // The counter ran natively: credit the rows Velox evaluated it
on.
+ metric += operatorMetrics.expressionStats.getOrDefault(slot, 0L)
Review Comment:
Only `IncrementMetricCall.functionName(slot)` builds names with this prefix,
and it already requires the slot to be between 0 and `maxCounters - 1`. Delta's
own `IncrementMetric` arrives as exactly `increment_metric`, so a malformed
slot name can't reach this. Falling back to output rows would also bring back
the overcounting this change fixes.
##########
gluten-substrait/src/main/scala/org/apache/gluten/expression/IncrementMetricCall.scala:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.expression
+
+import org.apache.spark.sql.catalyst.expressions.{Expression, UnaryExpression,
Unevaluable}
+import org.apache.spark.sql.types.DataType
+
+/**
+ * A native pass-through call that makes the backend count how many rows
reached this point of an
+ * expression tree. Delta's project transformer emits it in place of
`IncrementMetric` when the
+ * backend supports native counting; filters never carry one. `functionName`
is one of the backend's
+ * fixed counter slots and is unique to one SQL metric within its projection:
the backend keys its
+ * per-expression statistics by function name, and the metrics updater credits
the metric from that
+ * entry. Never evaluated on Spark; it only exists in the expression list
handed to the backend.
+ *
+ * Gluten maps it by class (`ExpressionNames.INCREMENT_METRIC`, which is also
the name to put in the
+ * expression blacklist), and `getTransformer` sends `functionName` to the
backend. Its `prettyName`
+ * is the slot name, never `increment_metric`, which is how Gluten recognizes
Delta's own
+ * `IncrementMetric` without depending on Delta.
+ */
+case class IncrementMetricCall(child: Expression, functionName: String)
+ extends UnaryExpression
+ with Unevaluable
+ with Transformable {
+
+ override def dataType: DataType = child.dataType
+
+ override def nullable: Boolean = child.nullable
+
+ override def prettyName: String = functionName
+
+ override protected def withNewChildInternal(newChild: Expression):
IncrementMetricCall =
+ copy(child = newChild)
+
+ override def getTransformer(
+ childrenTransformers: Seq[ExpressionTransformer]): ExpressionTransformer
=
+ GenericExpressionTransformer(functionName, childrenTransformers, this)
+}
+
+object IncrementMetricCall {
+
+ /** Prefix of every counter slot name. */
+ val functionNamePrefix: String = "increment_metric_"
+
+ /**
+ * Number of counter slots the backend registers at startup
(`increment_metric_0` to
+ * `increment_metric_31`). Mirrors `kIncrementMetricFunctionSlots` in the
Velox backend. A
+ * projection with more distinct metrics than this falls back.
+ */
+ val maxCounters: Int = 32
+
Review Comment:
Only Velox counts natively, and `supportNativeIncrementMetric` is already
the per-backend switch, so a `BackendSettingsApi` getter would just move this
copy of the C++ constant into `VeloxBackend`. Drift also fails safe: a slot
Velox hasn't registered fails native validation, and the projection falls back
to Spark with correct metrics. I'd add the getter if a second backend
implements counting.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
