[SPARK-19447] Make Range operator generate "recordsRead" metric

## What changes were proposed in this pull request?

The Range was modified to produce "recordsRead" metric instead of "generated rows". The tests were updated and partially moved to SQLMetricsSuite.

## How was this patch tested?

Unit tests.

Author: Ala Luszczak <ala@databricks.com>

Closes #16960 from ala/range-records-read.

1 files changed, 6 insertions, 12 deletions

diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveSerDeSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveSerDeSuite.scala
index 35c41b531c..7803ac39e5 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveSerDeSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveSerDeSuite.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql.hive.execution
 
 import org.scalatest.BeforeAndAfterAll
 
-import org.apache.spark.sql.execution.MetricsTestHelper
+import org.apache.spark.sql.execution.metric.InputOutputMetricsHelper
 import org.apache.spark.sql.hive.test.TestHive
 
 /**
@@ -49,21 +49,15 @@ class HiveSerDeSuite extends HiveComparisonTest with BeforeAndAfterAll {
 
   createQueryTest("Read Partitioned with AvroSerDe", "SELECT * FROM episodes_part")
 
-  test("Test input/generated/output metrics") {
+  test("Checking metrics correctness") {
     import TestHive._
 
     val episodesCnt = sql("select * from episodes").count()
-    val episodesRes = MetricsTestHelper.runAndGetMetrics(sql("select * from episodes").toDF())
-    assert(episodesRes.recordsRead === episodesCnt :: Nil)
-    assert(episodesRes.shuffleRecordsRead.sum === 0)
-    assert(episodesRes.generatedRows.isEmpty)
-    assert(episodesRes.outputRows === episodesCnt :: Nil)
+    val episodesRes = InputOutputMetricsHelper.run(sql("select * from episodes").toDF())
+    assert(episodesRes === (episodesCnt, 0L, episodesCnt) :: Nil)
 
     val serdeinsCnt = sql("select * from serdeins").count()
-    val serdeinsRes = MetricsTestHelper.runAndGetMetrics(sql("select * from serdeins").toDF())
-    assert(serdeinsRes.recordsRead === serdeinsCnt :: Nil)
-    assert(serdeinsRes.shuffleRecordsRead.sum === 0)
-    assert(serdeinsRes.generatedRows.isEmpty)
-    assert(serdeinsRes.outputRows === serdeinsCnt :: Nil)
+    val serdeinsRes = InputOutputMetricsHelper.run(sql("select * from serdeins").toDF())
+    assert(serdeinsRes === (serdeinsCnt, 0L, serdeinsCnt) :: Nil)
   }
 }