[#12342] improvement(core): add OCC for namespace hierarchy

catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java

@@ -46,7 +46,6 @@
 import static org.apache.gravitino.catalog.kafka.KafkaCatalogPropertiesMetadata.BOOTSTRAP_SERVERS;
 import static org.apache.gravitino.catalog.kafka.KafkaTopicPropertiesMetadata.PARTITION_COUNT;
 import static org.apache.gravitino.catalog.kafka.KafkaTopicPropertiesMetadata.REPLICATION_FACTOR;
-import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.when;
 
 import com.google.common.collect.ImmutableMap;
@@ -74,18 +73,16 @@
 import org.apache.gravitino.messaging.Topic;
 import org.apache.gravitino.messaging.TopicChange;
 import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.BaseMetalake;
 import org.apache.gravitino.meta.CatalogEntity;
+import org.apache.gravitino.meta.SchemaVersion;
 import org.apache.gravitino.storage.IdGenerator;
 import org.apache.gravitino.storage.RandomIdGenerator;
-import org.apache.gravitino.storage.relational.helper.CatalogIds;
-import org.apache.gravitino.storage.relational.service.CatalogMetaService;
-import org.apache.gravitino.storage.relational.service.MetalakeMetaService;
 import org.apache.kafka.common.config.TopicConfig;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
-import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 
 public class TestKafkaCatalogOperations extends KafkaClusterEmbedded {
@@ -141,7 +138,7 @@ public PropertiesMetadata modelVersionPropertiesMetadata()
   private static KafkaCatalogOperations kafkaCatalogOperations;
 
   @BeforeAll
-  public static void setUp() throws IllegalAccessException {
+  public static void setUp() throws IOException, IllegalAccessException {
     Config config = Mockito.mock(Config.class);
     Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
     Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 * 1000L);
@@ -178,35 +175,23 @@ public static void setUp() throws IllegalAccessException {
     Mockito.when(config.get(Configs.CACHE_IMPLEMENTATION)).thenReturn("caffeine");
     Mockito.when(config.get(Configs.CACHE_LOCK_SEGMENTS)).thenReturn(16);
 
-    // Mock
-    MetalakeMetaService metalakeMetaService = MetalakeMetaService.getInstance();
-    MetalakeMetaService spyMetaservice = Mockito.spy(metalakeMetaService);
-    doReturn(1L).when(spyMetaservice).getMetalakeIdByName(Mockito.anyString());
-
-    CatalogMetaService catalogMetaService = CatalogMetaService.getInstance();
-    CatalogMetaService spyCatalogMetaService = Mockito.spy(catalogMetaService);
-    doReturn(1L)
-        .when(spyCatalogMetaService)
-        .getCatalogIdByMetalakeIdAndName(Mockito.anyLong(), Mockito.anyString());
-    doReturn(new CatalogIds(1L, 1L))
-        .when(spyCatalogMetaService)
-        .getCatalogIdByMetalakeAndCatalogName(Mockito.anyString(), Mockito.anyString());
-
-    MockedStatic<MetalakeMetaService> metalakeMetaServiceMockedStatic =
-        Mockito.mockStatic(MetalakeMetaService.class);
-    MockedStatic<CatalogMetaService> catalogMetaServiceMockedStatic =
-        Mockito.mockStatic(CatalogMetaService.class);
-
-    metalakeMetaServiceMockedStatic
-        .when(MetalakeMetaService::getInstance)
-        .thenReturn(spyMetaservice);
-    catalogMetaServiceMockedStatic
-        .when(CatalogMetaService::getInstance)
-        .thenReturn(spyCatalogMetaService);
-
     store = EntityStoreFactory.createEntityStore(config);
     store.initialize(config);
     idGenerator = new RandomIdGenerator();
+
+    BaseMetalake metalake =
+        BaseMetalake.builder()
+            .withId(1L)
+            .withName(METALAKE_NAME)
+            .withVersion(SchemaVersion.V_0_1)
+            .withAuditInfo(
+                AuditInfo.builder()
+                    .withCreator("testKafkaUser")
+                    .withCreateTime(Instant.now())
+                    .build())
+            .build();
+    store.put(metalake, false);
+
     kafkaCatalogEntity =
         CatalogEntity.builder()
             .withId(1L)
@@ -221,6 +206,7 @@ public static void setUp() throws IllegalAccessException {
                     .withCreateTime(Instant.now())
                     .build())
             .build();
+    store.put(kafkaCatalogEntity, false);
 
     FieldUtils.writeField(GravitinoEnv.getInstance(), "config", config, true);
 
@@ -238,11 +224,11 @@ public static void tearDown() throws IOException {
   }
 
   @Test
-  public void testKafkaCatalogConfiguration() {
+  public void testKafkaCatalogConfiguration() throws IOException {
     String catalogName = "test_kafka_catalog_configuration";
     CatalogEntity catalogEntity =
         CatalogEntity.builder()
-            .withId(2L)
+            .withId(idGenerator.nextId())
             .withName(catalogName)
             .withNamespace(Namespace.of(METALAKE_NAME))
             .withType(MESSAGING)
@@ -254,6 +240,7 @@ public void testKafkaCatalogConfiguration() {
                     .build())
             .withProperties(MOCK_CATALOG_PROPERTIES)
             .build();
+    store.put(catalogEntity, false);
     KafkaCatalogOperations ops = new KafkaCatalogOperations(store, idGenerator);
     Assertions.assertNull(ops.adminClientConfig);
 
@@ -274,11 +261,11 @@ public void testKafkaCatalogConfiguration() {
   }
 
   @Test
-  public void testInitialization() {
+  public void testInitialization() throws IOException {
     String catalogName = "test_kafka_catalog_initialization";
     CatalogEntity catalogEntity =
         CatalogEntity.builder()
-            .withId(2L)
+            .withId(idGenerator.nextId())
             .withName(catalogName)
             .withNamespace(Namespace.of(METALAKE_NAME))
             .withType(MESSAGING)
@@ -290,6 +277,7 @@ public void testInitialization() {
                     .build())
             .withProperties(MOCK_CATALOG_PROPERTIES)
             .build();
+    store.put(catalogEntity, false);
     KafkaCatalogOperations ops = new KafkaCatalogOperations(store, idGenerator);
     ops.initialize(
         MOCK_CATALOG_PROPERTIES, catalogEntity.toCatalogInfo(), KAFKA_PROPERTIES_METADATA);

core/src/main/java/org/apache/gravitino/catalog/ManagedSchemaOperations.java

@@ -25,6 +25,7 @@
 import java.util.List;
 import java.util.Map;
 import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityAlreadyExistsException;
 import org.apache.gravitino.EntityStore;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
@@ -116,9 +117,11 @@ public Schema createSchema(NameIdentifier ident, String comment, Map<String, Str
                     .build())
             .build();
     try {
-      store().put(schemaEntity, true /* overwrite */);
+      store().put(schemaEntity, false /* overwrite */);
     } catch (IOException ioe) {
       throw new RuntimeException("Failed to create schema " + ident, ioe);
+    } catch (EntityAlreadyExistsException e) {
+      throw new SchemaAlreadyExistsException(e, "Schema %s already exists", ident);
     } catch (NoSuchEntityException e) {
       throw new NoSuchCatalogException(e, "Catalog %s does not exist", ident.namespace());
     }

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

@@ -86,10 +86,30 @@ Integer updateCatalogMeta(
       @Param("newCatalogMeta") CatalogPO newCatalogPO,
       @Param("oldCatalogMeta") CatalogPO oldCatalogPO);
 
+  /**
+   * Advances the catalog version when the expected OCC version still matches.
+   *
+   * @return the number of updated rows
+   */
+  @UpdateProvider(type = CatalogMetaSQLProviderFactory.class, method = "fenceCatalogMeta")
+  Integer fenceCatalogMeta(
+      @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion);
+
   @UpdateProvider(
       type = CatalogMetaSQLProviderFactory.class,
       method = "softDeleteCatalogMetasByCatalogId")
-  Integer softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long catalogId);
+  Integer softDeleteCatalogMetasByCatalogId(
+      @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion);
+
+  /**
+   * Soft-deletes catalogs whose identifiers and OCC versions still match.
+   *
+   * @return the number of deleted rows
+   */
+  @UpdateProvider(
+      type = CatalogMetaSQLProviderFactory.class,
+      method = "softDeleteCatalogMetasWithVersion")
+  Integer softDeleteCatalogMetasWithVersion(@Param("catalogMetas") List<CatalogPO> catalogPOs);
 
   @UpdateProvider(
       type = CatalogMetaSQLProviderFactory.class,

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

@@ -109,8 +109,21 @@ public static String updateCatalogMeta(
     return getProvider().updateCatalogMeta(newCatalogPO, oldCatalogPO);
   }
 
-  public static String softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long catalogId) {
-    return getProvider().softDeleteCatalogMetasByCatalogId(catalogId);
+  /** Returns SQL that advances a catalog OCC version conditionally. */
+  public static String fenceCatalogMeta(
+      @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().fenceCatalogMeta(catalogId, currentVersion);
+  }
+
+  public static String softDeleteCatalogMetasByCatalogId(
+      @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().softDeleteCatalogMetasByCatalogId(catalogId, currentVersion);
+  }
+
+  /** Returns SQL that soft-deletes catalogs using identifier-and-version pairs. */
+  public static String softDeleteCatalogMetasWithVersion(
+      @Param("catalogMetas") List<CatalogPO> catalogPOs) {
+    return getProvider().softDeleteCatalogMetasWithVersion(catalogPOs);
   }
 
   public static String softDeleteCatalogMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {

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

@@ -70,10 +70,20 @@ Integer updateMetalakeMeta(
       @Param("newMetalakeMeta") MetalakePO newMetalakePO,
       @Param("oldMetalakeMeta") MetalakePO oldMetalakePO);
 
+  /**
+   * Advances the metalake version when the expected OCC version still matches.
+   *
+   * @return the number of updated rows
+   */
+  @UpdateProvider(type = MetalakeMetaSQLProviderFactory.class, method = "fenceMetalakeMeta")
+  Integer fenceMetalakeMeta(
+      @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion);
+
   @UpdateProvider(
       type = MetalakeMetaSQLProviderFactory.class,
       method = "softDeleteMetalakeMetaByMetalakeId")
-  Integer softDeleteMetalakeMetaByMetalakeId(@Param("metalakeId") Long metalakeId);
+  Integer softDeleteMetalakeMetaByMetalakeId(
+      @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion);
 
   @DeleteProvider(
       type = MetalakeMetaSQLProviderFactory.class,

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

@@ -88,8 +88,15 @@ public static String updateMetalakeMeta(
     return getProvider().updateMetalakeMeta(newMetalakePO, oldMetalakePO);
   }
 
-  public static String softDeleteMetalakeMetaByMetalakeId(@Param("metalakeId") Long metalakeId) {
-    return getProvider().softDeleteMetalakeMetaByMetalakeId(metalakeId);
+  /** Returns SQL that advances a metalake OCC version conditionally. */
+  public static String fenceMetalakeMeta(
+      @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().fenceMetalakeMeta(metalakeId, currentVersion);
+  }
+
+  public static String softDeleteMetalakeMetaByMetalakeId(
+      @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().softDeleteMetalakeMetaByMetalakeId(metalakeId, currentVersion);
   }
 
   public static String deleteMetalakeMetasByLegacyTimeline(

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

@@ -102,11 +102,36 @@ SchemaPO selectSchemaByFullQualifiedName(
   Integer updateSchemaMeta(
       @Param("newSchemaMeta") SchemaPO newSchemaPO, @Param("oldSchemaMeta") SchemaPO oldSchemaPO);
 
+  /**
+   * Advances the schema version when the expected OCC version still matches.
+   *
+   * @return the number of updated rows
+   */
+  @UpdateProvider(type = SchemaMetaSQLProviderFactory.class, method = "fenceSchemaMeta")
+  Integer fenceSchemaMeta(
+      @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion);
+
   @UpdateProvider(
       type = SchemaMetaSQLProviderFactory.class,
       method = "softDeleteSchemaMetasBySchemaIds")
   Integer softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List<Long> schemaIds);
 
+  @UpdateProvider(
+      type = SchemaMetaSQLProviderFactory.class,
+      method = "softDeleteSchemaMetaBySchemaIdAndVersion")
+  Integer softDeleteSchemaMetaBySchemaIdAndVersion(
+      @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion);
+
+  /**
+   * Soft-deletes schemas whose identifiers and OCC versions still match.
+   *
+   * @return the number of deleted rows
+   */
+  @UpdateProvider(
+      type = SchemaMetaSQLProviderFactory.class,
+      method = "softDeleteSchemaMetasWithVersion")
+  Integer softDeleteSchemaMetasWithVersion(@Param("schemaMetas") List<SchemaPO> schemaPOs);
+
   @UpdateProvider(
       type = SchemaMetaSQLProviderFactory.class,
       method = "softDeleteSchemaMetasByMetalakeId")

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

@@ -116,10 +116,27 @@ public static String updateSchemaMeta(
     return getProvider().updateSchemaMeta(newSchemaPO, oldSchemaPO);
   }
 
+  /** Returns SQL that advances a schema OCC version conditionally. */
+  public static String fenceSchemaMeta(
+      @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().fenceSchemaMeta(schemaId, currentVersion);
+  }
+
   public static String softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List<Long> schemaIds) {
     return getProvider().softDeleteSchemaMetasBySchemaIds(schemaIds);
   }
 
+  public static String softDeleteSchemaMetaBySchemaIdAndVersion(
+      @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().softDeleteSchemaMetaBySchemaIdAndVersion(schemaId, currentVersion);
+  }
+
+  /** Returns SQL that soft-deletes schemas using identifier-and-version pairs. */
+  public static String softDeleteSchemaMetasWithVersion(
+      @Param("schemaMetas") List<SchemaPO> schemaPOs) {
+    return getProvider().softDeleteSchemaMetasWithVersion(schemaPOs);
+  }
+
   public static String softDeleteSchemaMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {
     return getProvider().softDeleteSchemaMetasByMetalakeId(metalakeId);
   }

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

@@ -206,25 +206,43 @@ public String updateCatalogMeta(
         + " last_version = #{newCatalogMeta.lastVersion},"
         + " deleted_at = #{newCatalogMeta.deletedAt}"
         + " WHERE catalog_id = #{oldCatalogMeta.catalogId}"
-        + " AND catalog_name = #{oldCatalogMeta.catalogName}"
-        + " AND metalake_id = #{oldCatalogMeta.metalakeId}"
-        + " AND type = #{oldCatalogMeta.type}"
-        + " AND provider = #{oldCatalogMeta.provider}"
-        + " AND (catalog_comment = #{oldCatalogMeta.catalogComment} "
-        + "   OR (catalog_comment IS NULL and #{oldCatalogMeta.catalogComment} IS NULL))"
-        + " AND properties = #{oldCatalogMeta.properties}"
-        + " AND audit_info = #{oldCatalogMeta.auditInfo}"
         + " AND current_version = #{oldCatalogMeta.currentVersion}"
-        + " AND last_version = #{oldCatalogMeta.lastVersion}"
         + " AND deleted_at = 0";
   }
 
-  public String softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long catalogId) {
+  /** Returns SQL that advances a catalog OCC version conditionally. */
+  public String fenceCatalogMeta(
+      @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion) {
+    return "UPDATE "
+        + TABLE_NAME
+        + " SET last_version = current_version + 1, current_version = current_version + 1"
+        + " WHERE catalog_id = #{catalogId}"
+        + " AND current_version = #{currentVersion} AND deleted_at = 0";
+  }
+
+  public String softDeleteCatalogMetasByCatalogId(
+      @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion) {
     return "UPDATE "
         + TABLE_NAME
         + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
         + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
-        + " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
+        + " WHERE catalog_id = #{catalogId}"
+        + " AND current_version = #{currentVersion} AND deleted_at = 0";
+  }
+
+  /** Returns SQL that soft-deletes catalogs using identifier-and-version pairs. */
+  public String softDeleteCatalogMetasWithVersion(
+      @Param("catalogMetas") List<CatalogPO> catalogPOs) {
+    return "<script>"
+        + "UPDATE "
+        + TABLE_NAME
+        + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+        + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
+        + " WHERE deleted_at = 0 AND "
+        + "<foreach collection='catalogMetas' item='item' separator=' OR ' open='(' close=')'>"
+        + "(catalog_id = #{item.catalogId} AND current_version = #{item.currentVersion})"
+        + "</foreach>"
+        + "</script>";
   }
 
   public String softDeleteCatalogMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {