Qualification tool: Add penalty for row conversions

core/src/main/scala/com/nvidia/spark/rapids/tool/qualification/QualOutputWriter.scala

@@ -391,6 +391,7 @@ object QualOutputWriter {
   val ESTIMATED_GPU_SPEEDUP = "Estimated GPU Speedup"
   val ESTIMATED_GPU_TIMESAVED = "Estimated GPU Time Saved"
   val STAGE_ESTIMATED_STR = "Stage Estimated"
+  val NUM_TRANSITIONS = "Number of CPU-GPU Transitions"
   val UNSUPPORTED_EXECS = "Unsupported Execs"
   val UNSUPPORTED_EXPRS = "Unsupported Expressions"
   val CLUSTER_TAGS = "Cluster Tags"
@@ -856,7 +857,8 @@ object QualOutputWriter {
       AVERAGE_SPEEDUP_STR -> AVERAGE_SPEEDUP_STR.size,
       STAGE_DUR_STR -> STAGE_DUR_STR.size,
       UNSUPPORTED_TASK_DURATION_STR -> UNSUPPORTED_TASK_DURATION_STR.size,
-      STAGE_ESTIMATED_STR -> STAGE_ESTIMATED_STR.size
+      STAGE_ESTIMATED_STR -> STAGE_ESTIMATED_STR.size,
+      NUM_TRANSITIONS -> NUM_TRANSITIONS.size
     )
     detailedHeadersAndFields
   }
@@ -878,7 +880,8 @@ object QualOutputWriter {
           headersAndSizes(AVERAGE_SPEEDUP_STR),
         info.stageTaskTime.toString -> headersAndSizes(STAGE_DUR_STR),
         info.unsupportedTaskDur.toString -> headersAndSizes(UNSUPPORTED_TASK_DURATION_STR),
-        info.estimated.toString -> headersAndSizes(STAGE_ESTIMATED_STR))
+        info.estimated.toString -> headersAndSizes(STAGE_ESTIMATED_STR),
+        info.numTransitions.toString -> headersAndSizes(NUM_TRANSITIONS))
       constructOutputRow(data, delimiter, prettyPrint)
     }
   }

core/src/main/scala/com/nvidia/spark/rapids/tool/qualification/Qualification.scala

