[#12221] feat(iceberg): Add retained table deletion coordinator

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}";
+  }
+}

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

@@ -0,0 +1,176 @@
+/*
+ * 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 java.util.List;
+import javax.annotation.Nullable;
+import org.apache.gravitino.storage.relational.po.TablePO;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
+import org.apache.ibatis.annotations.Update;
+
+/** Row-locked, generation-scoped mutations of a retained table root. */
+public interface TableDeletionMapper {
+
+  /**
+   * Returns and locks one exact live table root.
+   *
+   * @param tableId immutable table identifier
+   * @return live table root, or {@code null} when absent
+   */
+  @Nullable
+  @Select({
+    "SELECT table_id AS tableId, table_name AS tableName, metalake_id AS metalakeId,",
+    "catalog_id AS catalogId, schema_id AS schemaId, audit_info AS auditInfo,",
+    "current_version AS currentVersion, last_version AS lastVersion,",
+    "deleted_at AS deletedAt, deletion_id AS deletionId FROM table_meta",
+    "WHERE table_id = #{tableId} AND deleted_at = 0 AND deletion_id IS NULL FOR UPDATE"
+  })
+  TablePO selectLiveTableForUpdate(@Param("tableId") long tableId);
+
+  /**
+   * Returns the table root that points to an exact deletion action.
+   *
+   * @param deletionId opaque deletion identifier
+   * @return retained table root, or {@code null} when the pointer is absent
+   */
+  @Nullable
+  @Select({
+    "SELECT table_id AS tableId, table_name AS tableName, metalake_id AS metalakeId,",
+    "catalog_id AS catalogId, schema_id AS schemaId, audit_info AS auditInfo,",
+    "current_version AS currentVersion, last_version AS lastVersion,",
+    "deleted_at AS deletedAt, deletion_id AS deletionId FROM table_meta",
+    "WHERE deletion_id = #{deletionId} AND deleted_at > 0"
+  })
+  TablePO selectRetainedTable(@Param("deletionId") String deletionId);
+
+  /** Returns retained table roots under one exact schema identity. */
+  @Select({
+    "SELECT table_id AS tableId, table_name AS tableName, metalake_id AS metalakeId,",
+    "catalog_id AS catalogId, schema_id AS schemaId, audit_info AS auditInfo,",
+    "current_version AS currentVersion, last_version AS lastVersion,",
+    "deleted_at AS deletedAt, deletion_id AS deletionId FROM table_meta",
+    "WHERE schema_id = #{schemaId} AND deleted_at > 0 AND deletion_id IS NOT NULL",
+    "ORDER BY table_name, deletion_id"
+  })
+  List<TablePO> selectRetainedTables(@Param("schemaId") long schemaId);
+
+  /** Returns retained table roots for one exact schema identity and table name. */
+  @Select({
+    "SELECT table_id AS tableId, table_name AS tableName, metalake_id AS metalakeId,",
+    "catalog_id AS catalogId, schema_id AS schemaId, audit_info AS auditInfo,",
+    "current_version AS currentVersion, last_version AS lastVersion,",
+    "deleted_at AS deletedAt, deletion_id AS deletionId FROM table_meta",
+    "WHERE schema_id = #{schemaId} AND table_name = #{tableName}",
+    "AND deleted_at > 0 AND deletion_id IS NOT NULL",
+    "ORDER BY deletion_id"
+  })
+  List<TablePO> selectRetainedTablesByName(
+      @Param("schemaId") long schemaId, @Param("tableName") String tableName);
+
+  /**
+   * Returns and locks the table root that points to an exact deletion action.
+   *
+   * @param deletionId opaque deletion identifier
+   * @return retained table root, or {@code null} when the pointer is absent
+   */
+  @Nullable
+  @Select({
+    "SELECT table_id AS tableId, table_name AS tableName, metalake_id AS metalakeId,",
+    "catalog_id AS catalogId, schema_id AS schemaId, audit_info AS auditInfo,",
+    "current_version AS currentVersion, last_version AS lastVersion,",
+    "deleted_at AS deletedAt, deletion_id AS deletionId FROM table_meta",
+    "WHERE deletion_id = #{deletionId} AND deleted_at > 0 FOR UPDATE"
+  })
+  TablePO selectRetainedTableForUpdate(@Param("deletionId") String deletionId);
+
+  /**
+   * Returns an exact live table root after restoration.
+   *
+   * @param tableId immutable table identifier
+   * @return live table root, or {@code null} when absent
+   */
+  @Nullable
+  @Select({
+    "SELECT table_id AS tableId, table_name AS tableName, metalake_id AS metalakeId,",
+    "catalog_id AS catalogId, schema_id AS schemaId, audit_info AS auditInfo,",
+    "current_version AS currentVersion, last_version AS lastVersion,",
+    "deleted_at AS deletedAt, deletion_id AS deletionId FROM table_meta",
+    "WHERE table_id = #{tableId} AND deleted_at = 0 AND deletion_id IS NULL"
+  })
+  TablePO selectLiveTable(@Param("tableId") long tableId);
+
+  /** Returns the unchanged user owner of one retained table identity. */
+  @Nullable
+  @Select({
+    "SELECT u.user_name FROM owner_meta o",
+    "JOIN user_meta u ON u.user_id = o.owner_id AND u.deleted_at = 0",
+    "WHERE o.metadata_object_id = #{tableId} AND o.metadata_object_type = 'TABLE'",
+    "AND o.owner_type = 'USER' AND o.deleted_at = 0"
+  })
+  String selectRetainedUserOwnerName(@Param("tableId") long tableId);
+
+  /** Returns the unchanged group owner of one retained table identity. */
+  @Nullable
+  @Select({
+    "SELECT g.group_name FROM owner_meta o",
+    "JOIN group_meta g ON g.group_id = o.owner_id AND g.deleted_at = 0",
+    "WHERE o.metadata_object_id = #{tableId} AND o.metadata_object_type = 'TABLE'",
+    "AND o.owner_type = 'GROUP' AND o.deleted_at = 0"
+  })
+  String selectRetainedGroupOwnerName(@Param("tableId") long tableId);
+
+  /**
+   * Tombstones the exact live table root.
+   *
+   * @param tableId immutable table identifier
+   * @param schemaId immutable parent schema identifier
+   * @param tableName table name at deletion time
+   * @param tableVersion metadata version read from the locked live row
+   * @param deletedAt authoritative deletion time
+   * @param deletionId opaque deletion identifier
+   * @return number of updated rows
+   */
+  @Update({
+    "UPDATE table_meta SET deleted_at = #{deletedAt}, deletion_id = #{deletionId}",
+    "WHERE table_id = #{tableId} AND schema_id = #{schemaId}",
+    "AND table_name = #{tableName} AND current_version = #{tableVersion}",
+    "AND deleted_at = 0 AND deletion_id IS NULL"
+  })
+  int tombstoneTable(
+      @Param("tableId") long tableId,
+      @Param("schemaId") long schemaId,
+      @Param("tableName") String tableName,
+      @Param("tableVersion") long tableVersion,
+      @Param("deletedAt") long deletedAt,
+      @Param("deletionId") String deletionId);
+
+  /**
+   * Reactivates only the root that still points to the exact deletion action.
+   *
+   * @param tableId immutable table identifier
+   * @param deletionId opaque deletion identifier
+   * @return number of updated rows
+   */
+  @Update({
+    "UPDATE table_meta SET deleted_at = 0, deletion_id = NULL",
+    "WHERE table_id = #{tableId} AND deletion_id = #{deletionId} AND deleted_at > 0"
+  })
+  int restoreTable(@Param("tableId") long tableId, @Param("deletionId") String deletionId);
+}

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

