twalthr commented on code in PR #23412:
URL: https://github.com/apache/flink/pull/23412#discussion_r1326845652


##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/DynamicTableSourceSpec.java:
##########
@@ -95,29 +92,19 @@ private DynamicTableSource getTableSource(FlinkContext 
context, FlinkTypeFactory
                     FactoryUtil.createDynamicTableSource(
                             factory,
                             contextResolvedTable.getIdentifier(),
-                            resolvedCatalogTable,
+                            contextResolvedTable.getResolvedTable(),
                             loadOptionsFromCatalogTable(contextResolvedTable, 
context),
                             context.getTableConfig(),
                             context.getClassLoader(),
                             contextResolvedTable.isTemporary());
-            // validate DynamicSource and apply Metadata
-            DynamicSourceUtils.prepareDynamicSource(
-                    contextResolvedTable.getIdentifier().toString(),
-                    resolvedCatalogTable,
-                    tableSource,
-                    false,
-                    context.getTableConfig().getConfiguration());
 
             if (sourceAbilities != null) {
-                //  Note: use DynamicSourceUtils.createProducedType to produce 
the type info so that
-                //  keep consistent with sql2Rel phase which also called the 
method producing
-                //  deterministic format (PHYSICAL COLUMNS + METADATA COLUMNS) 
when converts a given
-                //  DynamicTableSource to a RelNode.
-                // TODO should do a refactor(e.g., add serialized input type 
info into each
-                //  SourceAbilitySpec so as to avoid this implicit logic 
dependency)
                 RowType newProducedType =
-                        DynamicSourceUtils.createProducedType(
-                                contextResolvedTable.getResolvedSchema(), 
tableSource);
+                        (RowType)
+                                contextResolvedTable
+                                        .getResolvedSchema()
+                                        .toSourceRowDataType()
+                                        .getLogicalType();

Review Comment:
   By just looking at the code, I would say `newProducedType` should be 
initialized with only physical columns. Not sure what the tests say to this. 
But `toSourceRowDataType` includes also computed column which makes no sense 
for the DynamicTableSource as the source doesn't know this concept. After the 
spec is applied in the loop, the type changes for `PHYSICAL COLUMNS + METADATA 
COLUMNS`. In any case, the producedType should always contain what comes out of 
the source.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Reply via email to