[GLUTEN-9003][CORE] Carry native counter counts into SQL metrics

backends-velox/src/main/java/org/apache/gluten/metrics/OperatorMetrics.java

@@ -17,6 +17,9 @@
 package org.apache.gluten.metrics;
 
 public class OperatorMetrics implements IOperatorMetrics {
+  /** Rows processed per expression, keyed by function name; see IncrementMetricCall. */
+  public java.util.Map<String, Long> expressionStats = new java.util.HashMap<>();
+
   public long inputRows;
   public long inputVectors;
   public long inputBytes;

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala

@@ -1417,6 +1417,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),
       Sig[MonotonicallyIncreasingID](ExpressionNames.MONOTONICALLY_INCREASING_ID),
       // For test purpose.
       Sig[VeloxDummyExpression](VeloxDummyExpression.VELOX_DUMMY_EXPRESSION)

backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala

@@ -51,8 +51,20 @@ object MetricsUtil extends Logging {
       .map(_.asLong())
       .getOrElse(0L)
 
+  private[metrics] 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
+  }
+
   private def operatorMetricFromJson(node: JsonNode): OperatorMetrics = {
     val metrics = new OperatorMetrics()
+    metrics.expressionStats = expressionStatsFromJson(node)
     metrics.inputRows = value(node, "inputRows")
     metrics.inputVectors = value(node, "inputVectors")
     metrics.inputBytes = value(node, "inputBytes")
@@ -207,7 +219,7 @@ object MetricsUtil extends Logging {
    * @return
    *   the merged metrics
    */
-  private def mergeMetrics(operatorMetrics: JList[OperatorMetrics]): OperatorMetrics = {
+  private[metrics] def mergeMetrics(operatorMetrics: JList[OperatorMetrics]): OperatorMetrics = {
     if (operatorMetrics.size() == 0) {
       return null
     }
@@ -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))
+        }
+    }
+    aggregated.expressionStats = expressionStats
     aggregated
   }
 

backends-velox/src/main/scala/org/apache/gluten/metrics/ProjectMetricsUpdater.scala

@@ -16,6 +16,8 @@
  */
 package org.apache.gluten.metrics
 
