From facfe831fade050f3166ae9d941d5e3e6fcb66c1 Mon Sep 17 00:00:00 2001
From: roryqi <roryqi@apache.org>
Date: Wed, 6 Nov 2024 17:41:35 +0800
Subject: [PATCH] [#5029] improvement(storage): Delete the related relations
 about the metadata object when it is deleted. (#5426)

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

Delete the related relations about the metadata object when it is
deleted.

### Why are the changes needed?

Fix: #5029

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

No.

### How was this patch tested?

Add UTs.
---
 .../mapper/SecurableObjectMapper.java         |  21 +-
 .../SecurableObjectSQLProviderFactory.java    |  19 +-
 .../mapper/TagMetadataObjectRelMapper.java    |  22 ++
 ...agMetadataObjectRelSQLProviderFactory.java |  20 ++
 .../base/OwnerMetaBaseSQLProvider.java        |  38 +--
 .../base/SecurableObjectBaseSQLProvider.java  |  79 ++++-
 .../TagMetadataObjectRelBaseSQLProvider.java  | 125 ++++++-
 .../OwnerMetaPostgreSQLProvider.java          |  24 +-
 .../SecurableObjectPostgreSQLProvider.java    |  81 ++++-
 ...agMetadataObjectRelPostgreSQLProvider.java | 112 ++++++
 .../service/CatalogMetaService.java           |  25 +-
 .../service/FilesetMetaService.java           |  14 +
 .../service/MetalakeMetaService.java          |   4 +-
 .../relational/service/SchemaMetaService.java |  25 +-
 .../relational/service/TableMetaService.java  |  21 +-
 .../relational/service/TopicMetaService.java  |  14 +
 .../storage/relational/TestJDBCBackend.java   | 116 +++++++
 .../service/TestOwnerMetaService.java         | 240 +++++++++++++
 .../service/TestSecurableObjects.java         | 209 ++++++++++++
 .../service/TestTagMetaService.java           | 323 ++++++++++++++++++
 20 files changed, 1474 insertions(+), 58 deletions(-)

diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SecurableObjectMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SecurableObjectMapper.java
index a5160f905eb..d4e8fdf0a06 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SecurableObjectMapper.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SecurableObjectMapper.java
@@ -58,8 +58,25 @@ void batchSoftDeleteSecurableObjects(
 
   @UpdateProvider(
       type = SecurableObjectSQLProviderFactory.class,
-      method = "softDeleteRoleMetasByMetalakeId")
-  void softDeleteRoleMetasByMetalakeId(@Param("metalakeId") Long metalakeId);
+      method = "softDeleteSecurableObjectsByMetalakeId")
+  void softDeleteSecurableObjectsByMetalakeId(@Param("metalakeId") Long metalakeId);
+
+  @UpdateProvider(
+      type = SecurableObjectSQLProviderFactory.class,
+      method = "softDeleteObjectRelsByMetadataObject")
+  void softDeleteObjectRelsByMetadataObject(
+      @Param("metadataObjectId") Long metadataObjectId,
+      @Param("metadataObjectType") String metadataObjectType);
+
+  @UpdateProvider(
+      type = SecurableObjectSQLProviderFactory.class,
+      method = "softDeleteObjectRelsByCatalogId")
+  void softDeleteObjectRelsByCatalogId(@Param("catalogId") Long catalogId);
+
+  @UpdateProvider(
+      type = SecurableObjectSQLProviderFactory.class,
+      method = "softDeleteObjectRelsBySchemaId")
+  void softDeleteObjectRelsBySchemaId(@Param("schemaId") Long schemaId);
 
   @SelectProvider(
       type = SecurableObjectSQLProviderFactory.class,
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SecurableObjectSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SecurableObjectSQLProviderFactory.java
index 4f664dabcca..dab4bcf7030 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SecurableObjectSQLProviderFactory.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SecurableObjectSQLProviderFactory.java
@@ -66,8 +66,23 @@ public static String softDeleteSecurableObjectsByRoleId(@Param("roleId") Long ro
     return getProvider().softDeleteSecurableObjectsByRoleId(roleId);
   }
 
-  public static String softDeleteRoleMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {
-    return getProvider().softDeleteRoleMetasByMetalakeId(metalakeId);
+  public static String softDeleteSecurableObjectsByMetalakeId(
+      @Param("metalakeId") Long metalakeId) {
+    return getProvider().softDeleteSecurableObjectsByMetalakeId(metalakeId);
+  }
+
+  public static String softDeleteObjectRelsByMetadataObject(
+      @Param("metadataObjectId") Long metadataObjectId,
+      @Param("metadataObjectType") String metadataObjectType) {
+    return getProvider().softDeleteObjectRelsByMetadataObject(metadataObjectId, metadataObjectType);
+  }
+
+  public static String softDeleteObjectRelsByCatalogId(@Param("catalogId") Long catalogId) {
+    return getProvider().softDeleteObjectRelsByCatalogId(catalogId);
+  }
+
+  public static String softDeleteObjectRelsBySchemaId(@Param("schemaId") Long schemaId) {
+    return getProvider().softDeleteObjectRelsBySchemaId(schemaId);
   }
 
   public static String listSecurableObjectsByRoleId(@Param("roleId") Long roleId) {
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelMapper.java
index 40b62955881..ed6ae54347c 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelMapper.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelMapper.java
@@ -75,6 +75,28 @@ Integer softDeleteTagMetadataObjectRelsByMetalakeAndTagName(
       method = "softDeleteTagMetadataObjectRelsByMetalakeId")
   void softDeleteTagMetadataObjectRelsByMetalakeId(@Param("metalakeId") Long metalakeId);
 
+  @UpdateProvider(
+      type = TagMetadataObjectRelSQLProviderFactory.class,
+      method = "softDeleteTagMetadataObjectRelsByMetadataObject")
+  void softDeleteTagMetadataObjectRelsByMetadataObject(
+      @Param("metadataObjectId") Long metadataObjectId,
+      @Param("metadataObjectType") String metadataObjectType);
+
+  @UpdateProvider(
+      type = TagMetadataObjectRelSQLProviderFactory.class,
+      method = "softDeleteTagMetadataObjectRelsByCatalogId")
+  void softDeleteTagMetadataObjectRelsByCatalogId(@Param("catalogId") Long catalogId);
+
+  @UpdateProvider(
+      type = TagMetadataObjectRelSQLProviderFactory.class,
+      method = "softDeleteTagMetadataObjectRelsBySchemaId")
+  void softDeleteTagMetadataObjectRelsBySchemaId(@Param("schemaId") Long schemaId);
+
+  @UpdateProvider(
+      type = TagMetadataObjectRelSQLProviderFactory.class,
+      method = "softDeleteTagMetadataObjectRelsByTableId")
+  void softDeleteTagMetadataObjectRelsByTableId(@Param("tableId") Long tableId);
+
   @DeleteProvider(
       type = TagMetadataObjectRelSQLProviderFactory.class,
       method = "deleteTagEntityRelsByLegacyTimeline")
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelSQLProviderFactory.java
index 6e7deb6f1ad..34be439f235 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelSQLProviderFactory.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelSQLProviderFactory.java
@@ -95,6 +95,26 @@ public static String softDeleteTagMetadataObjectRelsByMetalakeId(
     return getProvider().softDeleteTagMetadataObjectRelsByMetalakeId(metalakeId);
   }
 
+  public static String softDeleteTagMetadataObjectRelsByMetadataObject(
+      @Param("metadataObjectId") Long metadataObjectId,
+      @Param("metadataObjectType") String metadataObjectType) {
+    return getProvider()
+        .softDeleteTagMetadataObjectRelsByMetadataObject(metadataObjectId, metadataObjectType);
+  }
+
+  public static String softDeleteTagMetadataObjectRelsByCatalogId(
+      @Param("catalogId") Long catalogId) {
+    return getProvider().softDeleteTagMetadataObjectRelsByCatalogId(catalogId);
+  }
+
+  public static String softDeleteTagMetadataObjectRelsBySchemaId(@Param("schemaId") Long schemaId) {
+    return getProvider().softDeleteTagMetadataObjectRelsBySchemaId(schemaId);
+  }
+
+  public static String softDeleteTagMetadataObjectRelsByTableId(@Param("tableId") Long tableId) {
+    return getProvider().softDeleteTagMetadataObjectRelsByTableId(tableId);
+  }
+
   public static String deleteTagEntityRelsByLegacyTimeline(
       @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit) {
     return getProvider().deleteTagEntityRelsByLegacyTimeline(legacyTimeline, limit);
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/OwnerMetaBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/OwnerMetaBaseSQLProvider.java
index 4929b0cbcdc..ab89ddcd47a 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/OwnerMetaBaseSQLProvider.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/OwnerMetaBaseSQLProvider.java
@@ -123,30 +123,30 @@ public String softDeleteOwnerRelByCatalogId(@Param("catalogId") Long catalogId)
         + OWNER_TABLE_NAME
         + " ot SET ot.deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
         + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
-        + " WHERE EXISTS ("
+        + " WHERE ot.deleted_at = 0 AND EXISTS ("
         + " SELECT ct.catalog_id FROM "
         + CatalogMetaMapper.TABLE_NAME
-        + " ct WHERE ct.catalog_id = #{catalogId}  AND ct.deleted_at = 0 AND ot.deleted_at = 0 AND "
-        + "ct.catalog_id = ot.metadata_object_id AND ot.metadata_object_type = 'CATALOG'"
+        + " ct WHERE ct.catalog_id = #{catalogId} AND "
+        + " ct.catalog_id = ot.metadata_object_id AND ot.metadata_object_type = 'CATALOG'"
         + " UNION "
         + " SELECT st.catalog_id FROM "
         + SchemaMetaMapper.TABLE_NAME
-        + " st WHERE st.catalog_id = #{catalogId} AND st.deleted_at = 0 AND ot.deleted_at = 0 AND "
-        + "st.schema_id = ot.metadata_object_id AND ot.metadata_object_type = 'SCHEMA'"
+        + " st WHERE st.catalog_id = #{catalogId} AND "
+        + " st.schema_id = ot.metadata_object_id AND ot.metadata_object_type = 'SCHEMA'"
         + " UNION "
         + " SELECT tt.catalog_id FROM "
         + TopicMetaMapper.TABLE_NAME
-        + " tt WHERE tt.catalog_id = #{catalogId} AND tt.deleted_at = 0 AND ot.deleted_at = 0 AND "
-        + "tt.topic_id = ot.metadata_object_id AND ot.metadata_object_type = 'TOPIC'"
+        + " tt WHERE tt.catalog_id = #{catalogId} AND "
+        + " tt.topic_id = ot.metadata_object_id AND ot.metadata_object_type = 'TOPIC'"
         + " UNION "
         + " SELECT tat.catalog_id FROM "
         + TableMetaMapper.TABLE_NAME
-        + " tat WHERE tat.catalog_id = #{catalogId} AND tat.deleted_at = 0 AND ot.deleted_at = 0 AND "
-        + "tat.table_id = ot.metadata_object_id AND ot.metadata_object_type = 'TABLE'"
+        + " tat WHERE tat.catalog_id = #{catalogId} AND "
+        + " tat.table_id = ot.metadata_object_id AND ot.metadata_object_type = 'TABLE'"
         + " UNION "
         + " SELECT ft.catalog_id FROM "
         + FilesetMetaMapper.META_TABLE_NAME
-        + " ft WHERE ft.catalog_id = #{catalogId} AND ft.deleted_at = 0 AND ot.deleted_at = 0 AND"
+        + " ft WHERE ft.catalog_id = #{catalogId} AND"
         + " ft.fileset_id = ot.metadata_object_id AND ot.metadata_object_type = 'FILESET'"
         + ")";
   }
@@ -156,26 +156,26 @@ public String sotDeleteOwnerRelBySchemaId(@Param("schemaId") Long schemaId) {
         + OWNER_TABLE_NAME
         + " ot SET ot.deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
         + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
-        + " WHERE EXISTS ("
+        + " WHERE ot.deleted_at = 0 AND EXISTS ("
         + " SELECT st.schema_id FROM "
         + SchemaMetaMapper.TABLE_NAME
-        + " st WHERE st.schema_id = #{schemaId} AND st.deleted_at = 0 AND ot.deleted_at = 0 "
-        + "AND st.schema_id = ot.metadata_object_id AND ot.metadata_object_type = 'SCHEMA'"
+        + " st WHERE st.schema_id = #{schemaId} AND"
+        + " st.schema_id = ot.metadata_object_id AND ot.metadata_object_type = 'SCHEMA'"
         + " UNION "
         + " SELECT tt.schema_id FROM "
         + TopicMetaMapper.TABLE_NAME
-        + " tt WHERE tt.schema_id = #{schemaId} AND tt.deleted_at = 0 AND ot.deleted_at = 0 AND "
-        + "tt.topic_id = ot.metadata_object_id AND ot.metadata_object_type = 'TOPIC'"
+        + " tt WHERE tt.schema_id = #{schemaId} AND "
+        + " tt.topic_id = ot.metadata_object_id AND ot.metadata_object_type = 'TOPIC'"
         + " UNION "
         + " SELECT tat.schema_id FROM "
         + TableMetaMapper.TABLE_NAME
-        + " tat WHERE tat.schema_id = #{schemaId} AND tat.deleted_at = 0 AND ot.deleted_at = 0 AND "
-        + "tat.table_id = ot.metadata_object_id AND ot.metadata_object_type = 'TABLE'"
+        + " tat WHERE tat.schema_id = #{schemaId} AND "
+        + " tat.table_id = ot.metadata_object_id AND ot.metadata_object_type = 'TABLE'"
         + " UNION "
         + " SELECT ft.schema_id FROM "
         + FilesetMetaMapper.META_TABLE_NAME
-        + " ft WHERE ft.schema_id = #{schemaId} AND ft.deleted_at = 0 AND ot.deleted_at = 0 AND "
-        + "ft.fileset_id = ot.metadata_object_id AND ot.metadata_object_type = 'FILESET'"
+        + " ft WHERE ft.schema_id = #{schemaId} AND "
+        + " ft.fileset_id = ot.metadata_object_id AND ot.metadata_object_type = 'FILESET'"
         + ")";
   }
 
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SecurableObjectBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SecurableObjectBaseSQLProvider.java
index 5561f7cbb47..1c47741e050 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SecurableObjectBaseSQLProvider.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SecurableObjectBaseSQLProvider.java
@@ -22,6 +22,11 @@
 import static org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper.SECURABLE_OBJECT_TABLE_NAME;
 
 import java.util.List;
+import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TableMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper;
 import org.apache.gravitino.storage.relational.po.SecurableObjectPO;
 import org.apache.ibatis.annotations.Param;
 
@@ -73,7 +78,7 @@ public String softDeleteSecurableObjectsByRoleId(@Param("roleId") Long roleId) {
         + " WHERE role_id = #{roleId} AND deleted_at = 0";
   }
 
-  public String softDeleteRoleMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {
+  public String softDeleteSecurableObjectsByMetalakeId(@Param("metalakeId") Long metalakeId) {
     return "UPDATE "
         + SECURABLE_OBJECT_TABLE_NAME
         + " ob SET ob.deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
@@ -84,6 +89,78 @@ public String softDeleteRoleMetasByMetalakeId(@Param("metalakeId") Long metalake
         + " AND ro.deleted_at = 0) AND ob.deleted_at = 0";
   }
 
+  public String softDeleteObjectRelsByMetadataObject(
+      @Param("metadataObjectId") Long metadataObjectId,
+      @Param("metadataObjectType") String metadataObjectType) {
+    return "UPDATE "
+        + SECURABLE_OBJECT_TABLE_NAME
+        + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+        + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
+        + " WHERE metadata_object_id = #{metadataObjectId} AND deleted_at = 0"
+        + " AND type = #{metadataObjectType}";
+  }
+
+  public String softDeleteObjectRelsByCatalogId(@Param("catalogId") Long catalogId) {
+    return "UPDATE "
+        + SECURABLE_OBJECT_TABLE_NAME
+        + " sect SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+        + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
+        + " WHERE sect.deleted_at = 0 AND EXISTS ("
+        + " SELECT ct.catalog_id FROM "
+        + CatalogMetaMapper.TABLE_NAME
+        + " ct WHERE ct.catalog_id = #{catalogId}  AND "
+        + " ct.catalog_id = sect.metadata_object_id AND sect.type = 'CATALOG'"
+        + " UNION "
+        + " SELECT st.catalog_id FROM "
+        + SchemaMetaMapper.TABLE_NAME
+        + " st WHERE st.catalog_id = #{catalogId} AND "
+        + " st.schema_id = sect.metadata_object_id AND sect.type = 'SCHEMA'"
+        + " UNION "
+        + " SELECT tt.catalog_id FROM "
+        + TopicMetaMapper.TABLE_NAME
+        + " tt WHERE tt.catalog_id = #{catalogId} AND "
+        + " tt.topic_id = sect.metadata_object_id AND sect.type = 'TOPIC'"
+        + " UNION "
+        + " SELECT tat.catalog_id FROM "
+        + TableMetaMapper.TABLE_NAME
+        + " tat WHERE tat.catalog_id = #{catalogId} AND "
+        + " tat.table_id = sect.metadata_object_id AND sect.type = 'TABLE'"
+        + " UNION "
+        + " SELECT ft.catalog_id FROM "
+        + FilesetMetaMapper.META_TABLE_NAME
+        + " ft WHERE ft.catalog_id = #{catalogId} AND"
+        + " ft.fileset_id = sect.metadata_object_id AND sect.type = 'FILESET'"
+        + ")";
+  }
+
+  public String softDeleteObjectRelsBySchemaId(@Param("schemaId") Long schemaId) {
+    return "UPDATE "
+        + SECURABLE_OBJECT_TABLE_NAME
+        + " sect SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+        + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
+        + " WHERE sect.deleted_at = 0 AND EXISTS ("
+        + " SELECT st.schema_id FROM "
+        + SchemaMetaMapper.TABLE_NAME
+        + " st WHERE st.schema_id = #{schemaId} "
+        + " AND st.schema_id = sect.metadata_object_id AND sect.type = 'SCHEMA'"
+        + " UNION "
+        + " SELECT tt.schema_id FROM "
+        + TopicMetaMapper.TABLE_NAME
+        + " tt WHERE tt.schema_id = #{schemaId} AND "
+        + " tt.topic_id = sect.metadata_object_id AND sect.type = 'TOPIC'"
+        + " UNION "
+        + " SELECT tat.schema_id FROM "
+        + TableMetaMapper.TABLE_NAME
+        + " tat WHERE tat.schema_id = #{schemaId} AND "
+        + " tat.table_id = sect.metadata_object_id AND sect.type = 'TABLE'"
+        + " UNION "
+        + " SELECT ft.schema_id FROM "
+        + FilesetMetaMapper.META_TABLE_NAME
+        + " ft WHERE ft.schema_id = #{schemaId} AND "
+        + " ft.fileset_id = sect.metadata_object_id AND sect.type = 'FILESET'"
+        + ")";
+  }
+
   public String listSecurableObjectsByRoleId(@Param("roleId") Long roleId) {
     return "SELECT role_id as roleId, metadata_object_id as metadataObjectId,"
         + " type as type, privilege_names as privilegeNames,"
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetadataObjectRelBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetadataObjectRelBaseSQLProvider.java
index 5a9b066a006..6c1c7eb5fca 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetadataObjectRelBaseSQLProvider.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TagMetadataObjectRelBaseSQLProvider.java
@@ -18,11 +18,16 @@
  */
 package org.apache.gravitino.storage.relational.mapper.provider.base;
 
-import static org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME;
-
 import java.util.List;
+import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TableColumnMapper;
+import org.apache.gravitino.storage.relational.mapper.TableMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.TagMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper;
+import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper;
 import org.apache.gravitino.storage.relational.po.TagMetadataObjectRelPO;
 import org.apache.ibatis.annotations.Param;
 
@@ -40,7 +45,7 @@ public String listTagPOsByMetadataObjectIdAndType(
         + " FROM "
         + TagMetaMapper.TAG_TABLE_NAME
         + " tm JOIN "
-        + TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
         + " te ON tm.tag_id = te.tag_id"
         + " WHERE te.metadata_object_id = #{metadataObjectId}"
         + " AND te.metadata_object_type = #{metadataObjectType} AND te.deleted_at = 0"
@@ -60,7 +65,7 @@ public String getTagPOsByMetadataObjectAndTagName(
         + " FROM "
         + TagMetaMapper.TAG_TABLE_NAME
         + " tm JOIN "
-        + TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
         + " te ON tm.tag_id = te.tag_id"
         + " WHERE te.metadata_object_id = #{metadataObjectId}"
         + " AND te.metadata_object_type = #{metadataObjectType} AND tm.tag_name = #{tagName}"
@@ -74,7 +79,7 @@ public String listTagMetadataObjectRelsByMetalakeAndTagName(
         + " te.current_version as currentVersion, te.last_version as lastVersion,"
         + " te.deleted_at as deletedAt"
         + " FROM "
-        + TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
         + " te JOIN "
         + TagMetaMapper.TAG_TABLE_NAME
         + " tm JOIN "
@@ -88,7 +93,7 @@ public String batchInsertTagMetadataObjectRels(
       @Param("tagRels") List<TagMetadataObjectRelPO> tagRelPOs) {
     return "<script>"
         + "INSERT INTO "
-        + TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
         + "(tag_id, metadata_object_id, metadata_object_type, audit_info,"
         + " current_version, last_version, deleted_at)"
         + " VALUES "
@@ -110,7 +115,7 @@ public String batchDeleteTagMetadataObjectRelsByTagIdsAndMetadataObject(
       @Param("tagIds") List<Long> tagIds) {
     return "<script>"
         + "UPDATE "
-        + TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
         + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
         + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
         + " WHERE tag_id IN "
@@ -125,7 +130,7 @@ public String batchDeleteTagMetadataObjectRelsByTagIdsAndMetadataObject(
   public String softDeleteTagMetadataObjectRelsByMetalakeAndTagName(
       @Param("metalakeName") String metalakeName, @Param("tagName") String tagName) {
     return "UPDATE "
-        + TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
         + " te SET te.deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
         + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
         + " WHERE te.tag_id IN (SELECT tm.tag_id FROM "
@@ -138,7 +143,7 @@ public String softDeleteTagMetadataObjectRelsByMetalakeAndTagName(
 
   public String softDeleteTagMetadataObjectRelsByMetalakeId(@Param("metalakeId") Long metalakeId) {
     return "UPDATE "
-        + TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
         + " te SET te.deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
         + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
         + " WHERE EXISTS (SELECT * FROM "
@@ -147,10 +152,110 @@ public String softDeleteTagMetadataObjectRelsByMetalakeId(@Param("metalakeId") L
         + " AND tm.deleted_at = 0) AND te.deleted_at = 0";
   }
 
+  public String softDeleteTagMetadataObjectRelsByMetadataObject(
+      @Param("metadataObjectId") Long metadataObjectId,
+      @Param("metadataObjectType") String metadataObjectType) {
+    return " UPDATE "
+        + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+        + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
+        + " WHERE metadata_object_id = #{metadataObjectId} AND deleted_at = 0"
+        + " AND metadata_object_type = #{metadataObjectType}";
+  }
+
+  public String softDeleteTagMetadataObjectRelsByCatalogId(@Param("catalogId") Long catalogId) {
+    return " UPDATE "
+        + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + " tmt SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+        + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
+        + " WHERE tmt.deleted_at = 0 AND EXISTS ("
+        + " SELECT ct.catalog_id FROM "
+        + CatalogMetaMapper.TABLE_NAME
+        + " ct WHERE ct.catalog_id = #{catalogId}  AND "
+        + "ct.catalog_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'CATALOG'"
+        + " UNION "
+        + " SELECT st.catalog_id FROM "
+        + SchemaMetaMapper.TABLE_NAME
+        + " st WHERE st.catalog_id = #{catalogId} AND "
+        + "st.schema_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'SCHEMA'"
+        + " UNION "
+        + " SELECT tt.catalog_id FROM "
+        + TopicMetaMapper.TABLE_NAME
+        + " tt WHERE tt.catalog_id = #{catalogId} AND "
+        + "tt.topic_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'TOPIC'"
+        + " UNION "
+        + " SELECT tat.catalog_id FROM "
+        + TableMetaMapper.TABLE_NAME
+        + " tat WHERE tat.catalog_id = #{catalogId} AND "
+        + "tat.table_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'TABLE'"
+        + " UNION "
+        + " SELECT ft.catalog_id FROM "
+        + FilesetMetaMapper.META_TABLE_NAME
+        + " ft WHERE ft.catalog_id = #{catalogId}  AND"
+        + " ft.fileset_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'FILESET'"
+        + " UNION "
+        + " SELECT cot.catalog_id FROM "
+        + TableColumnMapper.COLUMN_TABLE_NAME
+        + " cot WHERE cot.catalog_id = #{catalogId} AND"
+        + " cot.column_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'COLUMN'"
+        + ")";
+  }
+
+  public String softDeleteTagMetadataObjectRelsBySchemaId(@Param("schemaId") Long schemaId) {
+    return " UPDATE "
+        + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + " tmt SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+        + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
+        + " WHERE tmt.deleted_at = 0 AND EXISTS ("
+        + " SELECT st.schema_id FROM "
+        + SchemaMetaMapper.TABLE_NAME
+        + " st WHERE st.schema_id = #{schemaId} AND "
+        + " st.schema_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'SCHEMA'"
+        + " UNION "
+        + " SELECT tt.schema_id FROM "
+        + TopicMetaMapper.TABLE_NAME
+        + " tt WHERE tt.schema_id = #{schemaId} AND "
+        + "tt.topic_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'TOPIC'"
+        + " UNION "
+        + " SELECT tat.schema_id FROM "
+        + TableMetaMapper.TABLE_NAME
+        + " tat WHERE tat.schema_id = #{schemaId} AND "
+        + "tat.table_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'TABLE'"
+        + " UNION "
+        + " SELECT ft.schema_id FROM "
+        + FilesetMetaMapper.META_TABLE_NAME
+        + " ft WHERE ft.schema_id = #{schemaId} AND "
+        + " ft.fileset_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'FILESET'"
+        + " UNION "
+        + " SELECT cot.schema_id FROM "
+        + TableColumnMapper.COLUMN_TABLE_NAME
+        + " cot WHERE cot.schema_id = #{schemaId} AND "
+        + " cot.column_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'COLUMN'"
+        + ")";
+  }
+
+  public String softDeleteTagMetadataObjectRelsByTableId(@Param("tableId") Long tableId) {
+    return " UPDATE "
+        + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + " tmt SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+        + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
+        + " WHERE tmt.deleted_at = 0 AND EXISTS ("
+        + " SELECT tat.table_id FROM "
+        + TableMetaMapper.TABLE_NAME
+        + " tat WHERE tat.table_id = #{tableId} AND "
+        + " tat.table_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'TABLE'"
+        + " UNION "
+        + " SELECT cot.table_id FROM "
+        + TableColumnMapper.COLUMN_TABLE_NAME
+        + " cot WHERE cot.table_id = #{tableId} AND"
+        + " cot.column_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'COLUMN'"
+        + ")";
+  }
+
   public String deleteTagEntityRelsByLegacyTimeline(
       @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit) {
     return "DELETE FROM "
-        + TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
         + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT #{limit}";
   }
 }
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/OwnerMetaPostgreSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/OwnerMetaPostgreSQLProvider.java
index c218c4f68cb..ba594f7732d 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/OwnerMetaPostgreSQLProvider.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/OwnerMetaPostgreSQLProvider.java
@@ -62,30 +62,30 @@ public String softDeleteOwnerRelByCatalogId(Long catalogId) {
         + OWNER_TABLE_NAME
         + " ot SET deleted_at = floor(extract(epoch from((current_timestamp -"
         + " timestamp '1970-01-01 00:00:00')*1000)))"
-        + " WHERE EXISTS ("
+        + " WHERE ot.deleted_at = 0 AND EXISTS ("
         + " SELECT ct.catalog_id FROM "
         + CatalogMetaMapper.TABLE_NAME
-        + " ct WHERE ct.catalog_id = #{catalogId}  AND ct.deleted_at = 0 AND ot.deleted_at = 0 AND "
+        + " ct WHERE ct.catalog_id = #{catalogId} AND "
         + "ct.catalog_id = ot.metadata_object_id AND ot.metadata_object_type = 'CATALOG'"
         + " UNION "
         + " SELECT st.catalog_id FROM "
         + SchemaMetaMapper.TABLE_NAME
-        + " st WHERE st.catalog_id = #{catalogId} AND st.deleted_at = 0 AND ot.deleted_at = 0 AND "
+        + " st WHERE st.catalog_id = #{catalogId} AND "
         + "st.schema_id = ot.metadata_object_id AND ot.metadata_object_type = 'SCHEMA'"
         + " UNION "
         + " SELECT tt.catalog_id FROM "
         + TopicMetaMapper.TABLE_NAME
-        + " tt WHERE tt.catalog_id = #{catalogId} AND tt.deleted_at = 0 AND ot.deleted_at = 0 AND "
+        + " tt WHERE tt.catalog_id = #{catalogId} AND "
         + "tt.topic_id = ot.metadata_object_id AND ot.metadata_object_type = 'TOPIC'"
         + " UNION "
         + " SELECT tat.catalog_id FROM "
         + TableMetaMapper.TABLE_NAME
-        + " tat WHERE tat.catalog_id = #{catalogId} AND tat.deleted_at = 0 AND ot.deleted_at = 0 AND "
+        + " tat WHERE tat.catalog_id = #{catalogId} AND "
         + "tat.table_id = ot.metadata_object_id AND ot.metadata_object_type = 'TABLE'"
         + " UNION "
         + " SELECT ft.catalog_id FROM "
         + FilesetMetaMapper.META_TABLE_NAME
-        + " ft WHERE ft.catalog_id = #{catalogId} AND ft.deleted_at = 0 AND ot.deleted_at = 0 AND"
+        + " ft WHERE ft.catalog_id = #{catalogId} AND"
         + " ft.fileset_id = ot.metadata_object_id AND ot.metadata_object_type = 'FILESET'"
         + ")";
   }
@@ -95,25 +95,25 @@ public String sotDeleteOwnerRelBySchemaId(Long schemaId) {
     return "UPDATE  "
         + OWNER_TABLE_NAME
         + " ot SET deleted_at = floor(extract(epoch from((current_timestamp - timestamp '1970-01-01 00:00:00')*1000))) "
-        + " WHERE EXISTS ("
+        + " WHERE ot.deleted_at = 0 AND EXISTS ("
         + " SELECT st.schema_id FROM "
         + SchemaMetaMapper.TABLE_NAME
-        + " st WHERE st.schema_id = #{schemaId} AND st.deleted_at = 0 AND ot.deleted_at = 0 "
-        + "AND st.schema_id = ot.metadata_object_id AND ot.metadata_object_type = 'SCHEMA'"
+        + " st WHERE st.schema_id = #{schemaId} "
+        + " AND st.schema_id = ot.metadata_object_id AND ot.metadata_object_type = 'SCHEMA'"
         + " UNION "
         + " SELECT tt.schema_id FROM "
         + TopicMetaMapper.TABLE_NAME
-        + " tt WHERE tt.schema_id = #{schemaId} AND tt.deleted_at = 0 AND ot.deleted_at = 0 AND "
+        + " tt WHERE tt.schema_id = #{schemaId} AND "
         + "tt.topic_id = ot.metadata_object_id AND ot.metadata_object_type = 'TOPIC'"
         + " UNION "
         + " SELECT tat.schema_id FROM "
         + TableMetaMapper.TABLE_NAME
-        + " tat WHERE tat.schema_id = #{schemaId} AND tat.deleted_at = 0 AND ot.deleted_at = 0 AND "
+        + " tat WHERE tat.schema_id = #{schemaId} AND "
         + "tat.table_id = ot.metadata_object_id AND ot.metadata_object_type = 'TABLE'"
         + " UNION "
         + " SELECT ft.schema_id FROM "
         + FilesetMetaMapper.META_TABLE_NAME
-        + " ft WHERE ft.schema_id = #{schemaId} AND ft.deleted_at = 0 AND ot.deleted_at = 0 AND "
+        + " ft WHERE ft.schema_id = #{schemaId} AND "
         + "ft.fileset_id = ot.metadata_object_id AND ot.metadata_object_type = 'FILESET'"
         + ")";
   }
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SecurableObjectPostgreSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SecurableObjectPostgreSQLProvider.java
index 20867ee4229..92352bcd95a 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SecurableObjectPostgreSQLProvider.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SecurableObjectPostgreSQLProvider.java
@@ -22,6 +22,11 @@
 import static org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper.SECURABLE_OBJECT_TABLE_NAME;
 
 import java.util.List;
+import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TableMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.provider.base.SecurableObjectBaseSQLProvider;
 import org.apache.gravitino.storage.relational.po.SecurableObjectPO;
 import org.apache.ibatis.annotations.Param;
@@ -54,7 +59,7 @@ public String softDeleteSecurableObjectsByRoleId(Long roleId) {
   }
 
   @Override
-  public String softDeleteRoleMetasByMetalakeId(Long metalakeId) {
+  public String softDeleteSecurableObjectsByMetalakeId(Long metalakeId) {
     return "UPDATE "
         + SECURABLE_OBJECT_TABLE_NAME
         + " ob SET deleted_at = floor(extract(epoch from((current_timestamp -"
@@ -64,4 +69,78 @@ public String softDeleteRoleMetasByMetalakeId(Long metalakeId) {
         + " ro WHERE ro.metalake_id = #{metalakeId} AND ro.role_id = ob.role_id"
         + " AND ro.deleted_at = 0) AND ob.deleted_at = 0";
   }
+
+  @Override
+  public String softDeleteObjectRelsByMetadataObject(
+      @Param("metadataObjectId") Long metadataObjectId,
+      @Param("metadataObjectType") String metadataObjectType) {
+    return "UPDATE "
+        + SECURABLE_OBJECT_TABLE_NAME
+        + " SET deleted_at = floor(extract(epoch from((current_timestamp -"
+        + " timestamp '1970-01-01 00:00:00')*1000)))"
+        + " WHERE metadata_object_id = #{metadataObjectId} AND deleted_at = 0 AND type = #{metadataObjectType}";
+  }
+
+  @Override
+  public String softDeleteObjectRelsByCatalogId(@Param("catalogId") Long catalogId) {
+    return "UPDATE "
+        + SECURABLE_OBJECT_TABLE_NAME
+        + " sect SET deleted_at = floor(extract(epoch from((current_timestamp -"
+        + " timestamp '1970-01-01 00:00:00')*1000)))"
+        + " WHERE sect.deleted_at = 0 AND EXISTS ("
+        + " SELECT ct.catalog_id FROM "
+        + CatalogMetaMapper.TABLE_NAME
+        + " ct WHERE ct.catalog_id = #{catalogId} AND "
+        + "ct.catalog_id = sect.metadata_object_id AND sect.type = 'CATALOG'"
+        + " UNION "
+        + " SELECT st.catalog_id FROM "
+        + SchemaMetaMapper.TABLE_NAME
+        + " st WHERE st.catalog_id = #{catalogId} AND "
+        + "st.schema_id = sect.metadata_object_id AND sect.type = 'SCHEMA'"
+        + " UNION "
+        + " SELECT tt.catalog_id FROM "
+        + TopicMetaMapper.TABLE_NAME
+        + " tt WHERE tt.catalog_id = #{catalogId}  AND "
+        + "tt.topic_id = sect.metadata_object_id AND sect.type = 'TOPIC'"
+        + " UNION "
+        + " SELECT tat.catalog_id FROM "
+        + TableMetaMapper.TABLE_NAME
+        + " tat WHERE tat.catalog_id = #{catalogId}  AND "
+        + "tat.table_id = sect.metadata_object_id AND sect.type = 'TABLE'"
+        + " UNION "
+        + " SELECT ft.catalog_id FROM "
+        + FilesetMetaMapper.META_TABLE_NAME
+        + " ft WHERE ft.catalog_id = #{catalogId}  AND"
+        + " ft.fileset_id = sect.metadata_object_id AND sect.type = 'FILESET'"
+        + ")";
+  }
+
+  @Override
+  public String softDeleteObjectRelsBySchemaId(@Param("schemaId") Long schemaId) {
+    return "UPDATE "
+        + SECURABLE_OBJECT_TABLE_NAME
+        + " sect SET deleted_at = floor(extract(epoch from((current_timestamp -"
+        + " timestamp '1970-01-01 00:00:00')*1000)))"
+        + " WHERE sect.deleted_at = 0 AND EXISTS ("
+        + " SELECT st.schema_id FROM "
+        + SchemaMetaMapper.TABLE_NAME
+        + " st WHERE st.schema_id = #{schemaId}  "
+        + "AND st.schema_id = sect.metadata_object_id AND sect.type = 'SCHEMA'"
+        + " UNION "
+        + " SELECT tt.schema_id FROM "
+        + TopicMetaMapper.TABLE_NAME
+        + " tt WHERE tt.schema_id = #{schemaId} AND "
+        + "tt.topic_id = sect.metadata_object_id AND sect.type = 'TOPIC'"
+        + " UNION "
+        + " SELECT tat.schema_id FROM "
+        + TableMetaMapper.TABLE_NAME
+        + " tat WHERE tat.schema_id = #{schemaId}  AND "
+        + "tat.table_id = sect.metadata_object_id AND sect.type = 'TABLE'"
+        + " UNION "
+        + " SELECT ft.schema_id FROM "
+        + FilesetMetaMapper.META_TABLE_NAME
+        + " ft WHERE ft.schema_id = #{schemaId} AND "
+        + "ft.fileset_id = sect.metadata_object_id AND sect.type = 'FILESET'"
+        + ")";
+  }
 }
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TagMetadataObjectRelPostgreSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TagMetadataObjectRelPostgreSQLProvider.java
index ee45f465f7f..827098b6e27 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TagMetadataObjectRelPostgreSQLProvider.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TagMetadataObjectRelPostgreSQLProvider.java
@@ -21,9 +21,17 @@
 import static org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME;
 
 import java.util.List;
+import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TableColumnMapper;
+import org.apache.gravitino.storage.relational.mapper.TableMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.TagMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper;
+import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.provider.base.TagMetadataObjectRelBaseSQLProvider;
+import org.apache.ibatis.annotations.Param;
 
 public class TagMetadataObjectRelPostgreSQLProvider extends TagMetadataObjectRelBaseSQLProvider {
   @Override
@@ -53,6 +61,110 @@ public String softDeleteTagMetadataObjectRelsByMetalakeId(Long metalakeId) {
         + " AND tm.deleted_at = 0) AND te.deleted_at = 0";
   }
 
+  @Override
+  public String softDeleteTagMetadataObjectRelsByMetadataObject(
+      @Param("metadataObjectId") Long metadataObjectId,
+      @Param("metadataObjectType") String metadataObjectType) {
+    return " UPDATE "
+        + TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + " SET deleted_at = floor(extract(epoch from((current_timestamp -"
+        + " timestamp '1970-01-01 00:00:00')*1000))) "
+        + " WHERE metadata_object_id = #{metadataObjectId} AND deleted_at = 0"
+        + " AND metadata_object_type = #{metadataObjectType}";
+  }
+
+  @Override
+  public String softDeleteTagMetadataObjectRelsByCatalogId(@Param("catalogId") Long catalogId) {
+    return " UPDATE "
+        + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + " tmt SET deleted_at =  floor(extract(epoch from((current_timestamp -"
+        + " timestamp '1970-01-01 00:00:00')*1000))) "
+        + " WHERE tmt.deleted_at = 0 AND EXISTS ("
+        + " SELECT ct.catalog_id FROM "
+        + CatalogMetaMapper.TABLE_NAME
+        + " ct WHERE ct.catalog_id = #{catalogId}  AND "
+        + "ct.catalog_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'CATALOG'"
+        + " UNION "
+        + " SELECT st.catalog_id FROM "
+        + SchemaMetaMapper.TABLE_NAME
+        + " st WHERE st.catalog_id = #{catalogId} AND "
+        + "st.schema_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'SCHEMA'"
+        + " UNION "
+        + " SELECT tt.catalog_id FROM "
+        + TopicMetaMapper.TABLE_NAME
+        + " tt WHERE tt.catalog_id = #{catalogId} AND "
+        + "tt.topic_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'TOPIC'"
+        + " UNION "
+        + " SELECT tat.catalog_id FROM "
+        + TableMetaMapper.TABLE_NAME
+        + " tat WHERE tat.catalog_id = #{catalogId} AND "
+        + "tat.table_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'TABLE'"
+        + " UNION "
+        + " SELECT ft.catalog_id FROM "
+        + FilesetMetaMapper.META_TABLE_NAME
+        + " ft WHERE ft.catalog_id = #{catalogId}  AND"
+        + " ft.fileset_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'FILESET'"
+        + " UNION "
+        + " SELECT cot.catalog_id FROM "
+        + TableColumnMapper.COLUMN_TABLE_NAME
+        + " cot WHERE cot.catalog_id = #{catalogId} AND"
+        + " cot.column_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'COLUMN'"
+        + ")";
+  }
+
+  @Override
+  public String softDeleteTagMetadataObjectRelsBySchemaId(@Param("schemaId") Long schemaId) {
+    return " UPDATE "
+        + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + " tmt SET deleted_at =  floor(extract(epoch from((current_timestamp -"
+        + " timestamp '1970-01-01 00:00:00')*1000))) "
+        + " WHERE tmt.deleted_at = 0 AND EXISTS ("
+        + " SELECT st.schema_id FROM "
+        + SchemaMetaMapper.TABLE_NAME
+        + " st WHERE st.schema_id = #{schemaId} AND "
+        + "st.schema_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'SCHEMA'"
+        + " UNION "
+        + " SELECT tt.schema_id FROM "
+        + TopicMetaMapper.TABLE_NAME
+        + " tt WHERE tt.schema_id = #{schemaId} AND "
+        + "tt.topic_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'TOPIC'"
+        + " UNION "
+        + " SELECT tat.schema_id FROM "
+        + TableMetaMapper.TABLE_NAME
+        + " tat WHERE tat.schema_id = #{schemaId} AND"
+        + " tat.table_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'TABLE'"
+        + " UNION "
+        + " SELECT ft.schema_id FROM "
+        + FilesetMetaMapper.META_TABLE_NAME
+        + " ft WHERE ft.schema_id = #{schemaId} AND"
+        + " ft.fileset_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'FILESET'"
+        + " UNION "
+        + " SELECT cot.schema_id FROM "
+        + TableColumnMapper.COLUMN_TABLE_NAME
+        + " cot WHERE cot.schema_id = #{schemaId} AND"
+        + " cot.column_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'COLUMN'"
+        + ")";
+  }
+
+  @Override
+  public String softDeleteTagMetadataObjectRelsByTableId(@Param("tableId") Long tableId) {
+    return " UPDATE "
+        + TagMetadataObjectRelMapper.TAG_METADATA_OBJECT_RELATION_TABLE_NAME
+        + " tmt SET deleted_at =  floor(extract(epoch from((current_timestamp -"
+        + " timestamp '1970-01-01 00:00:00')*1000))) "
+        + " WHERE tmt.deleted_at = 0 AND EXISTS ("
+        + " SELECT tat.table_id FROM "
+        + TableMetaMapper.TABLE_NAME
+        + " tat WHERE tat.table_id = #{tableId} AND "
+        + " tat.table_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'TABLE'"
+        + " UNION "
+        + " SELECT cot.table_id FROM "
+        + TableColumnMapper.COLUMN_TABLE_NAME
+        + " cot WHERE cot.table_id = #{tableId} AND "
+        + " cot.column_id = tmt.metadata_object_id AND tmt.metadata_object_type = 'COLUMN'"
+        + ")";
+  }
+
   @Override
   public String batchDeleteTagMetadataObjectRelsByTagIdsAndMetadataObject(
       Long metadataObjectId, String metadataObjectType, List<Long> tagIds) {
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java
index 6e6e8aaa41f..15f1d1a3cf1 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java
@@ -38,8 +38,10 @@
 import org.apache.gravitino.storage.relational.mapper.FilesetVersionMapper;
 import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
 import org.apache.gravitino.storage.relational.mapper.TableColumnMapper;
 import org.apache.gravitino.storage.relational.mapper.TableMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper;
 import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper;
 import org.apache.gravitino.storage.relational.po.CatalogPO;
 import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
@@ -230,8 +232,15 @@ public boolean deleteCatalog(NameIdentifier identifier, boolean cascade) {
                   mapper -> mapper.softDeleteTopicMetasByCatalogId(catalogId)),
           () ->
               SessionUtils.doWithoutCommit(
-                  OwnerMetaMapper.class,
-                  mapper -> mapper.softDeleteOwnerRelByCatalogId(catalogId)));
+                  OwnerMetaMapper.class, mapper -> mapper.softDeleteOwnerRelByCatalogId(catalogId)),
+          () ->
+              SessionUtils.doWithoutCommit(
+                  SecurableObjectMapper.class,
+                  mapper -> mapper.softDeleteObjectRelsByCatalogId(catalogId)),
+          () ->
+              SessionUtils.doWithoutCommit(
+                  TagMetadataObjectRelMapper.class,
+                  mapper -> mapper.softDeleteTagMetadataObjectRelsByCatalogId(catalogId)));
     } else {
       List<SchemaEntity> schemaEntities =
           SchemaMetaService.getInstance()
@@ -251,6 +260,18 @@ public boolean deleteCatalog(NameIdentifier identifier, boolean cascade) {
                   OwnerMetaMapper.class,
                   mapper ->
                       mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
+                          catalogId, MetadataObject.Type.CATALOG.name())),
+          () ->
+              SessionUtils.doWithoutCommit(
+                  SecurableObjectMapper.class,
+                  mapper ->
+                      mapper.softDeleteObjectRelsByMetadataObject(
+                          catalogId, MetadataObject.Type.CATALOG.name())),
+          () ->
+              SessionUtils.doWithoutCommit(
+                  TagMetadataObjectRelMapper.class,
+                  mapper ->
+                      mapper.softDeleteTagMetadataObjectRelsByMetadataObject(
                           catalogId, MetadataObject.Type.CATALOG.name())));
     }
 
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java
index 4d63c5f19b7..e049f436406 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java
@@ -33,6 +33,8 @@
 import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.FilesetVersionMapper;
 import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
+import org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper;
 import org.apache.gravitino.storage.relational.po.FilesetMaxVersionPO;
 import org.apache.gravitino.storage.relational.po.FilesetPO;
 import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
@@ -245,6 +247,18 @@ public boolean deleteFileset(NameIdentifier identifier) {
                 OwnerMetaMapper.class,
                 mapper ->
                     mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
+                        filesetId, MetadataObject.Type.FILESET.name())),
+        () ->
+            SessionUtils.doWithoutCommit(
+                SecurableObjectMapper.class,
+                mapper ->
+                    mapper.softDeleteObjectRelsByMetadataObject(
+                        filesetId, MetadataObject.Type.FILESET.name())),
+        () ->
+            SessionUtils.doWithoutCommit(
+                TagMetadataObjectRelMapper.class,
+                mapper ->
+                    mapper.softDeleteTagMetadataObjectRelsByMetadataObject(
                         filesetId, MetadataObject.Type.FILESET.name())));
 
     return true;
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java
index dab3ba1dc54..cf05af8129a 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java
@@ -231,7 +231,7 @@ public boolean deleteMetalake(NameIdentifier ident, boolean cascade) {
             () ->
                 SessionUtils.doWithoutCommit(
                     SecurableObjectMapper.class,
-                    mapper -> mapper.softDeleteRoleMetasByMetalakeId(metalakeId)),
+                    mapper -> mapper.softDeleteSecurableObjectsByMetalakeId(metalakeId)),
             () ->
                 SessionUtils.doWithoutCommit(
                     TagMetaMapper.class,
@@ -280,7 +280,7 @@ public boolean deleteMetalake(NameIdentifier ident, boolean cascade) {
             () ->
                 SessionUtils.doWithoutCommit(
                     SecurableObjectMapper.class,
-                    mapper -> mapper.softDeleteRoleMetasByMetalakeId(metalakeId)),
+                    mapper -> mapper.softDeleteSecurableObjectsByMetalakeId(metalakeId)),
             () ->
                 SessionUtils.doWithoutCommit(
                     TagMetaMapper.class,
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java
index d2b125f1c36..5d3fa1b4f01 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/SchemaMetaService.java
@@ -37,8 +37,10 @@
 import org.apache.gravitino.storage.relational.mapper.FilesetVersionMapper;
 import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
 import org.apache.gravitino.storage.relational.mapper.TableColumnMapper;
 import org.apache.gravitino.storage.relational.mapper.TableMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper;
 import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper;
 import org.apache.gravitino.storage.relational.po.SchemaPO;
 import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
@@ -216,8 +218,15 @@ public boolean deleteSchema(NameIdentifier identifier, boolean cascade) {
                     mapper -> mapper.softDeleteTopicMetasBySchemaId(schemaId)),
             () ->
                 SessionUtils.doWithoutCommit(
-                    OwnerMetaMapper.class,
-                    mapper -> mapper.softDeleteOwnerRelBySchemaId(schemaId)));
+                    OwnerMetaMapper.class, mapper -> mapper.softDeleteOwnerRelBySchemaId(schemaId)),
+            () ->
+                SessionUtils.doWithoutCommit(
+                    SecurableObjectMapper.class,
+                    mapper -> mapper.softDeleteObjectRelsBySchemaId(schemaId)),
+            () ->
+                SessionUtils.doWithoutCommit(
+                    TagMetadataObjectRelMapper.class,
+                    mapper -> mapper.softDeleteTagMetadataObjectRelsBySchemaId(schemaId)));
       } else {
         List<TableEntity> tableEntities =
             TableMetaService.getInstance()
@@ -251,6 +260,18 @@ public boolean deleteSchema(NameIdentifier identifier, boolean cascade) {
                     OwnerMetaMapper.class,
                     mapper ->
                         mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
+                            schemaId, MetadataObject.Type.SCHEMA.name())),
+            () ->
+                SessionUtils.doWithoutCommit(
+                    SecurableObjectMapper.class,
+                    mapper ->
+                        mapper.softDeleteObjectRelsByMetadataObject(
+                            schemaId, MetadataObject.Type.SCHEMA.name())),
+            () ->
+                SessionUtils.doWithoutCommit(
+                    TagMetadataObjectRelMapper.class,
+                    mapper ->
+                        mapper.softDeleteTagMetadataObjectRelsByMetadataObject(
                             schemaId, MetadataObject.Type.SCHEMA.name())));
       }
     }
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
index ed7afe748de..248dedd8a73 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java
@@ -33,7 +33,9 @@
 import org.apache.gravitino.exceptions.NoSuchEntityException;
 import org.apache.gravitino.meta.TableEntity;
 import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
 import org.apache.gravitino.storage.relational.mapper.TableMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper;
 import org.apache.gravitino.storage.relational.po.ColumnPO;
 import org.apache.gravitino.storage.relational.po.TablePO;
 import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
@@ -213,7 +215,7 @@ public boolean deleteTable(NameIdentifier identifier) {
     SessionUtils.doMultipleWithCommit(
         () ->
             deleteResult.set(
-                SessionUtils.doWithCommitAndFetchResult(
+                SessionUtils.doWithoutCommitAndFetchResult(
                     TableMetaMapper.class,
                     mapper -> mapper.softDeleteTableMetasByTableId(tableId))),
         () -> {
@@ -223,11 +225,20 @@ public boolean deleteTable(NameIdentifier identifier) {
                 mapper ->
                     mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
                         tableId, MetadataObject.Type.TABLE.name()));
-          }
-        },
-        () -> {
-          if (deleteResult.get() > 0) {
             TableColumnMetaService.getInstance().deleteColumnsByTableId(tableId);
+            SessionUtils.doWithoutCommit(
+                SecurableObjectMapper.class,
+                mapper ->
+                    mapper.softDeleteObjectRelsByMetadataObject(
+                        tableId, MetadataObject.Type.TABLE.name()));
+            SessionUtils.doWithoutCommit(
+                TagMetadataObjectRelMapper.class,
+                mapper ->
+                    mapper.softDeleteTagMetadataObjectRelsByMetadataObject(
+                        tableId, MetadataObject.Type.TABLE.name()));
+            SessionUtils.doWithoutCommit(
+                TagMetadataObjectRelMapper.class,
+                mapper -> mapper.softDeleteTagMetadataObjectRelsByTableId(tableId));
           }
         });
 
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java
index f13fc202abf..7bc933824aa 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java
@@ -31,6 +31,8 @@
 import org.apache.gravitino.exceptions.NoSuchEntityException;
 import org.apache.gravitino.meta.TopicEntity;
 import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
+import org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper;
 import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper;
 import org.apache.gravitino.storage.relational.po.TopicPO;
 import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
@@ -205,6 +207,18 @@ public boolean deleteTopic(NameIdentifier identifier) {
                 OwnerMetaMapper.class,
                 mapper ->
                     mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
+                        topicId, MetadataObject.Type.TOPIC.name())),
+        () ->
+            SessionUtils.doWithoutCommit(
+                SecurableObjectMapper.class,
+                mapper ->
+                    mapper.softDeleteObjectRelsByMetadataObject(
+                        topicId, MetadataObject.Type.TOPIC.name())),
+        () ->
+            SessionUtils.doWithoutCommit(
+                TagMetadataObjectRelMapper.class,
+                mapper ->
+                    mapper.softDeleteTagMetadataObjectRelsByMetadataObject(
                         topicId, MetadataObject.Type.TOPIC.name())));
 
     return true;
diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/TestJDBCBackend.java b/core/src/test/java/org/apache/gravitino/storage/relational/TestJDBCBackend.java
index bb7586cde95..ca564f73be7 100644
--- a/core/src/test/java/org/apache/gravitino/storage/relational/TestJDBCBackend.java
+++ b/core/src/test/java/org/apache/gravitino/storage/relational/TestJDBCBackend.java
@@ -1036,6 +1036,122 @@ private Integer countOwnerRel(Long metalakeId) {
     }
   }
 
+  protected Integer countAllObjectRel(Long roleId) {
+    try (SqlSession sqlSession =
+            SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+        Connection connection = sqlSession.getConnection();
+        Statement statement1 = connection.createStatement();
+        ResultSet rs1 =
+            statement1.executeQuery(
+                String.format(
+                    "SELECT count(*) FROM role_meta_securable_object WHERE role_id = %d",
+                    roleId))) {
+      if (rs1.next()) {
+        return rs1.getInt(1);
+      } else {
+        throw new RuntimeException("Doesn't contain data");
+      }
+    } catch (SQLException se) {
+      throw new RuntimeException("SQL execution failed", se);
+    }
+  }
+
+  protected Integer countActiveObjectRel(Long roleId) {
+    try (SqlSession sqlSession =
+            SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+        Connection connection = sqlSession.getConnection();
+        Statement statement1 = connection.createStatement();
+        ResultSet rs1 =
+            statement1.executeQuery(
+                String.format(
+                    "SELECT count(*) FROM role_meta_securable_object WHERE role_id = %d AND deleted_at = 0",
+                    roleId))) {
+      if (rs1.next()) {
+        return rs1.getInt(1);
+      } else {
+        throw new RuntimeException("Doesn't contain data");
+      }
+    } catch (SQLException se) {
+      throw new RuntimeException("SQL execution failed", se);
+    }
+  }
+
+  protected Integer countAllTagRel(Long tagId) {
+    try (SqlSession sqlSession =
+            SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+        Connection connection = sqlSession.getConnection();
+        Statement statement1 = connection.createStatement();
+        ResultSet rs1 =
+            statement1.executeQuery(
+                String.format("SELECT count(*) FROM tag_relation_meta WHERE tag_id = %d", tagId))) {
+      if (rs1.next()) {
+        return rs1.getInt(1);
+      } else {
+        throw new RuntimeException("Doesn't contain data");
+      }
+    } catch (SQLException se) {
+      throw new RuntimeException("SQL execution failed", se);
+    }
+  }
+
+  protected Integer countActiveTagRel(Long tagId) {
+    try (SqlSession sqlSession =
+            SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+        Connection connection = sqlSession.getConnection();
+        Statement statement1 = connection.createStatement();
+        ResultSet rs1 =
+            statement1.executeQuery(
+                String.format(
+                    "SELECT count(*) FROM tag_relation_meta WHERE tag_id = %d AND deleted_at = 0",
+                    tagId))) {
+      if (rs1.next()) {
+        return rs1.getInt(1);
+      } else {
+        throw new RuntimeException("Doesn't contain data");
+      }
+    } catch (SQLException se) {
+      throw new RuntimeException("SQL execution failed", se);
+    }
+  }
+
+  protected Integer countAllOwnerRel(Long ownerId) {
+    try (SqlSession sqlSession =
+            SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+        Connection connection = sqlSession.getConnection();
+        Statement statement1 = connection.createStatement();
+        ResultSet rs1 =
+            statement1.executeQuery(
+                String.format("SELECT count(*) FROM owner_meta WHERE owner_id = %d", ownerId))) {
+      if (rs1.next()) {
+        return rs1.getInt(1);
+      } else {
+        throw new RuntimeException("Doesn't contain data");
+      }
+    } catch (SQLException se) {
+      throw new RuntimeException("SQL execution failed", se);
+    }
+  }
+
+  protected Integer countActiveOwnerRel(long ownerId) {
+    try (SqlSession sqlSession =
+            SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+        Connection connection = sqlSession.getConnection();
+        Statement statement1 = connection.createStatement();
+        ResultSet rs1 =
+            statement1.executeQuery(
+                String.format(
+                    "SELECT count(*) FROM owner_meta WHERE owner_id = %d AND deleted_at = 0",
+                    ownerId))) {
+      if (rs1.next()) {
+        return rs1.getInt(1);
+      } else {
+        throw new RuntimeException("Doesn't contain data");
+      }
+    } catch (SQLException se) {
+      throw new RuntimeException("SQL execution failed", se);
+    }
+  }
+
   public static BaseMetalake createBaseMakeLake(Long id, String name, AuditInfo auditInfo) {
     return BaseMetalake.builder()
         .withId(id)
diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestOwnerMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestOwnerMetaService.java
index 5625edf62d8..89dea63a591 100644
--- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestOwnerMetaService.java
+++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestOwnerMetaService.java
@@ -216,4 +216,244 @@ void testDifferentEntities() throws IOException {
     Assertions.assertTrue(entity instanceof UserEntity);
     Assertions.assertEquals("user", ((UserEntity) entity).name());
   }
+
+  @Test
+  public void testDeleteMetadataObject() throws IOException {
+    String metalakeName = "metalake";
+    AuditInfo auditInfo =
+        AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+    BaseMetalake metalake =
+        createBaseMakeLake(RandomIdGenerator.INSTANCE.nextId(), metalakeName, auditInfo);
+    backend.insert(metalake, false);
+
+    CatalogEntity catalog =
+        createCatalog(
+            RandomIdGenerator.INSTANCE.nextId(), Namespace.of("metalake"), "catalog", auditInfo);
+    backend.insert(catalog, false);
+
+    SchemaEntity schema =
+        createSchemaEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog"),
+            "schema",
+            auditInfo);
+    backend.insert(schema, false);
+
+    FilesetEntity fileset =
+        createFilesetEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "fileset",
+            auditInfo);
+    backend.insert(fileset, false);
+    TableEntity table =
+        createTableEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "table",
+            auditInfo);
+    backend.insert(table, false);
+    TopicEntity topic =
+        createTopicEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "topic",
+            auditInfo);
+    backend.insert(topic, false);
+
+    UserEntity user =
+        createUserEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            AuthorizationUtils.ofUserNamespace(metalakeName),
+            "user",
+            auditInfo);
+    backend.insert(user, false);
+
+    OwnerMetaService.getInstance()
+        .setOwner(catalog.nameIdentifier(), catalog.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(schema.nameIdentifier(), schema.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(table.nameIdentifier(), table.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(fileset.nameIdentifier(), fileset.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(topic.nameIdentifier(), topic.type(), user.nameIdentifier(), user.type());
+
+    Assertions.assertEquals(5, countAllOwnerRel(user.id()));
+    Assertions.assertEquals(5, countActiveOwnerRel(user.id()));
+
+    // Test to delete table
+    TableMetaService.getInstance().deleteTable(table.nameIdentifier());
+    Assertions.assertEquals(5, countAllOwnerRel(user.id()));
+    Assertions.assertEquals(4, countActiveOwnerRel(user.id()));
+
+    // Test to delete topic
+    TopicMetaService.getInstance().deleteTopic(topic.nameIdentifier());
+    Assertions.assertEquals(5, countAllOwnerRel(user.id()));
+    Assertions.assertEquals(3, countActiveOwnerRel(user.id()));
+
+    // Test to delete fileset
+    FilesetMetaService.getInstance().deleteFileset(fileset.nameIdentifier());
+    Assertions.assertEquals(5, countAllOwnerRel(user.id()));
+    Assertions.assertEquals(2, countActiveOwnerRel(user.id()));
+
+    // Test to delete schema
+    SchemaMetaService.getInstance().deleteSchema(schema.nameIdentifier(), false);
+    Assertions.assertEquals(5, countAllOwnerRel(user.id()));
+    Assertions.assertEquals(1, countActiveOwnerRel(user.id()));
+
+    // Test to delete catalog
+    CatalogMetaService.getInstance().deleteCatalog(catalog.nameIdentifier(), false);
+    Assertions.assertEquals(5, countAllOwnerRel(user.id()));
+    Assertions.assertEquals(0, countActiveOwnerRel(user.id()));
+
+    // Test to delete catalog with cascade mode
+    catalog =
+        createCatalog(
+            RandomIdGenerator.INSTANCE.nextId(), Namespace.of("metalake"), "catalog", auditInfo);
+    backend.insert(catalog, false);
+
+    schema =
+        createSchemaEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog"),
+            "schema",
+            auditInfo);
+    backend.insert(schema, false);
+
+    fileset =
+        createFilesetEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "fileset",
+            auditInfo);
+    backend.insert(fileset, false);
+    table =
+        createTableEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "table",
+            auditInfo);
+    backend.insert(table, false);
+    topic =
+        createTopicEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "topic",
+            auditInfo);
+    backend.insert(topic, false);
+
+    OwnerMetaService.getInstance()
+        .setOwner(catalog.nameIdentifier(), catalog.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(schema.nameIdentifier(), schema.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(table.nameIdentifier(), table.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(fileset.nameIdentifier(), fileset.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(topic.nameIdentifier(), topic.type(), user.nameIdentifier(), user.type());
+
+    CatalogMetaService.getInstance().deleteCatalog(catalog.nameIdentifier(), true);
+    Assertions.assertEquals(10, countAllOwnerRel(user.id()));
+    Assertions.assertEquals(0, countActiveOwnerRel(user.id()));
+
+    // Test to delete schema with cascade mode
+    catalog =
+        createCatalog(
+            RandomIdGenerator.INSTANCE.nextId(), Namespace.of("metalake"), "catalog", auditInfo);
+    backend.insert(catalog, false);
+
+    schema =
+        createSchemaEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog"),
+            "schema",
+            auditInfo);
+    backend.insert(schema, false);
+
+    fileset =
+        createFilesetEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "fileset",
+            auditInfo);
+    backend.insert(fileset, false);
+    table =
+        createTableEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "table",
+            auditInfo);
+    backend.insert(table, false);
+    topic =
+        createTopicEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "topic",
+            auditInfo);
+    backend.insert(topic, false);
+
+    OwnerMetaService.getInstance()
+        .setOwner(schema.nameIdentifier(), schema.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(catalog.nameIdentifier(), catalog.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(table.nameIdentifier(), table.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(fileset.nameIdentifier(), fileset.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(topic.nameIdentifier(), topic.type(), user.nameIdentifier(), user.type());
+
+    SchemaMetaService.getInstance().deleteSchema(schema.nameIdentifier(), true);
+    Assertions.assertEquals(15, countAllOwnerRel(user.id()));
+    Assertions.assertEquals(1, countActiveOwnerRel(user.id()));
+
+    // Test to delete user
+    schema =
+        createSchemaEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog"),
+            "schema",
+            auditInfo);
+    backend.insert(schema, false);
+
+    fileset =
+        createFilesetEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "fileset",
+            auditInfo);
+    backend.insert(fileset, false);
+    table =
+        createTableEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "table",
+            auditInfo);
+    backend.insert(table, false);
+    topic =
+        createTopicEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "topic",
+            auditInfo);
+    backend.insert(topic, false);
+
+    OwnerMetaService.getInstance()
+        .setOwner(schema.nameIdentifier(), schema.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(catalog.nameIdentifier(), catalog.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(table.nameIdentifier(), table.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(fileset.nameIdentifier(), fileset.type(), user.nameIdentifier(), user.type());
+    OwnerMetaService.getInstance()
+        .setOwner(topic.nameIdentifier(), topic.type(), user.nameIdentifier(), user.type());
+
+    UserMetaService.getInstance().deleteUser(user.nameIdentifier());
+    Assertions.assertEquals(20, countAllOwnerRel(user.id()));
+    Assertions.assertEquals(0, countActiveOwnerRel(user.id()));
+  }
 }
diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSecurableObjects.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSecurableObjects.java
index 49a5e9cc587..d18893472b6 100644
--- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSecurableObjects.java
+++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSecurableObjects.java
@@ -118,4 +118,213 @@ public void testAllTypeSecurableObjects() throws IOException {
     Assertions.assertDoesNotThrow(() -> roleMetaService.insertRole(role1, false));
     Assertions.assertEquals(role1, roleMetaService.getRoleByIdentifier(role1.nameIdentifier()));
   }
+
+  @Test
+  public void testDeleteMetadataObject() throws IOException {
+    String metalakeName = "metalake";
+    AuditInfo auditInfo =
+        AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+    BaseMetalake metalake =
+        createBaseMakeLake(RandomIdGenerator.INSTANCE.nextId(), metalakeName, auditInfo);
+    backend.insert(metalake, false);
+
+    CatalogEntity catalog =
+        createCatalog(
+            RandomIdGenerator.INSTANCE.nextId(), Namespace.of("metalake"), "catalog", auditInfo);
+    backend.insert(catalog, false);
+
+    SchemaEntity schema =
+        createSchemaEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog"),
+            "schema",
+            auditInfo);
+    backend.insert(schema, false);
+
+    FilesetEntity fileset =
+        createFilesetEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "fileset",
+            auditInfo);
+    backend.insert(fileset, false);
+    TableEntity table =
+        createTableEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "table",
+            auditInfo);
+    backend.insert(table, false);
+    TopicEntity topic =
+        createTopicEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "topic",
+            auditInfo);
+    backend.insert(topic, false);
+
+    SecurableObject catalogObject =
+        SecurableObjects.ofCatalog(
+            "catalog",
+            Lists.newArrayList(Privileges.UseCatalog.allow(), Privileges.CreateSchema.deny()));
+
+    SecurableObject schemaObject =
+        SecurableObjects.ofSchema(
+            catalogObject, "schema", Lists.newArrayList(Privileges.UseSchema.allow()));
+    SecurableObject tableObject =
+        SecurableObjects.ofTable(
+            schemaObject, "table", Lists.newArrayList(Privileges.SelectTable.allow()));
+    SecurableObject filesetObject =
+        SecurableObjects.ofFileset(
+            schemaObject, "fileset", Lists.newArrayList(Privileges.ReadFileset.allow()));
+    SecurableObject topicObject =
+        SecurableObjects.ofTopic(
+            schemaObject, "topic", Lists.newArrayList(Privileges.ConsumeTopic.deny()));
+
+    RoleEntity role1 =
+        createRoleEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            AuthorizationUtils.ofRoleNamespace(metalakeName),
+            "role1",
+            auditInfo,
+            Lists.newArrayList(
+                catalogObject, schemaObject, tableObject, filesetObject, topicObject),
+            ImmutableMap.of("k1", "v1"));
+
+    roleMetaService.insertRole(role1, false);
+
+    Assertions.assertEquals(5, countAllObjectRel(role1.id()));
+    Assertions.assertEquals(5, countActiveObjectRel(role1.id()));
+
+    // Test to delete table
+    TableMetaService.getInstance().deleteTable(table.nameIdentifier());
+    Assertions.assertEquals(5, countAllObjectRel(role1.id()));
+    Assertions.assertEquals(4, countActiveObjectRel(role1.id()));
+
+    // Test to delete topic
+    TopicMetaService.getInstance().deleteTopic(topic.nameIdentifier());
+    Assertions.assertEquals(5, countAllObjectRel(role1.id()));
+    Assertions.assertEquals(3, countActiveObjectRel(role1.id()));
+
+    // Test to delete fileset
+    FilesetMetaService.getInstance().deleteFileset(fileset.nameIdentifier());
+    Assertions.assertEquals(5, countAllObjectRel(role1.id()));
+    Assertions.assertEquals(2, countActiveObjectRel(role1.id()));
+
+    // Test to delete schema
+    SchemaMetaService.getInstance().deleteSchema(schema.nameIdentifier(), false);
+    Assertions.assertEquals(5, countAllObjectRel(role1.id()));
+    Assertions.assertEquals(1, countActiveObjectRel(role1.id()));
+
+    // Test to delete catalog
+    CatalogMetaService.getInstance().deleteCatalog(catalog.nameIdentifier(), false);
+    Assertions.assertEquals(5, countAllObjectRel(role1.id()));
+    Assertions.assertEquals(0, countActiveObjectRel(role1.id()));
+
+    roleMetaService.deleteRole(role1.nameIdentifier());
+
+    // Test to delete catalog with cascade mode
+    catalog =
+        createCatalog(
+            RandomIdGenerator.INSTANCE.nextId(), Namespace.of("metalake"), "catalog", auditInfo);
+    backend.insert(catalog, false);
+
+    schema =
+        createSchemaEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog"),
+            "schema",
+            auditInfo);
+    backend.insert(schema, false);
+
+    fileset =
+        createFilesetEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "fileset",
+            auditInfo);
+    backend.insert(fileset, false);
+    table =
+        createTableEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "table",
+            auditInfo);
+    backend.insert(table, false);
+    topic =
+        createTopicEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "topic",
+            auditInfo);
+    backend.insert(topic, false);
+    role1 =
+        createRoleEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            AuthorizationUtils.ofRoleNamespace(metalakeName),
+            "role1",
+            auditInfo,
+            Lists.newArrayList(
+                catalogObject, schemaObject, tableObject, filesetObject, topicObject),
+            ImmutableMap.of("k1", "v1"));
+
+    roleMetaService.insertRole(role1, false);
+
+    CatalogMetaService.getInstance().deleteCatalog(catalog.nameIdentifier(), true);
+    Assertions.assertEquals(5, countAllObjectRel(role1.id()));
+    Assertions.assertEquals(0, countActiveObjectRel(role1.id()));
+
+    roleMetaService.deleteRole(role1.nameIdentifier());
+
+    // Test to delete schema with cascade mode
+    catalog =
+        createCatalog(
+            RandomIdGenerator.INSTANCE.nextId(), Namespace.of("metalake"), "catalog", auditInfo);
+    backend.insert(catalog, false);
+
+    schema =
+        createSchemaEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog"),
+            "schema",
+            auditInfo);
+    backend.insert(schema, false);
+
+    fileset =
+        createFilesetEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "fileset",
+            auditInfo);
+    backend.insert(fileset, false);
+    table =
+        createTableEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "table",
+            auditInfo);
+    backend.insert(table, false);
+    topic =
+        createTopicEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of("metalake", "catalog", "schema"),
+            "topic",
+            auditInfo);
+    backend.insert(topic, false);
+    role1 =
+        createRoleEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            AuthorizationUtils.ofRoleNamespace(metalakeName),
+            "role1",
+            auditInfo,
+            Lists.newArrayList(
+                catalogObject, schemaObject, tableObject, filesetObject, topicObject),
+            ImmutableMap.of("k1", "v1"));
+
+    roleMetaService.insertRole(role1, false);
+
+    SchemaMetaService.getInstance().deleteSchema(schema.nameIdentifier(), true);
+    Assertions.assertEquals(5, countAllObjectRel(role1.id()));
+    Assertions.assertEquals(1, countActiveObjectRel(role1.id()));
+  }
 }
diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTagMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTagMetaService.java
index 5bb76ad9cc5..8194a17061c 100644
--- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTagMetaService.java
+++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTagMetaService.java
@@ -23,6 +23,7 @@
 import java.time.Instant;
 import java.util.List;
 import java.util.Map;
+import org.apache.commons.compress.utils.Lists;
 import org.apache.gravitino.Entity;
 import org.apache.gravitino.EntityAlreadyExistsException;
 import org.apache.gravitino.MetadataObject;
@@ -33,9 +34,13 @@
 import org.apache.gravitino.meta.AuditInfo;
 import org.apache.gravitino.meta.BaseMetalake;
 import org.apache.gravitino.meta.CatalogEntity;
+import org.apache.gravitino.meta.ColumnEntity;
+import org.apache.gravitino.meta.FilesetEntity;
 import org.apache.gravitino.meta.SchemaEntity;
 import org.apache.gravitino.meta.TableEntity;
 import org.apache.gravitino.meta.TagEntity;
+import org.apache.gravitino.meta.TopicEntity;
+import org.apache.gravitino.rel.types.Types;
 import org.apache.gravitino.storage.RandomIdGenerator;
 import org.apache.gravitino.storage.relational.TestJDBCBackend;
 import org.apache.gravitino.tag.TagManager;