@@ -34,7 +34,8 @@ class Qualification(outputDir: String, numRows: Int, hadoopConf: Configuration,
     timeout: Option[Long], nThreads: Int, order: String,
     pluginTypeChecker: PluginTypeChecker, reportReadSchema: Boolean,
     printStdout: Boolean, uiEnabled: Boolean, enablePB: Boolean,
-    reportSqlLevel: Boolean, maxSQLDescLength: Int, mlOpsEnabled:Boolean) extends Logging {
+    reportSqlLevel: Boolean, maxSQLDescLength: Int, mlOpsEnabled:Boolean,
+    ignoreTransitions: Boolean) extends Logging {
 
   private val allApps = new ConcurrentLinkedQueue[QualificationSummaryInfo]()
 
@@ -166,7 +167,7 @@ class Qualification(outputDir: String, numRows: Int, hadoopConf: Configuration,
     try {
       val startTime = System.currentTimeMillis()
       val appResult = QualificationAppInfo.createApp(path, hadoopConf, pluginTypeChecker,
-        reportSqlLevel, mlOpsEnabled)
+        reportSqlLevel, mlOpsEnabled, ignoreTransitions)
       val qualAppResult = appResult match {
         case Left(errorMessage: String) =>
           // Case when an error occurred during QualificationAppInfo creation

core/src/main/scala/com/nvidia/spark/rapids/tool/qualification/QualificationArgs.scala

@@ -112,6 +112,11 @@ Usage: java -cp rapids-4-spark-tools_2.12-<version>.jar:$SPARK_HOME/jars/*
     opt[Boolean](required = false,
       descr = "Whether to parse ML functions in the eventlogs. Default is false.",
       default = Some(false))
+  val ignoreTransitions: ScallopOption[Boolean] =
+    opt[Boolean](required = false,
+      descr = "Whether to ignore durations for ColumnarToRow and RowToColumnar transitions " +
+        "in the eventlogs while calculating the speedup. Default is false.",
+      default = Some(false))
   val sparkProperty: ScallopOption[List[String]] =
     opt[List[String]](required = false,
       descr = "Filter applications based on certain Spark properties that were set during " +

core/src/main/scala/com/nvidia/spark/rapids/tool/qualification/QualificationMain.scala

@@ -60,6 +60,7 @@ object QualificationMain extends Logging {
     val reportSqlLevel = appArgs.perSql.getOrElse(false)
     val platform = appArgs.platform.getOrElse("onprem")
     val mlOpsEnabled = appArgs.mlFunctions.getOrElse(false)
+    val ignoreTransitions = appArgs.ignoreTransitions.getOrElse(false)
 
     val hadoopConf = RapidsToolsConfUtil.newHadoopConf
 
@@ -93,7 +94,7 @@ object QualificationMain extends Logging {
 
     val qual = new Qualification(outputDirectory, numOutputRows, hadoopConf, timeout,
       nThreads, order, pluginTypeChecker, reportReadSchema, printStdout, uiEnabled,
-      enablePB, reportSqlLevel, maxSQLDescLength, mlOpsEnabled)
+      enablePB, reportSqlLevel, maxSQLDescLength, mlOpsEnabled, ignoreTransitions)
     val res = qual.qualifyApps(filteredLogs)
     (0, res)
   }

core/src/main/scala/org/apache/spark/sql/rapids/tool/qualification/QualificationAppInfo.scala

@@ -16,6 +16,8 @@
 
 package org.apache.spark.sql.rapids.tool.qualification
 
+import java.util.concurrent.TimeUnit
+
 import scala.collection.mutable.{ArrayBuffer, HashMap}
 
 import com.nvidia.spark.rapids.tool.EventLogInfo
@@ -37,7 +39,8 @@ class QualificationAppInfo(
     pluginTypeChecker: PluginTypeChecker,
     reportSqlLevel: Boolean,
     perSqlOnly: Boolean = false,
-    mlOpsEnabled: Boolean = false)
+    mlOpsEnabled: Boolean = false,
+    ignoreTransitions: Boolean = false)
   extends AppBase(eventLogInfo, hadoopConf) with Logging {
 
   var appId: String = ""
@@ -51,6 +54,8 @@ class QualificationAppInfo(
     HashMap.empty[Long, StageTaskQualificationSummary]
   val stageIdToTaskEndSum: HashMap[Long, StageTaskQualificationSummary] =
     HashMap.empty[Long, StageTaskQualificationSummary]
+  val stageIdToGpuCpuTransitions: HashMap[Int, Int] = HashMap.empty[Int, Int]
+  var execsNoStageTransitions: Int = 0
 
   val stageIdToSqlID: HashMap[Int, Long] = HashMap.empty[Int, Long]
   val sqlIDtoFailures: HashMap[Long, ArrayBuffer[String]] = HashMap.empty[Long, ArrayBuffer[String]]
@@ -161,11 +166,11 @@ class QualificationAppInfo(
   }
 
   private def calculateSQLSupportedTaskDuration(all: Seq[StageQualSummaryInfo]): Long = {
-    all.map(s => s.stageTaskTime - s.unsupportedTaskDur).sum
+    all.map(s => s.stageTaskTime - s.unsupportedTaskDur).sum - calculateNoExecsStageDurations(all)
   }
 
   private def calculateSQLUnsupportedTaskDuration(all: Seq[StageQualSummaryInfo]): Long = {
-    all.map(_.unsupportedTaskDur).sum
+    all.map(_.unsupportedTaskDur).sum  + calculateNoExecsStageDurations(all)
   }
 
   private def calculateSpeedupFactor(all: Seq[StageQualSummaryInfo]): Double = {
@@ -174,6 +179,23 @@ class QualificationAppInfo(
     res
   }
 
+  private def calculateNoExecsStageDurations(all: Seq[StageQualSummaryInfo]): Long = {
+    // If there are Execs not associated with any stage, then some of the Execs may not be
+    // supported on GPU.  We need to estimate the duration of these Execs and add it to the
+    // unsupportedTaskDur. We estimate the duration by taking the average of the unsupportedTaskDur
+    // of all the stages and multiplying it by the number of Execs that are not associated with
+    // any stage. We multiply with a penalty factor of 0.05
+    // TODO: Need to come up with better heuristics for penalty factor.
+    val unsupportedTasksize= all.map(_.unsupportedTaskDur).size
+    if (execsNoStageTransitions != 0 && unsupportedTasksize != 0) {
+      execsNoStageTransitions * (
+        all.map(_.unsupportedTaskDur).sum / unsupportedTasksize) * 0.05
+    }.toLong
+    else {
+      0L
+    }
+  }
+
   private def getAllReadFileFormats: Seq[String] = {
     dataSourceInfo.map { ds =>
       s"${ds.format.toLowerCase()}[${ds.schema}]"
@@ -241,13 +263,79 @@ class QualificationAppInfo(
     stages.map { stageId =>
       val stageTaskTime = stageIdToTaskEndSum.get(stageId)
         .map(_.totalTaskDuration).getOrElse(0L)
+      val numTransitions = ignoreTransitions match {
+        case false => stageIdToGpuCpuTransitions.getOrElse(stageId, 0)
+        case true => 0
+      }
+      // val numTransitions = stageIdToGpuCpuTransitions.getOrElse(stageId, 0)
+      val transitionsTime = numTransitions match {
+        case 0 => 0L // no transitions
+        case gpuCpuTransitions if gpuCpuTransitions > 0 =>
+          // Duration to transfer data from GPU to CPU and vice versa.
+          // Assuming it's a PCI-E Gen3, but also assuming that some of the result could be
+          // spilled to disk.
+          // Duration in Spark metrics is in milliseconds and CPU-GPU transfer rate is in bytes/sec.
+          // So we need to convert the transitions time to milliseconds.
+          val totalBytesRead =
+            stageIdToTaskEndSum.get(stageId).map(_.totalbytesRead).getOrElse(0L)
+          if (totalBytesRead > 0) {
+            TimeUnit.SECONDS.toMillis(
+              totalBytesRead / QualificationAppInfo.CPU_GPU_TRANSFER_RATE) * gpuCpuTransitions
+          } else {
+            0L
+          }
+        case _ => 0L
+      }
+      // Update totaltaskduration of stageIdToTaskEndSum to include transitions time
+      val stageIdToTasksMetrics = stageIdToTaskEndSum.get(stageId).orElse(None)
+      if (stageIdToTasksMetrics.isDefined) {
+        stageIdToTasksMetrics.get.totalTaskDuration += transitionsTime
+      }
       StageQualSummaryInfo(stageId, allSpeedupFactorAvg, stageTaskTime,
-        eachStageUnsupported, estimated)
+        eachStageUnsupported + transitionsTime, estimated, numTransitions)
     }.toSet
   }
 
   def summarizeStageLevel(execInfos: Seq[ExecInfo], sqlID: Long): Set[StageQualSummaryInfo] = {
     val (allStagesToExecs, execsNoStage) = getStageToExec(execInfos)
+
+    // Get the total number of transitions between CPU and GPU for each stage and
+    // store it in a Map.
+    allStagesToExecs.foreach { case (stageId, execs) =>
+      // Flatten all the Execs within a stage.
+      // Example: Exchange;WholeStageCodegen (14);Exchange;WholeStageCodegen (13);Exchange
+      // will be flattened to Exchange;Sort;Exchange;Sort;SortMergeJoin;SortMergeJoin;Exchange;
+      val allExecs = execs.map(x => if (x.exec.startsWith("WholeStage")) {
+        x.children.getOrElse(Seq.empty)
+      } else {
+        Seq(x)
+      }).flatten.reverse
+
+      // If it's a shuffle stage, then we need to keep the first and last Exchange and remove
+      // all the intermediate Exchanges as input size is captured in Exchange node.
+      val dedupedExecs = if (allExecs.size > 2) {
+        allExecs.head +:
+          allExecs.tail.init.filter(x => x.exec != "Exchange") :+ allExecs.last
+      } else {
+        allExecs
+      }
+      // Create a list of transitions by zipping allExecs with itself but with the first element
+      // This will create a list of adjacent pairs.
+      // Example: If allExecs = (ScanExec, FilterExec, SortExec, ProjectExec), then it will create
+      // a list of tuples as follows:
+      // (ScanExec, FilterExec), (FilterExec, SortExec), (SortExec, ProjectExec)
+      val transitions = dedupedExecs.zip(dedupedExecs.drop(1)).count {
+        // If the current execution (currExec) is supported, and the next execution (nextExec)
+        // is not supported, or if the current execution is not supported and the next execution
+        // is supported, then we consider this as a transition.
+        case (currExec, nextExec) => (currExec.isSupported && !nextExec.isSupported) ||
+          (!currExec.isSupported && nextExec.isSupported)
+      }
+      stageIdToGpuCpuTransitions(stageId) = transitions
+    }
+    if (execsNoStage.nonEmpty) {
+      execsNoStageTransitions += execsNoStage.filterNot(exec => exec.isSupported).size
+    }
     if (allStagesToExecs.isEmpty) {
       // use job level
       // also get the job ids associated with the SQLId
@@ -670,7 +758,8 @@ class StageTaskQualificationSummary(
     val stageAttemptId: Int,
     var executorRunTime: Long,
     var executorCPUTime: Long,
-    var totalTaskDuration: Long)
+    var totalTaskDuration: Long,
+    var totalbytesRead: Long)
 
 case class QualApplicationInfo(
     appName: String,
@@ -736,7 +825,8 @@ case class StageQualSummaryInfo(
     averageSpeedup: Double,
     stageTaskTime: Long,
     unsupportedTaskDur: Long,
-    estimated: Boolean = false)
+    estimated: Boolean = false,
+    numTransitions: Int)
 
 object QualificationAppInfo extends Logging {
   // define recommendation constants
@@ -746,6 +836,13 @@ object QualificationAppInfo extends Logging {
   val NOT_APPLICABLE = "Not Applicable"
   val LOWER_BOUND_RECOMMENDED = 1.3
   val LOWER_BOUND_STRONGLY_RECOMMENDED = 2.5
+  // Below is the total time taken whenever there are ColumnarToRow or RowToColumnar transitions
+  // This includes the time taken to convert the data from one format to another and the time taken
+  // to transfer the data from CPU to GPU and vice versa. Current transfer rate is 10MB/s and is
+  // based on the testing on few eventlogs.
+  // TODO: Need to test this on more eventlogs including NDS queries
+  //  and come up with a better transfer rate.
+  val CPU_GPU_TRANSFER_RATE = 10000000L
 
   private def handleException(e: Exception, path: EventLogInfo): String = {
     val message: String = e match {
@@ -838,10 +935,11 @@ object QualificationAppInfo extends Logging {
       hadoopConf: Configuration,
       pluginTypeChecker: PluginTypeChecker,
       reportSqlLevel: Boolean,
-      mlOpsEnabled: Boolean): Either[String, QualificationAppInfo] = {
+      mlOpsEnabled: Boolean,
+      ignoreTransitions: Boolean): Either[String, QualificationAppInfo] = {
     try {
         val app = new QualificationAppInfo(Some(path), Some(hadoopConf), pluginTypeChecker,
-          reportSqlLevel, false, mlOpsEnabled)
+          reportSqlLevel, false, mlOpsEnabled, ignoreTransitions)
         logInfo(s"${path.eventLog.toString} has App: ${app.appId}")
         Right(app)
       } catch {

core/src/main/scala/org/apache/spark/sql/rapids/tool/qualification/QualificationEventProcessor.scala

@@ -72,16 +72,26 @@ class QualificationEventProcessor(app: QualificationAppInfo, perSqlOnly: Boolean
     super.doSparkListenerTaskEnd(app, event)
     // keep all stage task times to see for nonsql duration
     val taskSum = app.stageIdToTaskEndSum.getOrElseUpdate(event.stageId, {
-      new StageTaskQualificationSummary(event.stageId, event.stageAttemptId, 0, 0, 0)
+      new StageTaskQualificationSummary(event.stageId, event.stageAttemptId, 0, 0, 0, 0)
     })
     taskSum.executorRunTime += event.taskMetrics.executorRunTime
     taskSum.executorCPUTime += NANOSECONDS.toMillis(event.taskMetrics.executorCpuTime)
     taskSum.totalTaskDuration += event.taskInfo.duration
+    // Add the total bytes read from the task if it's available. This is from inputMetrics if
+    // it is reading from datasource, or shuffleReadMetrics if it is reading from shuffle.
+    val inputMetrics = event.taskMetrics.inputMetrics
+    if (inputMetrics != null) {
+      taskSum.totalbytesRead += inputMetrics.bytesRead
+    }
+    val shuffleReadMetrics = event.taskMetrics.shuffleReadMetrics
+    if (shuffleReadMetrics != null) {
+      taskSum.totalbytesRead += shuffleReadMetrics.totalBytesRead
+    }
 
     // Adds in everything (including failures)
     app.stageIdToSqlID.get(event.stageId).foreach { sqlID =>
       val taskSum = app.sqlIDToTaskEndSum.getOrElseUpdate(sqlID, {
-        new StageTaskQualificationSummary(event.stageId, event.stageAttemptId, 0, 0, 0)
+        new StageTaskQualificationSummary(event.stageId, event.stageAttemptId, 0, 0, 0, 0)
       })
       taskSum.executorRunTime += event.taskMetrics.executorRunTime
       taskSum.executorCPUTime += NANOSECONDS.toMillis(event.taskMetrics.executorCpuTime)

core/src/test/resources/QualificationExpectations/jdbc_expectation.csv

@@ -1,2 +1,2 @@
 App Name,App ID,Recommendation,Estimated GPU Speedup,Estimated GPU Duration,Estimated GPU Time Saved,SQL DF Duration,SQL Dataframe Task Duration,App Duration,GPU Opportunity,Executor CPU Time Percent,SQL Ids with Failures,Unsupported Read File Formats and Types,Unsupported Write Data Format,Complex Types,Nested Complex Types,Potential Problems,Longest SQL Duration,NONSQL Task Duration Plus Overhead,Unsupported Task Duration,Supported SQL DF Task Duration,Task Speedup Factor,App Duration Estimated,Unsupported Execs,Unsupported Expressions,Estimated Job Frequency (monthly)
-"Spark shell","app-20211019113801-0001","Not Recommended",1.0,569385.42,2581.57,3627,19894,571967,3503,28.41,"","JDBC[*]","","","","",1812,544575,677,19217,3.8,false,"Scan JDBCRelation(TBLS) [numPartitions=1];Execute CreateViewCommand;CollectLimit","",30
+"Spark shell","app-20211019113801-0001","Not Recommended",1.0,569387.57,2579.42,3627,19894,571967,3500,28.41,"","JDBC[*]","","","","",1812,544575,693,19201,3.8,false,"Scan JDBCRelation(TBLS) [numPartitions=1];Execute CreateViewCommand;CollectLimit","",30

core/src/test/resources/QualificationExpectations/qual_test_simple_expectation.csv

@@ -1,5 +1,5 @@
 App Name,App ID,Recommendation,Estimated GPU Speedup,Estimated GPU Duration,Estimated GPU Time Saved,SQL DF Duration,SQL Dataframe Task Duration,App Duration,GPU Opportunity,Executor CPU Time Percent,SQL Ids with Failures,Unsupported Read File Formats and Types,Unsupported Write Data Format,Complex Types,Nested Complex Types,Potential Problems,Longest SQL Duration,NONSQL Task Duration Plus Overhead,Unsupported Task Duration,Supported SQL DF Task Duration,Task Speedup Factor,App Duration Estimated,Unsupported Execs,Unsupported Expressions,Estimated Job Frequency (monthly)
-"Rapids Spark Profiling Tool Unit Tests","local-1622043423018","Recommended",1.92,8472.65,7846.34,12434,132257,16319,10589,37.7,"","","JSON","","","",7143,4717,19616,112641,3.86,false,"SerializeFromObject;Execute InsertIntoHadoopFsRelationCommand json;DeserializeToObject;Filter;MapElements;Scan","",1
-"Spark shell","local-1651187225439","Not Recommended",1.0,355483.43,153.56,760,180,355637,350,87.88,"","JSON[string:bigint:int]","","","","",498,343411,97,83,1.78,false,"SerializeFromObject;CollectLimit;DeserializeToObject;Scan json;Filter;MapElements","",1
-"Spark shell","local-1651188809790","Not Recommended",1.0,166199.97,15.02,911,283,166215,45,81.18,"","JSON[string:bigint:int]","","","","UDF",715,133608,269,14,1.5,false,"CollectLimit;Scan json;Project","UDF",1
-"Rapids Spark Profiling Tool Unit Tests","local-1623281204390","Not Recommended",1.0,6240.0,0.0,2032,4666,6240,0,46.27,"","JSON[string:bigint:int]","JSON","","","UDF",1209,5793,4664,2,1.0,false,"Execute InsertIntoHadoopFsRelationCommand json;LocalTableScan;Project;Scan json;Execute CreateViewCommand","UDF",1
+"Rapids Spark Profiling Tool Unit Tests","local-1622043423018","Recommended",1.92,8477.87,7841.12,12434,132257,16319,10582,37.7,"","","JSON","","","",7143,4717,19691,112566,3.86,false,"SerializeFromObject;Execute InsertIntoHadoopFsRelationCommand json;DeserializeToObject;Filter;MapElements;Scan","",1
+"Spark shell","local-1651187225439","Not Recommended",1.0,355490.83,146.16,760,180,355637,333,87.88,"","JSON[string:bigint:int]","","","","",498,343411,101,79,1.78,false,"SerializeFromObject;CollectLimit;DeserializeToObject;Scan json;Filter;MapElements","",1
+"Spark shell","local-1651188809790","Not Recommended",1.0,166213.92,1.07,911,283,166215,3,81.18,"","JSON[string:bigint:int]","","","","UDF",715,133608,282,1,1.5,false,"CollectLimit;Scan json;Project","UDF",1
+"Rapids Spark Profiling Tool Unit Tests","local-1623281204390","Not Recommended",1.0,6240.0,0.0,2032,4666,6240,-151,46.27,"","JSON[string:bigint:int]","JSON","","","UDF",1209,5793,5013,-347,1.0,false,"Execute InsertIntoHadoopFsRelationCommand json;LocalTableScan;Project;Scan json;Execute CreateViewCommand","UDF",1