[FEA] Skip Databricks Photon jobs at app level in Qualification tool

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

@@ -149,8 +149,12 @@ class Qualification(outputPath: String, numRows: Int, hadoopConf: Configuration,
       val appResult = QualificationAppInfo.createApp(path, hadoopConf, pluginTypeChecker,
         reportSqlLevel, mlOpsEnabled, penalizeTransitions)
       val qualAppResult = appResult match {
-        case Left(errorMessage: String) =>
-          // Case when an error occurred during QualificationAppInfo creation
+        case Left(FailureApp("skipped", errorMessage)) =>
+          // Case when encountered DataBricks Photon event log during QualificationAppInfo creation
+          progressBar.foreach(_.reportSkippedProcess())
+          SkippedQualAppResult(pathStr, errorMessage)
+        case Left(FailureApp(_, errorMessage)) =>
+          // Case when other error occurred during QualificationAppInfo creation
           progressBar.foreach(_.reportUnkownStatusProcess())
           UnknownQualAppResult(pathStr, "", errorMessage)
         case Right(app: QualificationAppInfo) =>
@@ -213,6 +217,8 @@ class Qualification(outputPath: String, numRows: Int, hadoopConf: Configuration,
         StatusSummaryInfo(path, "SUCCESS", appId, message)
       case FailureQualAppResult(path, message) =>
         StatusSummaryInfo(path, "FAILURE", "", message)
+      case SkippedQualAppResult(path, message) =>
+        StatusSummaryInfo(path, "SKIPPED", "", message)
       case UnknownQualAppResult(path, appId, message) =>
         StatusSummaryInfo(path, "UNKNOWN", appId, message)
       case qualAppResult: QualAppResult =>

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

@@ -478,6 +478,8 @@ object SupportedMLFuncsName {
 
 case class GpuEventLogException(message: String) extends Exception(message)
 
+case class PhotonEventLogException(message: String) extends Exception(message)
+
 // Class used a container to hold the information of the Tuple<sqlID, PlanInfo, SparkGraph>
 // to simplify arguments of methods and caching.
 case class SqlPlanInfoGraphEntry(

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

@@ -28,7 +28,7 @@ import org.apache.hadoop.conf.Configuration
 import org.apache.spark.internal.Logging
 import org.apache.spark.scheduler.{SparkListener, SparkListenerEnvironmentUpdate, SparkListenerEvent, SparkListenerJobStart}
 import org.apache.spark.sql.execution.SparkPlanInfo
-import org.apache.spark.sql.rapids.tool.{AppBase, ClusterSummary, GpuEventLogException, SqlPlanInfoGraphBuffer, SupportedMLFuncsName, ToolUtils}
+import org.apache.spark.sql.rapids.tool.{AppBase, ClusterSummary, GpuEventLogException, PhotonEventLogException, SqlPlanInfoGraphBuffer, SupportedMLFuncsName, ToolUtils}
 
 
 class QualificationAppInfo(
@@ -795,6 +795,10 @@ class QualificationAppInfo(
     val sqlPlanInfoGraphEntry = SqlPlanInfoGraphBuffer.createEntry(sqlID, planInfo)
     checkMetadataForReadSchema(sqlPlanInfoGraphEntry)
     for (node <- sqlPlanInfoGraphEntry.sparkPlanGraph.allNodes) {
+      if (node.name.contains("Photon")) {
+        throw PhotonEventLogException(
+          "Encountered Databricks Photon event log: skipping this file!")
+      }
       checkGraphNodeForReads(sqlID, node)
       val issues = findPotentialIssues(node.desc)
       if (issues.nonEmpty) {
@@ -949,6 +953,12 @@ case class StageQualSummaryInfo(
     stageWallclockDuration: Long = 0,
     unsupportedExecs: Seq[ExecInfo] = Seq.empty)
 
+// Case class to represent a failed QualificationAppInfo creation
+case class FailureApp(
+    status: String,
+    message: String
+)
+
 object QualificationAppInfo extends Logging {
   // define recommendation constants
   val RECOMMENDED = "Recommended"
@@ -963,20 +973,22 @@ object QualificationAppInfo extends Logging {
   // based on the testing on few candidate eventlogs.
   val CPU_GPU_TRANSFER_RATE = 1000000000L
 
-  private def handleException(e: Exception, path: EventLogInfo): String = {
-    val message: String = e match {
+  private def handleException(e: Exception, path: EventLogInfo): FailureApp = {
+    val (status, message): (String, String) = e match {
+      case photonLog: PhotonEventLogException =>
+        ("skipped", photonLog.message)
       case gpuLog: GpuEventLogException =>
-        gpuLog.message
+        ("unknow", gpuLog.message)
       case _: com.fasterxml.jackson.core.JsonParseException =>
-        s"Error parsing JSON: ${path.eventLog.toString}"
+        ("unknown", s"Error parsing JSON: ${path.eventLog.toString}")
       case _: IllegalArgumentException =>
-        s"Error parsing file: ${path.eventLog.toString}"
+        ("unknown", s"Error parsing file: ${path.eventLog.toString}")
       case _: Exception =>
         // catch all exceptions and skip that file
-        s"Got unexpected exception processing file: ${path.eventLog.toString}"
+        ("unknown", s"Got unexpected exception processing file: ${path.eventLog.toString}")
     }
 
-    s"${e.getClass.getSimpleName}: $message"
+    FailureApp(status, s"${e.getClass.getSimpleName}: $message")
   }
 
   def getRecommendation(totalSpeedup: Double,
@@ -1055,15 +1067,15 @@ object QualificationAppInfo extends Logging {
       pluginTypeChecker: PluginTypeChecker,
       reportSqlLevel: Boolean,
       mlOpsEnabled: Boolean,
-      penalizeTransitions: Boolean): Either[String, QualificationAppInfo] = {
+      penalizeTransitions: Boolean): Either[FailureApp, QualificationAppInfo] = {
     try {
-        val app = new QualificationAppInfo(Some(path), Some(hadoopConf), pluginTypeChecker,
-          reportSqlLevel, false, mlOpsEnabled, penalizeTransitions)
-        logInfo(s"${path.eventLog.toString} has App: ${app.appId}")
-        Right(app)
-      } catch {
-        case e: Exception =>
-          Left(handleException(e, path))
-      }
+      val app = new QualificationAppInfo(Some(path), Some(hadoopConf), pluginTypeChecker,
+        reportSqlLevel, false, mlOpsEnabled, penalizeTransitions)
+      logInfo(s"${path.eventLog.toString} has App: ${app.appId}")
+      Right(app)
+    } catch {
+      case e: Exception =>
+        Left(handleException(e, path))
+    }
   }
 }

core/src/main/scala/org/apache/spark/sql/rapids/tool/ui/ConsoleProgressBar.scala

@@ -27,14 +27,14 @@ import org.apache.spark.internal.Logging
  * ConsoleProgressBar shows the progress of the tool (Qualification/Profiler) in the console.
  * This class is inspired from org.apache.spark.ui.ConsoleProgressBar.
  *
- * The implementation defines three counters: succeeded, failed, and N/A.
+ * The implementation defines four counters: succeeded, failed, skipped, and N/A.
  * The list can be extended to add new counters during runtime which gives flexibility to have
  * custom statistics for different contexts.
  *
  * By default, the progress bar will be shown in the stdout.
  * Sample generated line:
- *   (.)+ Progress (\\d+)% [==> ] ((\\d+) succeeded + (\\d+) failed + (\\d+) N/A) / [(\\d+)]
- *   toolName Progress 71% [=======>   ] (83 succeeded + 2 failed + 0 N/A) / 119
+ *   (.)+ Progress (\\d+)% [==> ] ((\\d+) succeeded + (\\d+) failed + (\\d+) skipped + (\\d+) N/A) / [(\\d+)]
+ *   toolName Progress 71% [=======>   ] (81 succeeded + 3 failed + 1 skipped + 0 N/A) / 119
  *
  * At the end of execution, it dumps all the defined the counters.
  *
@@ -50,12 +50,14 @@ class ConsoleProgressBar(
   import ConsoleProgressBar._
   private val successCounter = new AtomicLong(0)
   private val failureCounter = new AtomicLong(0)
+  private val skippedCounter = new AtomicLong(0)
   private val statusNotAvailableCounter = new AtomicLong(0)
   private val totalCounter = new AtomicLong(0)
 
   private val metrics = mutable.LinkedHashMap[String, AtomicLong](
     PROCESS_SUCCESS_COUNT -> successCounter,
     PROCESS_FAILURE_COUNT -> failureCounter,
+    PROCESS_SKIPPED_COUNT -> skippedCounter,
     PROCESS_NOT_AVAILABLE_COUNT -> statusNotAvailableCounter,
     EXECUTION_TOTAL_COUNTER -> totalCounter)
 
@@ -127,6 +129,11 @@ class ConsoleProgressBar(
     totalCounter.incrementAndGet()
   }
 
+  def reportSkippedProcess(): Unit = {
+    skippedCounter.incrementAndGet()
+    totalCounter.incrementAndGet()
+  }
+
   def reportUnkownStatusProcess(): Unit = {
     statusNotAvailableCounter.incrementAndGet()
     totalCounter.incrementAndGet()
@@ -140,6 +147,10 @@ class ConsoleProgressBar(
     (1 to n).foreach(_ => reportFailedProcess())
   }
 
+  def reportSkippedProcesses(n: Int): Unit = {
+    (1 to n).foreach(_ => reportSkippedProcess())
+  }
+
   def reportUnknownStatusProcesses(n: Int): Unit = {
     (1 to n).foreach(_ => reportUnkownStatusProcess())
   }
@@ -158,6 +169,7 @@ class ConsoleProgressBar(
     val tailer =
       s"] (${successCounter.longValue()} succeeded + " +
         s"${failureCounter.longValue()} failed + " +
+        s"${skippedCounter.longValue()} skipped + " +
         s"${statusNotAvailableCounter.longValue()} N/A) / $totalCount"
     val w = TerminalWidth - header.length - tailer.length
     val bar = if (w > 0) {
@@ -231,6 +243,7 @@ object ConsoleProgressBar {
 
   val PROCESS_SUCCESS_COUNT = "process.success.count"
   val PROCESS_FAILURE_COUNT = "process.failure.count"
+  val PROCESS_SKIPPED_COUNT = "process.skipped.count"
   val PROCESS_NOT_AVAILABLE_COUNT = "process.NA.count"
   val EXECUTION_TOTAL_COUNTER = "execution.total.count"
 }

core/src/main/scala/org/apache/spark/sql/rapids/tool/util/OperationResult.scala

@@ -49,3 +49,6 @@ case class FailureQualAppResult(path: String, message: String)
 
 case class UnknownQualAppResult(path: String, appId: String, message: String)
   extends QualAppResult(path, message) {}
+
+case class SkippedQualAppResult(path: String, message: String)
+  extends QualAppResult(path, message) {}