[hive] hive create format table support query field comments. (#5070)

Author: zhuangchong

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FormatCatalogTable.java

@@ -26,6 +26,7 @@
 import org.apache.flink.table.connector.source.DynamicTableSource;
 import org.apache.flink.table.factories.DynamicTableFactory;
 import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.logical.RowType;
 
 import java.util.Collections;
 import java.util.HashMap;
@@ -57,9 +58,18 @@ public FormatTable table() {
 
     @Override
     public Schema getUnresolvedSchema() {
-        return Schema.newBuilder()
-                .fromRowDataType(fromLogicalToDataType(toLogicalType(table.rowType())))
-                .build();
+        final Schema.Builder builder = Schema.newBuilder();
+        RowType logicalType = toLogicalType(table.rowType());
+        logicalType
+                .getFields()
+                .forEach(
+                        field -> {
+                            builder.column(field.getName(), fromLogicalToDataType(field.getType()));
+                            if (field.getDescription().isPresent()) {
+                                builder.withComment(field.getDescription().get());
+                            }
+                        });
+        return builder.build();
     }
 
     @Override

paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveTableUtils.java

@@ -21,14 +21,11 @@
 import org.apache.paimon.options.Options;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.table.FormatTable.Format;
-import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.Pair;
 
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
 import org.apache.hadoop.hive.metastore.api.SerDeInfo;
 import org.apache.hadoop.hive.metastore.api.Table;
-import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -96,24 +93,11 @@ private static List<String> getFieldNames(List<FieldSchema> fieldSchemas) {
     public static RowType createRowType(Table table) {
         List<FieldSchema> allCols = new ArrayList<>(table.getSd().getCols());
         allCols.addAll(table.getPartitionKeys());
-        Pair<String[], DataType[]> columnInformation = extractColumnInformation(allCols);
-        return RowType.builder()
-                .fields(columnInformation.getRight(), columnInformation.getLeft())
-                .build();
-    }
 
-    private static Pair<String[], DataType[]> extractColumnInformation(List<FieldSchema> allCols) {
-        String[] colNames = new String[allCols.size()];
-        DataType[] colTypes = new DataType[allCols.size()];
-
-        for (int i = 0; i < allCols.size(); i++) {
-            FieldSchema fs = allCols.get(i);
-            colNames[i] = fs.getName();
-            colTypes[i] =
-                    HiveTypeUtils.toPaimonType(
-                            TypeInfoUtils.getTypeInfoFromTypeString(fs.getType()));
+        RowType.Builder builder = RowType.builder();
+        for (FieldSchema fs : allCols) {
+            builder.field(fs.getName(), HiveTypeUtils.toPaimonType(fs.getType()), fs.getComment());
         }
-
-        return Pair.of(colNames, colTypes);
+        return builder.build();
     }
 }

paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogFormatTableITCaseBase.java

@@ -42,6 +42,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.stream.Collectors;
 
 import static org.apache.paimon.options.CatalogOptions.FORMAT_TABLE_ENABLED;
@@ -119,46 +120,56 @@ public void evaluate() throws Throwable {
 
     @Test
     public void testCsvFormatTable() throws Exception {
-        hiveShell.execute("CREATE TABLE csv_table (a INT, b STRING)");
+        hiveShell.execute(
+                "CREATE TABLE csv_table (a INT COMMENT 'comment a', b STRING COMMENT 'comment b')");
         doTestFormatTable("csv_table");
     }
 
     @Test
     public void testCsvFormatTableWithDelimiter() throws Exception {
         hiveShell.execute(
-                "CREATE TABLE csv_table_delimiter (a INT, b STRING) ROW FORMAT DELIMITED FIELDS TERMINATED BY ';'");
+                "CREATE TABLE csv_table_delimiter (a INT COMMENT 'comment a', b STRING COMMENT 'comment b') ROW FORMAT DELIMITED FIELDS TERMINATED BY ';'");
         doTestFormatTable("csv_table_delimiter");
     }
 
     @Test
     public void testPartitionTable() throws Exception {
-        hiveShell.execute("CREATE TABLE partition_table (a INT) PARTITIONED BY (b STRING)");
+        hiveShell.execute(
+                "CREATE TABLE partition_table (a INT COMMENT 'comment a') PARTITIONED BY (b STRING COMMENT 'comment b')");
         doTestFormatTable("partition_table");
     }
 
     @Test
     public void testFlinkCreateCsvFormatTable() throws Exception {
         tEnv.executeSql(
-                        "CREATE TABLE flink_csv_table (a INT, b STRING) with ('type'='format-table', 'file.format'='csv')")
+                        "CREATE TABLE flink_csv_table (a INT COMMENT 'comment a', b STRING COMMENT 'comment b') with ('type'='format-table', 'file.format'='csv')")
                 .await();
         doTestFormatTable("flink_csv_table");
     }
 
     @Test
     public void testFlinkCreateFormatTableWithDelimiter() throws Exception {
         tEnv.executeSql(
-                "CREATE TABLE flink_csv_table_delimiter (a INT, b STRING) with ('type'='format-table', 'file.format'='csv', 'field-delimiter'=';')");
+                "CREATE TABLE flink_csv_table_delimiter (a INT COMMENT 'comment a', b STRING COMMENT 'comment b') with ('type'='format-table', 'file.format'='csv', 'field-delimiter'=';')");
         doTestFormatTable("flink_csv_table_delimiter");
     }
 
     @Test
     public void testFlinkCreatePartitionTable() throws Exception {
         tEnv.executeSql(
-                "CREATE TABLE flink_partition_table (a INT,b STRING) PARTITIONED BY (b) with ('type'='format-table', 'file.format'='csv')");
+                "CREATE TABLE flink_partition_table (a INT COMMENT 'comment a', b STRING COMMENT 'comment b') PARTITIONED BY (b) with ('type'='format-table', 'file.format'='csv')");
         doTestFormatTable("flink_partition_table");
     }
 
     private void doTestFormatTable(String tableName) throws Exception {
+        List<String> descResult =
+                collect("DESC " + tableName).stream()
+                        .map(Objects::toString)
+                        .collect(Collectors.toList());
+        assertThat(descResult)
+                .containsExactly(
+                        "+I[a, INT, true, null, null, null, comment a]",
+                        "+I[b, STRING, true, null, null, null, comment b]");
         hiveShell.execute(
                 String.format("INSERT INTO %s VALUES (100, 'Hive'), (200, 'Table')", tableName));
         assertThat(collect(String.format("SELECT * FROM %s", tableName)))