Qualification tool support for read and write execs and more, add mapping stage times to sql execs

tools/src/main/scala/com/nvidia/spark/rapids/tool/planparser/BatchScanExecParser.scala

@@ -0,0 +1,46 @@
+/*
+ * Copyright (c) 2022, NVIDIA CORPORATION.
+ *
+ * Licensed 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.
+ */
+
+package com.nvidia.spark.rapids.tool.planparser
+
+import com.nvidia.spark.rapids.tool.qualification.PluginTypeChecker
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.execution.ui.SparkPlanGraphNode
+import org.apache.spark.sql.rapids.tool.AppBase
+
+case class BatchScanExecParser(
+    node: SparkPlanGraphNode,
+    checker: PluginTypeChecker,
+    sqlID: Long,
+    app: AppBase) extends ExecParser with Logging {
+
+  val fullExecName = "BatchScanExec"
+
+  override def parse: ExecInfo = {
+    val accumId = node.metrics.find(_.name == "scan time").map(_.accumulatorId)
+    val maxDuration = SQLPlanParser.getTotalDuration(accumId, app)
+    val readInfo = ReadParser.parseReadNode(node)
+    // don't use the isExecSupported because we have finer grain.
+    val score = ReadParser.calculateReadScoreRatio(readInfo, checker)
+    val speedupFactor = checker.getSpeedupFactor(fullExecName)
+    val overallSpeedup = Math.max((speedupFactor * score).toInt, 1)
+
+    // TODO - add in parsing expressions - average speedup across?
+    ExecInfo(sqlID, s"${node.name} ${readInfo.format}", "", overallSpeedup,
+      maxDuration, node.id, score > 0, None)
+  }
+}

tools/src/main/scala/com/nvidia/spark/rapids/tool/planparser/BroadcastExchangeExecParser.scala

@@ -0,0 +1,51 @@
+/*
+ * Copyright (c) 2022, NVIDIA CORPORATION.
+ *
+ * Licensed 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.
+ */
+
+package com.nvidia.spark.rapids.tool.planparser
+
+import com.nvidia.spark.rapids.tool.qualification.PluginTypeChecker
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.execution.ui.SparkPlanGraphNode
+import org.apache.spark.sql.rapids.tool.AppBase
+
+case class BroadcastExchangeExecParser(
+    node: SparkPlanGraphNode,
+    checker: PluginTypeChecker,
+    sqlID: Long,
+    app: AppBase) extends ExecParser with Logging {
+
+  val fullExecName = node.name + "Exec"
+
+  override def parse: ExecInfo = {
+    // TODO - check the relation to see if really supported
+    val collectTimeId = node.metrics.find(_.name == "time to collect").map(_.accumulatorId)
+    val buildTimeId = node.metrics.find(_.name == "time to build").map(_.accumulatorId)
+    val broadcastTimeId = node.metrics.find(_.name == "time to broadcast").map(_.accumulatorId)
+    val maxCollectTime = SQLPlanParser.getDriverTotalDuration(collectTimeId, app)
+    val maxBuildTime = SQLPlanParser.getDriverTotalDuration(buildTimeId, app)
+    val maxBroadcastTime = SQLPlanParser.getDriverTotalDuration(broadcastTimeId, app)
+    val duration = (maxCollectTime ++ maxBuildTime ++ maxBroadcastTime).reduceOption(_ + _)
+    val (filterSpeedupFactor, isSupported) = if (checker.isExecSupported(fullExecName)) {
+      (checker.getSpeedupFactor(fullExecName), true)
+    } else {
+      (1, false)
+    }
+    // TODO - add in parsing expressions - average speedup across?
+    ExecInfo(sqlID, node.name, "", filterSpeedupFactor,
+      duration, node.id, isSupported, None)
+  }
+}

tools/src/main/scala/com/nvidia/spark/rapids/tool/planparser/CartesianProductExecParser.scala

