[FLINK-22559][table-planner] The consumed DataType of ExecSink should only consider physical columns

This closes apache#15864.

Author: leonardBang

flink-connector-kafka/src/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.