From f29cc7bea500b9e52d6515db98691976c1c5bca0 Mon Sep 17 00:00:00 2001
From: FANNG <xiaojing@datastrato.com>
Date: Wed, 24 Jan 2024 13:45:45 +0800
Subject: [PATCH] [#1546] fix(lakehouse-iceberg): fix create Iceberg table
 failed if comment is null (#1653)

### What changes were proposed in this pull request?
remove reserveProperties logic from CreateTableRequest since reserved
properties is handled by property system

### Why are the changes needed?

1. when build IcebergTable, comment equals to null is checked when
comment added to properties .
```
      if (null != comment) {
        icebergTable.properties.putIfAbsent(ICEBERG_COMMENT_FIELD_NAME, comment);
      }
```
2. when build CreateTableRequest, the previous implement remove all
reserved properties from property map including comment, and then add
comment to map, the bug happens, not checking the null, but this overall
code is redundant


Fix: #1546

### Does this PR introduce _any_ user-facing change?
no

### How was this patch tested?
add UT
---
 .../catalog/lakehouse/iceberg/IcebergTable.java   |  7 +------
 .../iceberg/ops/IcebergTableOpsHelper.java        |  8 --------
 .../lakehouse/iceberg/CatalogIcebergIT.java       | 15 +++++++++++++++
 3 files changed, 16 insertions(+), 14 deletions(-)

diff --git a/catalogs/catalog-lakehouse-iceberg/src/main/java/com/datastrato/gravitino/catalog/lakehouse/iceberg/IcebergTable.java b/catalogs/catalog-lakehouse-iceberg/src/main/java/com/datastrato/gravitino/catalog/lakehouse/iceberg/IcebergTable.java
index 6ed6de81731..238c1ed05ef 100644
--- a/catalogs/catalog-lakehouse-iceberg/src/main/java/com/datastrato/gravitino/catalog/lakehouse/iceberg/IcebergTable.java
+++ b/catalogs/catalog-lakehouse-iceberg/src/main/java/com/datastrato/gravitino/catalog/lakehouse/iceberg/IcebergTable.java
@@ -9,7 +9,6 @@
 import com.datastrato.gravitino.catalog.lakehouse.iceberg.converter.FromIcebergSortOrder;
 import com.datastrato.gravitino.catalog.lakehouse.iceberg.converter.ToIcebergPartitionSpec;
 import com.datastrato.gravitino.catalog.lakehouse.iceberg.converter.ToIcebergSortOrder;
-import com.datastrato.gravitino.catalog.lakehouse.iceberg.ops.IcebergTableOpsHelper;
 import com.datastrato.gravitino.catalog.rel.BaseTable;
 import com.datastrato.gravitino.meta.AuditInfo;
 import com.google.common.collect.Maps;
@@ -45,16 +44,12 @@ private IcebergTable() {}
 
   public CreateTableRequest toCreateTableRequest() {
     Schema schema = ConvertUtil.toIcebergSchema(this);
-
-    Map<String, String> resultProperties =
-        Maps.newHashMap(IcebergTableOpsHelper.removeReservedProperties(properties));
-    resultProperties.putIfAbsent(ICEBERG_COMMENT_FIELD_NAME, comment);
     CreateTableRequest.Builder builder =
         CreateTableRequest.builder()
             .withName(name)
             .withLocation(location)
             .withSchema(schema)
-            .setProperties(resultProperties)
+            .setProperties(properties)
             .withPartitionSpec(ToIcebergPartitionSpec.toPartitionSpec(schema, partitioning))
             .withWriteOrder(ToIcebergSortOrder.toSortOrder(schema, sortOrders));
     return builder.build();
diff --git a/catalogs/catalog-lakehouse-iceberg/src/main/java/com/datastrato/gravitino/catalog/lakehouse/iceberg/ops/IcebergTableOpsHelper.java b/catalogs/catalog-lakehouse-iceberg/src/main/java/com/datastrato/gravitino/catalog/lakehouse/iceberg/ops/IcebergTableOpsHelper.java
index 86d39b9716a..ca05aec8c74 100644
--- a/catalogs/catalog-lakehouse-iceberg/src/main/java/com/datastrato/gravitino/catalog/lakehouse/iceberg/ops/IcebergTableOpsHelper.java
+++ b/catalogs/catalog-lakehouse-iceberg/src/main/java/com/datastrato/gravitino/catalog/lakehouse/iceberg/ops/IcebergTableOpsHelper.java
@@ -28,9 +28,7 @@
 import com.google.common.collect.Lists;
 import java.util.Arrays;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
-import java.util.stream.Collectors;
 import javax.ws.rs.NotSupportedException;
 import lombok.Getter;
 import lombok.Setter;
@@ -309,12 +307,6 @@ public IcebergTableChange buildIcebergTableChanges(
     return icebergTableChange;
   }
 
-  public static Map<String, String> removeReservedProperties(Map<String, String> createProperties) {
-    return createProperties.entrySet().stream()
-        .filter(entry -> !IcebergReservedProperties.contains(entry.getKey()))
-        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-  }
-
   /**
    * Gravitino only supports single-level namespace storage management, which differs from Iceberg.
    * Therefore, we need to handle this difference here.
diff --git a/integration-test/src/test/java/com/datastrato/gravitino/integration/test/catalog/lakehouse/iceberg/CatalogIcebergIT.java b/integration-test/src/test/java/com/datastrato/gravitino/integration/test/catalog/lakehouse/iceberg/CatalogIcebergIT.java
index de771e688d4..a922af30260 100644
--- a/integration-test/src/test/java/com/datastrato/gravitino/integration/test/catalog/lakehouse/iceberg/CatalogIcebergIT.java
+++ b/integration-test/src/test/java/com/datastrato/gravitino/integration/test/catalog/lakehouse/iceberg/CatalogIcebergIT.java
@@ -326,6 +326,21 @@ void testOperationIcebergSchema() {
     Assertions.assertTrue(schemaNames.contains(schemaName));
   }
 
+  @Test
+  void testCreateTableWithNullComment() {
+    ColumnDTO[] columns = createColumns();
+    NameIdentifier tableIdentifier =
+        NameIdentifier.of(metalakeName, catalogName, schemaName, tableName);
+
+    TableCatalog tableCatalog = catalog.asTableCatalog();
+    Table createdTable =
+        tableCatalog.createTable(tableIdentifier, columns, null, null, null, null, null);
+    Assertions.assertNull(createdTable.comment());
+
+    Table loadTable = tableCatalog.loadTable(tableIdentifier);
+    Assertions.assertNull(loadTable.comment());
+  }
+
   @Test
   void testCreateAndLoadIcebergTable() {
     // Create table from Gravitino API