[#12453] improvement(core): add OCC for schema writes

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

@@ -44,7 +44,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;
@@ -72,18 +71,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 {
@@ -139,7 +136,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);
@@ -174,35 +171,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)
@@ -217,6 +202,7 @@ public static void setUp() throws IllegalAccessException {
                     .withCreateTime(Instant.now())
                     .build())
             .build();
+    store.put(kafkaCatalogEntity, false);
 
     FieldUtils.writeField(GravitinoEnv.getInstance(), "config", config, true);
 
@@ -234,11 +220,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)
@@ -250,6 +236,7 @@ public void testKafkaCatalogConfiguration() {
                     .build())
             .withProperties(MOCK_CATALOG_PROPERTIES)
             .build();
+    store.put(catalogEntity, false);
     KafkaCatalogOperations ops = new KafkaCatalogOperations(store, idGenerator);
     Assertions.assertNull(ops.adminClientConfig);
 
@@ -270,11 +257,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)
@@ -286,6 +273,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,13 @@ public Schema createSchema(NameIdentifier ident, String comment, Map<String, Str
                     .build())
             .build();
     try {
-      store().put(schemaEntity, true /* overwrite */);
+      // Insert only. Overwriting used to hide a lost race: two servers creating the same schema
+      // name would both "succeed", and the second one silently replaced the first one's schema.
+      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

@@ -85,6 +85,12 @@ CatalogPO selectCatalogMetaByName(
       method = "selectCatalogMetaByIdForUpdate")
   CatalogPO selectCatalogMetaByIdForUpdate(@Param("catalogId") Long catalogId);
 
+  /** Selects and share-locks an active catalog by ID for the current transaction. */
+  @SelectProvider(
+      type = CatalogMetaSQLProviderFactory.class,
+      method = "selectCatalogMetaByIdForShare")
+  CatalogPO selectCatalogMetaByIdForShare(@Param("catalogId") Long catalogId);
+
   @InsertProvider(type = CatalogMetaSQLProviderFactory.class, method = "insertCatalogMeta")
   void insertCatalogMeta(@Param("catalogMeta") CatalogPO catalogPO);
 

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

@@ -51,7 +51,15 @@ public static CatalogMetaBaseSQLProvider getProvider() {
 
   static class CatalogMetaMySQLProvider extends CatalogMetaBaseSQLProvider {}
 
-  static class CatalogMetaH2Provider extends CatalogMetaBaseSQLProvider {}
+  static class CatalogMetaH2Provider extends CatalogMetaBaseSQLProvider {
+    @Override
+    public String selectCatalogMetaByIdForShare(Long catalogId) {
+      // H2 has no shared row-lock syntax, so H2 backends fall back to an exclusive lock. Schema
+      // creations under one catalog therefore serialize on H2, and a slow creation can make a
+      // concurrent one hit H2's lock timeout instead of a clean conflict.
+      return selectCatalogMetaByIdForUpdate(catalogId);
+    }
+  }
 
   public static String listCatalogPOsByMetalakeName(@Param("metalakeName") String metalakeName) {
     return getProvider().listCatalogPOsByMetalakeName(metalakeName);
@@ -104,6 +112,11 @@ public static String selectCatalogMetaByIdForUpdate(@Param("catalogId") Long cat
     return getProvider().selectCatalogMetaByIdForUpdate(catalogId);
   }
 
+  /** Returns SQL that selects and share-locks an active catalog by ID. */
+  public static String selectCatalogMetaByIdForShare(@Param("catalogId") Long catalogId) {
+    return getProvider().selectCatalogMetaByIdForShare(catalogId);
+  }
+
   public static String insertCatalogMeta(@Param("catalogMeta") CatalogPO catalogPO) {
     return getProvider().insertCatalogMeta(catalogPO);
   }

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

@@ -86,6 +86,18 @@ SchemaPO selectSchemaByFullQualifiedName(
   @SelectProvider(type = SchemaMetaSQLProviderFactory.class, method = "selectSchemaMetaById")
   SchemaPO selectSchemaMetaById(@Param("schemaId") Long schemaId);
 
+  /** Selects and locks an active schema by ID for the current transaction. */
+  @SelectProvider(
+      type = SchemaMetaSQLProviderFactory.class,
+      method = "selectSchemaMetaByIdForUpdate")
+  SchemaPO selectSchemaMetaByIdForUpdate(@Param("schemaId") Long schemaId);
+
+  /** Selects and share-locks an active schema by ID for the current transaction. */
+  @SelectProvider(
+      type = SchemaMetaSQLProviderFactory.class,
+      method = "selectSchemaMetaByIdForShare")
+  SchemaPO selectSchemaMetaByIdForShare(@Param("schemaId") Long schemaId);
+
   @InsertProvider(type = SchemaMetaSQLProviderFactory.class, method = "insertSchemaMeta")
   void insertSchemaMeta(@Param("schemaMeta") SchemaPO schemaPO);
 
@@ -111,6 +123,19 @@ Integer updateSchemaMeta(
       method = "softDeleteSchemaMetasBySchemaIds")
   Integer softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List<Long> schemaIds);
 
+  /**
+   * Soft-deletes a schema, but only while it still carries the given version.
+   *
+   * @param schemaId the ID of the schema to delete
+   * @param currentVersion the version the caller read before deciding to delete
+   * @return 1 when the schema was deleted, 0 when it changed or is already gone
+   */
+  @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.
    *

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

@@ -49,7 +49,15 @@ public static SchemaMetaBaseSQLProvider getProvider() {
 
   static class SchemaMetaMySQLProvider extends SchemaMetaBaseSQLProvider {}
 
-  static class SchemaMetaH2Provider extends SchemaMetaBaseSQLProvider {}
+  static class SchemaMetaH2Provider extends SchemaMetaBaseSQLProvider {
+    @Override
+    public String selectSchemaMetaByIdForShare(Long schemaId) {
+      // H2 has no shared row-lock syntax, so H2 backends fall back to an exclusive lock. Writes of
+      // tables, views, filesets and the like under one schema therefore serialize on H2, and a slow
+      // write can make a concurrent one hit H2's lock timeout instead of a clean conflict.
+      return selectSchemaMetaByIdForUpdate(schemaId);
+    }
+  }
 
   public static String listSchemaPOsByFullQualifiedName(
       @Param("metalakeName") String metalakeName, @Param("catalogName") String catalogName) {
@@ -98,6 +106,16 @@ public static String selectSchemaMetaById(@Param("schemaId") Long schemaId) {
     return getProvider().selectSchemaMetaById(schemaId);
   }
 
+  /** Returns SQL that selects and locks an active schema by ID. */
+  public static String selectSchemaMetaByIdForUpdate(@Param("schemaId") Long schemaId) {
+    return getProvider().selectSchemaMetaByIdForUpdate(schemaId);
+  }
+
+  /** Returns SQL that selects and share-locks an active schema by ID. */
+  public static String selectSchemaMetaByIdForShare(@Param("schemaId") Long schemaId) {
+    return getProvider().selectSchemaMetaByIdForShare(schemaId);
+  }
+
   public static String insertSchemaMeta(@Param("schemaMeta") SchemaPO schemaPO) {
     return getProvider().insertSchemaMeta(schemaPO);
   }
@@ -125,6 +143,11 @@ public static String softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List<L
     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) {

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

@@ -147,6 +147,11 @@ public String selectCatalogMetaByIdForUpdate(@Param("catalogId") Long catalogId)
     return selectCatalogMetaById(catalogId) + " FOR UPDATE";
   }
 
+  /** Returns SQL that selects and share-locks an active catalog by ID. */
+  public String selectCatalogMetaByIdForShare(@Param("catalogId") Long catalogId) {
+    return selectCatalogMetaById(catalogId) + " LOCK IN SHARE MODE";
+  }
+
   public String insertCatalogMeta(@Param("catalogMeta") CatalogPO catalogPO) {
     return "INSERT INTO "
         + TABLE_NAME

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

@@ -182,6 +182,16 @@ public String selectSchemaMetaById(@Param("schemaId") Long schemaId) {
         + " WHERE schema_id = #{schemaId} AND deleted_at = 0";
   }
 
+  /** Returns SQL that selects and locks an active schema by ID. */
+  public String selectSchemaMetaByIdForUpdate(@Param("schemaId") Long schemaId) {
+    return selectSchemaMetaById(schemaId) + " FOR UPDATE";
+  }
+
+  /** Returns SQL that selects and share-locks an active schema by ID. */
+  public String selectSchemaMetaByIdForShare(@Param("schemaId") Long schemaId) {
+    return selectSchemaMetaById(schemaId) + " LOCK IN SHARE MODE";
+  }
+
   public String insertSchemaMeta(@Param("schemaMeta") SchemaPO schemaPO) {
     return "INSERT INTO "
         + TABLE_NAME
@@ -227,8 +237,12 @@ public String insertSchemaMetaOnDuplicateKeyUpdate(@Param("schemaMeta") SchemaPO
         + " schema_comment = #{schemaMeta.schemaComment},"
         + " properties = #{schemaMeta.properties},"
         + " audit_info = #{schemaMeta.auditInfo},"
-        + " current_version = #{schemaMeta.currentVersion},"
-        + " last_version = #{schemaMeta.lastVersion},"
+        // Move the version forward instead of writing the initial version again. Resetting it
+        // would let a slow alter or drop that still holds an older version pass its own version
+        // check later on. last_version is assigned first, so both columns are computed from the
+        // version the row had before this statement.
+        + " last_version = current_version + 1,"
+        + " current_version = current_version + 1,"
         + " deleted_at = #{schemaMeta.deletedAt}";
   }
 
@@ -265,12 +279,24 @@ public String batchInsertSchemaMetaOnDuplicateKeyUpdate(
         + " schema_comment = VALUES(schema_comment),"
         + " properties = VALUES(properties),"
         + " audit_info = VALUES(audit_info),"
-        + " current_version = VALUES(current_version),"
-        + " last_version = VALUES(last_version),"
+        // Move the version forward instead of writing the initial version again. Resetting it
+        // would let a slow alter or drop that still holds an older version pass its own version
+        // check later on. last_version is assigned first, so both columns are computed from the
+        // version the row had before this statement.
+        + " last_version = current_version + 1,"
+        + " current_version = current_version + 1,"
         + " deleted_at = VALUES(deleted_at)"
         + "</script>";
   }
 
+  /**
+   * Builds SQL that updates a schema only if nobody changed it in the meantime.
+   *
+   * <p>The WHERE clause used to repeat every column. Comparing the version alone is enough now,
+   * because every update moves the version forward, and it also avoids a MySQL trap: MySQL reports
+   * zero affected rows when an UPDATE writes the values a row already has, which the old SQL could
+   * not tell apart from a real conflict.
+   */
   public String updateSchemaMeta(
       @Param("newSchemaMeta") SchemaPO newSchemaPO, @Param("oldSchemaMeta") SchemaPO oldSchemaPO) {
     return "UPDATE "
@@ -285,15 +311,7 @@ public String updateSchemaMeta(
         + " last_version = #{newSchemaMeta.lastVersion},"
         + " deleted_at = #{newSchemaMeta.deletedAt}"
         + " WHERE schema_id = #{oldSchemaMeta.schemaId}"
-        + " AND schema_name = #{oldSchemaMeta.schemaName}"
-        + " AND metalake_id = #{oldSchemaMeta.metalakeId}"
-        + " AND catalog_id = #{oldSchemaMeta.catalogId}"
-        + " AND (schema_comment = #{oldSchemaMeta.schemaComment}"
-        + "   OR (schema_comment IS NULL and #{oldSchemaMeta.schemaComment} IS NULL))"
-        + " AND properties = #{oldSchemaMeta.properties}"
-        + " AND audit_info = #{oldSchemaMeta.auditInfo}"
         + " AND current_version = #{oldSchemaMeta.currentVersion}"
-        + " AND last_version = #{oldSchemaMeta.lastVersion}"
         + " AND deleted_at = 0";
   }
 
@@ -311,6 +329,16 @@ public String softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List<Long> sc
         + "</script>";
   }
 
+  public String softDeleteSchemaMetaBySchemaIdAndVersion(
+      @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion) {
+    return "UPDATE "
+        + TABLE_NAME
+        + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+        + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
+        + " WHERE schema_id = #{schemaId}"
+        + " AND current_version = #{currentVersion} AND deleted_at = 0";
+  }
+
   /** Returns SQL that soft-deletes schemas using identifier-and-version pairs. */
   public String softDeleteSchemaMetasWithVersion(@Param("schemaMetas") List<SchemaPO> schemaPOs) {
     return "<script>"