[#12221] feat(iceberg): Add conditional table undrop

core/src/main/java/org/apache/gravitino/cache/SupportsEntityCacheInvalidation.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.cache;
+
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.NameIdentifier;
+
+/** Capability for invalidating a locally cached metadata entity after an out-of-band mutation. */
+public interface SupportsEntityCacheInvalidation {
+
+  /**
+   * Invalidates one local entity-cache entry.
+   *
+   * @param identifier entity name identifier
+   * @param type entity type
+   * @return {@code true} when invalidation was accepted
+   */
+  boolean invalidateEntityCache(NameIdentifier identifier, Entity.EntityType type);
+}

core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java

@@ -49,6 +49,7 @@
 import org.apache.gravitino.cache.EntityCacheKey;
 import org.apache.gravitino.cache.EntityCacheRelationKey;
 import org.apache.gravitino.cache.NoOpsCache;
+import org.apache.gravitino.cache.SupportsEntityCacheInvalidation;
 import org.apache.gravitino.exceptions.NoSuchEntityException;
 import org.apache.gravitino.meta.GroupEntity;
 import org.apache.gravitino.meta.RoleEntity;
@@ -69,6 +70,7 @@ public class RelationalEntityStore
     implements EntityStore,
         SupportsRelationOperations,
         SupportsExternalIdOperations,
+        SupportsEntityCacheInvalidation,
         SupportsEntityChangeLog {
   private static final Logger LOGGER = LoggerFactory.getLogger(RelationalEntityStore.class);
   public static final ImmutableMap<String, String> RELATIONAL_BACKENDS =
@@ -77,6 +79,7 @@ public class RelationalEntityStore
   private RelationalBackend backend;
   private RelationalGarbageCollector garbageCollector;
   private EntityChangeLogPoller entityChangeLogPoller;
+  private TableEntityCacheChangeListener tableEntityCacheChangeListener;
   private EntityCache cache;
 
   @VisibleForTesting
@@ -108,6 +111,7 @@ public void initialize(Config config) throws RuntimeException {
             config.get(Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS),
             TimeUnit.SECONDS.toMillis(config.get(Configs.ENTITY_CHANGE_LOG_RETENTION_SECS)),
             TimeUnit.SECONDS.toMillis(config.get(Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)));
+    registerTableEntityCacheChangeListener();
     this.entityChangeLogPoller.start();
   }
 
@@ -274,9 +278,20 @@ public void unregisterEntityChangeLogListener(EntityChangeLogListener listener)
     entityChangeLogPoller.unregisterListener(listener);
   }
 
