[apache#1434] fix(core): Fix the bug of rename entity (apache#1437)

### What changes were proposed in this pull request?

Use the `overwrite` strategy when changing the content of name-id
mappings in case of possible bugs.

### Why are the changes needed?

If the name already existed in the name-id mappings, it would encounter
problems.

Fix: apache#1434 

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

N/A.

### How was this patch tested?

Add UT and IT.

core/src/main/java/com/datastrato/gravitino/storage/kv/KvNameMappingService.java

@@ -96,8 +96,14 @@ public boolean updateName(String oldName, String newName) throws IOException {
 
                   // Delete old name --> id mapping
                   transactionalKvBackend.delete(nameByte);
-
-                  transactionalKvBackend.put(getNameKey(newName), oldIdValue, false);
+                  // In case there exists the mapping of new_name --> id, so we should use
+                  // the overwritten strategy. In the following scenario, we should use the
+                  // overwritten strategy:
+                  // 1. Create name1
+                  // 2. Delete name1
+                  // 3. Create name2
+                  // 4. Rename name2 -> name1
+                  transactionalKvBackend.put(getNameKey(newName), oldIdValue, true);
                   transactionalKvBackend.put(
                       oldIdValue, newName.getBytes(StandardCharsets.UTF_8), true);
                   return true;

core/src/main/java/com/datastrato/gravitino/storage/kv/TransactionalKvBackendImpl.java

@@ -129,8 +129,7 @@ public void put(byte[] key, byte[] value, boolean overwrite)
       throws IOException, EntityAlreadyExistsException {
     byte[] oldValue = get(key);
     if (oldValue != null && !overwrite) {
-      throw new EntityAlreadyExistsException(
-          "Key already exists: " + ByteUtils.formatByteArray(key));
+      throw new EntityAlreadyExistsException("Key already exists: " + Bytes.wrap(key));
     }
     putPairs
         .get()

core/src/test/java/com/datastrato/gravitino/storage/kv/TestKvEntityStorage.java

@@ -887,4 +887,137 @@ void testStorageLayoutVersion() throws IOException {
       Assertions.assertEquals(StorageLayoutVersion.V1, entityStore.storageLayoutVersion);
     }
   }
+
+  @Test
+  void testDeleteAndRename() throws IOException {
+    Config config = Mockito.mock(Config.class);
+    File baseDir = new File(System.getProperty("java.io.tmpdir"));
+    File file = Files.createTempDirectory(baseDir.toPath(), "test").toFile();
+    file.deleteOnExit();
+    Mockito.when(config.get(ENTITY_STORE)).thenReturn("kv");
+    Mockito.when(config.get(ENTITY_KV_STORE)).thenReturn(DEFAULT_ENTITY_KV_STORE);
+    Mockito.when(config.get(Configs.ENTITY_SERDE)).thenReturn("proto");
+    Mockito.when(config.get(ENTRY_KV_ROCKSDB_BACKEND_PATH)).thenReturn(file.getAbsolutePath());
+    Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
+    Mockito.when(config.get(KV_DELETE_AFTER_TIME)).thenReturn(20 * 60 * 1000L);
+
+    try (EntityStore store = EntityStoreFactory.createEntityStore(config)) {
+      store.initialize(config);
+      Assertions.assertTrue(store instanceof KvEntityStore);
+      store.setSerDe(EntitySerDeFactory.createEntitySerDe(config.get(Configs.ENTITY_SERDE)));
+
+      AuditInfo auditInfo =
+          new AuditInfo.Builder().withCreator("creator").withCreateTime(Instant.now()).build();
+
+      BaseMetalake metalake1 = createBaseMakeLake("metalake1", auditInfo);
+      BaseMetalake metalake2 = createBaseMakeLake("metalake2", auditInfo);
+      BaseMetalake metalake3 = createBaseMakeLake("metalake3", auditInfo);
+
+      store.put(metalake1);
+      store.put(metalake2);
+      store.put(metalake3);
+
+      store.delete(NameIdentifier.of("metalake1"), EntityType.METALAKE);
+      store.delete(NameIdentifier.of("metalake2"), EntityType.METALAKE);
+      store.delete(NameIdentifier.of("metalake3"), EntityType.METALAKE);
+
+      // Rename metalake1 --> metalake2
+      store.put(metalake1);
+      store.update(
+          NameIdentifier.of("metalake1"),
+          BaseMetalake.class,
+          EntityType.METALAKE,
+          e -> createBaseMakeLake("metalake2", (AuditInfo) e.auditInfo()));
+
+      // Rename metalake3 --> metalake1
+      store.put(metalake3);
+      store.update(
+          NameIdentifier.of("metalake3"),
+          BaseMetalake.class,
+          EntityType.METALAKE,
+          e -> createBaseMakeLake("metalake1", (AuditInfo) e.auditInfo()));
+
+      // Rename metalake3 --> metalake2
+      store.put(metalake3);
+      store.delete(NameIdentifier.of("metalake2"), EntityType.METALAKE);
+      store.update(
+          NameIdentifier.of("metalake3"),
+          BaseMetalake.class,
+          EntityType.METALAKE,
+          e -> createBaseMakeLake("metalake2", (AuditInfo) e.auditInfo()));
+
+      // Finally, only metalake2 and metalake1 are left.
+      Assertions.assertDoesNotThrow(
+          () -> store.get(NameIdentifier.of("metalake2"), EntityType.METALAKE, BaseMetalake.class));
+      Assertions.assertDoesNotThrow(
+          () -> store.get(NameIdentifier.of("metalake1"), EntityType.METALAKE, BaseMetalake.class));
+      Assertions.assertThrows(
+          NoSuchEntityException.class,
+          () -> store.get(NameIdentifier.of("metalake3"), EntityType.METALAKE, BaseMetalake.class));
+
+      // Test catalog
+      CatalogEntity catalog1 = createCatalog(Namespace.of("metalake1"), "catalog1", auditInfo);
+      CatalogEntity catalog2 = createCatalog(Namespace.of("metalake1"), "catalog2", auditInfo);
+
+      store.put(catalog1);
+      store.put(catalog2);
+
+      store.delete(NameIdentifier.of("metalake1", "catalog1"), EntityType.CATALOG);
+      store.delete(NameIdentifier.of("metalake1", "catalog2"), EntityType.CATALOG);
+
+      store.put(catalog1);
+      // Should be OK;
+      store.update(
+          NameIdentifier.of("metalake1", "catalog1"),
+          CatalogEntity.class,
+          EntityType.CATALOG,
+          e -> createCatalog(Namespace.of("metalake1"), "catalog2", (AuditInfo) e.auditInfo()));
+
+      // Test schema
+      SchemaEntity schema1 =
+          createSchemaEntity(Namespace.of("metalake1", "catalog2"), "schema1", auditInfo);
+      SchemaEntity schema2 =
+          createSchemaEntity(Namespace.of("metalake1", "catalog2"), "schema2", auditInfo);
+
+      store.put(schema1);
+      store.put(schema2);
+
+      store.delete(NameIdentifier.of("metalake1", "catalog2", "schema1"), EntityType.SCHEMA);
+      store.delete(NameIdentifier.of("metalake1", "catalog2", "schema2"), EntityType.SCHEMA);
+
+      store.put(schema1);
+      store.update(
+          NameIdentifier.of("metalake1", "catalog2", "schema1"),
+          SchemaEntity.class,
+          EntityType.SCHEMA,
+          e ->
+              createSchemaEntity(
+                  Namespace.of("metalake1", "catalog2"), "schema2", (AuditInfo) e.auditInfo()));
+
+      // Test table
+      TableEntity table1 =
+          createTableEntity(Namespace.of("metalake1", "catalog2", "schema2"), "table1", auditInfo);
+      TableEntity table2 =
+          createTableEntity(Namespace.of("metalake1", "catalog2", "schema2"), "table2", auditInfo);
+
+      store.put(table1);
+      store.put(table2);
+
+      store.delete(
+          NameIdentifier.of("metalake1", "catalog2", "schema2", "table1"), EntityType.TABLE);
+      store.delete(
+          NameIdentifier.of("metalake1", "catalog2", "schema2", "table2"), EntityType.TABLE);
+
+      store.put(table1);
+      store.update(
+          NameIdentifier.of("metalake1", "catalog2", "schema2", "table1"),
+          TableEntity.class,
+          EntityType.TABLE,
+          e ->
+              createTableEntity(
+                  Namespace.of("metalake1", "catalog2", "schema2"),
+                  "table2",
+                  (AuditInfo) e.auditInfo()));
+    }
+  }
 }

core/src/test/java/com/datastrato/gravitino/storage/kv/TestKvNameMappingService.java

@@ -104,6 +104,31 @@ public void testUpdateName() throws Exception {
     }
   }
 
