[SPARK-30409][SPARK-29173][SQL][TESTS] Use NoOp datasource in SQL benchmarks

dev/run-benchmarks.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+
+#
+# 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.
+#
+
+import os
+from sparktestsupport.shellutils import run_cmd
+
+benchmarks = [
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.AggregateBenchmark'],
+    ['avro/test', 'org.apache.spark.sql.execution.benchmark.AvroReadBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.BloomFilterBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.DataSourceReadBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.DateTimeBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.ExtractBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.FilterPushdownBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.InExpressionBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.IntervalBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.JoinBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.MakeDateTimeBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.MiscBenchmark'],
+    ['hive/test', 'org.apache.spark.sql.execution.benchmark.ObjectHashAggregateExecBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.OrcNestedSchemaPruningBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.OrcV2NestedSchemaPruningBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.ParquetNestedSchemaPruningBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.RangeBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.TPCDSQueryBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.UDFBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.WideSchemaBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.benchmark.WideTableBenchmark'],
+    ['hive/test', 'org.apache.spark.sql.hive.orc.OrcReadBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.datasources.csv.CSVBenchmark'],
+    ['sql/test', 'org.apache.spark.sql.execution.datasources.json.JsonBenchmark']
+]
+
+print('Set SPARK_GENERATE_BENCHMARK_FILES=1')
+os.environ['SPARK_GENERATE_BENCHMARK_FILES'] = '1'
+
+for b in benchmarks:
+    print("Run benchmark: %s" % b[1])
+    run_cmd(['build/sbt', '%s:runMain %s' % (b[0], b[1])])

external/avro/src/test/scala/org/apache/spark/sql/execution/benchmark/AvroReadBenchmark.scala

