[GLUTEN-8010][CORE] Don't generate native metrics if transformer don't generate relNode

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

@@ -38,6 +38,9 @@ object MetricsUtil extends Logging {
           j.metricsUpdater(),
           // must put the buildPlan first
           Seq(treeifyMetricsUpdaters(j.buildPlan), treeifyMetricsUpdaters(j.streamedPlan)))
+      case t: TransformSupport if t.metricsUpdater() == MetricsUpdater.None =>
+        assert(t.children.size == 1, "MetricsUpdater.None can only be used on unary operator")
+        treeifyMetricsUpdaters(t.children.head)
       case t: TransformSupport =>
         MetricsUpdaterTree(t.metricsUpdater(), t.children.map(treeifyMetricsUpdaters))
       case _ =>

gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicPhysicalOperatorTransformer.scala

@@ -61,8 +61,11 @@ abstract class FilterExecTransformerBase(val cond: Expression, val input: SparkP
     case _ => false
   }
 
-  override def metricsUpdater(): MetricsUpdater =
+  override def metricsUpdater(): MetricsUpdater = if (getRemainingCondition == null) {
+    MetricsUpdater.None
+  } else {
     BackendsApiManager.getMetricsApiInstance.genFilterTransformerMetricsUpdater(metrics)
+  }
 
   def getRelNode(
       context: SubstraitContext,
@@ -149,15 +152,15 @@ abstract class FilterExecTransformerBase(val cond: Expression, val input: SparkP
 
   override protected def doTransform(context: SubstraitContext): TransformContext = {
     val childCtx = child.asInstanceOf[TransformSupport].transform(context)
-    val remainingCondition = getRemainingCondition
-    val operatorId = context.nextOperatorId(this.nodeName)
-    if (remainingCondition == null) {
+    if (metricsUpdater == MetricsUpdater.None) {
       // The computing for this filter is not needed.
-      context.registerEmptyRelToOperator(operatorId)
       // Since some columns' nullability will be removed after this filter, we need to update the
       // outputAttributes of child context.
       return TransformContext(output, childCtx.root)
     }
+
+    val operatorId = context.nextOperatorId(this.nodeName)
+    val remainingCondition = getRemainingCondition
     val currRel = getRelNode(
       context,
       remainingCondition,

gluten-substrait/src/main/scala/org/apache/gluten/execution/ExpandExecTransformer.scala

@@ -48,8 +48,11 @@ case class ExpandExecTransformer(
     AttributeSet.fromAttributeSets(projections.flatten.map(_.references))
   }
 
-  override def metricsUpdater(): MetricsUpdater =
+  override def metricsUpdater(): MetricsUpdater = if (projections == null || projections.isEmpty) {
+    MetricsUpdater.None
+  } else {
     BackendsApiManager.getMetricsApiInstance.genExpandTransformerMetricsUpdater(metrics)
+  }
 
   // The GroupExpressions can output data with arbitrary partitioning, so set it
   // as UNKNOWN partitioning
@@ -112,13 +115,12 @@ case class ExpandExecTransformer(
 
   override protected def doTransform(context: SubstraitContext): TransformContext = {
     val childCtx = child.asInstanceOf[TransformSupport].transform(context)
-    val operatorId = context.nextOperatorId(this.nodeName)
-    if (projections == null || projections.isEmpty) {
+    if (metricsUpdater == MetricsUpdater.None) {
       // The computing for this Expand is not needed.
-      context.registerEmptyRelToOperator(operatorId)
       return childCtx
     }
 
+    val operatorId = context.nextOperatorId(this.nodeName)
     val currRel =
       getRelNode(context, projections, child.output, operatorId, childCtx.root, validation = false)
     assert(currRel != null, "Expand Rel should be valid")

gluten-substrait/src/main/scala/org/apache/gluten/execution/SortExecTransformer.scala

@@ -44,8 +44,11 @@ case class SortExecTransformer(
   @transient override lazy val metrics =
     BackendsApiManager.getMetricsApiInstance.genSortTransformerMetrics(sparkContext)
 
-  override def metricsUpdater(): MetricsUpdater =
+  override def metricsUpdater(): MetricsUpdater = if (sortOrder == null || sortOrder.isEmpty) {
+    MetricsUpdater.None
+  } else {
     BackendsApiManager.getMetricsApiInstance.genSortTransformerMetricsUpdater(metrics)
+  }
 
   override def output: Seq[Attribute] = child.output
 
@@ -103,13 +106,12 @@ case class SortExecTransformer(
 
   override protected def doTransform(context: SubstraitContext): TransformContext = {
     val childCtx = child.asInstanceOf[TransformSupport].transform(context)
-    val operatorId = context.nextOperatorId(this.nodeName)
-    if (sortOrder == null || sortOrder.isEmpty) {
+    if (metricsUpdater == MetricsUpdater.None) {
       // The computing for this project is not needed.
-      context.registerEmptyRelToOperator(operatorId)
       return childCtx
     }
 
+    val operatorId = context.nextOperatorId(this.nodeName)
     val currRel =
       getRelNode(context, sortOrder, child.output, operatorId, childCtx.root, validation = false)
     assert(currRel != null, "Sort Rel should be valid")

gluten-substrait/src/main/scala/org/apache/gluten/execution/WindowExecTransformer.scala

@@ -51,8 +51,13 @@ case class WindowExecTransformer(
   @transient override lazy val metrics =
     BackendsApiManager.getMetricsApiInstance.genWindowTransformerMetrics(sparkContext)
 
-  override def metricsUpdater(): MetricsUpdater =
+  override def metricsUpdater(): MetricsUpdater = if (
+    windowExpression == null || windowExpression.isEmpty
+  ) {
+    MetricsUpdater.None
+  } else {
     BackendsApiManager.getMetricsApiInstance.genWindowTransformerMetricsUpdater(metrics)
+  }
 
   override def output: Seq[Attribute] = child.output ++ windowExpression.map(_.toAttribute)
 
@@ -177,13 +182,12 @@ case class WindowExecTransformer(
 
   override protected def doTransform(context: SubstraitContext): TransformContext = {
     val childCtx = child.asInstanceOf[TransformSupport].transform(context)
-    val operatorId = context.nextOperatorId(this.nodeName)
-    if (windowExpression == null || windowExpression.isEmpty) {
+    if (metricsUpdater == MetricsUpdater.None) {
       // The computing for this operator is not needed.
-      context.registerEmptyRelToOperator(operatorId)
       return childCtx
     }
 
+    val operatorId = context.nextOperatorId(this.nodeName)
     val currRel =
       getWindowRel(context, child.output, operatorId, childCtx.root, validation = false)
     assert(currRel != null, "Window Rel should be valid")

gluten-substrait/src/main/scala/org/apache/gluten/substrait/SubstraitContext.scala

@@ -112,19 +112,6 @@ class SubstraitContext extends Serializable {
     id
   }
 
-  /**
-   * Register empty rel list to certain operator id. Used when the computing of a Spark transformer
-   * is omitted.
-   * @param operatorId
-   *   operator id
-   */
-  def registerEmptyRelToOperator(operatorId: JLong): Unit = {
-    if (!operatorToRelsMap.containsKey(operatorId)) {
-      val rels = new JArrayList[JLong]()
-      operatorToRelsMap.put(operatorId, rels)
-    }
-  }
-
   /**
    * Return the registered map.
    * @return