Merge dev into main

Signed-off-by: spark-rapids automation <[email protected]>

core/pom.xml

@@ -23,7 +23,7 @@
     <artifactId>rapids-4-spark-tools_2.12</artifactId>
     <name>RAPIDS Accelerator for Apache Spark tools</name>
     <description>RAPIDS Accelerator for Apache Spark tools</description>
-    <version>24.02.4</version>
+    <version>24.02.5-SNAPSHOT</version>
     <packaging>jar</packaging>
     <url>http://github.com/NVIDIA/spark-rapids-tools</url>
 

core/src/main/scala/com/nvidia/spark/rapids/tool/planparser/DatabricksParseHelper.scala

@@ -39,6 +39,9 @@ object DatabricksParseHelper extends Logging {
   val PROP_WORKER_TYPE_ID_KEY = "spark.databricks.workerNodeTypeId"
   val PROP_DRIVER_TYPE_ID_KEY = "spark.databricks.driverNodeTypeId"
 
+  val SUB_PROP_CLUSTER_ID = "ClusterId"
+  val SUB_PROP_JOB_ID = "JobId"
+  val SUB_PROP_RUN_NAME = "RunName"
   /**
    * Checks if the properties indicate that the application is a Photon app.
    * This ca be checked by looking for keywords in one of the keys defined in PHOTON_SPARK_PROPS
@@ -52,6 +55,10 @@ object DatabricksParseHelper extends Logging {
     }
   }
 
+  def getSparkVersion(properties: collection.Map[String, String]): String = {
+    properties.getOrElse(PROP_TAG_CLUSTER_SPARK_VERSION_KEY, "")
+  }
+
   /**
    * Try to get the JobId from the cluster name. Parse the clusterName string which
    * looks like:

core/src/main/scala/com/nvidia/spark/rapids/tool/planparser/SQLPlanParser.scala

@@ -432,8 +432,8 @@ object SQLPlanParser extends Logging {
   def getStagesInSQLNode(node: SparkPlanGraphNode, app: AppBase): Set[Int] = {
     val nodeAccums = node.metrics.map(_.accumulatorId)
     nodeAccums.flatMap { nodeAccumId =>
-      app.accumulatorToStages.get(nodeAccumId)
-    }.flatten.toSet
+      app.stageManager.getStagesIdsByAccumId(nodeAccumId)
+    }.toSet
   }
 
   // Set containing execs that refers to other expressions. We need this to be a list to allow