Use Abstract class for common functions

Author: charlesmyu

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

@@ -21,6 +21,7 @@ public class Spark212Instrumentation extends AbstractSparkInstrumentation {
   public String[] helperClassNames() {
     return new String[] {
       packageName + ".AbstractDatadogSparkListener",
+      packageName + ".AbstractSparkPlanUtils",
       packageName + ".DatabricksParentContext",
       packageName + ".OpenlineageParentContext",
       packageName + ".DatadogSpark212Listener",
@@ -31,7 +32,7 @@ public String[] helperClassNames() {
       packageName + ".SparkSQLUtils",
       packageName + ".SparkSQLUtils$SparkPlanInfoForStage",
       packageName + ".SparkSQLUtils$AccumulatorWithStage",
-      packageName + ".Spark212PlanUtils",
+      packageName + ".Spark212PlanUtils"
     };
   }
 
@@ -109,14 +110,14 @@ public static void exit(
         @Advice.Return(readOnly = false) SparkPlanInfo planInfo,
         @Advice.Argument(0) SparkPlan plan) {
       if (planInfo.metadata().size() == 0) {
+        Spark212PlanUtils planUtils = new Spark212PlanUtils();
         HashMap<String, String> args = new HashMap<>();
         planInfo =
             new SparkPlanInfo(
                 planInfo.nodeName(),
                 planInfo.simpleString(),
                 planInfo.children(),
-                args.$plus$plus(
-                    JavaConverters.mapAsScalaMap(Spark212PlanUtils.extractPlanProduct(plan))),
+                args.$plus$plus(JavaConverters.mapAsScalaMap(planUtils.extractPlanProduct(plan))),
                 planInfo.metrics());
       }
     }

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

@@ -3,13 +3,13 @@
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
-import org.apache.spark.sql.execution.SparkPlan;
+import org.apache.spark.sql.catalyst.trees.TreeNode;
 import scala.collection.JavaConverters;
 
 // An extension of how Spark translates `SparkPlan`s to `SparkPlanInfo`, see here:
 // https://github.com/apache/spark/blob/v3.5.0/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlanInfo.scala#L54
-public class Spark212PlanUtils {
-  public static Map<String, String> extractPlanProduct(SparkPlan plan) {
+public class Spark212PlanUtils extends AbstractSparkPlanUtils {
+  public Map<String, String> extractPlanProduct(TreeNode plan) {
     HashMap<String, String> args = new HashMap<>();
     HashMap<String, String> unparsed = new HashMap<>();
 
@@ -19,7 +19,7 @@ public static Map<String, String> extractPlanProduct(SparkPlan plan) {
       Object obj = it.next();
       String key = String.format("_dd.unknown_key.%d", i);
 
-      String val = CommonSparkPlanUtils.parsePlanProduct(obj);
+      String val = parsePlanProduct(obj);
       if (val != null) {
         args.put(key, val);
       } else {

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

@@ -21,6 +21,7 @@ public class Spark213Instrumentation extends AbstractSparkInstrumentation {
   public String[] helperClassNames() {
     return new String[] {
       packageName + ".AbstractDatadogSparkListener",
+      packageName + ".AbstractSparkPlanUtils",
       packageName + ".DatabricksParentContext",
       packageName + ".OpenlineageParentContext",
       packageName + ".DatadogSpark213Listener",
@@ -31,8 +32,7 @@ public String[] helperClassNames() {
       packageName + ".SparkSQLUtils",
       packageName + ".SparkSQLUtils$SparkPlanInfoForStage",
       packageName + ".SparkSQLUtils$AccumulatorWithStage",
-      packageName + ".Spark213PlanUtils",
-      packageName + ".CommonSparkPlanUtils",
+      packageName + ".Spark213PlanUtils"
     };
   }
 
@@ -111,13 +111,13 @@ public static void exit(
         @Advice.Return(readOnly = false) SparkPlanInfo planInfo,
         @Advice.Argument(0) SparkPlan plan) {
       if (planInfo.metadata().size() == 0) {
+        Spark213PlanUtils planUtils = new Spark213PlanUtils();
         planInfo =
             new SparkPlanInfo(
                 planInfo.nodeName(),
                 planInfo.simpleString(),
                 planInfo.children(),
-                HashMap.from(
-                    JavaConverters.asScala(Spark213PlanUtils.extractPlanProduct(plan)).toList()),
+                HashMap.from(JavaConverters.asScala(planUtils.extractPlanProduct(plan)).toList()),
                 planInfo.metrics());
       }
     }

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

@@ -3,13 +3,13 @@
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
-import org.apache.spark.sql.execution.SparkPlan;
+import org.apache.spark.sql.catalyst.trees.TreeNode;
 import scala.collection.JavaConverters;
 
 // An extension of how Spark translates `SparkPlan`s to `SparkPlanInfo`, see here:
 // https://github.com/apache/spark/blob/v3.5.0/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlanInfo.scala#L54
-public class Spark213PlanUtils {
-  public static Map<String, String> extractPlanProduct(SparkPlan plan) {
+public class Spark213PlanUtils extends AbstractSparkPlanUtils {
+  public Map<String, String> extractPlanProduct(TreeNode plan) {
     HashMap<String, String> args = new HashMap<>();
     HashMap<String, String> unparsed = new HashMap<>();
 
@@ -19,7 +19,7 @@ public static Map<String, String> extractPlanProduct(SparkPlan plan) {
       Object obj = it.next();
       String key = plan.productElementName(i);
 
-      String val = CommonSparkPlanUtils.parsePlanProduct(obj);
+      String val = parsePlanProduct(obj);
       if (val != null) {
         args.put(key, val);
       } else {

…entation/spark/CommonSparkPlanUtils.java …tation/spark/AbstractSparkPlanUtils.javadd-java-agent/instrumentation/spark/src/main/java/datadog/trace/instrumentation/spark/CommonSparkPlanUtils.java renamed to dd-java-agent/instrumentation/spark/src/main/java/datadog/trace/instrumentation/spark/AbstractSparkPlanUtils.java dd-java-agent/instrumentation/spark/src/main/java/datadog/trace/instrumentation/spark/CommonSparkPlanUtils.java renamed to dd-java-agent/instrumentation/spark/src/main/java/datadog/trace/instrumentation/spark/AbstractSparkPlanUtils.java

@@ -1,12 +1,16 @@
 package datadog.trace.instrumentation.spark;
 
 import java.util.ArrayList;
+import java.util.Map;
 import org.apache.spark.sql.catalyst.plans.QueryPlan;
+import org.apache.spark.sql.catalyst.trees.TreeNode;
 import scala.Option;
 import scala.collection.Iterable;
 
-public class CommonSparkPlanUtils {
-  public static String parsePlanProduct(Object value) {
+public abstract class AbstractSparkPlanUtils {
+  public abstract Map<String, String> extractPlanProduct(TreeNode node);
+
+  public String parsePlanProduct(Object value) {
     if (value == null) {
       return "null";
     } else if (value instanceof Iterable) {

dd-java-agent/instrumentation/spark/src/testFixtures/groovy/datadog/trace/instrumentation/spark/AbstractSpark24SqlTest.groovy

@@ -183,13 +183,13 @@ abstract class AbstractSpark24SqlTest extends InstrumentationSpecification {
     var simpleString = actual["nodeDetailString"]
     var values = []
     var child = "N/A"
-    actual["meta"].each() { key, value ->
+    actual["meta"].each { key, value ->
       if (key == "_dd.unparsed") {
         values.add("\"_dd.unparsed\": \"any\"")
         child = value
       } else if (value instanceof List) {
         var list = []
-        value.each() { it ->
+        value.each { it ->
           list.add("\"$it\"")
         }
         def prettyList = "[\n    " + list.join(", \n    ") + "\n  ]"
@@ -201,7 +201,7 @@ abstract class AbstractSpark24SqlTest extends InstrumentationSpecification {
         values.add("\"$key\": \"$value\"")
       }
     }
-    values.sort() { it }
+    values.sort { it }
     def prettyValues = "\n\"meta\": {\n  " + values.join(", \n  ") + "\n},"
     if (values.size() == 1) {
       prettyValues = "\n\"meta\": {" + values.join(", ") + "},"