chore: extract comparison tool from fuzzer

Author: comphead

dev/benchmarks/tpcbench.py

@@ -111,12 +111,12 @@ def main(benchmark: str, data_path: str, query_path: str, iterations: int, outpu
                             # coming across for running DDL stmt
                             if len(df.columns) > 0:
                                 output_path = f"{write_path}/q{query}"
-                                # sort by all columns to have predictable output dataset for comparison
-                                df_sorted = df.orderBy(*df.columns)
                                 # rename same column names for output
-                                # output doesn't allow non unique column names
                                 # a, a, b, b => a, a_1, b, b_1
-                                dedup_columns(df_sorted).coalesce(1).write.mode("overwrite").parquet(output_path)
+                                # output doesn't allow non unique column names
+                                deduped = dedup_columns(df)
+                                # sort by all columns to have predictable output dataset for comparison
+                                deduped.orderBy(*deduped.columns).coalesce(1).write.mode("overwrite").parquet(output_path)
                                 print(f"Query {query} results written to {output_path}")
                             else:
                                 print(f"Skipping write: DataFrame has no schema for {output_path}")

fuzz-testing/README.md

@@ -113,7 +113,7 @@ The example below is for TPC-H queries results generated by pure Spark and Comet
 ```shell
 $SPARK_HOME/bin/spark-submit \    
     --master $SPARK_MASTER \
-    --class org.apache.comet.fuzz.ComparisonToolMain     
+    --class org.apache.comet.fuzz.ComparisonTool     
     target/comet-fuzz-spark3.5_2.12-0.12.0-SNAPSHOT-jar-with-dependencies.jar \
     compareParquet --input-spark-folder=/tmp/tpch/spark --input-comet-folder=/tmp/tpch/comet
 ```

fuzz-testing/src/main/scala/org/apache/comet/fuzz/ComparisonTool.scala

@@ -23,7 +23,7 @@ import java.io.File
 
 import org.rogach.scallop.{ScallopConf, ScallopOption, Subcommand}
 
-import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.{functions, SparkSession}
 
 class ComparisonToolConf(arguments: Seq[String]) extends ScallopConf(arguments) {
   object compareParquet extends Subcommand("compareParquet") {
@@ -36,7 +36,7 @@ class ComparisonToolConf(arguments: Seq[String]) extends ScallopConf(arguments)
   verify()
 }
 
-object ComparisonToolMain {
+object ComparisonTool {
 
   lazy val spark: SparkSession = SparkSession
     .builder()
@@ -108,22 +108,14 @@ object ComparisonToolMain {
             // Read Spark parquet files
             spark.conf.set("spark.comet.enabled", "false")
             val sparkDf = spark.read.parquet(sparkSubfolderPath.getAbsolutePath)
-            val sparkRows = sparkDf.collect()
-            val sparkPlan = sparkDf.queryExecution.executedPlan.toString
+            val sparkRows = sparkDf.orderBy(sparkDf.columns.map(functions.col): _*).collect()
 
             // Read Comet parquet files
             val cometDf = spark.read.parquet(cometSubfolderPath.getAbsolutePath)
-            val cometRows = cometDf.collect()
-            val cometPlan = cometDf.queryExecution.executedPlan.toString
+            val cometRows = cometDf.orderBy(cometDf.columns.map(functions.col): _*).collect()
 
             // Compare the results
-            QueryComparison.assertSameRows(
-              sparkRows,
-              cometRows,
-              sqlText = s"Reading parquet from subfolder: $subfolderName",
-              sparkPlan,
-              cometPlan,
-              output)
+            QueryComparison.assertSameRows(sparkRows, cometRows, output)
 
             output.write(s"Subfolder $subfolderName: ${sparkRows.length} rows matched\n\n")
 

fuzz-testing/src/main/scala/org/apache/comet/fuzz/QueryRunner.scala

@@ -26,6 +26,8 @@ import scala.io.Source
 
 import org.apache.spark.sql.{Row, SparkSession}
 
+import org.apache.comet.fuzz.QueryComparison.showPlans
+
 object QueryRunner {
 
   def createOutputMdFile(): BufferedWriter = {
@@ -80,17 +82,21 @@ object QueryRunner {
               val cometRows = df.collect()
               val cometPlan = df.queryExecution.executedPlan.toString
 
-              val success = QueryComparison.assertSameRows(
-                sparkRows,
-                cometRows,
-                sqlText = sql,
-                sparkPlan,
-                cometPlan,
-                output = w)
+              var success = QueryComparison.assertSameRows(sparkRows, cometRows, output = w)
+
+              // check that the plan contains Comet operators
+              if (!cometPlan.contains("Comet")) {
+                success = false
+                w.write("[ERROR] Comet did not accelerate any part of the plan\n")
+              }
+
+              QueryComparison.showSQL(w, sql)
 
               if (success) {
                 cometSuccessCount += 1
               } else {
+                // show plans for failed queries
+                showPlans(w, sparkPlan, cometPlan)
                 cometFailureCount += 1
               }
 
@@ -142,9 +148,6 @@ object QueryComparison {
   def assertSameRows(
       sparkRows: Array[Row],
       cometRows: Array[Row],
-      sqlText: String,
-      sparkPlan: String,
-      cometPlan: String,
       output: BufferedWriter): Boolean = {
     var success = true
     if (sparkRows.length == cometRows.length) {
@@ -160,8 +163,6 @@ object QueryComparison {
         for (j <- 0 until l.length) {
           if (!same(l(j), r(j))) {
             success = false
-            showSQL(output, sqlText)
-            showPlans(output, sparkPlan, cometPlan)
             output.write(s"First difference at row $i:\n")
             output.write("Spark: `" + formatRow(l) + "`\n")
             output.write("Comet: `" + formatRow(r) + "`\n")
@@ -172,23 +173,12 @@ object QueryComparison {
       }
     } else {
       success = false
-      showSQL(output, sqlText)
-      showPlans(output, sparkPlan, cometPlan)
       output.write(
         s"[ERROR] Spark produced ${sparkRows.length} rows and " +
           s"Comet produced ${cometRows.length} rows.\n")
     }
 
-    // check that the plan contains Comet operators
-    if (!cometPlan.contains("Comet")) {
-      success = false
-      showSQL(output, sqlText)
-      showPlans(output, sparkPlan, cometPlan)
-      output.write("[ERROR] Comet did not accelerate any part of the plan\n")
-    }
-
     success
-
   }
 
   private def same(l: Any, r: Any): Boolean = {
@@ -252,7 +242,7 @@ object QueryComparison {
     w.write("```\n")
   }
 
-  private def showPlans(w: BufferedWriter, sparkPlan: String, cometPlan: String): Unit = {
+  def showPlans(w: BufferedWriter, sparkPlan: String, cometPlan: String): Unit = {
     w.write("### Spark Plan\n")
     w.write(s"```\n$sparkPlan\n```\n")
     w.write("### Comet Plan\n")