[SPARK-41442][SQL] Only update SQLMetric value if merging with valid metric (apache#1643)

### What changes were proposed in this pull request?

This patch updates how `SQLMetric` merges two invalid instances where the value is both -1.

### Why are the changes needed?

We use -1 as initial value of `SQLMetric`, and change it to 0 while merging with other `SQLMetric` instances. A `SQLMetric` will be treated as invalid and filtered out later.

While we are developing with Spark, it is trouble behavior that two invalid `SQLMetric` instances merge to a valid `SQLMetric` because merging will set the value to 0.

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

Existing tests.

Closes apache#38969 from viirya/minor_sql_metrics.

Authored-by: Liang-Chi Hsieh <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: viirya

sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLMetrics.scala

@@ -55,8 +55,10 @@ class SQLMetric(val metricType: String, initValue: Long = 0L) extends Accumulato
 
   override def merge(other: AccumulatorV2[Long, Long]): Unit = other match {
     case o: SQLMetric =>
-      if (_value < 0) _value = 0
-      if (o.value > 0) _value += o.value
+      if (o.value > 0) {
+        if (_value < 0) _value = 0
+        _value += o.value
+      }
     case _ => throw QueryExecutionErrors.cannotMergeClassWithOtherClassError(
       this.getClass.getName, other.getClass.getName)
   }

sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala

@@ -1929,8 +1929,15 @@ class AdaptiveQueryExecSuite
         assert(aqeReads.length == 2)
         aqeReads.foreach { c =>
           val stats = c.child.asInstanceOf[QueryStageExec].getRuntimeStatistics
-          assert(stats.sizeInBytes >= 0)
+          val rowCount = stats.rowCount.get
           assert(stats.rowCount.get >= 0)
+          if (rowCount == 0) {
+            // For empty relation, the query stage doesn't serialize any bytes.
+            // The SQLMetric keeps initial value.
+            assert(stats.sizeInBytes == -1)
+          } else {
+            assert(stats.sizeInBytes > 0)
+          }
         }
       }
     }

sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala

@@ -774,7 +774,11 @@ class SQLMetricsSuite extends SharedSparkSession with SQLMetricsTestUtils
 
     testMetricsInSparkPlanOperator(exchanges.head,
       Map("dataSize" -> 3200, "shuffleRecordsWritten" -> 100))
-    testMetricsInSparkPlanOperator(exchanges(1), Map("dataSize" -> 0, "shuffleRecordsWritten" -> 0))
+    // `testData2.filter($"b" === 0)` is an empty relation.
+    // The exchange doesn't serialize any bytes.
+    // The SQLMetric keeps initial value.
+    testMetricsInSparkPlanOperator(exchanges(1),
+      Map("dataSize" -> -1, "shuffleRecordsWritten" -> 0))
   }
 
   test("Add numRows to metric of BroadcastExchangeExec") {