+  @Test
+  void testUpdateNameWithExistingName() throws Exception {
+    try (KvEntityStore kvEntityStore = getKvEntityStore(getConfig())) {
+      NameMappingService nameMappingService = kvEntityStore.nameMappingService;
+      IdGenerator idGenerator = getIdGeneratorByReflection(nameMappingService);
+      Mockito.doReturn(1L).when(idGenerator).nextId();
+      long name1IdRead = nameMappingService.getOrCreateIdFromName("name1");
+      Assertions.assertNotNull(nameMappingService.getIdByName("name1"));
+
+      Mockito.doReturn(2L).when(idGenerator).nextId();
+      long name2IdRead = nameMappingService.getOrCreateIdFromName("name2");
+      Assertions.assertNotNull(nameMappingService.getIdByName("name1"));
+      Assertions.assertNotEquals(name1IdRead, name2IdRead);
+
+      // Update name1 to an existing name like name2.
+      boolean result = nameMappingService.updateName("name1", "name2");
+      Assertions.assertTrue(result);
+
+      Long name2Id = nameMappingService.getIdByName("name2");
+      Assertions.assertEquals(1L, name2Id);
+
+      Assertions.assertNull(nameMappingService.getIdByName("name1"));
+    }
+  }
+
   @Test
   public void testBindAndUnBind() throws Exception {
     try (KvEntityStore kvEntityStore = getKvEntityStore(getConfig())) {

integration-test/src/test/java/com/datastrato/gravitino/integration/test/catalog/hive/CatalogHiveIT.java

@@ -1105,4 +1105,25 @@ void testAlterEntityName() {
       Assertions.assertNotNull(table);
     }
   }
+
+  @Test
+  void testDropAndRename() {
+    String metalakeName1 = GravitinoITUtils.genRandomName("CatalogHiveIT_metalake1");
+    String metalakeName2 = GravitinoITUtils.genRandomName("CatalogHiveIT_metalake2");
+
+    client.createMetalake(NameIdentifier.of(metalakeName1), "comment", Collections.emptyMap());
+    client.createMetalake(NameIdentifier.of(metalakeName2), "comment", Collections.emptyMap());
+
+    client.dropMetalake(NameIdentifier.of(metalakeName1));
+    client.dropMetalake(NameIdentifier.of(metalakeName2));
+
+    client.createMetalake(NameIdentifier.of(metalakeName1), "comment", Collections.emptyMap());
+
+    client.alterMetalake(NameIdentifier.of(metalakeName1), MetalakeChange.rename(metalakeName2));
+
+    client.loadMetalake(NameIdentifier.of(metalakeName2));
+
+    Assertions.assertThrows(
+        NoSuchMetalakeException.class, () -> client.loadMetalake(NameIdentifier.of(metalakeName1)));
+  }
 }