[FLINK-21549][table-planner-blink] Remove DynamicTableSourceSpecJsonD…

…eserializer and set ClassLoader and Configuration to DynamicTableSourceSpec when creating ExecNodeGraph This closes apache#15062
alec-brooks · Mar 7, 2021 · df72f25 · df72f25
1 parent 9de6d8d
commit df72f25
Show file tree

Hide file tree

Showing 5 changed files with 21 additions and 68 deletions.
diff --git a/...java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecTableSourceScan.java b/...java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecTableSourceScan.java
@@ -33,15 +33,13 @@
 import org.apache.flink.table.planner.delegation.PlannerBase;
 import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
 import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase;
-import org.apache.flink.table.planner.plan.nodes.exec.serde.DynamicTableSourceSpecJsonDeserializer;
 import org.apache.flink.table.planner.plan.nodes.exec.spec.DynamicTableSourceSpec;
 import org.apache.flink.table.runtime.connector.source.ScanRuntimeProviderContext;
 import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.RowType;
 
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 
 import java.util.Collections;
 
@@ -52,7 +50,6 @@ public abstract class CommonExecTableSourceScan extends ExecNodeBase<RowData> {
     public static final String FIELD_NAME_SCAN_TABLE_SOURCE = "scanTableSource";
 
     @JsonProperty(FIELD_NAME_SCAN_TABLE_SOURCE)
-    @JsonDeserialize(using = DynamicTableSourceSpecJsonDeserializer.class)
     private final DynamicTableSourceSpec tableSourceSpec;
 
     protected CommonExecTableSourceScan(
@@ -64,6 +61,10 @@ protected CommonExecTableSourceScan(
         this.tableSourceSpec = tableSourceSpec;
     }
 
+    public DynamicTableSourceSpec getTableSourceSpec() {
+        return tableSourceSpec;
+    }
+
     @Override
     protected Transformation<RowData> translateToPlanInternal(PlannerBase planner) {
         final StreamExecutionEnvironment env = planner.getExecEnv();
@@ -103,8 +104,4 @@ protected abstract Transformation<RowData> createInputFormatTransformation(
             InputFormat<RowData, ?> inputFormat,
             InternalTypeInfo<RowData> outputTypeInfo,
             String name);
-
-    public DynamicTableSourceSpec getTableSourceSpec() {
-        return tableSourceSpec;
-    }
 }
diff --git a/...che/flink/table/planner/plan/nodes/exec/serde/DynamicTableSourceSpecJsonDeserializer.java b/...che/flink/table/planner/plan/nodes/exec/serde/DynamicTableSourceSpecJsonDeserializer.java
diff --git a/.../org/apache/flink/table/planner/plan/nodes/exec/serde/ExecNodeGraphJsonPlanGenerator.java b/.../org/apache/flink/table/planner/plan/nodes/exec/serde/ExecNodeGraphJsonPlanGenerator.java
@@ -98,7 +98,7 @@ public static ExecNodeGraph generateExecNodeGraph(String jsonPlan, SerdeContext
         mapper.registerModule(module);
 
         final JsonPlanGraph jsonPlanGraph = mapper.readValue(jsonPlan, JsonPlanGraph.class);
-        return jsonPlanGraph.convertToExecNodeGraph();
+        return jsonPlanGraph.convertToExecNodeGraph(serdeCtx);
     }
 
     private static void registerSerializers(SimpleModule module) {
@@ -237,7 +237,7 @@ public void visit(ExecNode<?> node) {
             return new JsonPlanGraph(execGraph.getFlinkVersion(), allNodes, allEdges);
         }
 
-        public ExecNodeGraph convertToExecNodeGraph() {
+        public ExecNodeGraph convertToExecNodeGraph(SerdeContext serdeCtx) {
             Map<Integer, ExecNode<?>> idToExecNodes = new HashMap<>();
             for (ExecNode<?> execNode : nodes) {
                 int id = execNode.getId();
@@ -248,6 +248,10 @@ public ExecNodeGraph convertToExecNodeGraph() {
                                     id, execNode.getDescription()));
                 }
                 if (execNode instanceof StreamExecTableSourceScan) {
+                    DynamicTableSourceSpec tableSourceSpec =
+                            ((StreamExecTableSourceScan) execNode).getTableSourceSpec();
+                    tableSourceSpec.setReadableConfig(serdeCtx.getConfiguration());
+                    tableSourceSpec.setClassLoader(serdeCtx.getClassLoader());
                     applyProjectionPushDown((StreamExecTableSourceScan) execNode);
                 }
                 idToExecNodes.put(id, execNode);

diff --git a/...c/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/CatalogTableSpecBase.java b/...c/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/CatalogTableSpecBase.java
@@ -86,6 +86,12 @@ public ClassLoader getClassLoader() {
         return classLoader;
     }
 
+    @VisibleForTesting
+    @JsonIgnore
+    public ReadableConfig getReadableConfig() {
+        return configuration;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {

diff --git a/...org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSourceSpecSerdeTest.java b/...org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSourceSpecSerdeTest.java
@@ -43,7 +43,7 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertNull;
 
 /** Tests for {@link DynamicTableSourceSpec} serialization and deserialization. */
 @RunWith(Parameterized.class)
@@ -62,8 +62,6 @@ public void testDynamicTableSourceSpecSerde() throws IOException {
                         FlinkSqlOperatorTable.instance());
         ObjectMapper mapper = JsonSerdeUtil.createObjectMapper(serdeCtx);
         SimpleModule module = new SimpleModule();
-        module.addDeserializer(
-                DynamicTableSourceSpec.class, new DynamicTableSourceSpecJsonDeserializer());
         mapper.registerModule(module);
         StringWriter writer = new StringWriter(100);
         try (JsonGenerator gen = mapper.getFactory().createGenerator(writer)) {
@@ -72,7 +70,10 @@ public void testDynamicTableSourceSpecSerde() throws IOException {
         String json = writer.toString();
         DynamicTableSourceSpec actual = mapper.readValue(json, DynamicTableSourceSpec.class);
         assertEquals(spec, actual);
-        assertSame(classLoader, actual.getClassLoader());
+        assertNull(actual.getReadableConfig());
+        actual.setReadableConfig(serdeCtx.getConfiguration());
+        assertNull(actual.getClassLoader());
+        actual.setClassLoader(serdeCtx.getClassLoader());
         assertNotNull(actual.getScanTableSource());
     }