[FLINK-38796][table] UnresolvedMetadataColumn misses type during serializing to string

Author: snuyanzin

flink-python/pyflink/table/tests/test_schema.py

@@ -51,9 +51,9 @@ def test_schema_basic(self):
   `id` INT NOT NULL,
   `counter` INT NOT NULL,
   `payload` [ROW<name STRING, age INT, flag BOOLEAN>],
-  `topic` METADATA VIRTUAL,
+  `topic` STRING METADATA VIRTUAL,
   `ts` AS [orig_ts - INTERVAL '60' MINUTE],
-  `orig_ts` METADATA FROM 'timestamp',
+  `orig_ts` TIMESTAMP(3) METADATA FROM 'timestamp',
   `proctime` AS [PROCTIME()],
   WATERMARK FOR `ts` AS [ts - INTERVAL '5' SECOND],
   CONSTRAINT `primary_constraint` PRIMARY KEY (`id`) NOT ENFORCED

flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/SchemaResolutionTest.java

@@ -388,9 +388,9 @@ void testUnresolvedSchemaString() {
                                 + "  `id` INT NOT NULL COMMENT 'people id',\n"
                                 + "  `counter` INT NOT NULL,\n"
                                 + "  `payload` [ROW<name STRING, age INT, flag BOOLEAN>],\n"
-                                + "  `topic` METADATA VIRTUAL COMMENT 'kafka topic',\n"
+                                + "  `topic` STRING METADATA VIRTUAL COMMENT 'kafka topic',\n"
                                 + "  `ts` AS [orig_ts - INTERVAL '60' MINUTE] COMMENT 'rowtime',\n"
-                                + "  `orig_ts` METADATA FROM 'timestamp' COMMENT 'the ''origin'' timestamp',\n"
+                                + "  `orig_ts` TIMESTAMP(3) METADATA FROM 'timestamp' COMMENT 'the ''origin'' timestamp',\n"
                                 + "  `proctime` AS [PROCTIME()],\n"
                                 + "  WATERMARK FOR `ts` AS [ts - INTERVAL '5' SECOND],\n"
                                 + "  CONSTRAINT `primary_constraint` PRIMARY KEY (`id`) NOT ENFORCED,\n"

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/Schema.java

@@ -948,8 +948,7 @@ public boolean isVirtual() {
         @Override
         public String toString() {
             final StringBuilder sb = new StringBuilder();
-            sb.append(super.toString());
-            sb.append(" METADATA");
+            sb.append(String.format("%s %s METADATA", super.toString(), dataType.toString()));
             if (metadataKey != null) {
                 sb.append(" FROM '");
                 sb.append(EncodingUtils.escapeSingleQuotes(metadataKey));

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlMaterializedTableNodeToOperationConverterTest.java

@@ -952,7 +952,7 @@ private static Collection<TestSpec> alterSuccessCase() {
                                 + "  `c1` AS [CURRENT_TIMESTAMP],\n"
                                 + "  `a` BIGINT NOT NULL,\n"
                                 + "  `b` STRING,\n"
-                                + "  `topic` METADATA VIRTUAL COMMENT 'kafka topic',\n"
+                                + "  `topic` STRING METADATA VIRTUAL COMMENT 'kafka topic',\n"
                                 + "  `c` INT,\n"
                                 + "  `d` STRING,\n"
                                 + "  WATERMARK FOR `c1` AS [`c1` - INTERVAL '1' SECOND],\n"