[#4845] fix(core): Fix bugs when updating metalake or catalog with nullable comment.

core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaMapper.java

@@ -150,7 +150,8 @@ CatalogPO selectCatalogMetaByMetalakeIdAndName(
           + " AND metalake_id = #{oldCatalogMeta.metalakeId}"
           + " AND type = #{oldCatalogMeta.type}"
           + " AND provider = #{oldCatalogMeta.provider}"
-          + " AND catalog_comment = #{oldCatalogMeta.catalogComment}"
+          + " AND (catalog_comment = #{oldCatalogMeta.catalogComment} "
+          + "   OR (catalog_comment IS NULL and #{oldCatalogMeta.catalogComment} IS NULL))"
           + " AND properties = #{oldCatalogMeta.properties}"
           + " AND audit_info = #{oldCatalogMeta.auditInfo}"
           + " AND current_version = #{oldCatalogMeta.currentVersion}"

core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaMapper.java

@@ -134,7 +134,8 @@ public interface MetalakeMetaMapper {
           + " last_version = #{newMetalakeMeta.lastVersion}"
           + " WHERE metalake_id = #{oldMetalakeMeta.metalakeId}"
           + " AND metalake_name = #{oldMetalakeMeta.metalakeName}"
-          + " AND metalake_comment = #{oldMetalakeMeta.metalakeComment}"
+          + " AND (metalake_comment = #{oldMetalakeMeta.metalakeComment} "
+          + "  OR (metalake_comment IS NULL and #{oldMetalakeMeta.metalakeComment} IS NULL))"
           + " AND properties = #{oldMetalakeMeta.properties}"
           + " AND audit_info = #{oldMetalakeMeta.auditInfo}"
           + " AND schema_version = #{oldMetalakeMeta.schemaVersion}"

core/src/test/java/org/apache/gravitino/storage/relational/TestJDBCBackend.java

@@ -84,6 +84,7 @@
 import org.apache.gravitino.utils.NamespaceUtil;
 import org.apache.ibatis.session.SqlSession;
 import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -434,6 +435,96 @@ public void testUpdateAlreadyExistsException() throws IOException {
                 e -> createTopicEntity(topicCopy.id(), topicCopy.namespace(), "topic", auditInfo)));
   }
 
+  @Test
+  void testUpdateMetalakeWithNullableComment() throws IOException {
+    AuditInfo auditInfo =
+        AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+
+    BaseMetalake metalake =
+        BaseMetalake.builder()
+            .withId(RandomIdGenerator.INSTANCE.nextId())
+            .withName("metalake" + RandomIdGenerator.INSTANCE.nextId())
+            .withAuditInfo(auditInfo)
+            .withComment(null)
+            .withProperties(null)
+            .withVersion(SchemaVersion.V_0_1)
+            .build();
+
+    backend.insert(metalake, false);
+
+    backend.update(
+        metalake.nameIdentifier(),
+        Entity.EntityType.METALAKE,
+        e ->
+            BaseMetalake.builder()
+                .withId(metalake.id())
+                .withName(metalake.name())
+                .withAuditInfo(auditInfo)
+                .withComment("comment")
+                .withProperties(metalake.properties())
+                .withVersion(metalake.getVersion())
+                .build());
+
+    BaseMetalake updatedMetalake =
+        backend.get(metalake.nameIdentifier(), Entity.EntityType.METALAKE);
+    Assertions.assertNotNull(updatedMetalake.comment());
+
+    backend.delete(metalake.nameIdentifier(), Entity.EntityType.METALAKE, false);
+  }
+
+  @Test
+  void testUpdateCatalogWithNullableComment() throws IOException {
+    AuditInfo auditInfo =
+        AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+
+    String metalakeName = "metalake" + RandomIdGenerator.INSTANCE.nextId();
+    BaseMetalake metalake =
+        BaseMetalake.builder()
+            .withId(RandomIdGenerator.INSTANCE.nextId())
+            .withName(metalakeName)
+            .withAuditInfo(auditInfo)
+            .withComment("")
+            .withProperties(null)
+            .withVersion(SchemaVersion.V_0_1)
+            .build();
+    backend.insert(metalake, false);
+
+    CatalogEntity catalog =
+        CatalogEntity.builder()
+            .withId(RandomIdGenerator.INSTANCE.nextId())
+            .withNamespace(NamespaceUtil.ofCatalog(metalakeName))
+            .withName("catalog")
+            .withAuditInfo(auditInfo)
+            .withComment(null)
+            .withProperties(null)
+            .withType(Catalog.Type.RELATIONAL)
+            .withProvider("test")
+            .build();
+
+    backend.insert(catalog, false);
+
+    backend.update(
+        catalog.nameIdentifier(),
+        Entity.EntityType.CATALOG,
+        e ->
+            CatalogEntity.builder()
+                .withId(catalog.id())
+                .withNamespace(catalog.namespace())
+                .withName(catalog.name())
+                .withAuditInfo(auditInfo)
+                .withComment("comment")
+                .withProperties(catalog.getProperties())
+                .withType(Catalog.Type.RELATIONAL)
+                .withProvider("test")
+                .build());
+
+    CatalogEntity updatedCatalog = backend.get(catalog.nameIdentifier(), Entity.EntityType.CATALOG);
+    Assertions.assertNotNull(updatedCatalog.getComment());
+
+    backend.delete(catalog.nameIdentifier(), Entity.EntityType.CATALOG, false);
+    backend.delete(metalake.nameIdentifier(), Entity.EntityType.METALAKE, false);
+  }
+
   @Test
   public void testMetaLifeCycleFromCreationToDeletion() throws IOException {
     AuditInfo auditInfo =

integration-test/src/test/java/org/apache/gravitino/integration/test/client/CatalogIT.java

@@ -26,6 +26,7 @@
 import java.util.Map;
 import org.apache.commons.lang.ArrayUtils;
 import org.apache.gravitino.Catalog;
+import org.apache.gravitino.CatalogChange;
 import org.apache.gravitino.client.GravitinoMetalake;
 import org.apache.gravitino.integration.test.container.ContainerSuite;
 import org.apache.gravitino.integration.test.container.HiveContainer;
@@ -278,4 +279,25 @@ public void testCreateCatalogWithPackage() {
     Assertions.assertTrue(
         exception.getMessage().contains("Invalid package path: /tmp/none_exist_path_to_package"));
   }
+
+  @Test
+  void testUpdateCatalogWithNullableComment() {
+    String catalogName = GravitinoITUtils.genRandomName("catalog");
+    Assertions.assertFalse(metalake.catalogExists(catalogName));
+
+    Map<String, String> properties = Maps.newHashMap();
+    properties.put("metastore.uris", hmsUri);
+    Catalog catalog =
+        metalake.createCatalog(catalogName, Catalog.Type.RELATIONAL, "hive", null, properties);
+    Assertions.assertTrue(metalake.catalogExists(catalogName));
+
+    Assertions.assertEquals(catalogName, catalog.name());
+    Assertions.assertEquals(null, catalog.comment());
+
+    Catalog updatedCatalog =
+        metalake.alterCatalog(catalogName, CatalogChange.updateComment("new catalog comment"));
+    Assertions.assertEquals("new catalog comment", updatedCatalog.comment());
+
+    metalake.dropCatalog(catalogName);
+  }
 }

integration-test/src/test/java/org/apache/gravitino/integration/test/client/MetalakeIT.java

@@ -194,6 +194,20 @@ public void testDropMetalakes() {
     assertFalse(client.dropMetalake(metalakeA.name()), "metalake should be non-existent");
   }
 
+  @Test
+  public void testUpdateMetalakeWithNullableComment() {
+    client.createMetalake(metalakeNameA, null, Collections.emptyMap());
+    GravitinoMetalake metalake = client.loadMetalake(metalakeNameA);
+    assertEquals(metalakeNameA, metalake.name());
+    assertEquals(null, metalake.comment());
+
+    MetalakeChange[] changes =
+        new MetalakeChange[] {MetalakeChange.updateComment("new metalake comment")};
+    GravitinoMetalake updatedMetalake = client.alterMetalake(metalakeNameA, changes);
+    assertEquals("new metalake comment", updatedMetalake.comment());
+    client.dropMetalake(metalakeNameA);
+  }
+
   public void dropMetalakes() {
     GravitinoMetalake[] metaLakes = client.listMetalakes();
     for (GravitinoMetalake metalake : metaLakes) {