+import org.apache.gluten.expression.IncrementMetricCall
+
 import org.apache.spark.sql.execution.metric.SQLMetric
 
 class ProjectMetricsUpdater(
@@ -38,6 +40,10 @@ class ProjectMetricsUpdater(
         case (name, metric) =>
           name match {
             case "increment_metric" => metric += operatorMetrics.outputRows
+            case name if name.startsWith(IncrementMetricCall.functionNamePrefix) =>
+              // The counter ran natively: credit the rows Velox evaluated it on.
+              metric += Option(
+                operatorMetrics.expressionStats.get(name)).map(_.longValue()).getOrElse(0L)
             case _ => // do nothing
           }
       }

backends-velox/src/test/scala/org/apache/gluten/metrics/MetricsUtilSuite.scala

@@ -0,0 +1,69 @@
+/*
+ * 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 com.fasterxml.jackson.databind.ObjectMapper
+import org.scalatest.funsuite.AnyFunSuite
+
+import java.util.{Arrays => JArrays}
+
+import scala.collection.JavaConverters._
+
+class MetricsUtilSuite extends AnyFunSuite {
+
+  private val mapper = new ObjectMapper()
+
+  private def expressionStats(json: String): Map[String, Long] =
+    MetricsUtil
+      .expressionStatsFromJson(mapper.readTree(json))
+      .asScala
+      .map { case (name, rows) => name -> rows.longValue() }
+      .toMap
+
+  private def operatorMetrics(expressionStats: (String, Long)*): OperatorMetrics = {
+    val metrics = new OperatorMetrics()
+    expressionStats.foreach {
+      case (name, rows) => metrics.expressionStats.put(name, java.lang.Long.valueOf(rows))
+    }
+    metrics
+  }
+
+  test("expression stats are read from an operator's metrics") {
+    val json =
+      """{"outputRows": 4, "expressionStats": {"increment_metric_0": 3, "increment_metric_1": 0}}"""
+    assert(
+      expressionStats(json) === Map("increment_metric_0" -> 3L, "increment_metric_1" -> 0L))
+  }
+
+  test("an operator without expression stats has none") {
+    assert(expressionStats("""{"outputRows": 4}""").isEmpty)
+    assert(expressionStats("""{"outputRows": 4, "expressionStats": {}}""").isEmpty)
+  }
+
+  test("merging operator metrics sums the expression stats of each function") {
+    val merged = MetricsUtil.mergeMetrics(
+      JArrays.asList(
+        operatorMetrics("increment_metric_0" -> 3L, "increment_metric_1" -> 1L),
+        operatorMetrics("increment_metric_0" -> 2L),
+        operatorMetrics()))
+    assert(
+      merged.expressionStats.asScala.map { case (name, rows) => name -> rows.longValue() } ===
+        Map("increment_metric_0" -> 5L, "increment_metric_1" -> 1L))
+    // A concrete map, like the one every OperatorMetrics starts with.
+    assert(merged.expressionStats.isInstanceOf[java.util.HashMap[_, _]])
+  }
+}

backends-velox/src/test/scala/org/apache/gluten/metrics/ProjectMetricsUpdaterSuite.scala

@@ -0,0 +1,75 @@
+/*
+ * 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.spark.sql.execution.metric.SQLMetric
+
+import org.scalatest.funsuite.AnyFunSuite
+
+class ProjectMetricsUpdaterSuite extends AnyFunSuite {
+
+  private def projectMetrics(): Map[String, SQLMetric] = Seq(
+    "numOutputRows",
+    "outputVectors",
+    "outputBytes",
+    "cpuCount",
+    "wallNanos",
+    "peakMemoryBytes",
+    "numMemoryAllocations",
+    "loadLazyVectorTime").map(_ -> new SQLMetric("sum")).toMap
+
+  private def operatorMetrics(
+      outputRows: Long,
+      expressionStats: (String, Long)*): OperatorMetrics = {
+    val metrics = new OperatorMetrics()
+    metrics.outputRows = outputRows
+    expressionStats.foreach {
+      case (name, rows) => metrics.expressionStats.put(name, java.lang.Long.valueOf(rows))
+    }
+    metrics
+  }
+
+  test("a native counter is credited with the rows its function processed") {
+    val updated = new SQLMetric("sum")
+    val copied = new SQLMetric("sum")
+    val updater = new ProjectMetricsUpdater(
+      projectMetrics(),
+      Seq("increment_metric_0" -> updated, "increment_metric_1" -> copied))
+    updater.updateNativeMetrics(
+      operatorMetrics(
+        10,
+        "increment_metric_0" -> 3L,
+        "increment_metric_1" -> 7L,
+        "increment_metric_2" -> 10L))
+    assert(updated.value === 3)
+    assert(copied.value === 7)
+  }
+
+  test("a native counter that never ran is credited with zero rows") {
+    val metric = new SQLMetric("sum")
+    val updater = new ProjectMetricsUpdater(projectMetrics(), Seq("increment_metric_0" -> metric))
+    updater.updateNativeMetrics(operatorMetrics(10))
+    assert(metric.value === 0)
+  }
+
+  test("a counter removed from the native plan is credited with the output rows") {
+    val metric = new SQLMetric("sum")
+    val updater = new ProjectMetricsUpdater(projectMetrics(), Seq("increment_metric" -> metric))
+    updater.updateNativeMetrics(operatorMetrics(10, "increment_metric_0" -> 3L))
+    assert(metric.value === 10)
+  }
+}

gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala

@@ -62,6 +62,13 @@ trait BackendSettingsApi {
 
   def supportNativeRowIndexColumn(): Boolean = true
 
+  /**
+   * Whether the backend evaluates Delta's `IncrementMetric` natively through
+   * [[org.apache.gluten.expression.IncrementMetricCall]] and reports the processed-row count of
+   * each counter function in its operator metrics.
+   */
+  def supportNativeIncrementMetric(): Boolean = false
+
   def supportSortExec(): Boolean = false
 
   def supportSortMergeJoinExec(): Boolean = true

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 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.
+ *
+ * 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
+
+  /** Name of the counter slot `slot`, which must be below [[maxCounters]]. */
+  def functionName(slot: Int): String = {
+    require(slot >= 0 && slot < maxCounters, s"counter slot $slot out of range")
+    functionNamePrefix + slot
+  }
+}

shims/common/src/main/scala/org/apache/gluten/expression/ExpressionNames.scala

@@ -361,6 +361,7 @@ object ExpressionNames {
   final val AT_LEAST_N_NON_NULLS = "at_least_n_non_nulls"
   final val ASSERT_TRUE = "assert_true"
   final val ASSERT_NOT_NULL = "assert_not_null"
+  final val INCREMENT_METRIC = "increment_metric"
   final val NULLIF = "nullif"
   final val NVL = "nvl"
   final val NVL2 = "nvl2"