[hotfix] Pull legacyUidsEnabled as an abstract method to CommonExecTableSourceScan

Author: dawidwys

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecTableSourceScan.java

@@ -124,7 +124,7 @@ public String getDynamicFilteringDataListenerID() {
     protected Transformation<RowData> translateToPlanInternal(
             PlannerBase planner, ExecNodeConfig config) {
         final Transformation<RowData> transformation =
-                super.createTransformation(planner, config, false);
+                super.translateToPlanInternal(planner, config);
         // the boundedness has been checked via the runtime provider already, so we can safely
         // declare all legacy transformations as bounded to make the stream graph generator happy
         ExecNodeUtil.makeLegacySourceTransformationsBounded(transformation);
@@ -164,6 +164,11 @@ public Transformation<RowData> createInputFormatTransformation(
         return env.addSource(function, operatorName, outputTypeInfo).getTransformation();
     }
 
+    @Override
+    protected final boolean legacyUidsEnabled() {
+        return false;
+    }
+
     public BatchExecTableSourceScan copyAndRemoveInputs() {
         BatchExecTableSourceScan tableSourceScan =
                 new BatchExecTableSourceScan(

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecTableSourceScan.java

@@ -112,8 +112,9 @@ public DynamicTableSourceSpec getTableSourceSpec() {
         return tableSourceSpec;
     }
 
-    protected Transformation<RowData> createTransformation(
-            PlannerBase planner, ExecNodeConfig config, boolean legacyUidsEnabled) {
+    @Override
+    protected Transformation<RowData> translateToPlanInternal(
+            PlannerBase planner, ExecNodeConfig config) {
         final Transformation<RowData> sourceTransform;
         final StreamExecutionEnvironment env = planner.getExecEnv();
         final TransformationMetadata metadata =
@@ -186,15 +187,15 @@ protected Transformation<RowData> createTransformation(
                     ((DataStreamScanProvider) provider)
                             .produceDataStream(createProviderContext(metadata, config), env)
                             .getTransformation();
-            if (legacyUidsEnabled) {
+            if (legacyUidsEnabled()) {
                 metadata.fill(sourceTransform);
             }
             sourceTransform.setOutputType(outputTypeInfo);
         } else if (provider instanceof TransformationScanProvider) {
             sourceTransform =
                     ((TransformationScanProvider) provider)
                             .createTransformation(createProviderContext(metadata, config));
-            if (legacyUidsEnabled) {
+            if (legacyUidsEnabled()) {
                 metadata.fill(sourceTransform);
             }
             sourceTransform.setOutputType(outputTypeInfo);
@@ -369,4 +370,6 @@ protected abstract Transformation<RowData> createInputFormatTransformation(
             InputFormat<RowData, ?> inputFormat,
             InternalTypeInfo<RowData> outputTypeInfo,
             String operatorName);
+
+    protected abstract boolean legacyUidsEnabled();
 }

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecTableSourceScan.java

@@ -25,9 +25,7 @@
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.connector.source.ScanTableSource;
 import org.apache.flink.table.data.RowData;
-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.ExecNodeConfig;
 import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeContext;
 import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeMetadata;
 import org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecTableSourceScan;
@@ -94,12 +92,6 @@ public StreamExecTableSourceScan(
                 description);
     }
 
-    @Override
-    protected Transformation<RowData> translateToPlanInternal(
-            PlannerBase planner, ExecNodeConfig config) {
-        return createTransformation(planner, config, legacyUidsEnabled());
-    }
-
     @Override
     public Transformation<RowData> createInputFormatTransformation(
             StreamExecutionEnvironment env,
@@ -111,7 +103,8 @@ public Transformation<RowData> createInputFormatTransformation(
         return env.createInput(inputFormat, outputTypeInfo).name(operatorName).getTransformation();
     }
 
-    private boolean legacyUidsEnabled() {
+    @Override
+    protected final boolean legacyUidsEnabled() {
         return getVersion() == 1;
     }
 }