+  @Override
+  public boolean invalidateEntityCache(NameIdentifier identifier, Entity.EntityType entityType) {
+    return cache.invalidate(identifier, entityType);
+  }
+
+  void registerTableEntityCacheChangeListener() {
+    this.tableEntityCacheChangeListener = new TableEntityCacheChangeListener(this);
+    this.entityChangeLogPoller.registerListener(tableEntityCacheChangeListener);
+  }
+
   @Override
   public void close() throws IOException {
     cache.clear();
+    entityChangeLogPoller.unregisterListener(tableEntityCacheChangeListener);
     entityChangeLogPoller.close();
     garbageCollector.close();
     backend.close();

core/src/main/java/org/apache/gravitino/storage/relational/TableEntityCacheChangeListener.java

@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational;
+
+import java.util.List;
+import java.util.Locale;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.cache.SupportsEntityCacheInvalidation;
+import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Invalidates local table entities when this server observes a peer's table change log. */
+final class TableEntityCacheChangeListener implements EntityChangeLogListener {
+
+  private static final Logger LOG = LoggerFactory.getLogger(TableEntityCacheChangeListener.class);
+
+  private final SupportsEntityCacheInvalidation cacheInvalidation;
+
+  TableEntityCacheChangeListener(SupportsEntityCacheInvalidation cacheInvalidation) {
+    this.cacheInvalidation = cacheInvalidation;
+  }
+
+  @Override
+  public void onEntityChange(List<EntityChangeRecord> changes) {
+    for (EntityChangeRecord change : changes) {
+      if (change.getEntityType() == null
+          || !Entity.EntityType.TABLE.name().equals(change.getEntityType().toUpperCase(Locale.ROOT))
+          || StringUtils.isBlank(change.getFullName())) {
+        continue;
+      }
+
+      try {
+        cacheInvalidation.invalidateEntityCache(
+            NameIdentifier.parse(change.getFullName()), Entity.EntityType.TABLE);
+      } catch (RuntimeException e) {
+        LOG.warn("Failed to invalidate table cache from change log: {}", change.getFullName(), e);
+      }
+    }
+  }
+}

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

@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.mapper;
+
+import javax.annotation.Nullable;
+import org.apache.gravitino.storage.relational.po.EntityDeletionPO;
+import org.apache.ibatis.annotations.DeleteProvider;
+import org.apache.ibatis.annotations.InsertProvider;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.SelectProvider;
+
+/** MyBatis mapper for active metadata deletion generations. */
+public interface EntityDeletionMapper {
+
+  /** Active deletion-generation table name. */
+  String TABLE_NAME = "entity_deletion";
+
+  /**
+   * Inserts one active deletion generation.
+   *
+   * @param deletion deletion generation to insert
+   */
+  @InsertProvider(type = EntityDeletionSQLProvider.class, method = "insertEntityDeletion")
+  void insertEntityDeletion(@Param("deletion") EntityDeletionPO deletion);
+
+  /**
+   * Selects one exact deletion generation.
+   *
+   * @param deletionId opaque deletion identifier
+   * @return persisted deletion, or {@code null} when absent
+   */
+  @Nullable
+  @SelectProvider(type = EntityDeletionSQLProvider.class, method = "selectEntityDeletion")
+  EntityDeletionPO selectEntityDeletion(@Param("deletionId") String deletionId);
+
+  /**
+   * Selects and locks one exact deletion action.
+   *
+   * @param deletionId opaque deletion identifier
+   * @return persisted deletion, or {@code null} when absent
+   */
+  @Nullable
+  @SelectProvider(type = EntityDeletionSQLProvider.class, method = "selectEntityDeletionForUpdate")
+  EntityDeletionPO selectEntityDeletionForUpdate(@Param("deletionId") String deletionId);
+
+  /**
+   * Deletes an exact action after the caller has locked and validated it.
+   *
+   * @param deletionId opaque deletion identifier
+   * @return number of deleted rows
+   */
+  @DeleteProvider(type = EntityDeletionSQLProvider.class, method = "deleteEntityDeletion")
+  int deleteEntityDeletion(@Param("deletionId") String deletionId);
+}

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

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.mapper;
+
+import static org.apache.gravitino.storage.relational.mapper.EntityDeletionMapper.TABLE_NAME;
+
+import org.apache.gravitino.storage.relational.po.EntityDeletionPO;
+import org.apache.ibatis.annotations.Param;
+
+/** Portable SQL provider for active metadata deletion generations. */
+public class EntityDeletionSQLProvider {
+
+  private static final String SELECT_COLUMNS =
+      "deletion_id AS deletionId, state,"
+          + " retention_expires_at AS retentionExpiresAt, purge_job_id AS purgeJobId";
+
+  /**
+   * Builds the insert for one deletion generation.
+   *
+   * @param deletion deletion generation to insert
+   * @return parameterized insert SQL
+   */
+  public static String insertEntityDeletion(@Param("deletion") EntityDeletionPO deletion) {
+    return "INSERT INTO "
+        + TABLE_NAME
+        + " (deletion_id, state, retention_expires_at, purge_job_id)"
+        + " VALUES (#{deletion.deletionId}, #{deletion.state},"
+        + " #{deletion.retentionExpiresAt}, #{deletion.purgeJobId})";
+  }
+
+  /**
+   * Builds an exact deletion-generation lookup.
+   *
+   * @param deletionId opaque deletion identifier
+   * @return parameterized select SQL
+   */
+  public static String selectEntityDeletion(@Param("deletionId") String deletionId) {
+    return "SELECT "
+        + SELECT_COLUMNS
+        + " FROM "
+        + TABLE_NAME
+        + " WHERE deletion_id = #{deletionId}";
+  }
+
+  /**
+   * Builds an exact deletion-action locking lookup.
+   *
+   * @param deletionId opaque deletion identifier
+   * @return parameterized select SQL
+   */
+  public static String selectEntityDeletionForUpdate(@Param("deletionId") String deletionId) {
+    return selectEntityDeletion(deletionId) + " FOR UPDATE";
+  }
+
+  /**
+   * Builds an exact deletion-action removal after it has been locked and validated.
+   *
+   * @param deletionId opaque deletion identifier
+   * @return parameterized delete SQL
+   */
+  public static String deleteEntityDeletion(@Param("deletionId") String deletionId) {
+    return "DELETE FROM " + TABLE_NAME + " WHERE deletion_id = #{deletionId}";
+  }
+}