@@ -64,7 +64,7 @@ object AvroReadBenchmark extends SqlBasedBenchmark {
         prepareTable(dir, spark.sql(s"SELECT CAST(value as ${dataType.sql}) id FROM t1"))
 
         benchmark.addCase("Sum") { _ =>
-          spark.sql("SELECT sum(id) FROM avroTable").collect()
+          spark.sql("SELECT sum(id) FROM avroTable").noop()
         }
 
         benchmark.run()
@@ -85,7 +85,7 @@ object AvroReadBenchmark extends SqlBasedBenchmark {
           spark.sql("SELECT CAST(value AS INT) AS c1, CAST(value as STRING) AS c2 FROM t1"))
 
         benchmark.addCase("Sum of columns") { _ =>
-          spark.sql("SELECT sum(c1), sum(length(c2)) FROM avroTable").collect()
+          spark.sql("SELECT sum(c1), sum(length(c2)) FROM avroTable").noop()
         }
 
         benchmark.run()
@@ -104,15 +104,15 @@ object AvroReadBenchmark extends SqlBasedBenchmark {
         prepareTable(dir, spark.sql("SELECT value % 2 AS p, value AS id FROM t1"), Some("p"))
 
         benchmark.addCase("Data column") { _ =>
-          spark.sql("SELECT sum(id) FROM avroTable").collect()
+          spark.sql("SELECT sum(id) FROM avroTable").noop()
         }
 
         benchmark.addCase("Partition column") { _ =>
-          spark.sql("SELECT sum(p) FROM avroTable").collect()
+          spark.sql("SELECT sum(p) FROM avroTable").noop()
         }
 
         benchmark.addCase("Both columns") { _ =>
-          spark.sql("SELECT sum(p), sum(id) FROM avroTable").collect()
+          spark.sql("SELECT sum(p), sum(id) FROM avroTable").noop()
         }
 
         benchmark.run()
@@ -130,7 +130,7 @@ object AvroReadBenchmark extends SqlBasedBenchmark {
         prepareTable(dir, spark.sql("SELECT CAST((id % 200) + 10000 as STRING) AS c1 FROM t1"))
 
         benchmark.addCase("Sum of string length") { _ =>
-          spark.sql("SELECT sum(length(c1)) FROM avroTable").collect()
+          spark.sql("SELECT sum(length(c1)) FROM avroTable").noop()
         }
 
         benchmark.run()
@@ -155,7 +155,7 @@ object AvroReadBenchmark extends SqlBasedBenchmark {
 
         benchmark.addCase("Sum of string length") { _ =>
           spark.sql("SELECT SUM(LENGTH(c2)) FROM avroTable " +
-            "WHERE c1 IS NOT NULL AND c2 IS NOT NULL").collect()
+            "WHERE c1 IS NOT NULL AND c2 IS NOT NULL").noop()
         }
 
         benchmark.run()
@@ -178,7 +178,7 @@ object AvroReadBenchmark extends SqlBasedBenchmark {
         prepareTable(dir, spark.sql("SELECT * FROM t1"))
 
         benchmark.addCase("Sum of single column") { _ =>
-          spark.sql(s"SELECT sum(c$middle) FROM avroTable").collect()
+          spark.sql(s"SELECT sum(c$middle) FROM avroTable").noop()
         }
 
         benchmark.run()

sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/AggregateBenchmark.scala

@@ -48,19 +48,19 @@ object AggregateBenchmark extends SqlBasedBenchmark {
     runBenchmark("aggregate without grouping") {
       val N = 500L << 22
       codegenBenchmark("agg w/o group", N) {
-        spark.range(N).selectExpr("sum(id)").collect()
+        spark.range(N).selectExpr("sum(id)").noop()
       }
     }
 
     runBenchmark("stat functions") {
       val N = 100L << 20
 
       codegenBenchmark("stddev", N) {
-        spark.range(N).groupBy().agg("id" -> "stddev").collect()
+        spark.range(N).groupBy().agg("id" -> "stddev").noop()
       }
 
       codegenBenchmark("kurtosis", N) {
-        spark.range(N).groupBy().agg("id" -> "kurtosis").collect()
+        spark.range(N).groupBy().agg("id" -> "kurtosis").noop()
       }
     }
 
@@ -70,7 +70,7 @@ object AggregateBenchmark extends SqlBasedBenchmark {
       val benchmark = new Benchmark("Aggregate w keys", N, output = output)
 
       def f(): Unit = {
-        spark.range(N).selectExpr("(id & 65535) as k").groupBy("k").sum().collect()
+        spark.range(N).selectExpr("(id & 65535) as k").groupBy("k").sum().noop()
       }
 
       benchmark.addCase("codegen = F", numIters = 2) { _ =>
@@ -107,7 +107,7 @@ object AggregateBenchmark extends SqlBasedBenchmark {
       spark.range(N).selectExpr("id", "floor(rand() * 10000) as k")
         .createOrReplaceTempView("test")
 
-      def f(): Unit = spark.sql("select k, k, sum(id) from test group by k, k").collect()
+      def f(): Unit = spark.sql("select k, k, sum(id) from test group by k, k").noop()
 
       benchmark.addCase("codegen = F", numIters = 2) { _ =>
         withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false") {
@@ -142,7 +142,7 @@ object AggregateBenchmark extends SqlBasedBenchmark {
       val benchmark = new Benchmark("Aggregate w string key", N, output = output)
 
       def f(): Unit = spark.range(N).selectExpr("id", "cast(id & 1023 as string) as k")
-        .groupBy("k").count().collect()
+        .groupBy("k").count().noop()
 
       benchmark.addCase("codegen = F", numIters = 2) { _ =>
         withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false") {
@@ -177,7 +177,7 @@ object AggregateBenchmark extends SqlBasedBenchmark {
       val benchmark = new Benchmark("Aggregate w decimal key", N, output = output)
 
       def f(): Unit = spark.range(N).selectExpr("id", "cast(id & 65535 as decimal) as k")
-        .groupBy("k").count().collect()
+        .groupBy("k").count().noop()
 
       benchmark.addCase("codegen = F") { _ =>
         withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false") {
@@ -222,7 +222,7 @@ object AggregateBenchmark extends SqlBasedBenchmark {
           "id > 1023 as k6")
         .groupBy("k1", "k2", "k3", "k4", "k5", "k6")
         .sum()
-        .collect()
+        .noop()
 
       benchmark.addCase("codegen = F") { _ =>
         withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false") {
@@ -282,7 +282,7 @@ object AggregateBenchmark extends SqlBasedBenchmark {
           "case when id > 1800 and id <= 1900 then 1 else 0 end as v18")
         .groupBy("k1", "k2", "k3")
         .sum()
-        .collect()
+        .noop()
 
       benchmark.addCase("codegen = F") { _ =>
         withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false") {
@@ -315,7 +315,7 @@ object AggregateBenchmark extends SqlBasedBenchmark {
 
       codegenBenchmark("cube", N) {
         spark.range(N).selectExpr("id", "id % 1000 as k1", "id & 256 as k2")
-          .cube("k1", "k2").sum("id").collect()
+          .cube("k1", "k2").sum("id").noop()
       }
     }
 

sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/BloomFilterBenchmark.scala

@@ -70,10 +70,10 @@ object BloomFilterBenchmark extends SqlBasedBenchmark {
       runBenchmark(s"ORC Read") {
         val benchmark = new Benchmark(s"Read a row from ${scaleFactor}M rows", N, output = output)
         benchmark.addCase("Without bloom filter") { _ =>
-          spark.read.orc(path + "/withoutBF").where("value = 0").count
+          spark.read.orc(path + "/withoutBF").where("value = 0").noop()
         }
         benchmark.addCase("With bloom filter") { _ =>
-          spark.read.orc(path + "/withBF").where("value = 0").count
+          spark.read.orc(path + "/withBF").where("value = 0").noop()
         }
         benchmark.run()
       }