Use SparkPlanInfo constructor that is compatible with Databricks' Spark fork

Author: charlesmyu

dd-java-agent/instrumentation/spark/spark_2.12/src/main/java/datadog/trace/instrumentation/spark/Spark212Instrumentation.java

@@ -110,21 +110,38 @@ public static void exit(
                 .toMap(Predef.$conforms());
         try {
           Constructor<?> targetCtor = null;
+          Object[] args = null;
           for (Constructor<?> c : SparkPlanInfo.class.getConstructors()) {
             if (c.getParameterCount() == 5) {
               targetCtor = c;
+              args =
+                  new Object[] {
+                    planInfo.nodeName(),
+                    planInfo.simpleString(),
+                    planInfo.children(),
+                    meta,
+                    planInfo.metrics()
+                  };
               break;
-            }
-          }
-          if (targetCtor != null) {
-            Object newInst =
-                targetCtor.newInstance(
+            } else if (c.getParameterCount() == 8) {
+              // Databricks fork of Spark has a different SparkPlanInfo constructor
+              targetCtor = c;
+              args =
+                  new Object[] {
                     planInfo.nodeName(),
                     planInfo.simpleString(),
                     planInfo.children(),
                     meta,
-                    planInfo.metrics());
-            planInfo = (SparkPlanInfo) newInst;
+                    planInfo.metrics(),
+                    SparkPlanInfo.class.getMethod("estRowCount").invoke(new Object[] {}),
+                    SparkPlanInfo.class.getMethod("rddScopeId").invoke(new Object[] {}),
+                    SparkPlanInfo.class.getMethod("explainId").invoke(new Object[] {})
+                  };
+              break;
+            }
+          }
+          if (targetCtor != null && args != null) {
+            planInfo = (SparkPlanInfo) targetCtor.newInstance(args);
           }
         } catch (Throwable ignored) {
         }

dd-java-agent/instrumentation/spark/spark_2.13/src/main/java/datadog/trace/instrumentation/spark/Spark213Instrumentation.java

@@ -110,21 +110,38 @@ public static void exit(
             HashMap.from(JavaConverters.asScala(planUtils.extractFormattedProduct(plan)));
         try {
           Constructor<?> targetCtor = null;
+          Object[] args = null;
           for (Constructor<?> c : SparkPlanInfo.class.getConstructors()) {
             if (c.getParameterCount() == 5) {
               targetCtor = c;
+              args =
+                  new Object[] {
+                    planInfo.nodeName(),
+                    planInfo.simpleString(),
+                    planInfo.children(),
+                    meta,
+                    planInfo.metrics()
+                  };
               break;
-            }
-          }
-          if (targetCtor != null) {
-            Object newInst =
-                targetCtor.newInstance(
+            } else if (c.getParameterCount() == 8) {
+              // Databricks fork of Spark has a different SparkPlanInfo constructor
+              targetCtor = c;
+              args =
+                  new Object[] {
                     planInfo.nodeName(),
                     planInfo.simpleString(),
                     planInfo.children(),
                     meta,
-                    planInfo.metrics());
-            planInfo = (SparkPlanInfo) newInst;
+                    planInfo.metrics(),
+                    SparkPlanInfo.class.getMethod("estRowCount").invoke(new Object[] {}),
+                    SparkPlanInfo.class.getMethod("rddScopeId").invoke(new Object[] {}),
+                    SparkPlanInfo.class.getMethod("explainId").invoke(new Object[] {})
+                  };
+              break;
+            }
+          }
+          if (targetCtor != null && args != null) {
+            planInfo = (SparkPlanInfo) targetCtor.newInstance(args);
           }
         } catch (Throwable ignored) {
         }