[FLINK-21679][table] Set output type for transformations from SourceP…

…rovider and DataStreamScanProvider in CommonExecTableSourceScan This closes apache#15120
silenceland · Mar 10, 2021 · d97c68b · d97c68b
1 parent a3ffaa8
commit d97c68b
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 6 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
@@ -85,10 +85,14 @@ protected Transformation<RowData> translateToPlanInternal(PlannerBase planner) {
         } else if (provider instanceof SourceProvider) {
             Source<RowData, ?, ?> source = ((SourceProvider) provider).createSource();
             // TODO: Push down watermark strategy to source scan
-            return env.fromSource(source, WatermarkStrategy.noWatermarks(), operatorName)
+            return env.fromSource(
+                            source, WatermarkStrategy.noWatermarks(), operatorName, outputTypeInfo)
                     .getTransformation();
         } else if (provider instanceof DataStreamScanProvider) {
-            return ((DataStreamScanProvider) provider).produceDataStream(env).getTransformation();
+            Transformation<RowData> transformation =
+                    ((DataStreamScanProvider) provider).produceDataStream(env).getTransformation();
+            transformation.setOutputType(outputTypeInfo);
+            return transformation;
         } else {
             throw new UnsupportedOperationException(
                     provider.getClass().getSimpleName() + " is unsupported now.");

diff --git a/...planner-blink/src/test/java/org/apache/flink/table/planner/factories/TestFileFactory.java b/...planner-blink/src/test/java/org/apache/flink/table/planner/factories/TestFileFactory.java
@@ -31,7 +31,6 @@
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.table.connector.ChangelogMode;
 import org.apache.flink.table.connector.sink.DynamicTableSink;
 import org.apache.flink.table.connector.sink.SinkProvider;
@@ -45,7 +44,6 @@
 import org.apache.flink.table.factories.DynamicTableSinkFactory;
 import org.apache.flink.table.factories.DynamicTableSourceFactory;
 import org.apache.flink.table.filesystem.FileSystemOptions;
-import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
 
 import java.io.BufferedReader;
 import java.io.IOException;
@@ -198,7 +196,11 @@ public void close() throws IOException {
 
         @Override
         public TypeInformation<RowData> getProducedType() {
-            return InternalTypeInfo.ofFields(DataTypes.STRING().getLogicalType());
+            // For ScanTableSource, the output type is determined by the planner,
+            // and the result of this method will not be used.
+            // The purpose of returning null is to verify that the planner can
+            // handle the output type correctly.
+            return null;
         }
     }
 

diff --git a/...-blink/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java b/...-blink/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java
@@ -725,7 +725,7 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderCon
                             @Override
                             public DataStream<RowData> produceDataStream(
                                     StreamExecutionEnvironment execEnv) {
-                                return execEnv.addSource(function, type);
+                                return execEnv.addSource(function);
                             }
 
                             @Override