diff --git a/core/src/main/scala/com/nvidia/spark/rapids/SparkRapidsBuildInfoEvent.scala b/core/src/main/scala/com/nvidia/spark/rapids/SparkRapidsBuildInfoEvent.scala
index 019ed8bf4..3ece8ce3e 100644
--- a/core/src/main/scala/com/nvidia/spark/rapids/SparkRapidsBuildInfoEvent.scala
+++ b/core/src/main/scala/com/nvidia/spark/rapids/SparkRapidsBuildInfoEvent.scala
@@ -17,6 +17,7 @@
 package com.nvidia.spark.rapids
 
 import org.apache.spark.scheduler.SparkListenerEvent
+import org.apache.spark.sql.rapids.tool.annotation.ToolsReflection
 
 
 /**
@@ -30,4 +31,9 @@ case class SparkRapidsBuildInfoEvent(
   sparkRapidsJniBuildInfo: Map[String, String],
   cudfBuildInfo: Map[String, String],
   sparkRapidsPrivateBuildInfo: Map[String, String]
-) extends SparkListenerEvent
+) extends SparkListenerEvent {
+  @ToolsReflection("BD-3.2.1", "Ignore")
+  override val eventTime: Long = 0
+  @ToolsReflection("BD-3.2.1", "Ignore")
+  override val eventType: String = ""
+}
diff --git a/core/src/main/scala/org/apache/spark/scheduler/SparkListenerEvent.scala b/core/src/main/scala/org/apache/spark/scheduler/SparkListenerEvent.scala
new file mode 100644
index 000000000..7b209e55a
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/scheduler/SparkListenerEvent.scala
@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.scheduler
+
+import com.fasterxml.jackson.annotation.JsonTypeInfo
+
+import org.apache.spark.annotation.DeveloperApi
+
+/**
+ * This code is mostly copied from org.apache.spark.scheduler.SparkListenerEvent
+ * to make it compatible at runtime with custom Spark implementation that defines abstract methods
+ * in the trait.
+ *
+ * This class is packaged due to a bug in Scala 2.12 that links the method
+ * to the abstract trait, which might not exist in the classpath.
+ * See the related Scala issues:
+ *   https://github.com/scala/bug/issues/10477
+ *   https://github.com/scala/scala-dev/issues/219
+ *   https://github.com/scala/scala-dev/issues/268
+ */
+@DeveloperApi
+@JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, include = JsonTypeInfo.As.PROPERTY, property = "Event")
+trait SparkListenerEvent extends ToolsListenerEventExtraAPIs {
+  /* Whether output this event to the event log */
+  protected[spark] def logEvent: Boolean = true
+}
diff --git a/core/src/main/scala/org/apache/spark/scheduler/ToolsListenerEventExtraAPIs.scala b/core/src/main/scala/org/apache/spark/scheduler/ToolsListenerEventExtraAPIs.scala
new file mode 100644
index 000000000..b45a3ea63
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/scheduler/ToolsListenerEventExtraAPIs.scala
@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.scheduler
+
+import org.apache.spark.sql.rapids.tool.annotation.ToolsReflection
+
+/**
+ * Base trait for events related to SparkRapids build info. This used to add extra APIs that are
+ * not defined in the base Spark trait. This is a work around to be compatible in
+ * runtime with custom Spark implementations that define abstract methods in the trait.
+ * see https://github.com/NVIDIA/spark-rapids-tools/issues/1360
+ */
+trait ToolsListenerEventExtraAPIs {
+  @ToolsReflection("BD-3.2.1",
+    "Ignore the implementation: Definition for an abstract field in the SparkListenerEvent.")
+  val eventTime: Long = 0
+  @ToolsReflection("BD-3.2.1",
+    "Ignore the implementation: Definition for an abstract field in the SparkListenerEvent.")
+  val eventType: String = ""
+}
diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/annotation/ToolsReflection.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/annotation/ToolsReflection.scala
new file mode 100644
index 000000000..95a74b697
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/annotation/ToolsReflection.scala
@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.sql.rapids.tool.annotation
+
+import scala.annotation.StaticAnnotation
+import scala.annotation.meta.{beanGetter, beanSetter, field, getter, param, setter}
+
+
+/**
+ * This code is mostly copied from org.apache.spark.annotation.Since
+ * Reason is copied here because it is being private to Spark packages which makes it
+ * inaccessible for Non-Spark packages.
+ *
+ * A Scala annotation that indicates entities that are used for reeflection in Tools to match
+ * different Spark runtime APIs
+ */
+@param @field @getter @setter @beanGetter @beanSetter
+class ToolsReflection(source: String, comment: String) extends StaticAnnotation
diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/ToolsPlanGraph.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/ToolsPlanGraph.scala
index a112da0b9..1e4243595 100644
--- a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/ToolsPlanGraph.scala
+++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/ToolsPlanGraph.scala
@@ -19,229 +19,59 @@ package org.apache.spark.sql.rapids.tool.util
 import java.util.concurrent.atomic.AtomicLong
 
 import scala.collection.mutable
