[lake/paimon] Support passing Fluss table comment through to the Paimon table (#1999)

Author: LiebingYu

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java

@@ -203,6 +203,10 @@ public static Schema toPaimonSchema(TableDescriptor tableDescriptor) {
                 .getCustomProperties()
                 .forEach((k, v) -> setFlussPropertyToPaimon(k, v, options));
         schemaBuilder.options(options.toMap());
+
+        // set comment
+        tableDescriptor.getComment().ifPresent(schemaBuilder::comment);
+
         return schemaBuilder.build();
     }
 

fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/LakeEnabledTableCreateITCase.java

@@ -151,6 +151,7 @@ void testCreateLakeEnabledTable() throws Exception {
                         .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
                         .customProperties(customProperties)
                         .distributedBy(BUCKET_NUM, "log_c1", "log_c2")
+                        .comment("test comment")
                         .build();
         TablePath logTablePath = TablePath.of(DATABASE, "log_table");
         admin.createTable(logTablePath, logTable, false).get();
@@ -902,6 +903,8 @@ private void verifyPaimonTable(
         // now, check schema
         RowType paimonRowType = paimonTable.rowType();
         assertThat(paimonRowType).isEqualTo(expectedRowType);
+
+        assertThat(paimonTable.comment()).isEqualTo(flussTable.getComment());
     }
 
     private TableDescriptor createTableDescriptor(