This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 402dd20435 [#12452] improvement(core): add OCC for catalog writes
(#12455)
402dd20435 is described below
commit 402dd2043560101495e3cb876f3e61c55ac5cf5c
Author: Qi Yu <[email protected]>
AuthorDate: Wed Aug 19 19:41:44 2026 +0800
[#12452] improvement(core): add OCC for catalog writes (#12455)
### What changes were proposed in this pull request?
Add database-backed optimistic concurrency control and transaction
boundaries for catalog writes.
- Advance the catalog OCC version on every alter and on an overwrite
insert, and guard alter and drop with a compare-and-set on the observed
version, classifying a failed CAS as either a stale-version conflict or
an already-missing entity.
- Protect catalog creation with a shared lock on the parent metalake row
on MySQL and PostgreSQL, without changing the parent version. H2 uses an
exclusive lock because it has no shared row-lock syntax, so catalog
creations under one metalake serialize on H2.
- Keep the catalog CAS and the non-empty check or the cascade cleanup in
one database transaction, and CAS-delete descendant schemas with their
observed identifier-and-version pairs.
- Keep the drop idempotent when another server deletes the catalog
mid-request: a later store read such as listing schemas reports the
catalog as missing, and the cached catalog wrapper is discarded.
(`store.delete` maps a missing entity to `false` on its own.)
- Comment the concurrency-critical statements, so the reason for their
ordering is readable next to the code.
Rebased on current `main` (on top of #12374). Second of three PRs
replacing #12350.
**Known window:** until #12456 lands, a schema created concurrently with
a catalog drop can still slip through, because schema writes do not yet
take the parent catalog row lock. #12456 adds that lock, which closes
the window for both the cascade cleanup and the non-empty check.
### Why are the changes needed?
Managed catalog operations previously consisted of multiple independent
reads and writes. Concurrent alter, create, and drop requests could
overwrite newer metadata, create a catalog below a metalake that was
being deleted, or run partial cascade cleanup.
Fix: #12452
### Does this PR introduce _any_ user-facing change?
Concurrent catalog version conflicts are reported as HTTP 409. If the
observed entity was deleted or renamed away, alter reports not found and
drop preserves its idempotent false result.
### How was this patch tested?
- `./gradlew :core:test :core:javadoc -PskipITs` (H2)
- New tests in `TestCatalogMetaService`, `TestCatalogManager`,
`TestPOConverters`, including
`testOverwriteInsertAdvancesCurrentVersion`, which checks that an
overwrite moves the version forward and that a writer holding the
pre-overwrite version no longer passes its CAS.
- MySQL and PostgreSQL coverage is left to CI
(`-PskipDockerTests=false`).
---------
Co-authored-by: Jerry Shao <[email protected]>
---
.../apache/gravitino/catalog/CatalogManager.java | 20 +-
.../relational/mapper/CatalogMetaMapper.java | 16 +-
.../mapper/CatalogMetaSQLProviderFactory.java | 10 +-
.../relational/mapper/MetalakeMetaMapper.java | 6 +
.../mapper/MetalakeMetaSQLProviderFactory.java | 15 +-
.../relational/mapper/SchemaMetaMapper.java | 5 -
.../mapper/SchemaMetaSQLProviderFactory.java | 4 -
.../provider/base/CatalogMetaBaseSQLProvider.java | 36 ++-
.../provider/base/MetalakeMetaBaseSQLProvider.java | 5 +
.../provider/base/SchemaMetaBaseSQLProvider.java | 8 -
.../postgresql/CatalogMetaPostgreSQLProvider.java | 29 +-
.../postgresql/MetalakeMetaPostgreSQLProvider.java | 6 +
.../postgresql/SchemaMetaPostgreSQLProvider.java | 8 -
.../relational/service/CatalogMetaService.java | 223 ++++++++++----
.../relational/service/MetalakeMetaService.java | 10 +-
.../storage/relational/utils/POConverters.java | 8 +-
.../gravitino/catalog/TestCatalogManager.java | 73 +++++
.../TestCatalogMetaPostgreSQLProvider.java | 59 ++++
.../relational/service/TestCatalogMetaService.java | 323 +++++++++++++++++++++
.../storage/relational/utils/TestPOConverters.java | 2 +
20 files changed, 748 insertions(+), 118 deletions(-)
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 72c0139cf4..fd0dcd2afb 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
@@ -625,9 +625,18 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
LockType.WRITE,
() -> {
checkMetalake(metalakeIdent, store);
- boolean needClean = true;
+ boolean needClean = false;
try {
- store.put(e, false /* overwrite */);
+ try {
+ store.put(e, false /* overwrite */);
+ } catch (NoSuchEntityException e1) {
+ // The relational store locks and rechecks the parent metalake
while inserting the
+ // catalog. A concurrent drop or rename can therefore make the
metalake disappear
+ // after checkMetalake() succeeds but before this insert starts.
+ LOG.warn("Metalake {} does not exist", metalakeIdent, e1);
+ throw new NoSuchMetalakeException(e1, "Metalake %s does not
exist", metalakeIdent);
+ }
+ needClean = true;
CatalogWrapper wrapper =
catalogCache.get(ident, id -> createCatalogWrapper(e,
mergedConfig));
@@ -984,6 +993,13 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
} 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 553a33a71a..7cf0195b3e 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 @@ public interface CatalogMetaMapper {
@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 @@ public interface CatalogMetaMapper {
@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 9b3154fa9e..d2afccb0d6 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 class CatalogMetaSQLProviderFactory {
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 class CatalogMetaSQLProviderFactory {
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 665ae02b79..2adcce1a28 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 11f64ad662..8e3737b986 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 class MetalakeMetaSQLProviderFactory {
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 class MetalakeMetaSQLProviderFactory {
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 989f1ebd99..9c7ab2b4d9 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 @@ public interface SchemaMetaMapper {
method = "softDeleteSchemaMetasBySchemaIds")
Integer softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List<Long>
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 30c2ee9dfc..557bee15f8 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 class SchemaMetaSQLProviderFactory {
return getProvider().softDeleteSchemaMetasBySchemaIds(schemaIds);
}
- public static String softDeleteSchemaMetasByCatalogId(@Param("catalogId")
Long catalogId) {
- return getProvider().softDeleteSchemaMetasByCatalogId(catalogId);
- }
-
/** Returns SQL that soft-deletes schemas using identifier-and-version
pairs. */
public static String softDeleteSchemaMetasWithVersion(
@Param("schemaMetas") List<SchemaPO> 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 f50b9c203d..e6f8f03c18 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 class CatalogMetaBaseSQLProvider {
+ " 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 class CatalogMetaBaseSQLProvider {
+ " 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.
+ *
+ * <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 updateCatalogMeta(
@Param("newCatalogMeta") CatalogPO newCatalogPO,
@Param("oldCatalogMeta") CatalogPO oldCatalogPO) {
@@ -211,25 +228,18 @@ public class CatalogMetaBaseSQLProvider {
+ " 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 a7a78f4b7b..f8e99a3990 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 class MetalakeMetaBaseSQLProvider {
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 ee72e38324..9ee36cc52d 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 class SchemaMetaBaseSQLProvider {
+ "</script>";
}
- 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<SchemaPO> schemaPOs) {
return "<script>"
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java
index 9ff0849817..1a29a27522 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java
@@ -27,11 +27,12 @@ import org.apache.ibatis.annotations.Param;
public class CatalogMetaPostgreSQLProvider extends CatalogMetaBaseSQLProvider {
@Override
- public String softDeleteCatalogMetasByCatalogId(Long catalogId) {
+ public String softDeleteCatalogMetasByCatalogId(Long catalogId, Long
currentVersion) {
return "UPDATE "
+ TABLE_NAME
+ " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000
AS BIGINT)"
- + " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
+ + " WHERE catalog_id = #{catalogId}"
+ + " AND current_version = #{currentVersion} AND deleted_at = 0";
}
/** {@inheritDoc} */
@@ -86,8 +87,18 @@ public class CatalogMetaPostgreSQLProvider extends
CatalogMetaBaseSQLProvider {
+ " 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. The column has to be written as <table>.<column>
here: on this side of
+ // ON CONFLICT a bare name could mean either the stored row or the
rejected one, and
+ // PostgreSQL refuses it as ambiguous. The table-qualified name is the
stored row, and
+ // PostgreSQL computes every assignment from it.
+ + " current_version = "
+ + TABLE_NAME
+ + ".current_version + 1,"
+ + " last_version = "
+ + TABLE_NAME
+ + ".current_version + 1,"
+ " deleted_at = #{catalogMeta.deletedAt}";
}
@@ -108,17 +119,7 @@ public class CatalogMetaPostgreSQLProvider extends
CatalogMetaBaseSQLProvider {
+ " 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 (CAST(catalog_comment AS VARCHAR) IS NULL AND "
- + " CAST(#{oldCatalogMeta.catalogComment} AS VARCHAR) 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";
}
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/MetalakeMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/MetalakeMetaPostgreSQLProvider.java
index 20a92d1063..074ce44429 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/MetalakeMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/MetalakeMetaPostgreSQLProvider.java
@@ -25,6 +25,12 @@ import org.apache.gravitino.storage.relational.po.MetalakePO;
import org.apache.ibatis.annotations.Param;
public class MetalakeMetaPostgreSQLProvider extends
MetalakeMetaBaseSQLProvider {
+
+ @Override
+ public String selectMetalakeMetaByIdForShare(Long metalakeId) {
+ return selectMetalakeMetaById(metalakeId) + " FOR SHARE";
+ }
+
@Override
public String softDeleteMetalakeMetaByMetalakeId(Long metalakeId, Long
currentVersion) {
return "UPDATE "
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SchemaMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SchemaMetaPostgreSQLProvider.java
index d805e52c0b..42e5d4e665 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SchemaMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SchemaMetaPostgreSQLProvider.java
@@ -125,14 +125,6 @@ public class SchemaMetaPostgreSQLProvider extends
SchemaMetaBaseSQLProvider {
+ "</script>";
}
- @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<SchemaPO> 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 9f63b9189f..5cd9ca66e3 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 com.google.common.base.Preconditions;
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.Namespace;
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.FilesetMetaMapper;
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.TagMetadataObjectRelMapper
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 class CatalogMetaService {
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 class CatalogMetaService {
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 class CatalogMetaService {
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 class CatalogMetaService {
ViewMetaMapper.class,
mapper -> mapper.softDeleteViewMetasByCatalogId(catalogId)));
} else {
- List<SchemaEntity> 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<SchemaPO> 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,
@@ -402,4 +429,96 @@ public class CatalogMetaService {
return POConverters.fromCatalogPOs(catalogPOs,
firstIdent.namespace());
});
}
+
+ /**
+ * 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.
+ *
+ * <p>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.
+ *
+ * <p>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<SchemaPO> 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);
+ }
+ }
}
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 cc42229f73..58c8c054fe 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 @@ public class MetalakeMetaService {
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 2a05888fd7..3085c1ed6d 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 class POConverters {
*/
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 c55c404469..b53313f026 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.Catalog;
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.Capability;
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;
@@ -391,6 +394,41 @@ public class TestCatalogManager {
Assertions.assertNull(catalogManager.getCatalogCache().getIfPresent(failedIdent));
}
+ @Test
+ void testCreateCatalogReturnsNoSuchMetalakeWhenParentDisappears() throws
Exception {
+ InMemoryEntityStore store = Mockito.spy(new InMemoryEntityStore());
+ store.initialize(config);
+ store.put(metalakeEntity, true);
+
+ NoSuchEntityException missingMetalake =
+ new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ EntityType.METALAKE.name().toLowerCase(),
+ metalake);
+ Mockito.doThrow(missingMetalake).when(store).put(any(CatalogEntity.class),
eq(false));
+
+ CatalogManager manager =
+ new CatalogManager(config, store, new RandomIdGenerator(), new
SecretManager(config));
+ NameIdentifier ident = NameIdentifier.of(metalake,
"concurrent_parent_drop");
+ Map<String, String> props =
+ ImmutableMap.of(
+ PROPERTY_KEY1, "value1", PROPERTY_KEY2, "value2",
PROPERTY_KEY5_PREFIX + "1", "value3");
+
+ try {
+ NoSuchMetalakeException exception =
+ Assertions.assertThrows(
+ NoSuchMetalakeException.class,
+ () ->
+ manager.createCatalog(
+ ident, Catalog.Type.RELATIONAL, provider, "comment",
props));
+ Assertions.assertSame(missingMetalake, exception.getCause());
+ Mockito.verify(store, Mockito.never()).delete(ident, EntityType.CATALOG,
true);
+ } finally {
+ manager.close();
+ store.close();
+ }
+ }
+
@Test
public void testCreateCatalogValidatesBackendConnection() {
Map<String, String> okProps =
@@ -850,6 +888,26 @@ public class TestCatalogManager {
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<String, String> 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 +1041,7 @@ public class TestCatalogManager {
private final AtomicReference<EntityChangeLogListener>
unregisteredListener =
new AtomicReference<>();
private boolean returnFalseForCatalogDelete;
+ private boolean throwMissingCatalogForSchemaList;
@Override
public boolean delete(NameIdentifier ident, EntityType entityType, boolean
cascade)
@@ -993,6 +1052,20 @@ public class TestCatalogManager {
return super.delete(ident, entityType, cascade);
}
+ @Override
+ public <E extends Entity & HasIdentifier> List<E> list(
+ Namespace namespace, Class<E> 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/mapper/provider/postgresql/TestCatalogMetaPostgreSQLProvider.java
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestCatalogMetaPostgreSQLProvider.java
new file mode 100644
index 0000000000..90b5d8e257
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestCatalogMetaPostgreSQLProvider.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.mapper.provider.postgresql;
+
+import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestCatalogMetaPostgreSQLProvider {
+
+ @Test
+ void testOverwriteInsertQualifiesVersionColumns() {
+ String conflictClause = conflictClause();
+
+ // PostgreSQL rejects a bare column name on this side of ON CONFLICT,
because it could mean
+ // either the stored row or the rejected one. Both assignments must name
the table.
+ Assertions.assertTrue(
+ conflictClause.contains(CatalogMetaMapper.TABLE_NAME +
".current_version + 1"),
+ () -> "Version columns must be table-qualified in ON CONFLICT, but
got: " + conflictClause);
+ Assertions.assertFalse(
+ conflictClause.matches(".*[^.\\w]current_version\\s*\\+.*"),
+ () -> "Found an unqualified current_version reference in: " +
conflictClause);
+ }
+
+ @Test
+ void testOverwriteInsertAdvancesBothVersionColumns() {
+ String conflictClause = conflictClause();
+
+ // An overwrite must never write the initial version back, or a stale
writer could still pass
+ // its own version check afterwards.
+ Assertions.assertTrue(
+ conflictClause.contains(
+ "current_version = " + CatalogMetaMapper.TABLE_NAME +
".current_version + 1"));
+ Assertions.assertTrue(
+ conflictClause.contains(
+ "last_version = " + CatalogMetaMapper.TABLE_NAME +
".current_version + 1"));
+ }
+
+ private String conflictClause() {
+ String sql = new
CatalogMetaPostgreSQLProvider().insertCatalogMetaOnDuplicateKeyUpdate(null);
+ return sql.substring(sql.indexOf("ON CONFLICT"));
+ }
+}
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 de5520908a..cbc598903e 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.ResultSet;
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.rel.types.Types;
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 class TestCatalogMetaService extends TestJDBCBackend
{
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<Throwable> 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<Throwable> 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 @@ public class TestCatalogMetaService extends
TestJDBCBackend {
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 class TestCatalogMetaService extends
TestJDBCBackend {
assertEquals(0, countActiveTagRelForMetadataObject(function.id(),
"FUNCTION"));
}
+ private List<Throwable> insertCatalogsConcurrently(CatalogEntity first,
CatalogEntity second)
+ throws Exception {
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ CountDownLatch ready = new CountDownLatch(2);
+ CountDownLatch start = new CountDownLatch(1);
+ try {
+ Future<Throwable> firstResult =
+ executor.submit(
+ () -> {
+ ready.countDown();
+ start.await();
+ try {
+ CatalogMetaService.getInstance().insertCatalog(first, false);
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+ Future<Throwable> 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 22a5389506..10e4ac042f 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 class TestPOConverters {
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());
}
