Copilot commented on code in PR #13358:
URL: https://github.com/apache/gluten/pull/13358#discussion_r4327452758
##########
backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala:
##########
@@ -1404,6 +1404,7 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi with
Logging {
Sig[VeloxBloomFilterAggregate](ExpressionNames.BLOOM_FILTER_AGG),
Sig[MapFilter](ExpressionNames.MAP_FILTER),
Sig[AssertNotNull](ExpressionNames.ASSERT_NOT_NULL),
+ Sig[IncrementMetricCall](ExpressionNames.INCREMENT_METRIC),
Review Comment:
This registers `IncrementMetricCall` under `increment_metric`, but
`IncrementMetricCall.prettyName` currently evaluates to the slot name (e.g.,
`increment_metric_0`). With a static mapping like this, planning may never hit
this signature for native counters. Align the registry keying with how
`IncrementMetricCall` is matched at translation time (either change the
expression’s `prettyName` to `increment_metric`, or introduce prefix-aware
matching for `increment_metric_` slots).
##########
gluten-substrait/src/main/scala/org/apache/gluten/expression/IncrementMetricCall.scala:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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 transformers emit it in place of `IncrementMetric`
when the backend
+ * supports native counting. `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.
+ */
+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)
Review Comment:
`prettyName` is set to `functionName` (e.g., `increment_metric_0`), but the
Velox expression registry entry added in this PR is keyed on
`ExpressionNames.INCREMENT_METRIC` (`increment_metric`). If expression
translation/lookup is driven by `prettyName` (as it is for many Spark
expressions), `IncrementMetricCall` may not be recognized/translated because
its name won’t match the registered signature. To fix this, either (a) make
`prettyName` return the stable name `increment_metric` and keep the slot name
for backend emission via `functionName`/transformer args, or (b) update the
expression registry to support matching the `increment_metric_` prefix (e.g., a
custom resolver instead of a single static name).
##########
backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala:
##########
@@ -51,8 +51,20 @@ object MetricsUtil extends Logging {
.map(_.asLong())
.getOrElse(0L)
+ private def expressionStatsFromJson(node: JsonNode): JMap[String,
java.lang.Long] = {
+ val stats = new java.util.HashMap[String, java.lang.Long]()
+ Option(node.get("expressionStats")).foreach {
+ exprStats =>
+ exprStats.fields().asScala.foreach {
+ entry => stats.put(entry.getKey,
java.lang.Long.valueOf(entry.getValue.asLong()))
+ }
+ }
+ stats
+ }
Review Comment:
New JSON parsing logic for `expressionStats` is introduced here, but the
added test suite covers only `ProjectMetricsUpdater` behavior. Consider adding
a focused unit test that validates `expressionStats` parsing and aggregation
(e.g., missing field, empty object, multiple entries, and summing across
multiple operator nodes), to prevent regressions in metrics JSON handling.
##########
gluten-substrait/src/main/scala/org/apache/gluten/expression/IncrementMetricCall.scala:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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 transformers emit it in place of `IncrementMetric`
when the backend
+ * supports native counting. `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.
+ */
+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
Review Comment:
`prettyName` is set to `functionName` (e.g., `increment_metric_0`), but the
Velox expression registry entry added in this PR is keyed on
`ExpressionNames.INCREMENT_METRIC` (`increment_metric`). If expression
translation/lookup is driven by `prettyName` (as it is for many Spark
expressions), `IncrementMetricCall` may not be recognized/translated because
its name won’t match the registered signature. To fix this, either (a) make
`prettyName` return the stable name `increment_metric` and keep the slot name
for backend emission via `functionName`/transformer args, or (b) update the
expression registry to support matching the `increment_metric_` prefix (e.g., a
custom resolver instead of a single static name).
##########
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 = scala.collection.mutable.HashMap.empty[String, Long]
+ operatorMetrics.asScala.foreach {
+ metrics =>
+ metrics.expressionStats.asScala.foreach {
+ case (name, rows) =>
+ expressionStats.update(name, expressionStats.getOrElse(name, 0L) +
rows)
+ }
+ }
+ aggregated.expressionStats =
+ expressionStats.map { case (name, rows) => name ->
java.lang.Long.valueOf(rows) }.asJava
Review Comment:
This assigns `aggregated.expressionStats` using `asJava` over a Scala map
produced via `.map(...)`. Depending on Scala/JDK collection converter behavior,
this can yield a view/wrapper rather than a concrete `java.util.HashMap`, and
its mutability characteristics can differ from the `OperatorMetrics` default
(`new java.util.HashMap<>()`). To keep `OperatorMetrics.expressionStats`
consistently a concrete Java map (and avoid subtle wrapper behavior), consider
constructing a `java.util.HashMap[String, java.lang.Long]` directly during
aggregation and assigning that.
--
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]
