diff --git a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java index 72c0139cf43..4feb42b56f1 100644 --- a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java +++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java @@ -984,6 +984,13 @@ public boolean dropCatalog(NameIdentifier ident, boolean force) } catch (NoSuchMetalakeException | NoSuchCatalogException ignored) { return false; + } catch (NoSuchEntityException ignored) { + // Another server deleted the catalog after it was loaded above, so a later store read + // such as listing its schemas no longer finds it. The drop stays idempotent, but the + // wrapper cached by loadCatalogAndWrap has to be discarded. store.delete itself never + // reaches here: it maps a missing entity to false on its own. + catalogCache.invalidate(ident); + return false; } catch (GravitinoRuntimeException e) { throw e; } catch (Exception e) { diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaMapper.java index 553a33a71a9..7cf0195b3ea 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaMapper.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaMapper.java @@ -79,6 +79,12 @@ CatalogPO selectCatalogMetaByName( @SelectProvider(type = CatalogMetaSQLProviderFactory.class, method = "selectCatalogMetaById") CatalogPO selectCatalogMetaById(@Param("catalogId") Long catalogId); + /** Returns an active catalog by ID and locks it. */ + @SelectProvider( + type = CatalogMetaSQLProviderFactory.class, + method = "selectCatalogMetaByIdForUpdate") + CatalogPO selectCatalogMetaByIdForUpdate(@Param("catalogId") Long catalogId); + @InsertProvider(type = CatalogMetaSQLProviderFactory.class, method = "insertCatalogMeta") void insertCatalogMeta(@Param("catalogMeta") CatalogPO catalogPO); @@ -92,10 +98,18 @@ Integer updateCatalogMeta( @Param("newCatalogMeta") CatalogPO newCatalogPO, @Param("oldCatalogMeta") CatalogPO oldCatalogPO); + /** + * Soft-deletes a catalog, but only while it still carries the given version. + * + * @param catalogId the ID of the catalog to delete + * @param currentVersion the version the caller read before deciding to delete + * @return 1 when the catalog was deleted, 0 when it changed or is already gone + */ @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. diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaSQLProviderFactory.java index 9b3154fa9ec..d2afccb0d61 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaSQLProviderFactory.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaSQLProviderFactory.java @@ -99,6 +99,11 @@ public static String selectCatalogMetaById(@Param("catalogId") Long catalogId) { return getProvider().selectCatalogMetaById(catalogId); } + /** Builds SQL that returns and locks an active catalog by ID. */ + public static String selectCatalogMetaByIdForUpdate(@Param("catalogId") Long catalogId) { + return getProvider().selectCatalogMetaByIdForUpdate(catalogId); + } + public static String insertCatalogMeta(@Param("catalogMeta") CatalogPO catalogPO) { return getProvider().insertCatalogMeta(catalogPO); } @@ -114,8 +119,9 @@ public static String updateCatalogMeta( return getProvider().updateCatalogMeta(newCatalogPO, oldCatalogPO); } - public static String softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long catalogId) { - return getProvider().softDeleteCatalogMetasByCatalogId(catalogId); + 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. */ diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaMapper.java index 665ae02b794..2adcce1a283 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaMapper.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaMapper.java @@ -53,6 +53,12 @@ public interface MetalakeMetaMapper { method = "selectMetalakeMetaByIdForUpdate") MetalakePO selectMetalakeMetaByIdForUpdate(@Param("metalakeId") Long metalakeId); + /** Returns an active metalake by ID and locks it for shared access. */ + @SelectProvider( + type = MetalakeMetaSQLProviderFactory.class, + method = "selectMetalakeMetaByIdForShare") + MetalakePO selectMetalakeMetaByIdForShare(@Param("metalakeId") Long metalakeId); + @SelectProvider( type = MetalakeMetaSQLProviderFactory.class, method = "listMetalakePOsByMetalakeIds") diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaSQLProviderFactory.java index 11f64ad662b..8e3737b9860 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaSQLProviderFactory.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaSQLProviderFactory.java @@ -51,7 +51,15 @@ public static MetalakeMetaBaseSQLProvider getProvider() { static class MetalakeMetaMySQLProvider extends MetalakeMetaBaseSQLProvider {} - static class MetalakeMetaH2Provider extends MetalakeMetaBaseSQLProvider {} + static class MetalakeMetaH2Provider extends MetalakeMetaBaseSQLProvider { + @Override + public String selectMetalakeMetaByIdForShare(Long metalakeId) { + // H2 has no shared row-lock syntax, so H2 backends fall back to an exclusive lock. Catalog + // creations under one metalake therefore serialize on H2, and a slow creation can make a + // concurrent one hit H2's lock timeout instead of a clean conflict. + return selectMetalakeMetaByIdForUpdate(metalakeId); + } + } public String listMetalakePOs() { return getProvider().listMetalakePOs(); @@ -70,6 +78,11 @@ public static String selectMetalakeMetaByIdForUpdate(@Param("metalakeId") Long m return getProvider().selectMetalakeMetaByIdForUpdate(metalakeId); } + /** Builds SQL that returns an active metalake by ID and locks it for shared access. */ + public static String selectMetalakeMetaByIdForShare(@Param("metalakeId") Long metalakeId) { + return getProvider().selectMetalakeMetaByIdForShare(metalakeId); + } + public static String selectMetalakeIdMetaByName(@Param("metalakeName") String metalakeName) { return getProvider().selectMetalakeIdMetaByName(metalakeName); } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaMapper.java index 989f1ebd99e..9c7ab2b4d9b 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaMapper.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaMapper.java @@ -111,11 +111,6 @@ Integer updateSchemaMeta( method = "softDeleteSchemaMetasBySchemaIds") Integer softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List schemaIds); - @UpdateProvider( - type = SchemaMetaSQLProviderFactory.class, - method = "softDeleteSchemaMetasByCatalogId") - Integer softDeleteSchemaMetasByCatalogId(@Param("catalogId") Long catalogId); - /** * Soft-deletes schemas whose identifiers and OCC versions still match. * diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.java index 30c2ee9dfc2..557bee15f81 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.java @@ -125,10 +125,6 @@ public static String softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List schemaPOs) { diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java index f50b9c203d6..e6f8f03c183 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java @@ -142,6 +142,11 @@ public String selectCatalogMetaById(@Param("catalogId") Long catalogId) { + " WHERE catalog_id = #{catalogId} AND deleted_at = 0"; } + /** Builds SQL that returns and locks an active catalog by ID. */ + public String selectCatalogMetaByIdForUpdate(@Param("catalogId") Long catalogId) { + return selectCatalogMetaById(catalogId) + " FOR UPDATE"; + } + public String insertCatalogMeta(@Param("catalogMeta") CatalogPO catalogPO) { return "INSERT INTO " + TABLE_NAME @@ -190,11 +195,23 @@ public String insertCatalogMetaOnDuplicateKeyUpdate(@Param("catalogMeta") Catalo + " catalog_comment = #{catalogMeta.catalogComment}," + " properties = #{catalogMeta.properties}," + " audit_info = #{catalogMeta.auditInfo}," - + " current_version = #{catalogMeta.currentVersion}," - + " last_version = #{catalogMeta.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 = #{catalogMeta.deletedAt}"; } + /** + * Builds SQL that updates a catalog only if nobody changed it in the meantime. + * + *

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 updateCatalogMeta( @Param("newCatalogMeta") CatalogPO newCatalogPO, @Param("oldCatalogMeta") CatalogPO oldCatalogPO) { @@ -211,25 +228,18 @@ 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) { + 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. */ diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/MetalakeMetaBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/MetalakeMetaBaseSQLProvider.java index a7a78f4b7b5..f8e99a39907 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/MetalakeMetaBaseSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/MetalakeMetaBaseSQLProvider.java @@ -64,6 +64,11 @@ public String selectMetalakeMetaByIdForUpdate(@Param("metalakeId") Long metalake return selectMetalakeMetaById(metalakeId) + " FOR UPDATE"; } + /** Builds SQL that returns an active metalake by ID and locks it for shared access. */ + public String selectMetalakeMetaByIdForShare(@Param("metalakeId") Long metalakeId) { + return selectMetalakeMetaById(metalakeId) + " LOCK IN SHARE MODE"; + } + public String selectMetalakeIdMetaByName(@Param("metalakeName") String metalakeName) { return "SELECT metalake_id as metalakeId" + " FROM " diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java index ee72e383249..9ee36cc52da 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java @@ -311,14 +311,6 @@ public String softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List sc + ""; } - public String softDeleteSchemaMetasByCatalogId(@Param("catalogId") Long catalogId) { - 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"; - } - /** Returns SQL that soft-deletes schemas using identifier-and-version pairs. */ public String softDeleteSchemaMetasWithVersion(@Param("schemaMetas") List schemaPOs) { return ""; } - @Override - public String softDeleteSchemaMetasByCatalogId(Long catalogId) { - return "UPDATE " - + TABLE_NAME - + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)" - + " WHERE catalog_id = #{catalogId} AND deleted_at = 0"; - } - /** {@inheritDoc} */ @Override public String softDeleteSchemaMetasWithVersion(List schemaPOs) { 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 9f63b9189f3..b7413b50dec 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 @@ -24,7 +24,6 @@ import java.io.IOException; import java.util.List; import java.util.Objects; -import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; import java.util.stream.Collectors; import org.apache.gravitino.Entity; @@ -35,7 +34,6 @@ import org.apache.gravitino.exceptions.NoSuchEntityException; import org.apache.gravitino.exceptions.NonEmptyEntityException; import org.apache.gravitino.meta.CatalogEntity; -import org.apache.gravitino.meta.SchemaEntity; import org.apache.gravitino.metrics.Monitored; import org.apache.gravitino.storage.relational.helper.CatalogIds; import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper; @@ -43,6 +41,7 @@ import org.apache.gravitino.storage.relational.mapper.FilesetVersionMapper; import org.apache.gravitino.storage.relational.mapper.FunctionMetaMapper; import org.apache.gravitino.storage.relational.mapper.FunctionVersionMetaMapper; +import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; import org.apache.gravitino.storage.relational.mapper.ModelMetaMapper; import org.apache.gravitino.storage.relational.mapper.ModelVersionAliasRelMapper; import org.apache.gravitino.storage.relational.mapper.ModelVersionMetaMapper; @@ -57,6 +56,8 @@ import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper; import org.apache.gravitino.storage.relational.mapper.ViewMetaMapper; import org.apache.gravitino.storage.relational.po.CatalogPO; +import org.apache.gravitino.storage.relational.po.MetalakePO; +import org.apache.gravitino.storage.relational.po.SchemaPO; import org.apache.gravitino.storage.relational.utils.ExceptionUtils; import org.apache.gravitino.storage.relational.utils.POConverters; import org.apache.gravitino.storage.relational.utils.SessionUtils; @@ -179,20 +180,35 @@ public void insertCatalog(CatalogEntity catalogEntity, boolean overwrite) throws try { NameIdentifierUtil.checkCatalog(catalogEntity.nameIdentifier()); - String metalake = NameIdentifierUtil.getMetalake(catalogEntity.nameIdentifier()); - Long metalakeId = - EntityIdService.getEntityId(NameIdentifier.of(metalake), Entity.EntityType.METALAKE); - - SessionUtils.doWithCommit( - CatalogMetaMapper.class, - mapper -> { - CatalogPO po = POConverters.initializeCatalogPOWithVersion(catalogEntity, metalakeId); - if (overwrite) { - mapper.insertCatalogMetaOnDuplicateKeyUpdate(po); - } else { - mapper.insertCatalogMeta(po); - } - }); + String metalakeName = NameIdentifierUtil.getMetalake(catalogEntity.nameIdentifier()); + // This read runs before the transaction below, so it only tells us the metalake ID and name + // we start from. The metalake may still be dropped or renamed right after it. That is why + // lockMetalakeForCatalogCreate checks the row again inside the transaction. + MetalakePO metalakePO = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalakeName)); + if (metalakePO == null) { + throw new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.METALAKE.name().toLowerCase(), + metalakeName); + } + + SessionUtils.doMultipleWithCommit( + () -> lockMetalakeForCatalogCreate(metalakePO), + () -> + SessionUtils.doWithoutCommit( + CatalogMetaMapper.class, + mapper -> { + CatalogPO po = + POConverters.initializeCatalogPOWithVersion( + catalogEntity, metalakePO.getMetalakeId()); + if (overwrite) { + mapper.insertCatalogMetaOnDuplicateKeyUpdate(po); + } else { + mapper.insertCatalogMeta(po); + } + })); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.CATALOG, catalogEntity.nameIdentifier().toString()); @@ -220,29 +236,33 @@ public CatalogEntity updateCatalog( newEntity.id(), oldCatalogEntity.id()); - AtomicInteger updateResult = new AtomicInteger(0); try { SessionUtils.doMultipleWithCommit( - () -> - updateResult.set( - SessionUtils.getWithoutCommit( - CatalogMetaMapper.class, - mapper -> - mapper.updateCatalogMeta( - POConverters.updateCatalogPOWithVersion( - oldCatalogPO, newEntity, oldCatalogPO.getMetalakeId()), - oldCatalogPO)))); + () -> { + // The UPDATE only matches the row if its version is still the one we read above, and + // it writes the next version. So two servers that read the same catalog cannot both + // apply their change: the second one updates no row. + int updated = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, + mapper -> + mapper.updateCatalogMeta( + POConverters.updateCatalogPOWithVersion( + oldCatalogPO, newEntity, oldCatalogPO.getMetalakeId()), + oldCatalogPO)); + if (updated == 0) { + // Zero rows can mean two different things: someone else changed the catalog, or the + // catalog is gone. Let catalogWriteFailure tell them apart and pick the error. + throw catalogWriteFailure(identifier, oldCatalogPO); + } + }); } catch (RuntimeException re) { ExceptionUtils.checkSQLException( re, Entity.EntityType.CATALOG, newEntity.nameIdentifier().toString()); throw re; } - if (updateResult.get() > 0) { - return newEntity; - } else { - throw new IOException("Failed to update the entity: " + identifier); - } + return newEntity; } @Monitored( @@ -252,18 +272,20 @@ public boolean deleteCatalog(NameIdentifier identifier, boolean cascade) { NameIdentifierUtil.checkCatalog(identifier); String catalogName = identifier.name(); - long catalogId = EntityIdService.getEntityId(identifier, Entity.EntityType.CATALOG); + // Read the whole row, not just the ID, because the delete below needs the version we saw. + CatalogPO catalogPO = getCatalogPOByName(identifier.namespace().level(0), catalogName); + long catalogId = catalogPO.getCatalogId(); if (cascade) { SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - CatalogMetaMapper.class, - mapper -> mapper.softDeleteCatalogMetasByCatalogId(catalogId)), - () -> - SessionUtils.doWithoutCommit( - SchemaMetaMapper.class, - mapper -> mapper.softDeleteSchemaMetasByCatalogId(catalogId)), + () -> { + // Delete the parent first, then its children. The parent delete locks the catalog row, + // and schema writes lock that same row before they touch a schema, so no schema can be + // added or removed after this point. Anything that goes wrong later in this + // transaction rolls this soft delete back with it. + deleteCatalogWithVersion(identifier, catalogPO); + deleteSchemasWithVersions(identifier, catalogId); + }, () -> SessionUtils.doWithoutCommit( TableMetaMapper.class, @@ -328,19 +350,24 @@ public boolean deleteCatalog(NameIdentifier identifier, boolean cascade) { ViewMetaMapper.class, mapper -> mapper.softDeleteViewMetasByCatalogId(catalogId))); } else { - List schemaEntities = - SchemaMetaService.getInstance() - .listSchemasByNamespace( - NamespaceUtil.ofSchema(identifier.namespace().level(0), catalogName)); - if (!schemaEntities.isEmpty()) { - throw new NonEmptyEntityException( - "Entity %s has sub-entities, you should remove sub-entities first", identifier); - } SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - CatalogMetaMapper.class, - mapper -> mapper.softDeleteCatalogMetasByCatalogId(catalogId)), + () -> { + // Delete the catalog first and check for schemas afterwards. This order looks odd, but + // it is what makes the check safe: the delete locks the catalog row, and schema + // creation locks the same row before inserting. So a create either finishes before this + // delete, in which case the check below sees its schema, or it waits until this + // transaction ends. Checking first would leave a gap where a schema can be inserted + // between the check and the delete. If the check does find a schema, the exception + // rolls the soft delete back. + deleteCatalogWithVersion(identifier, catalogPO); + List schemaPOs = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.listSchemaPOsByCatalogId(catalogId)); + if (!schemaPOs.isEmpty()) { + throw new NonEmptyEntityException( + "Entity %s has sub-entities, you should remove sub-entities first", identifier); + } + }, () -> SessionUtils.doWithoutCommit( OwnerMetaMapper.class, @@ -374,6 +401,98 @@ public boolean deleteCatalog(NameIdentifier identifier, boolean cascade) { return true; } + /** + * Soft-deletes the catalog only if its version is still the one the caller read. A drop that + * loses the race to another writer must not delete a catalog it never saw. + */ + private void deleteCatalogWithVersion(NameIdentifier identifier, CatalogPO observedCatalogPO) { + int deleted = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, + mapper -> + mapper.softDeleteCatalogMetasByCatalogId( + observedCatalogPO.getCatalogId(), observedCatalogPO.getCurrentVersion())); + if (deleted == 0) { + throw catalogWriteFailure(identifier, observedCatalogPO); + } + } + + /** + * Holds the parent metalake row for the rest of the transaction, so the catalog cannot be created + * below a metalake that is going away. + * + *

The lock is shared, not exclusive: many catalogs can be created under the same metalake at + * the same time. Dropping a metalake takes an exclusive lock on this row, so a drop and a create + * cannot overlap. Whoever gets the row first wins, and the loser either sees the metalake gone or + * inserts under a metalake that is still there. + * + *

The name is compared again because the ID alone cannot tell a rename apart: the caller + * looked the metalake up by name, so a renamed row means the name in the request no longer + * exists. + */ + private void lockMetalakeForCatalogCreate(MetalakePO observedMetalakePO) { + MetalakePO currentMetalakePO = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, + mapper -> mapper.selectMetalakeMetaByIdForShare(observedMetalakePO.getMetalakeId())); + if (currentMetalakePO == null + || !Objects.equals( + currentMetalakePO.getMetalakeName(), observedMetalakePO.getMetalakeName())) { + throw new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.METALAKE.name().toLowerCase(), + observedMetalakePO.getMetalakeName()); + } + } + + /** + * Decides which error a failed compare-and-set should report. The write matched no row either + * because someone else changed the catalog, which is a conflict, or because the catalog was + * deleted or renamed away, which is a missing entity. + */ + private RuntimeException catalogWriteFailure( + NameIdentifier identifier, CatalogPO observedCatalogPO) { + // Sessions run at READ_COMMITTED, so a plain read would already see the latest committed row. + // The locking read additionally waits for a writer that is still in flight, so a rename or + // delete that has not committed yet is classified as not-found instead of as a stale-version + // conflict. The lock is taken on the error path of a transaction that is about to roll back. + CatalogPO currentCatalogPO = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, + mapper -> mapper.selectCatalogMetaByIdForUpdate(observedCatalogPO.getCatalogId())); + if (currentCatalogPO == null + || !Objects.equals(currentCatalogPO.getCatalogName(), observedCatalogPO.getCatalogName()) + || !Objects.equals(currentCatalogPO.getMetalakeId(), observedCatalogPO.getMetalakeId())) { + return new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.CATALOG.name().toLowerCase(), + identifier.name()); + } + return ExceptionUtils.concurrentModification(Entity.EntityType.CATALOG, identifier); + } + + /** + * Soft-deletes every schema of the catalog, each one guarded by the version read here. The caller + * must already hold the catalog row, so no schema can appear or disappear in between. + */ + private void deleteSchemasWithVersions(NameIdentifier catalogIdentifier, Long catalogId) { + List schemaPOs = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.listSchemaPOsByCatalogId(catalogId)); + if (schemaPOs.isEmpty()) { + return; + } + int deleted = + SessionUtils.getWithoutCommit( + SchemaMetaMapper.class, mapper -> mapper.softDeleteSchemaMetasWithVersion(schemaPOs)); + // A smaller count means one of these schemas was altered by someone who did not take the + // catalog row lock. Never commit half a cascade: roll the whole transaction back instead. + if (deleted != schemaPOs.size()) { + throw ExceptionUtils.concurrentChildModification( + Entity.EntityType.SCHEMA, Entity.EntityType.CATALOG, catalogIdentifier); + } + } + @Monitored( metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, baseMetricName = "deleteCatalogMetasByLegacyTimeline") 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 cc42229f73d..58c8c054fec 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 @@ -407,11 +407,11 @@ void deleteMetalakeWithVersion(NameIdentifier identifier, Long metalakeId, Long private RuntimeException metalakeWriteFailure( NameIdentifier identifier, Long metalakeId, String observedName) { - // Use a locking read to see the latest committed row. Under MySQL REPEATABLE READ, a plain - // SELECT can return an old snapshot that still contains a row another writer already deleted - // or renamed. We would then report a version conflict instead of a missing metalake. The CAS - // UPDATE above already waits for the same row lock, so the other writer has finished before - // this read runs. + // Sessions run at READ_COMMITTED, so a plain read would already see the latest committed row. + // The locking read additionally waits for a writer that is still in flight, so a delete or + // rename that has not committed yet is reported as a missing metalake instead of as a stale + // version conflict. The lock is taken on the error path of a transaction that is about to roll + // back. MetalakePO currentMetalakePO = SessionUtils.getWithoutCommit( MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByIdForUpdate(metalakeId)); diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java b/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java index 2a05888fd7d..3085c1ed6d3 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java @@ -234,9 +234,11 @@ public static CatalogPO initializeCatalogPOWithVersion( */ public static CatalogPO updateCatalogPOWithVersion( CatalogPO oldCatalogPO, CatalogEntity newCatalog, Long metalakeId) { - Long lastVersion = oldCatalogPO.getLastVersion(); - // Will set the version to the last version + 1 when having some fields need be multiple version - Long nextVersion = lastVersion; + // Every update moves the version forward, even when nothing else changes. The version is what + // the UPDATE compares against, so a version that stands still would let two servers overwrite + // each other. Both columns get the same value because a catalog keeps no old versions to + // address, unlike a fileset. + Long nextVersion = oldCatalogPO.getCurrentVersion() + 1; try { return CatalogPO.builder() .withCatalogId(newCatalog.id()) diff --git a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java index c55c4044692..de28d3ea1dc 100644 --- a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java +++ b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java @@ -45,9 +45,11 @@ import org.apache.gravitino.CatalogChange; import org.apache.gravitino.Config; import org.apache.gravitino.Configs; +import org.apache.gravitino.Entity; import org.apache.gravitino.Entity.EntityType; import org.apache.gravitino.EntityStore; import org.apache.gravitino.GravitinoEnv; +import org.apache.gravitino.HasIdentifier; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; import org.apache.gravitino.Schema; @@ -57,6 +59,7 @@ import org.apache.gravitino.connector.capability.CapabilityResult; import org.apache.gravitino.exceptions.CatalogAlreadyExistsException; import org.apache.gravitino.exceptions.NoSuchCatalogException; +import org.apache.gravitino.exceptions.NoSuchEntityException; import org.apache.gravitino.exceptions.NoSuchMetalakeException; import org.apache.gravitino.exceptions.NoSuchSchemaException; import org.apache.gravitino.lock.LockManager; @@ -850,6 +853,26 @@ void testDropCatalogDoesNotMarkLocalMutationWhenStoreReturnsFalse() throws Excep manager.close(); } + @Test + void testDropCatalogReturnsFalseWhenConcurrentDeleteWins() throws Exception { + ChangeLogAwareEntityStore store = new ChangeLogAwareEntityStore(); + store.initialize(config); + store.put(metalakeEntity, true); + + CatalogManager manager = + new CatalogManager(config, store, new RandomIdGenerator(), new SecretManager(config)); + NameIdentifier ident = NameIdentifier.of("metalake", "concurrently_deleted"); + Map props = + ImmutableMap.of( + PROPERTY_KEY1, "value1", PROPERTY_KEY2, "value2", PROPERTY_KEY5_PREFIX + "1", "value3"); + manager.createCatalog(ident, Catalog.Type.RELATIONAL, provider, "comment", props); + store.throwMissingCatalogForSchemaList = true; + + Assertions.assertFalse(manager.dropCatalog(ident, true)); + Assertions.assertNull(manager.getCatalogCache().getIfPresent(ident)); + manager.close(); + } + @Test void testFailedCreateCatalogCleanupMarksLocalMutation() throws Exception { ChangeLogAwareEntityStore store = new ChangeLogAwareEntityStore(); @@ -983,6 +1006,7 @@ private static class ChangeLogAwareEntityStore extends InMemoryEntityStore private final AtomicReference unregisteredListener = new AtomicReference<>(); private boolean returnFalseForCatalogDelete; + private boolean throwMissingCatalogForSchemaList; @Override public boolean delete(NameIdentifier ident, EntityType entityType, boolean cascade) @@ -993,6 +1017,20 @@ public boolean delete(NameIdentifier ident, EntityType entityType, boolean casca return super.delete(ident, entityType, cascade); } + @Override + public List list( + Namespace namespace, Class cl, EntityType entityType) throws IOException { + // Mirrors the relational store: listing the schemas of a catalog that another server has + // already deleted resolves the parent catalog id first and reports the catalog as missing. + if (throwMissingCatalogForSchemaList && entityType == EntityType.SCHEMA) { + throw new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + EntityType.CATALOG.name().toLowerCase(), + namespace.level(namespace.length() - 1)); + } + return super.list(namespace, cl, entityType); + } + @Override public void registerEntityChangeLogListener(EntityChangeLogListener listener) { this.listener.set(listener); diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestCatalogMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestCatalogMetaService.java index de5520908af..cbc598903eb 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestCatalogMetaService.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestCatalogMetaService.java @@ -29,12 +29,22 @@ import java.sql.SQLException; import java.sql.Statement; import java.time.Instant; +import java.util.Arrays; import java.util.List; +import java.util.Objects; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import org.apache.gravitino.Catalog; import org.apache.gravitino.Entity; import org.apache.gravitino.EntityAlreadyExistsException; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; +import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.NonEmptyEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.CatalogEntity; import org.apache.gravitino.meta.ColumnEntity; @@ -50,7 +60,11 @@ import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.relational.TestJDBCBackend; import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper; +import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; +import org.apache.gravitino.storage.relational.po.CatalogPO; +import org.apache.gravitino.storage.relational.po.MetalakePO; import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper; +import org.apache.gravitino.storage.relational.utils.POConverters; import org.apache.gravitino.storage.relational.utils.SessionUtils; import org.apache.gravitino.utils.NameIdentifierUtil; import org.apache.gravitino.utils.NamespaceUtil; @@ -88,6 +102,84 @@ public void testInsertAlreadyExistsException() throws IOException { assertThrows(EntityAlreadyExistsException.class, () -> backend.insert(catalogCopy, false)); } + @TestTemplate + public void testInsertCatalogLocksMetalakeWithoutChangingVersion() throws IOException { + MetalakePO beforeInsert = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalakeName)); + CatalogEntity catalog = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + "catalog_fence", + auditInfo); + backend.insert(catalog, false); + + MetalakePO afterInsert = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalakeName)); + assertEquals(beforeInsert.getCurrentVersion(), afterInsert.getCurrentVersion()); + assertEquals(beforeInsert.getLastVersion(), afterInsert.getLastVersion()); + + CatalogEntity duplicate = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + catalog.name(), + auditInfo); + assertThrows(EntityAlreadyExistsException.class, () -> backend.insert(duplicate, false)); + + MetalakePO afterFailure = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalakeName)); + assertEquals(afterInsert.getCurrentVersion(), afterFailure.getCurrentVersion()); + assertEquals(afterInsert.getLastVersion(), afterFailure.getLastVersion()); + } + + @TestTemplate + public void testConcurrentSameNameCatalogCreateReportsAlreadyExists() throws Exception { + CatalogEntity first = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + "concurrent_catalog", + auditInfo); + CatalogEntity second = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + first.name(), + auditInfo); + + List results = insertCatalogsConcurrently(first, second); + assertEquals(1, results.stream().filter(Objects::isNull).count()); + Throwable failure = results.stream().filter(Objects::nonNull).findFirst().orElseThrow(); + Assertions.assertTrue( + failure instanceof EntityAlreadyExistsException, + () -> "Expected EntityAlreadyExistsException, but got " + failure); + } + + @TestTemplate + public void testConcurrentDifferentCatalogCreatesBothSucceed() throws Exception { + CatalogEntity first = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + "concurrent_catalog_1", + auditInfo); + CatalogEntity second = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + "concurrent_catalog_2", + auditInfo); + + List results = insertCatalogsConcurrently(first, second); + Assertions.assertTrue( + results.stream().allMatch(Objects::isNull), + () -> "Concurrent catalog creates failed: " + results); + } + @TestTemplate public void testUpdateAlreadyExistsException() throws IOException { CatalogEntity catalog = @@ -149,6 +241,184 @@ void testUpdateCatalogWithNullableComment() throws IOException { Assertions.assertNotNull(updatedCatalog.getComment()); } + @TestTemplate + public void testAlterAndDeleteUseCurrentVersion() throws IOException { + CatalogEntity catalog = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + "catalog_occ", + auditInfo); + backend.insert(catalog, false); + CatalogPO oldPO = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, mapper -> mapper.selectCatalogMetaById(catalog.id())); + CatalogEntity updatedCatalog = + CatalogEntity.builder() + .withId(catalog.id()) + .withName(catalog.name()) + .withNamespace(catalog.namespace()) + .withAuditInfo(auditInfo) + .withComment("updated") + .withProperties(catalog.getProperties()) + .withType(catalog.getType()) + .withProvider(catalog.getProvider()) + .build(); + CatalogPO newPO = + POConverters.updateCatalogPOWithVersion(oldPO, updatedCatalog, oldPO.getMetalakeId()); + + int updated = + SessionUtils.doWithCommitAndFetchResult( + CatalogMetaMapper.class, mapper -> mapper.updateCatalogMeta(newPO, oldPO)); + int staleUpdate = + SessionUtils.doWithCommitAndFetchResult( + CatalogMetaMapper.class, mapper -> mapper.updateCatalogMeta(newPO, oldPO)); + int staleDelete = + SessionUtils.doWithCommitAndFetchResult( + CatalogMetaMapper.class, + mapper -> + mapper.softDeleteCatalogMetasByCatalogId(catalog.id(), oldPO.getCurrentVersion())); + assertEquals(1, updated); + assertEquals(0, staleUpdate); + assertEquals(0, staleDelete); + assertTrue(backend.exists(catalog.nameIdentifier(), Entity.EntityType.CATALOG)); + int deleted = + SessionUtils.doWithCommitAndFetchResult( + CatalogMetaMapper.class, + mapper -> + mapper.softDeleteCatalogMetasByCatalogId(catalog.id(), newPO.getCurrentVersion())); + assertEquals(1, deleted); + } + + @TestTemplate + public void testOverwriteInsertAdvancesCurrentVersion() throws IOException { + CatalogEntity catalog = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + "catalog_overwrite_occ", + auditInfo); + backend.insert(catalog, false); + CatalogPO initialPO = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, mapper -> mapper.selectCatalogMetaById(catalog.id())); + + backend.insert(catalog, true); + + CatalogPO overwrittenPO = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, mapper -> mapper.selectCatalogMetaById(catalog.id())); + assertEquals(initialPO.getCurrentVersion() + 1, overwrittenPO.getCurrentVersion().longValue()); + assertEquals( + overwrittenPO.getCurrentVersion().longValue(), overwrittenPO.getLastVersion().longValue()); + + // A writer that observed the catalog before the overwrite must not pass its compare-and-set. + int staleDelete = + SessionUtils.doWithCommitAndFetchResult( + CatalogMetaMapper.class, + mapper -> + mapper.softDeleteCatalogMetasByCatalogId( + catalog.id(), initialPO.getCurrentVersion())); + assertEquals(0, staleDelete); + } + + @TestTemplate + public void testAlterReportsOptimisticLockConflict() throws IOException { + CatalogEntity catalog = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + "catalog_alter_conflict", + auditInfo); + backend.insert(catalog, false); + + assertThrows( + OptimisticLockException.class, + () -> + CatalogMetaService.getInstance() + .updateCatalog( + catalog.nameIdentifier(), + entity -> { + CatalogEntity current = (CatalogEntity) entity; + CatalogPO currentPO = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, + mapper -> mapper.selectCatalogMetaById(current.id())); + CatalogEntity competingUpdate = + copyCatalogWithComment(current, "competing update"); + CatalogPO competingPO = + POConverters.updateCatalogPOWithVersion( + currentPO, competingUpdate, currentPO.getMetalakeId()); + SessionUtils.doWithCommitAndFetchResult( + CatalogMetaMapper.class, + mapper -> mapper.updateCatalogMeta(competingPO, currentPO)); + return copyCatalogWithComment(current, "requested update"); + })); + } + + @TestTemplate + public void testAlterReportsNoSuchWhenCatalogIsDeletedConcurrently() throws IOException { + CatalogEntity catalog = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + "catalog_alter_deleted", + auditInfo); + backend.insert(catalog, false); + + assertThrows( + NoSuchEntityException.class, + () -> + CatalogMetaService.getInstance() + .updateCatalog( + catalog.nameIdentifier(), + entity -> { + CatalogEntity current = (CatalogEntity) entity; + CatalogPO currentPO = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, + mapper -> mapper.selectCatalogMetaById(current.id())); + SessionUtils.doWithCommitAndFetchResult( + CatalogMetaMapper.class, + mapper -> + mapper.softDeleteCatalogMetasByCatalogId( + current.id(), currentPO.getCurrentVersion())); + return copyCatalogWithComment(current, "requested update"); + })); + } + + @TestTemplate + public void testNonCascadeDeleteRollsBackCatalogFence() throws IOException { + CatalogEntity catalog = + createCatalog( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofCatalog(metalakeName), + "catalog_non_empty", + auditInfo); + backend.insert(catalog, false); + SchemaEntity schema = + createSchemaEntity( + RandomIdGenerator.INSTANCE.nextId(), + NamespaceUtil.ofSchema(metalakeName, catalog.name()), + "schema", + auditInfo); + backend.insert(schema, false); + CatalogPO beforeDelete = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, mapper -> mapper.selectCatalogMetaById(catalog.id())); + + assertThrows( + NonEmptyEntityException.class, + () -> CatalogMetaService.getInstance().deleteCatalog(catalog.nameIdentifier(), false)); + + CatalogPO afterDelete = + SessionUtils.getWithoutCommit( + CatalogMetaMapper.class, mapper -> mapper.selectCatalogMetaById(catalog.id())); + assertEquals(beforeDelete.getCurrentVersion(), afterDelete.getCurrentVersion()); + assertTrue(backend.exists(catalog.nameIdentifier(), Entity.EntityType.CATALOG)); + assertTrue(backend.exists(schema.nameIdentifier(), Entity.EntityType.SCHEMA)); + } + @TestTemplate public void testMetaLifeCycleFromCreationToDeletion() throws IOException { CatalogEntity catalog = @@ -303,6 +573,59 @@ public void testDeleteCatalogCascadeRemovesTagRelations() throws IOException { assertEquals(0, countActiveTagRelForMetadataObject(function.id(), "FUNCTION")); } + private List insertCatalogsConcurrently(CatalogEntity first, CatalogEntity second) + throws Exception { + ExecutorService executor = Executors.newFixedThreadPool(2); + CountDownLatch ready = new CountDownLatch(2); + CountDownLatch start = new CountDownLatch(1); + try { + Future firstResult = + executor.submit( + () -> { + ready.countDown(); + start.await(); + try { + CatalogMetaService.getInstance().insertCatalog(first, false); + return null; + } catch (Throwable throwable) { + return throwable; + } + }); + Future secondResult = + executor.submit( + () -> { + ready.countDown(); + start.await(); + try { + CatalogMetaService.getInstance().insertCatalog(second, false); + return null; + } catch (Throwable throwable) { + return throwable; + } + }); + assertTrue(ready.await(30, TimeUnit.SECONDS)); + start.countDown(); + return Arrays.asList( + firstResult.get(30, TimeUnit.SECONDS), secondResult.get(30, TimeUnit.SECONDS)); + } finally { + start.countDown(); + executor.shutdownNow(); + } + } + + private CatalogEntity copyCatalogWithComment(CatalogEntity catalog, String comment) { + return CatalogEntity.builder() + .withId(catalog.id()) + .withName(catalog.name()) + .withNamespace(catalog.namespace()) + .withType(catalog.getType()) + .withProvider(catalog.getProvider()) + .withComment(comment) + .withProperties(catalog.getProperties()) + .withAuditInfo(auditInfo) + .build(); + } + private void associateTag(TagEntity tag, NameIdentifier ident, Entity.EntityType type) throws IOException { TagMetaService.getInstance() diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java index 22a53895065..10e4ac042fe 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java @@ -681,6 +681,8 @@ public void testUpdateCatalogPOVersion() { assertEquals(1, initPO.getCurrentVersion()); assertEquals(1, initPO.getLastVersion()); assertEquals(0, initPO.getDeletedAt()); + assertEquals(2, updatePO.getCurrentVersion()); + assertEquals(2, updatePO.getLastVersion()); assertEquals("this is test2", updatePO.getCatalogComment()); }