@@ -22,6 +22,7 @@
 import java.util.List;
 import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper;
+import org.apache.gravitino.storage.relational.mapper.EntityDeletionMapper;
 import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.FilesetVersionMapper;
 import org.apache.gravitino.storage.relational.mapper.FunctionMetaMapper;
@@ -44,6 +45,7 @@
 import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
 import org.apache.gravitino.storage.relational.mapper.StatisticMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.TableColumnMapper;
+import org.apache.gravitino.storage.relational.mapper.TableDeletionMapper;
 import org.apache.gravitino.storage.relational.mapper.TableMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.TableVersionMapper;
 import org.apache.gravitino.storage.relational.mapper.TagMetaMapper;
@@ -61,6 +63,7 @@ public class DefaultMapperPackageProvider implements MapperPackageProvider {
   public List<Class<?>> getMapperClasses() {
     return ImmutableList.of(
         CatalogMetaMapper.class,
+        EntityDeletionMapper.class,
         EntityChangeLogMapper.class,
         FilesetMetaMapper.class,
         FilesetVersionMapper.class,
@@ -84,6 +87,7 @@ public List<Class<?>> getMapperClasses() {
         SecurableObjectMapper.class,
         StatisticMetaMapper.class,
         TableColumnMapper.class,
+        TableDeletionMapper.class,
         TableMetaMapper.class,
         TagMetadataObjectRelMapper.class,
         TagMetaMapper.class,

core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TableMetaBaseSQLProvider.java

@@ -292,7 +292,8 @@ public String deleteTableMetasByLegacyTimeline(
       @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit) {
     return "DELETE FROM "
         + TABLE_NAME
-        + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT #{limit}";
+        + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline}"
+        + " AND deletion_id IS NULL LIMIT #{limit}";
   }
 
   public String selectTableByFullQualifiedName(

core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TableMetaPostgreSQLProvider.java

@@ -100,6 +100,7 @@ public String deleteTableMetasByLegacyTimeline(
         + TABLE_NAME
         + " WHERE table_id IN (SELECT table_id FROM "
         + TABLE_NAME
-        + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT #{limit})";
+        + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline}"
+        + " AND deletion_id IS NULL LIMIT #{limit})";
   }
 }