@@ -27,7 +27,7 @@ case class CartesianProductExecParser(
 
   val fullExecName = node.name + "Exec"
 
-  override def parse: Seq[ExecInfo] = {
+  override def parse: ExecInfo = {
     // CartesianProduct doesn't have duration
     val duration = None
     val (speedupFactor, isSupported) = if (checker.isExecSupported(fullExecName)) {
@@ -36,7 +36,6 @@ case class CartesianProductExecParser(
       (1, false)
     }
     // TODO - add in parsing expressions - average speedup across?
-    Seq(ExecInfo(sqlID, node.name, "", speedupFactor,
-      duration, node.id, isSupported, None))
+    ExecInfo(sqlID, node.name, "", speedupFactor, duration, node.id, isSupported, None)
   }
 }

tools/src/main/scala/com/nvidia/spark/rapids/tool/planparser/CoalesceExecParser.scala

@@ -27,7 +27,7 @@ case class CoalesceExecParser(
 
   val fullExecName = node.name + "Exec"
 
-  override def parse: Seq[ExecInfo] = {
+  override def parse: ExecInfo = {
     // coalesce doesn't have duration
     val duration = None
     val (speedupFactor, isSupported) = if (checker.isExecSupported(fullExecName)) {
@@ -36,7 +36,7 @@ case class CoalesceExecParser(
       (1, false)
     }
     // TODO - add in parsing expressions - average speedup across?
-    Seq(ExecInfo(sqlID, node.name, "", speedupFactor,
-      duration, node.id, isSupported, None))
+    ExecInfo(sqlID, node.name, "", speedupFactor,
+      duration, node.id, isSupported, None)
   }
 }

tools/src/main/scala/com/nvidia/spark/rapids/tool/planparser/CollectLimitExecParser.scala

@@ -27,7 +27,7 @@ case class CollectLimitExecParser(
 
   val fullExecName = node.name + "Exec"
 
-  override def parse: Seq[ExecInfo] = {
+  override def parse: ExecInfo = {
     // collect doesn't have duration
     val duration = None
     val (speedupFactor, isSupported) = if (checker.isExecSupported(fullExecName)) {
@@ -36,7 +36,7 @@ case class CollectLimitExecParser(
       (1, false)
     }
     // TODO - add in parsing expressions - average speedup across?
-    Seq(ExecInfo(sqlID, node.name, "", speedupFactor,
-      duration, node.id, isSupported, None))
+    ExecInfo(sqlID, node.name, "", speedupFactor,
+      duration, node.id, isSupported, None)
   }
 }

tools/src/main/scala/com/nvidia/spark/rapids/tool/planparser/CustomShuffleReaderExecParser.scala

@@ -0,0 +1,42 @@
+/*
+ * Copyright (c) 2022, NVIDIA CORPORATION.
+ *
+ * Licensed 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.
+ */
+
+package com.nvidia.spark.rapids.tool.planparser
+
+import com.nvidia.spark.rapids.tool.qualification.PluginTypeChecker
+
+import org.apache.spark.sql.execution.ui.SparkPlanGraphNode
+
+case class CustomShuffleReaderExecParser(
+    node: SparkPlanGraphNode,
+    checker: PluginTypeChecker,
+    sqlID: Long) extends ExecParser {
+
+  // note this is called either AQEShuffleRead and CustomShuffleReader depending
+  // on the Spark version, our supported ops list it as CustomShuffleReader
+  val fullExecName = "CustomShuffleReaderExec"
+
+  override def parse: ExecInfo = {
+    // doesn't have duration
+    val duration = None
+    val (speedupFactor, isSupported) = if (checker.isExecSupported(fullExecName)) {
+      (checker.getSpeedupFactor(fullExecName), true)
+    } else {
+      (1, false)
+    }
+    ExecInfo(sqlID, node.name, "", speedupFactor, duration, node.id, isSupported, None)
+  }
+}

tools/src/main/scala/com/nvidia/spark/rapids/tool/planparser/DataWritingCommandExecParser.scala

@@ -0,0 +1,43 @@
+/*
+ * Copyright (c) 2022, NVIDIA CORPORATION.
+ *
+ * Licensed 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.
+ */
+
+package com.nvidia.spark.rapids.tool.planparser
+
+import com.nvidia.spark.rapids.tool.qualification.PluginTypeChecker
+
+import org.apache.spark.sql.execution.ui.SparkPlanGraphNode
+
+case class DataWritingCommandExecParser(
+    node: SparkPlanGraphNode,
+    checker: PluginTypeChecker,
+    sqlID: Long) extends ExecParser {
+
+  // hardcode because InsertIntoHadoopFsRelationCommand uses this same exec
+  // and InsertIntoHadoopFsRelationCommand doesn't have an entry in the
+  // supported execs file
+  val fullExecName = "DataWritingCommandExec"
+
+  override def parse: ExecInfo = {
+    val writeFormat = node.desc.split(",")(2)
+    val writeSupported = checker.isWriteFormatsupported(writeFormat)
+    val duration = None
+    val speedupFactor = checker.getSpeedupFactor(fullExecName)
+    val finalSpeedup = if (writeSupported) speedupFactor else 1
+    // TODO - add in parsing expressions - average speedup across?
+    ExecInfo(sqlID, s"${node.name.trim} ${writeFormat.toLowerCase.trim}", "", finalSpeedup,
+      duration, node.id, writeSupported, None)
+  }
+}

tools/src/main/scala/com/nvidia/spark/rapids/tool/planparser/ExecParser.scala

@@ -17,6 +17,6 @@
 package com.nvidia.spark.rapids.tool.planparser
 
 trait ExecParser {
-  def parse: Seq[ExecInfo]
+  def parse: ExecInfo
   val fullExecName: String
 }

tools/src/main/scala/com/nvidia/spark/rapids/tool/planparser/ExpandExecParser.scala

@@ -27,7 +27,7 @@ case class ExpandExecParser(
 
   val fullExecName = node.name + "Exec"
 
-  override def parse: Seq[ExecInfo] = {
+  override def parse: ExecInfo = {
     // Expand doesn't have duration
     val duration = None
     val (speedupFactor, isSupported) = if (checker.isExecSupported(fullExecName)) {
@@ -36,7 +36,6 @@ case class ExpandExecParser(
       (1, false)
     }
     // TODO - add in parsing expressions - average speedup across?
-    Seq(ExecInfo(sqlID, node.name, "", speedupFactor,
-      duration, node.id, isSupported, None))
+    ExecInfo(sqlID, node.name, "", speedupFactor, duration, node.id, isSupported, None)
   }
 }

tools/src/main/scala/com/nvidia/spark/rapids/tool/planparser/FileSourceScanExecParser.scala

@@ -0,0 +1,47 @@
+/*
+ * Copyright (c) 2022, NVIDIA CORPORATION.
+ *
+ * Licensed 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.
+ */
+
+package com.nvidia.spark.rapids.tool.planparser
+
+import com.nvidia.spark.rapids.tool.qualification.PluginTypeChecker
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.execution.ui.SparkPlanGraphNode
+import org.apache.spark.sql.rapids.tool.AppBase
+
+case class FileSourceScanExecParser(
+    node: SparkPlanGraphNode,
+    checker: PluginTypeChecker,
+    sqlID: Long,
+    app: AppBase) extends ExecParser with Logging {
+
+  // The node name for Scans is Scan <format> so here we hardcode
+  val fullExecName = "FileSourceScanExec"
+
+  override def parse: ExecInfo = {
+    val accumId = node.metrics.find(_.name == "scan time").map(_.accumulatorId)
+    val maxDuration = SQLPlanParser.getTotalDuration(accumId, app)
+
+    val readInfo = ReadParser.parseReadNode(node)
+    // don't use the isExecSupported because we have finer grain.
+    val score = ReadParser.calculateReadScoreRatio(readInfo, checker)
+    val speedupFactor = checker.getSpeedupFactor(fullExecName)
+    val overallSpeedup = Math.max((speedupFactor * score).toInt, 1)
+
+    // TODO - add in parsing expressions - average speedup across?
+    ExecInfo(sqlID, node.name, "", overallSpeedup, maxDuration, node.id, score > 0, None)
+  }
+}

tools/src/main/scala/com/nvidia/spark/rapids/tool/planparser/FilterExecParser.scala

@@ -27,7 +27,7 @@ case class FilterExecParser(
 
   val fullExecName = node.name + "Exec"
 
-  override def parse: Seq[ExecInfo] = {
+  override def parse: ExecInfo = {
     // filter doesn't have duration
     val duration = None
     val (speedupFactor, isSupported) = if (checker.isExecSupported(fullExecName)) {
@@ -36,7 +36,6 @@ case class FilterExecParser(
       (1, false)
     }
     // TODO - add in parsing expressions - average speedup across?
-    Seq(ExecInfo(sqlID, node.name, "", speedupFactor,
-      duration, node.id, isSupported, None))
+    ExecInfo(sqlID, node.name, "", speedupFactor, duration, node.id, isSupported, None)
   }
 }

tools/src/main/scala/com/nvidia/spark/rapids/tool/planparser/GenerateExecParser.scala

@@ -27,7 +27,7 @@ case class GenerateExecParser(
 
   val fullExecName = node.name + "Exec"
 
-  override def parse: Seq[ExecInfo] = {
+  override def parse: ExecInfo = {
     // Generate doesn't have duration
     val duration = None
     val (speedupFactor, isSupported) = if (checker.isExecSupported(fullExecName)) {
@@ -36,7 +36,6 @@ case class GenerateExecParser(
       (1, false)
     }
     // TODO - add in parsing expressions - average speedup across?
-    Seq(ExecInfo(sqlID, node.name, "", speedupFactor,
-      duration, node.id, isSupported, None))
+    ExecInfo(sqlID, node.name, "", speedupFactor, duration, node.id, isSupported, None)
   }
 }

tools/src/main/scala/com/nvidia/spark/rapids/tool/planparser/GlobalLimitExecParser.scala

@@ -27,7 +27,7 @@ case class GlobalLimitExecParser(
 
   val fullExecName = node.name + "Exec"
 
-  override def parse: Seq[ExecInfo] = {
+  override def parse: ExecInfo = {
     // GlobalLimit doesn't have duration
     val duration = None
     val (speedupFactor, isSupported) = if (checker.isExecSupported(fullExecName)) {
@@ -36,7 +36,6 @@ case class GlobalLimitExecParser(
       (1, false)
     }
     // TODO - add in parsing expressions - average speedup across?
-    Seq(ExecInfo(sqlID, node.name, "", speedupFactor,
-      duration, node.id, isSupported, None))
+    ExecInfo(sqlID, node.name, "", speedupFactor, duration, node.id, isSupported, None)
   }
 }