@@ -658,4 +663,322 @@ public void testListAssociatedMetadataObjectsForTag() throws IOException {
 
     Assertions.assertEquals(0, metadataObjects5.size());
   }
+
+  @Test
+  public void testDeleteMetadataObjectForTag() throws IOException {
+    BaseMetalake metalake =
+        createBaseMakeLake(RandomIdGenerator.INSTANCE.nextId(), metalakeName, auditInfo);
+    backend.insert(metalake, false);
+
+    CatalogEntity catalog =
+        createCatalog(
+            RandomIdGenerator.INSTANCE.nextId(), Namespace.of(metalakeName), "catalog1", auditInfo);
+    backend.insert(catalog, false);
+
+    SchemaEntity schema =
+        createSchemaEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of(metalakeName, catalog.name()),
+            "schema1",
+            auditInfo);
+    backend.insert(schema, false);
+
+    ColumnEntity column =
+        ColumnEntity.builder()
+            .withId(RandomIdGenerator.INSTANCE.nextId())
+            .withName("column1")
+            .withPosition(0)
+            .withAutoIncrement(false)
+            .withNullable(false)
+            .withDataType(Types.IntegerType.get())
+            .withAuditInfo(auditInfo)
+            .build();
+
+    List<ColumnEntity> columns = Lists.newArrayList();
+    columns.add(column);
+
+    TableEntity table =
+        TableEntity.builder()
+            .withName("table")
+            .withNamespace(Namespace.of(metalakeName, catalog.name(), schema.name()))
+            .withId(RandomIdGenerator.INSTANCE.nextId())
+            .withColumns(columns)
+            .withAuditInfo(auditInfo)
+            .build();
+
+    backend.insert(table, false);
+
+    TopicEntity topic =
+        createTopicEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of(metalakeName, catalog.name(), schema.name()),
+            "topic1",
+            auditInfo);
+    backend.insert(topic, false);
+
+    FilesetEntity fileset =
+        createFilesetEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of(metalakeName, catalog.name(), schema.name()),
+            "fileset1",
+            auditInfo);
+    backend.insert(fileset, false);
+
+    TagMetaService tagMetaService = TagMetaService.getInstance();
+    TagEntity tagEntity1 =
+        TagEntity.builder()
+            .withId(RandomIdGenerator.INSTANCE.nextId())
+            .withName("tag1")
+            .withNamespace(TagManager.ofTagNamespace(metalakeName))
+            .withComment("comment")
+            .withProperties(props)
+            .withAuditInfo(auditInfo)
+            .build();
+    tagMetaService.insertTag(tagEntity1, false);
+    tagMetaService.associateTagsWithMetadataObject(
+        catalog.nameIdentifier(),
+        catalog.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+    tagMetaService.associateTagsWithMetadataObject(
+        schema.nameIdentifier(),
+        schema.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+    tagMetaService.associateTagsWithMetadataObject(
+        table.nameIdentifier(),
+        table.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+    tagMetaService.associateTagsWithMetadataObject(
+        topic.nameIdentifier(),
+        topic.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+    tagMetaService.associateTagsWithMetadataObject(
+        fileset.nameIdentifier(),
+        fileset.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+    NameIdentifier columnIdentifier =
+        NameIdentifier.of(Namespace.fromString(table.nameIdentifier().toString()), column.name());
+    tagMetaService.associateTagsWithMetadataObject(
+        columnIdentifier,
+        column.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+
+    Assertions.assertEquals(6, countActiveTagRel(tagEntity1.id()));
+    Assertions.assertEquals(6, countAllTagRel(tagEntity1.id()));
+
+    // Test to drop a table
+    TableMetaService.getInstance().deleteTable(table.nameIdentifier());
+    Assertions.assertEquals(4, countActiveTagRel(tagEntity1.id()));
+    Assertions.assertEquals(6, countAllTagRel(tagEntity1.id()));
+
+    // Test to drop a topic
+    TopicMetaService.getInstance().deleteTopic(topic.nameIdentifier());
+    Assertions.assertEquals(3, countActiveTagRel(tagEntity1.id()));
+    Assertions.assertEquals(6, countAllTagRel(tagEntity1.id()));
+
+    // Test to drop a fileset
+    FilesetMetaService.getInstance().deleteFileset(fileset.nameIdentifier());
+    Assertions.assertEquals(2, countActiveTagRel(tagEntity1.id()));
+    Assertions.assertEquals(6, countAllTagRel(tagEntity1.id()));
+
+    // Test to drop a schema
+    SchemaMetaService.getInstance().deleteSchema(schema.nameIdentifier(), false);
+    Assertions.assertEquals(1, countActiveTagRel(tagEntity1.id()));
+    Assertions.assertEquals(6, countAllTagRel(tagEntity1.id()));
+
+    // Test to drop a catalog
+    CatalogMetaService.getInstance().deleteCatalog(catalog.nameIdentifier(), false);
+    Assertions.assertEquals(0, countActiveTagRel(tagEntity1.id()));
+    Assertions.assertEquals(6, countAllTagRel(tagEntity1.id()));
+
+    // Test to drop a catalog using cascade mode
+    catalog =
+        createCatalog(
+            RandomIdGenerator.INSTANCE.nextId(), Namespace.of(metalakeName), "catalog1", auditInfo);
+    backend.insert(catalog, false);
+
+    schema =
+        createSchemaEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of(metalakeName, catalog.name()),
+            "schema1",
+            auditInfo);
+    backend.insert(schema, false);
+
+    column =
+        ColumnEntity.builder()
+            .withId(RandomIdGenerator.INSTANCE.nextId())
+            .withName("column1")
+            .withPosition(0)
+            .withAutoIncrement(false)
+            .withNullable(false)
+            .withDataType(Types.IntegerType.get())
+            .withAuditInfo(auditInfo)
+            .build();
+
+    columns = Lists.newArrayList();
+    columns.add(column);
+
+    table =
+        TableEntity.builder()
+            .withName("table")
+            .withNamespace(Namespace.of(metalakeName, catalog.name(), schema.name()))
+            .withId(RandomIdGenerator.INSTANCE.nextId())
+            .withColumns(columns)
+            .withAuditInfo(auditInfo)
+            .build();
+
+    backend.insert(table, false);
+
+    topic =
+        createTopicEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of(metalakeName, catalog.name(), schema.name()),
+            "topic1",
+            auditInfo);
+    backend.insert(topic, false);
+
+    fileset =
+        createFilesetEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of(metalakeName, catalog.name(), schema.name()),
+            "fileset1",
+            auditInfo);
+    backend.insert(fileset, false);
+
+    tagMetaService.associateTagsWithMetadataObject(
+        catalog.nameIdentifier(),
+        catalog.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+    tagMetaService.associateTagsWithMetadataObject(
+        schema.nameIdentifier(),
+        schema.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+    tagMetaService.associateTagsWithMetadataObject(
+        table.nameIdentifier(),
+        table.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+    tagMetaService.associateTagsWithMetadataObject(
+        topic.nameIdentifier(),
+        topic.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+    tagMetaService.associateTagsWithMetadataObject(
+        fileset.nameIdentifier(),
+        fileset.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+    columnIdentifier =
+        NameIdentifier.of(Namespace.fromString(table.nameIdentifier().toString()), column.name());
+    tagMetaService.associateTagsWithMetadataObject(
+        columnIdentifier,
+        column.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+
+    CatalogMetaService.getInstance().deleteCatalog(catalog.nameIdentifier(), true);
+    Assertions.assertEquals(0, countActiveTagRel(tagEntity1.id()));
+    Assertions.assertEquals(12, countAllTagRel(tagEntity1.id()));
+
+    // Test to drop a schema using cascade mode
+    catalog =
+        createCatalog(
+            RandomIdGenerator.INSTANCE.nextId(), Namespace.of(metalakeName), "catalog1", auditInfo);
+    backend.insert(catalog, false);
+
+    schema =
+        createSchemaEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of(metalakeName, catalog.name()),
+            "schema1",
+            auditInfo);
+    backend.insert(schema, false);
+
+    column =
+        ColumnEntity.builder()
+            .withId(RandomIdGenerator.INSTANCE.nextId())
+            .withName("column1")
+            .withPosition(0)
+            .withAutoIncrement(false)
+            .withNullable(false)
+            .withDataType(Types.IntegerType.get())
+            .withAuditInfo(auditInfo)
+            .build();
+
+    columns = Lists.newArrayList();
+    columns.add(column);
+
+    table =
+        TableEntity.builder()
+            .withName("table")
+            .withNamespace(Namespace.of(metalakeName, catalog.name(), schema.name()))
+            .withId(RandomIdGenerator.INSTANCE.nextId())
+            .withColumns(columns)
+            .withAuditInfo(auditInfo)
+            .build();
+
+    backend.insert(table, false);
+
+    topic =
+        createTopicEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of(metalakeName, catalog.name(), schema.name()),
+            "topic1",
+            auditInfo);
+    backend.insert(topic, false);
+
+    fileset =
+        createFilesetEntity(
+            RandomIdGenerator.INSTANCE.nextId(),
+            Namespace.of(metalakeName, catalog.name(), schema.name()),
+            "fileset1",
+            auditInfo);
+    backend.insert(fileset, false);
+
+    tagMetaService.associateTagsWithMetadataObject(
+        catalog.nameIdentifier(),
+        catalog.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+    tagMetaService.associateTagsWithMetadataObject(
+        schema.nameIdentifier(),
+        schema.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+    tagMetaService.associateTagsWithMetadataObject(
+        table.nameIdentifier(),
+        table.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+    tagMetaService.associateTagsWithMetadataObject(
+        topic.nameIdentifier(),
+        topic.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+    tagMetaService.associateTagsWithMetadataObject(
+        fileset.nameIdentifier(),
+        fileset.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+    columnIdentifier =
+        NameIdentifier.of(Namespace.fromString(table.nameIdentifier().toString()), column.name());
+    tagMetaService.associateTagsWithMetadataObject(
+        columnIdentifier,
+        column.type(),
+        new NameIdentifier[] {tagEntity1.nameIdentifier()},
+        new NameIdentifier[0]);
+
+    // Test to drop a schema
+    SchemaMetaService.getInstance().deleteSchema(schema.nameIdentifier(), true);
+    Assertions.assertEquals(1, countActiveTagRel(tagEntity1.id()));
+    Assertions.assertEquals(18, countAllTagRel(tagEntity1.id()));
+  }
 }