[FLINK-22559][table-planner] The consumed DataType of ExecSink should…

… only consider physical columns
apache · May 8, 2021 · b2bc1f4 · b2bc1f4
1 parent f39af7e
commit b2bc1f4
Show file tree

Hide file tree

Showing 2 changed files with 1 addition and 3 deletions.
diff --git a/...c/test/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaTableITCase.java b/...c/test/java/org/apache/flink/streaming/connectors/kafka/table/UpsertKafkaTableITCase.java
@@ -23,7 +23,6 @@
 import org.apache.flink.table.utils.LegacyRowResource;
 import org.apache.flink.types.Row;
 
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -198,7 +197,6 @@ public void testSourceSinkWithKeyAndPartialValue() throws Exception {
     }
 
     @Test
-    @Ignore // FLINK-22559
     public void testKafkaSourceSinkWithKeyAndFullValue() throws Exception {
         // we always use a different topic name for each parameterized topic,
         // in order to make sure the topic can be created.

diff --git a/...k/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java b/...k/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
@@ -319,6 +319,6 @@ private InternalTypeInfo<RowData> getInputTypeInfo() {
     }
 
     private RowType getConsumedRowType(ResolvedSchema schema) {
-        return (RowType) schema.toSinkRowDataType().getLogicalType();
+        return (RowType) schema.toPhysicalRowDataType().getLogicalType();
     }
 }