-import scala.reflect.runtime.universe._
 
 import org.apache.spark.sql.execution.SparkPlanInfo
 import org.apache.spark.sql.execution.ui.{SparkPlanGraph, SparkPlanGraphCluster, SparkPlanGraphEdge, SparkPlanGraphNode, SQLPlanMetric}
 import org.apache.spark.sql.rapids.tool.store.AccumNameRef
-
-class DBReflectionEntry[T](mirror: Mirror, className: String, paramsSize: Option[Int] = None) {
-  // Get the class symbol
-  private val classSymbol = mirror.staticClass(className)
-  private val reflectiveClass = mirror.reflectClass(classSymbol)
-  // Get the constructor method symbol
-  val constr: MethodSymbol = createConstructor(classSymbol, paramsSize)
-
-  // If the paramsCount is defined, we select the the constructor that has parameters size equal to
-  // that value
-  private def createConstructor(symbol: ClassSymbol, paramsCount: Option[Int]): MethodSymbol = {
-    paramsCount match {
-      case None =>
-        // return the primary constructor
-        symbol.primaryConstructor.asMethod
-      case Some(count) =>
-        // return the constructor with given  parameter size
-        val constructors = symbol.info.decls.filter(_.isConstructor)
-          .map(_.asMethod)
-          .filter(_.paramLists.flatten.size == count)
-        val constructor = constructors.headOption.getOrElse {
-          throw new IllegalArgumentException(
-            s"No constructor found with exactly $count parameters for class[$className]")
-        }
-        constructor
-    }
-  }
-
-  def createInstanceFromList(args: List[_]): T = {
-    reflectiveClass
-      .reflectConstructor(constr)(args: _*)
-      .asInstanceOf[T]
-  }
-}
-
-case class DBGraphNodeStub(m: Mirror)
-  extends DBReflectionEntry[org.apache.spark.sql.execution.ui.SparkPlanGraphNode](
-    m, "org.apache.spark.sql.execution.ui.SparkPlanGraphNode") {
-  // DataBricks has different constructor of the sparkPlanGraphNode
-  // [(long,java.lang.String,java.lang.String,scala.collection.Seq,java.lang.String,
-  // boolean,scala.Option,scala.Option)] and
-  // [final long id, final java.lang.String name, final java.lang.String desc,
-  // final scala.collection.Seq<org.apache.spark.sql.execution.ui.SQLPlanMetric> metrics,
-  // final java.lang.String rddScopeId, final boolean started,
-  // final scala.Option<scala.math.BigInt> estRowCount)
-
-  // For 10.4 --> only 1 constructor and has 6 arguments (not 7)
-  // (final long id, final java.lang.String name, final java.lang.String desc,
-  // final scala.collection.Seq<org.apache.spark.sql.execution.ui.SQLPlanMetric> metrics,
-  // final java.lang.String rddScopeId, final scala.Option<scala.math.BigInt> estRowCount
-
-  // DB10.4 has constructor with 6 arguments.
-  private val isDB104OrOlder: Boolean = constr.paramLists.flatten.size < 7
-
-  def createInstance(id: Long, name: String, desc: String,
-      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphNode = {
-    // Define argument values
-    val argValues = if (isDB104OrOlder) {
-      List(id, name, desc, metrics, "", None)
-    } else {
-      List(id, name, desc, metrics, "", false, None, None)
-    }
-    createInstanceFromList(argValues)
-  }
-}
-
-case class DBGraphSQLMetricStub(m: Mirror)
-  extends DBReflectionEntry[org.apache.spark.sql.execution.ui.SQLPlanMetric](
-    m, "org.apache.spark.sql.execution.ui.SQLPlanMetric") {
-  // DataBricks has different constructor of the sparkPlanGraphNode
-  //Array(final java.lang.String name, final long accumulatorId,
-  // final java.lang.String metricType, final boolean experimental)
-
-  // for 10.4 it is only one constructor with 3 arguments.
-  // final java.lang.String name, final long accumulatorId, final java.lang.String metricType
-  private val isDB104OrOlder: Boolean = constr.paramLists.flatten.size < 4
-  def createInstance(name: String,
-      accumulatorId: Long,
-      metricType: String): SQLPlanMetric = {
-    val argValues = if (isDB104OrOlder) {
-      List(name, accumulatorId, metricType)
-    } else {
-      List(name, accumulatorId, metricType, false)
-    }
-    createInstanceFromList(argValues)
-  }
-}
-
-case class DBGraphClusterStub(m: Mirror)
-  extends DBReflectionEntry[org.apache.spark.sql.execution.ui.SparkPlanGraphCluster](
-    m, "org.apache.spark.sql.execution.ui.SparkPlanGraphCluster") {
-  // DataBricks has different constructor of the sparkPlanGraphNode
-  // (final long id, final java.lang.String name, final java.lang.String desc,
-  // final ArrayBuffer<org.apache.spark.sql.execution.ui.SparkPlanGraphNode> nodes,
-  // final scala.collection.Seq<org.apache.spark.sql.execution.ui.SQLPlanMetric> metrics,
-  // final java.lang.String rddScopeId)
-
-  // 10.4 is the same as other versions
-  // (final long id, final java.lang.String name, final java.lang.String desc,
-  // final ArrayBuffer<org.apache.spark.sql.execution.ui.SparkPlanGraphNode> nodes,
-  // final Seq<org.apache.spark.sql.execution.ui.SQLPlanMetric> metrics,
-  // final java.lang.String rddScopeId
-  def createInstance(id: Long,
-      name: String,
-      desc: String,
-      nodes: mutable.ArrayBuffer[SparkPlanGraphNode],
-      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphCluster = {
-    val argValues = List(id, name, desc, nodes, metrics, "")
-    createInstanceFromList(argValues)
-  }
-}
-
-// All versions in DB accept 2 parameters for constructor. So we stick to that version
-// by passing (2) to the parent class
-case class DBGraphEdgeStub(m: Mirror)
-  extends DBReflectionEntry[org.apache.spark.sql.execution.ui.SparkPlanGraphEdge](
-    m, "org.apache.spark.sql.execution.ui.SparkPlanGraphEdge", Option(2)) {
-  // DataBricks has different constructor of the sparkPlanGraphNode
-  // (final long fromId, final long toId,
-  // final scala.Option<java.lang.Object> numOutputRowsId)
-  //
-  // for 10.4 only one constructor with two arguments
-  // final long fromId, final long toId)
-
-  def createInstance(fromId: Long, toId: Long): SparkPlanGraphEdge = {
-    val argValues = List(fromId, toId)
-    createInstanceFromList(argValues)
-  }
-}
-
-// Container class to hold snapshot of the reflection fields instead of recalculating them every
-// time we call the constructor
-case class DBReflectionContainer() {
-  private val mirror = runtimeMirror(getClass.getClassLoader)
-  private val nodeStub = DBGraphNodeStub(mirror)
-  private val clusterStub = DBGraphClusterStub(mirror)
-  private val edgeStub = DBGraphEdgeStub(mirror)
-  private val metricStub = DBGraphSQLMetricStub(mirror)
-
-  def constructNode(id: Long, name: String, desc: String,
-      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphNode = {
-    nodeStub.createInstance(id, name, desc, metrics)
-  }
-
-  def constructSQLPlanMetric(name: String,
-      accumulatorId: Long,
-      metricType: String): SQLPlanMetric = {
-    metricStub.createInstance(name, accumulatorId, metricType)
-  }
-
-  def constructCluster(id: Long,
-      name: String,
-      desc: String,
-      nodes: mutable.ArrayBuffer[SparkPlanGraphNode],
-      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphCluster = {
-    clusterStub.createInstance(id, name, desc, nodes, metrics)
-  }
-
-  def constructEdge(fromId: Long, toId: Long): SparkPlanGraphEdge = {
-    edgeStub.createInstance(fromId, toId)
-  }
-}
+import org.apache.spark.sql.rapids.tool.util.stubs.{GraphReflectionAPI, GraphReflectionAPIHelper}
 
 /**
  * This code is mostly copied from org.apache.spark.sql.execution.ui.SparkPlanGraph
  * with changes to handle GPU nodes. Without this special handle, the default SparkPlanGraph
- * would not be able to recognize reused/exchange nodes leading to duplicating nodes.
+ * would not recognize reused/exchange nodes leading to duplicating nodes.
  *
  * Build a SparkPlanGraph from the root of a SparkPlan tree.
  */
 object ToolsPlanGraph {
-  // TODO: We should have a util to detect if the runtime is Databricks.
-  //      This can be achieved by checking for spark properties
-  //      spark.databricks.clusterUsageTags.clusterAllTags
-  private lazy val dbRuntimeReflection = DBReflectionContainer()
-  // By default call the Spark constructor. If this fails, we fall back to the DB constructor
+  // Captures the API loaded at runtime if any.
+  var api: GraphReflectionAPI = _
+
+  // The actual code used to build the graph. If the API is not available, then fallback to the
+  // Spark default API.
+  private lazy val graphBuilder: SparkPlanInfo => SparkPlanGraph = {
+    GraphReflectionAPIHelper.api match {
+      case Some(_) =>
+        // set the api to the available one
+        api = GraphReflectionAPIHelper.api.get
+        (planInfo: SparkPlanInfo) => {
+          val nodeIdGenerator = new AtomicLong(0)
+          val nodes = mutable.ArrayBuffer[SparkPlanGraphNode]()
+          val edges = mutable.ArrayBuffer[SparkPlanGraphEdge]()
+          val exchanges = mutable.HashMap[SparkPlanInfo, SparkPlanGraphNode]()
+          buildSparkPlanGraphNode(planInfo, nodeIdGenerator, nodes, edges, null, null, exchanges)
+          new SparkPlanGraph(nodes, edges)
+        }
+      case _ =>
+        // Fallback to the default SparkPlanGraph
+        (planInfo: SparkPlanInfo) => {
+          SparkPlanGraph(planInfo)
+        }
+    }
+  }
+
+  // used for testing purpose
   def constructGraphNode(id: Long, name: String, desc: String,
       metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphNode = {
-    try {
-      new SparkPlanGraphNode(id, name, desc, metrics)
-    } catch {
-      case _: java.lang.NoSuchMethodError =>
-        dbRuntimeReflection.constructNode(id, name, desc, metrics)
-    }
+    GraphReflectionAPIHelper.api.get.constructNode(id, name, desc, metrics)
   }
 
+  // Normalize the accumName before creating it.
   private def constructSQLPlanMetric(name: String,
       accumulatorId: Long,
       metricType: String): SQLPlanMetric = {
     val accNameRef = AccumNameRef.getOrCreateAccumNameRef(name)
-    try {
-      SQLPlanMetric(accNameRef.value, accumulatorId, metricType)
-    } catch {
-      case _: java.lang.NoSuchMethodError =>
-        dbRuntimeReflection.constructSQLPlanMetric(
-          accNameRef.value, accumulatorId, metricType)
-    }
-  }
-
-  private def constructCluster(id: Long,
-      name: String,
-      desc: String,
-      nodes: mutable.ArrayBuffer[SparkPlanGraphNode],
-      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphCluster = {
-    try {
-      new SparkPlanGraphCluster(id, name, desc, nodes, metrics)
-    } catch {
-      case _: java.lang.NoSuchMethodError =>
-        dbRuntimeReflection.constructCluster(id, name, desc, nodes, metrics)
-    }
-  }
-  private def constructEdge(fromId: Long, toId: Long): SparkPlanGraphEdge = {
-    try {
-      SparkPlanGraphEdge(fromId, toId)
-    } catch {
-      case _: java.lang.NoSuchMethodError =>
-        dbRuntimeReflection.constructEdge(fromId, toId)
-    }
+    GraphReflectionAPIHelper.api.get.constructSQLPlanMetric(accNameRef.value, accumulatorId,
+      metricType)
   }
 
   /**
@@ -249,15 +79,10 @@ object ToolsPlanGraph {
    */
   def apply(planInfo: SparkPlanInfo): SparkPlanGraph = {
     try {
-      val nodeIdGenerator = new AtomicLong(0)
-      val nodes = mutable.ArrayBuffer[SparkPlanGraphNode]()
-      val edges = mutable.ArrayBuffer[SparkPlanGraphEdge]()
-      val exchanges = mutable.HashMap[SparkPlanInfo, SparkPlanGraphNode]()
-      buildSparkPlanGraphNode(planInfo, nodeIdGenerator, nodes, edges, null, null, exchanges)
-      new SparkPlanGraph(nodes, edges)
+      graphBuilder(planInfo)
     } catch {
       // If the construction of the graph fails due to NoSuchMethod, then it is possible the
-      // runtime is DB and we fallback to the loaded runtime jars
+      // runtime is not compatible with the constructors and we fallback to the loaded runtime jars
       case _ : java.lang.NoSuchMethodError | _ : java.lang.IllegalArgumentException =>
         SparkPlanGraph(planInfo)
     }
@@ -285,7 +110,7 @@ object ToolsPlanGraph {
           constructSQLPlanMetric(metric.name, metric.accumulatorId, metric.metricType)
         }
 
-        val cluster = constructCluster(
+        val cluster = api.constructCluster(
           nodeIdGenerator.getAndIncrement(),
           planInfo.nodeName,
           planInfo.simpleString,
@@ -302,7 +127,7 @@ object ToolsPlanGraph {
         if (exchanges.contains(planInfo.children.head)) {
           // Point to the re-used exchange
           val node = exchanges(planInfo.children.head)
-          edges += constructEdge(node.id, parent.id)
+          edges += api.constructEdge(node.id, parent.id)
         } else {
           buildSparkPlanGraphNode(
             planInfo.children.head, nodeIdGenerator, nodes, edges, parent, null, exchanges)
@@ -316,7 +141,7 @@ object ToolsPlanGraph {
       case "Subquery" | "SubqueryBroadcast" if exchanges.contains(planInfo) =>
         // Point to the re-used subquery
         val node = exchanges(planInfo)
-        edges += constructEdge(node.id, parent.id)
+        edges += api.constructEdge(node.id, parent.id)
       case "ReusedSubquery" =>
         // Re-used subquery might appear before the original subquery, so skip this node and let
         // the previous `case` make sure the re-used and the original point to the same node.
@@ -325,12 +150,12 @@ object ToolsPlanGraph {
       case "ReusedExchange" if exchanges.contains(planInfo.children.head) =>
         // Point to the re-used exchange
         val node = exchanges(planInfo.children.head)
-        edges += constructEdge(node.id, parent.id)
+        edges += api.constructEdge(node.id, parent.id)
       case name =>
         val metrics = planInfo.metrics.map { metric =>
           constructSQLPlanMetric(metric.name, metric.accumulatorId, metric.metricType)
         }
-        val node = constructGraphNode(nodeIdGenerator.getAndIncrement(),
+        val node = api.constructNode(nodeIdGenerator.getAndIncrement(),
           planInfo.nodeName, planInfo.simpleString, metrics)
         if (subgraph == null) {
           nodes += node
@@ -342,7 +167,7 @@ object ToolsPlanGraph {
         }
 
         if (parent != null) {
-          edges += constructEdge(node.id, parent.id)
+          edges += api.constructEdge(node.id, parent.id)
         }
         planInfo.children.foreach(
           buildSparkPlanGraphNode(_, nodeIdGenerator, nodes, edges, node, subgraph, exchanges))
diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/DefaultGraphReflectionAPI.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/DefaultGraphReflectionAPI.scala
new file mode 100644
index 000000000..9f0a956e2
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/DefaultGraphReflectionAPI.scala
@@ -0,0 +1,51 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.sql.rapids.tool.util.stubs
+
+import scala.collection.mutable
+
+import org.apache.spark.sql.execution.ui.{SparkPlanGraphCluster, SparkPlanGraphEdge, SparkPlanGraphNode, SQLPlanMetric}
+
+/**
+ * Implementation that uses the SparkPlanGraphNode, SparkPlanGraphCluster, SparkPlanGraphEdge and
+ * SQLPlanMetric classes from the Spark codebase. This API is used to reflectively construct these
+ * classes at runtime.
+ */
+class DefaultGraphReflectionAPI extends GraphReflectionAPI {
+  def constructNode(id: Long, name: String, desc: String,
+      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphNode = {
+    new SparkPlanGraphNode(id, name, desc, metrics)
+  }
+
+  def constructSQLPlanMetric(name: String,
+      accumulatorId: Long,
+      metricType: String): SQLPlanMetric = {
+    SQLPlanMetric(name, accumulatorId, metricType)
+  }
+
+  def constructCluster(id: Long,
+      name: String,
+      desc: String,
+      nodes: mutable.ArrayBuffer[SparkPlanGraphNode],
+      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphCluster = {
+    new SparkPlanGraphCluster(id, name, desc, nodes, metrics)
+  }
+
+  def constructEdge(fromId: Long, toId: Long): SparkPlanGraphEdge = {
+    SparkPlanGraphEdge(fromId, toId)
+  }
+}
diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/GraphReflectionAPI.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/GraphReflectionAPI.scala
new file mode 100644
index 000000000..8e6324707
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/GraphReflectionAPI.scala
@@ -0,0 +1,45 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.sql.rapids.tool.util.stubs
+
+import scala.collection.mutable
+import scala.reflect.runtime.universe.runtimeMirror
+
+import org.apache.spark.sql.execution.ui.{SparkPlanGraphCluster, SparkPlanGraphEdge, SparkPlanGraphNode, SQLPlanMetric}
+
+/**
+ * API to define methods used to construct SparkPlanGraphNode, SparkPlanGraphCluster,
+ * SparkPlanGraphEdge and SQLPlanMetric objects at runtime.
+ */
+trait GraphReflectionAPI {
+  protected val mirror = runtimeMirror(getClass.getClassLoader)
+
+  def constructNode(id: Long, name: String, desc: String,
+      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphNode
+
+  def constructSQLPlanMetric(name: String,
+      accumulatorId: Long,
+      metricType: String): SQLPlanMetric
+
+  def constructCluster(id: Long,
+      name: String,
+      desc: String,
+      nodes: mutable.ArrayBuffer[SparkPlanGraphNode],
+      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphCluster
+
+  def constructEdge(fromId: Long, toId: Long): SparkPlanGraphEdge
+}
diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/GraphReflectionAPIHelper.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/GraphReflectionAPIHelper.scala
new file mode 100644
index 000000000..871e9630f
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/GraphReflectionAPIHelper.scala
@@ -0,0 +1,60 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.sql.rapids.tool.util.stubs
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.rapids.tool.util.stubs.bd.BDGraphReflectionAPI
+import org.apache.spark.sql.rapids.tool.util.stubs.db.DBGraphReflectionAPI
+
+/**
+ * Helper object to load the correct runtime GraphReflectionAPI implementation.
+ * This is used to improve performance by avoiding try-catch blocks on every single component.
+ */
+object GraphReflectionAPIHelper extends Logging {
+  /**
+   * Iterate through the available GraphReflectionAPI implementations and return the first one that
+   * can be used to construct a SparkPlan Graph Node.
+   * @return the first GraphReflectionAPI that is compatible with the current Spark runtime.
+   */
+  private def loadRuntimeGraphAPI(): Option[GraphReflectionAPI] = {
+    // defines the available GraphReflectionAPI implementations
+    val allAPIS = Map(
+      "Default Graph API" -> new DefaultGraphReflectionAPI(),
+      "BD Graph API" -> BDGraphReflectionAPI(),
+      "DB Graph API" -> DBGraphReflectionAPI()
+    )
+    // Finds the first compatible API by creating a dummy node.
+    val res = allAPIS.find { entry =>
+      try {
+        // Create a dummy node and captures the exception if the API is not compatible
+        entry._2.constructNode(0, "node1", "descr", Seq.empty)
+        true
+      } catch {
+        case _: Throwable => false
+      }
+    }
+    // Log this information or show an error to be aware of incompatible runtimes.
+    if (res.isDefined) {
+      logInfo(s"Using runtime API [${res.get._1}] to Construct SparkPlan Graph")
+    } else {
+      logError("No runtime Graph API found. Falling to the spark runtime constructor")
+    }
+    res.map(_._2)
+  }
+  // caches the API to avoid re-creating it multiple times.
+  lazy val api: Option[GraphReflectionAPI] = loadRuntimeGraphAPI()
+}
diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/GraphReflectionEntry.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/GraphReflectionEntry.scala
new file mode 100644
index 000000000..7681fb870
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/GraphReflectionEntry.scala
@@ -0,0 +1,61 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.sql.rapids.tool.util.stubs
+
+import scala.reflect.runtime.universe.{ClassSymbol, MethodSymbol, Mirror}
+
+/**
+ * A reflection entry for a class that can be instantiated at runtime.
+ * @param mirror The mirror to use for reflection (created once per classloader)
+ * @param className The name of the class to instantiate
+ * @param paramsSize The number of parameters to the constructor, if known. Some Spark releases
+ *                   like Databricks have different overrides that differ by count of arguments.
+ * @tparam T The class to instantiate
+ */
+class GraphReflectionEntry[T](mirror: Mirror, className: String, paramsSize: Option[Int] = None) {
+  // Get the class symbol
+  private val classSymbol = mirror.staticClass(className)
+  private val reflectiveClass = mirror.reflectClass(classSymbol)
+  // Get the constructor method symbol
+  val constr: MethodSymbol = createConstructor(classSymbol, paramsSize)
+
+  // If the paramsCount is defined, we select the constructor that has parameters size equal to
+  // that value
+  private def createConstructor(symbol: ClassSymbol, paramsCount: Option[Int]): MethodSymbol = {
+    paramsCount match {
+      case None =>
+        // return the primary constructor
+        symbol.primaryConstructor.asMethod
+      case Some(count) =>
+        // return the constructor with given parameter size
+        val constructors = symbol.info.decls.filter(_.isConstructor)
+          .map(_.asMethod)
+          .filter(_.paramLists.flatten.size == count)
+        val constructor = constructors.headOption.getOrElse {
+          throw new IllegalArgumentException(
+            s"No constructor found with exactly $count parameters for class[$className]")
+        }
+        constructor
+    }
+  }
+
+  def createInstanceFromList(args: List[_]): T = {
+    reflectiveClass
+      .reflectConstructor(constr)(args: _*)
+      .asInstanceOf[T]
+  }
+}
diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/bd/BDGraphClusterStub.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/bd/BDGraphClusterStub.scala
new file mode 100644
index 000000000..30657a277
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/bd/BDGraphClusterStub.scala
@@ -0,0 +1,37 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.sql.rapids.tool.util.stubs.bd
+
+import scala.collection.mutable
+import scala.reflect.runtime.universe.Mirror
+
+import org.apache.spark.sql.execution.ui.{SparkPlanGraphCluster, SparkPlanGraphNode, SQLPlanMetric}
+import org.apache.spark.sql.rapids.tool.annotation.ToolsReflection
+import org.apache.spark.sql.rapids.tool.util.stubs.GraphReflectionEntry
+
+case class BDGraphClusterStub(m: Mirror)
+  extends GraphReflectionEntry[org.apache.spark.sql.execution.ui.SparkPlanGraphCluster](
+    m, "org.apache.spark.sql.execution.ui.SparkPlanGraphCluster") {
+  @ToolsReflection("BD-3.2.1", "Defines an extra argument planId: Int in the constructor")
+  def createInstance(id: Long,
+      name: String,
+      desc: String,
+      nodes: mutable.ArrayBuffer[SparkPlanGraphNode],
+      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphCluster = {
+    createInstanceFromList(List(id, name, desc, nodes, metrics, 0))
+  }
+}
diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/bd/BDGraphNodeStub.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/bd/BDGraphNodeStub.scala
new file mode 100644
index 000000000..dda259be8
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/bd/BDGraphNodeStub.scala
@@ -0,0 +1,35 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.sql.rapids.tool.util.stubs.bd
+
+import scala.reflect.runtime.universe.Mirror
+
+import org.apache.spark.sql.execution.ui.{SparkPlanGraphNode, SQLPlanMetric}
+import org.apache.spark.sql.rapids.tool.annotation.ToolsReflection
+import org.apache.spark.sql.rapids.tool.util.stubs.GraphReflectionEntry
+
+case class BDGraphNodeStub(m: Mirror)
+  extends GraphReflectionEntry[org.apache.spark.sql.execution.ui.SparkPlanGraphNode](
+    m, "org.apache.spark.sql.execution.ui.SparkPlanGraphNode") {
+
+  @ToolsReflection("BD-3.2.1", "Defines an extra argument planId: Int in the constructor")
+  def createInstance(id: Long, name: String, desc: String,
+      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphNode = {
+    // Define argument values
+    createInstanceFromList(List(id, name, desc, metrics, 0))
+  }
+}
diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/bd/BDGraphReflectionAPI.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/bd/BDGraphReflectionAPI.scala
new file mode 100644
index 000000000..f67a20c83
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/bd/BDGraphReflectionAPI.scala
@@ -0,0 +1,42 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.sql.rapids.tool.util.stubs.bd
+
+import scala.collection.mutable
+
+import org.apache.spark.sql.execution.ui.{SparkPlanGraphCluster, SparkPlanGraphNode, SQLPlanMetric}
+import org.apache.spark.sql.rapids.tool.annotation.ToolsReflection
+import org.apache.spark.sql.rapids.tool.util.stubs.DefaultGraphReflectionAPI
+
+@ToolsReflection("BD-3.2.1", "Reflection to add extra arguments to the Node/Cluster constructors")
+case class BDGraphReflectionAPI() extends DefaultGraphReflectionAPI {
+  private val nodeStub = BDGraphNodeStub(mirror)
+  private val clusterStub = BDGraphClusterStub(mirror)
+
+  override def constructNode(id: Long, name: String, desc: String,
+      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphNode = {
+    nodeStub.createInstance(id, name, desc, metrics)
+  }
+
+  override def constructCluster(id: Long,
+      name: String,
+      desc: String,
+      nodes: mutable.ArrayBuffer[SparkPlanGraphNode],
+      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphCluster = {
+    clusterStub.createInstance(id, name, desc, nodes, metrics)
+  }
+}
diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/db/DBGraphClusterStub.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/db/DBGraphClusterStub.scala
new file mode 100644
index 000000000..162947d14
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/db/DBGraphClusterStub.scala
@@ -0,0 +1,47 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.sql.rapids.tool.util.stubs.db
+
+import scala.collection.mutable
+import scala.reflect.runtime.universe.Mirror
+
+import org.apache.spark.sql.execution.ui.{SparkPlanGraphCluster, SparkPlanGraphNode, SQLPlanMetric}
+import org.apache.spark.sql.rapids.tool.util.stubs.GraphReflectionEntry
+
+case class DBGraphClusterStub(m: Mirror)
+  extends GraphReflectionEntry[org.apache.spark.sql.execution.ui.SparkPlanGraphCluster](
+    m, "org.apache.spark.sql.execution.ui.SparkPlanGraphCluster") {
+  // DataBricks has different constructor of the sparkPlanGraphNode
+  // (final long id, final java.lang.String name, final java.lang.String desc,
+  // final ArrayBuffer<org.apache.spark.sql.execution.ui.SparkPlanGraphNode> nodes,
+  // final scala.collection.Seq<org.apache.spark.sql.execution.ui.SQLPlanMetric> metrics,
+  // final java.lang.String rddScopeId)
+
+  // 10.4 is the same as other versions
+  // (final long id, final java.lang.String name, final java.lang.String desc,
+  // final ArrayBuffer<org.apache.spark.sql.execution.ui.SparkPlanGraphNode> nodes,
+  // final Seq<org.apache.spark.sql.execution.ui.SQLPlanMetric> metrics,
+  // final java.lang.String rddScopeId
+  def createInstance(id: Long,
+      name: String,
+      desc: String,
+      nodes: mutable.ArrayBuffer[SparkPlanGraphNode],
+      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphCluster = {
+    val argValues = List(id, name, desc, nodes, metrics, "")
+    createInstanceFromList(argValues)
+  }
+}
diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/db/DBGraphEdgeStub.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/db/DBGraphEdgeStub.scala
new file mode 100644
index 000000000..4d36b4e95
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/db/DBGraphEdgeStub.scala
@@ -0,0 +1,41 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.sql.rapids.tool.util.stubs.db
+
+import scala.reflect.runtime.universe.Mirror
+
+import org.apache.spark.sql.execution.ui.SparkPlanGraphEdge
+import org.apache.spark.sql.rapids.tool.annotation.ToolsReflection
+import org.apache.spark.sql.rapids.tool.util.stubs.GraphReflectionEntry
+
+@ToolsReflection("DataBricks",
+  "All DB accept 2 parameters. We enforce that constructor by passing (2) to the parent class")
+case class DBGraphEdgeStub(m: Mirror)
+  extends GraphReflectionEntry[org.apache.spark.sql.execution.ui.SparkPlanGraphEdge](
+    m, "org.apache.spark.sql.execution.ui.SparkPlanGraphEdge", Option(2)) {
+  // DataBricks has different constructor of the sparkPlanGraphEdge
+  // (final long fromId, final long toId,
+  // final scala.Option<java.lang.Object> numOutputRowsId)
+  //
+  // for 10.4 only one constructor with two arguments
+  // final long fromId, final long toId)
+
+  def createInstance(fromId: Long, toId: Long): SparkPlanGraphEdge = {
+    val argValues = List(fromId, toId)
+    createInstanceFromList(argValues)
+  }
+}
diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/db/DBGraphNodeStub.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/db/DBGraphNodeStub.scala
new file mode 100644
index 000000000..cba08bb25
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/db/DBGraphNodeStub.scala
@@ -0,0 +1,53 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.sql.rapids.tool.util.stubs.db
+
+import scala.reflect.runtime.universe.Mirror
+
+import org.apache.spark.sql.execution.ui.{SparkPlanGraphNode, SQLPlanMetric}
+import org.apache.spark.sql.rapids.tool.util.stubs.GraphReflectionEntry
+
+case class DBGraphNodeStub(m: Mirror)
+  extends GraphReflectionEntry[org.apache.spark.sql.execution.ui.SparkPlanGraphNode](
+    m, "org.apache.spark.sql.execution.ui.SparkPlanGraphNode") {
+  // DataBricks has different constructor of the sparkPlanGraphNode
+  // [(long,java.lang.String,java.lang.String,scala.collection.Seq,java.lang.String,
+  // boolean,scala.Option,scala.Option)] and
+  // [final long id, final java.lang.String name, final java.lang.String desc,
+  // final scala.collection.Seq<org.apache.spark.sql.execution.ui.SQLPlanMetric> metrics,
+  // final java.lang.String rddScopeId, final boolean started,
+  // final scala.Option<scala.math.BigInt> estRowCount)
+
+  // For 10.4 --> only 1 constructor and has 6 arguments (not 7)
+  // (final long id, final java.lang.String name, final java.lang.String desc,
+  // final scala.collection.Seq<org.apache.spark.sql.execution.ui.SQLPlanMetric> metrics,
+  // final java.lang.String rddScopeId, final scala.Option<scala.math.BigInt> estRowCount
+
+  // DB10.4 has constructor with 6 arguments.
+  private val isDB104OrOlder: Boolean = constr.paramLists.flatten.size < 7
+
+  def createInstance(id: Long, name: String, desc: String,
+      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphNode = {
+    // Define argument values
+    val argValues = if (isDB104OrOlder) {
+      List(id, name, desc, metrics, "", None)
+    } else {
+      List(id, name, desc, metrics, "", false, None, None)
+    }
+    createInstanceFromList(argValues)
+  }
+}
diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/db/DBGraphReflectionAPI.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/db/DBGraphReflectionAPI.scala
new file mode 100644
index 000000000..42716d8d9
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/db/DBGraphReflectionAPI.scala
@@ -0,0 +1,59 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.sql.rapids.tool.util.stubs.db
+
+import scala.collection.mutable
+
+import org.apache.spark.sql.execution.ui.{SparkPlanGraphCluster, SparkPlanGraphEdge, SparkPlanGraphNode, SQLPlanMetric}
+import org.apache.spark.sql.rapids.tool.annotation.ToolsReflection
+import org.apache.spark.sql.rapids.tool.util.stubs.DefaultGraphReflectionAPI
+
+/**
+ * Implementation that uses the SparkPlanGraphNode, SparkPlanGraphCluster, SparkPlanGraphEdge and
+ * SQLPlanMetric classes from the DataBricks codeBase.
+ * This API is used to reflectively construct these classes at runtime.
+ */
+@ToolsReflection("DataBricks", "DB defines different constructors for SparkGraphComponents")
+case class DBGraphReflectionAPI() extends DefaultGraphReflectionAPI {
+  private val nodeStub = DBGraphNodeStub(mirror)
+  private val clusterStub = DBGraphClusterStub(mirror)
+  private val edgeStub = DBGraphEdgeStub(mirror)
+  private val metricStub = DBGraphSQLMetricStub(mirror)
+
+  override def constructNode(id: Long, name: String, desc: String,
+      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphNode = {
+    nodeStub.createInstance(id, name, desc, metrics)
+  }
+
+  override def constructSQLPlanMetric(name: String,
+      accumulatorId: Long,
+      metricType: String): SQLPlanMetric = {
+    metricStub.createInstance(name, accumulatorId, metricType)
+  }
+
+  override def constructCluster(id: Long,
+      name: String,
+      desc: String,
+      nodes: mutable.ArrayBuffer[SparkPlanGraphNode],
+      metrics: collection.Seq[SQLPlanMetric]): SparkPlanGraphCluster = {
+    clusterStub.createInstance(id, name, desc, nodes, metrics)
+  }
+
+  override def constructEdge(fromId: Long, toId: Long): SparkPlanGraphEdge = {
+    edgeStub.createInstance(fromId, toId)
+  }
+}
diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/db/DBGraphSQLMetricStub.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/db/DBGraphSQLMetricStub.scala
new file mode 100644
index 000000000..b2591ad7d
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/db/DBGraphSQLMetricStub.scala
@@ -0,0 +1,45 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.sql.rapids.tool.util.stubs.db
+
+import scala.reflect.runtime.universe.Mirror
+
+import org.apache.spark.sql.execution.ui.SQLPlanMetric
+import org.apache.spark.sql.rapids.tool.util.stubs.GraphReflectionEntry
+
+case class DBGraphSQLMetricStub(m: Mirror)
+  extends GraphReflectionEntry[org.apache.spark.sql.execution.ui.SQLPlanMetric](
+    m, "org.apache.spark.sql.execution.ui.SQLPlanMetric") {
+  // DataBricks has different constructor of the sparkPlanGraphNode
+  //Array(final java.lang.String name, final long accumulatorId,
+  // final java.lang.String metricType, final boolean experimental)
+
+  // for 10.4 it is only one constructor with 3 arguments.
+  // final java.lang.String name, final long accumulatorId, final java.lang.String metricType
+  private val isDB104OrOlder: Boolean = constr.paramLists.flatten.size < 4
+
+  def createInstance(name: String,
+      accumulatorId: Long,
+      metricType: String): SQLPlanMetric = {
+    val argValues = if (isDB104OrOlder) {
+      List(name, accumulatorId, metricType)
+    } else {
+      List(name, accumulatorId, metricType, false)
+    }
+    createInstanceFromList(argValues)
+  }
+}
diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/package.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/package.scala
new file mode 100644
index 000000000..5174763b0
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/stubs/package.scala
@@ -0,0 +1,25 @@
+/*
+ * Copyright (c) 2024, 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 org.apache.spark.sql.rapids.tool.util
+
+/**
+ * Stubs used for reflection at runtime to work with different Spark implementations.
+ * This will include stubs created for each Spark flavor when needed.
+ */
+package object stubs {
+
+}