[Improve][Iceberg] Support table comment for catalog (#7936)

Author: hailin0

seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/catalog/IcebergCatalog.java

@@ -63,6 +63,8 @@
 
 @Slf4j
 public class IcebergCatalog implements Catalog {
+    public static final String PROPS_TABLE_COMMENT = "comment";
+
     private final String catalogName;
     private final ReadonlyConfig readonlyConfig;
     private final IcebergCatalogLoader icebergCatalogLoader;
@@ -257,14 +259,17 @@ public CatalogTable toCatalogTable(Table icebergTable, TablePath tablePath) {
                 icebergTable.spec().fields().stream()
                         .map(PartitionField::name)
                         .collect(Collectors.toList());
-
+        String comment =
+                Optional.ofNullable(icebergTable.properties())
+                        .map(e -> e.get(PROPS_TABLE_COMMENT))
+                        .orElse(null);
         return CatalogTable.of(
                 org.apache.seatunnel.api.table.catalog.TableIdentifier.of(
                         catalogName, tablePath.getDatabaseName(), tablePath.getTableName()),
                 builder.build(),
                 icebergTable.properties(),
                 partitionKeys,
-                null,
+                comment,
                 catalogName);
     }
 

seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/utils/SchemaUtils.java

@@ -28,6 +28,7 @@
 import org.apache.seatunnel.api.table.catalog.exception.DatabaseNotExistException;
 import org.apache.seatunnel.api.table.catalog.exception.TableAlreadyExistException;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.connectors.seatunnel.iceberg.catalog.IcebergCatalog;
 import org.apache.seatunnel.connectors.seatunnel.iceberg.config.SinkConfig;
 import org.apache.seatunnel.connectors.seatunnel.iceberg.data.IcebergTypeMapper;
 import org.apache.seatunnel.connectors.seatunnel.iceberg.sink.schema.SchemaAddColumn;
@@ -105,6 +106,8 @@ public static Table autoCreateTable(
         SinkConfig config = new SinkConfig(readonlyConfig);
         // build auto create table
         Map<String, String> options = new HashMap<>(table.getOptions());
+        Optional.ofNullable(table.getComment())
+                .map(e -> options.put(IcebergCatalog.PROPS_TABLE_COMMENT, e));
         // override
         options.putAll(config.getAutoCreateProps());
         return createTable(catalog, toIcebergTableIdentifier(tablePath), config, schema, options);

seatunnel-connectors-v2/connector-iceberg/src/test/java/org/apache/seatunnel/connectors/seatunnel/iceberg/catalog/IcebergCatalogTest.java

@@ -194,7 +194,8 @@ CatalogTable buildAllTypesTable(TableIdentifier tableIdentifier) {
         TableSchema schema = builder.build();
         HashMap<String, String> options = new HashMap<>();
         options.put("write.parquet.compression-codec", "zstd");
+        options.put("comment", "test");
         return CatalogTable.of(
-                tableIdentifier, schema, options, Collections.singletonList("dt_col"), "null");
+                tableIdentifier, schema, options, Collections.singletonList("dt_col"), "test");
     }
 }