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 7bbe3a58dd [#12451] improvement(core): add OCC for metalake writes
(#12454)
7bbe3a58dd is described below
commit 7bbe3a58dd4a303d553989500ccab5580251eb3b
Author: Qi Yu <[email protected]>
AuthorDate: Mon Aug 17 13:48:50 2026 +0800
[#12451] improvement(core): add OCC for metalake writes (#12454)
### What changes were proposed in this pull request?
Add database-backed optimistic concurrency control and transaction
boundaries for metalake writes.
- Advance the metalake OCC version on every alter, and guard alter and
delete with a compare-and-set on the observed version.
- Classify a failed CAS as either a stale-version conflict or an
already-missing entity.
- Keep the root CAS and the non-empty check or the cascade cleanup in
one database transaction. Change-log emission now happens at the
`JDBCBackend` mutation boundary after #12374, so this PR no longer
touches it.
- Lock the catalog rows before a metalake cascade snapshot, then
CAS-delete descendant catalogs and schemas with their observed
identifier-and-version pairs, so a concurrent child write is reported
instead of silently dropped.
- Add the shared `OptimisticLockException` factories used by the
follow-up catalog and schema PRs.
Rebased on current `main` (on top of #12374). This is the first of three
PRs that replace #12350, which reviewers found too large. The stack is
metalake -> catalog -> schema; each PR is independently green. Two
cross-entity tests in `TestMetalakeMetaService` (concurrent schema alter
during a metalake cascade, and metalake cascade racing a schema create)
land with the schema PR, because the behaviour they assert only exists
once schema writes take the catalog row lock and bump the schema
version.
### Why are the changes needed?
Managed metalake operations previously consisted of multiple independent
reads and writes. Concurrent alter and delete requests could overwrite
newer metadata, and a cascade delete could run partial cleanup while
another writer was still modifying descendants.
Fix: #12451
### Does this PR introduce _any_ user-facing change?
Concurrent metalake 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 :catalogs:catalog-fileset:test
:catalogs:catalog-kafka:test -PskipITs` (H2)
- New tests in `TestMetalakeMetaService`, `TestMetalakeManager`,
`TestExceptionUtils`, `TestPOConverters`.
- MySQL and PostgreSQL coverage for the new `<foreach>` CAS delete and
the PostgreSQL lock syntax is left to CI (`-PskipDockerTests=false`).
---------
Co-authored-by: Jerry Shao <[email protected]>
---
.../apache/gravitino/metalake/MetalakeManager.java | 4 +-
.../relational/mapper/CatalogMetaMapper.java | 15 +-
.../mapper/CatalogMetaSQLProviderFactory.java | 11 +-
.../relational/mapper/MetalakeMetaMapper.java | 19 ++-
.../mapper/MetalakeMetaSQLProviderFactory.java | 10 +-
.../relational/mapper/SchemaMetaMapper.java | 17 +-
.../mapper/SchemaMetaSQLProviderFactory.java | 15 +-
.../provider/base/CatalogMetaBaseSQLProvider.java | 18 ++-
.../provider/base/MetalakeMetaBaseSQLProvider.java | 18 +--
.../provider/base/SchemaMetaBaseSQLProvider.java | 28 +++-
.../postgresql/CatalogMetaPostgreSQLProvider.java | 13 +-
.../postgresql/MetalakeMetaPostgreSQLProvider.java | 13 +-
.../postgresql/SchemaMetaPostgreSQLProvider.java | 16 +-
.../relational/service/MetalakeMetaService.java | 163 ++++++++++++++-----
.../storage/relational/utils/ExceptionUtils.java | 36 +++++
.../storage/relational/utils/POConverters.java | 6 +-
.../gravitino/metalake/TestMetalakeManager.java | 20 +++
.../service/TestMetalakeMetaService.java | 173 +++++++++++++++++++++
.../relational/utils/TestExceptionUtils.java | 50 ++++++
.../storage/relational/utils/TestPOConverters.java | 2 +
20 files changed, 555 insertions(+), 92 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java
b/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java
index 1bc5dfd13b..8f4a038ae4 100644
--- a/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java
+++ b/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java
@@ -358,7 +358,9 @@ public class MetalakeManager implements MetalakeDispatcher,
Closeable {
}
return store.delete(ident, EntityType.METALAKE, true);
- } catch (NoSuchMetalakeException e) {
+ } catch (NoSuchMetalakeException | NoSuchEntityException e) {
+ // Another server may have completed the drop after the initial
existence check.
+ // Dropping an already-removed metalake remains an idempotent
false result.
return false;
} catch (IOException 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 9f19d1a6a8..553a33a71a 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
@@ -47,6 +47,12 @@ public interface CatalogMetaMapper {
@SelectProvider(type = CatalogMetaSQLProviderFactory.class, method =
"listCatalogPOsByMetalakeId")
List<CatalogPO> listCatalogPOsByMetalakeId(@Param("metalakeId") Long
metalakeId);
+ /** Selects and locks all active catalogs in a metalake for the current
transaction. */
+ @SelectProvider(
+ type = CatalogMetaSQLProviderFactory.class,
+ method = "listCatalogPOsByMetalakeIdForUpdate")
+ List<CatalogPO> listCatalogPOsByMetalakeIdForUpdate(@Param("metalakeId")
Long metalakeId);
+
@SelectProvider(type = CatalogMetaSQLProviderFactory.class, method =
"listCatalogPOsByCatalogIds")
List<CatalogPO> listCatalogPOsByCatalogIds(@Param("catalogIds") List<Long>
catalogIds);
@@ -91,10 +97,15 @@ public interface CatalogMetaMapper {
method = "softDeleteCatalogMetasByCatalogId")
Integer softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long
catalogId);
+ /**
+ * Soft-deletes catalogs whose identifiers and OCC versions still match.
+ *
+ * @return the number of deleted rows
+ */
@UpdateProvider(
type = CatalogMetaSQLProviderFactory.class,
- method = "softDeleteCatalogMetasByMetalakeId")
- Integer softDeleteCatalogMetasByMetalakeId(@Param("metalakeId") Long
metalakeId);
+ method = "softDeleteCatalogMetasWithVersion")
+ Integer softDeleteCatalogMetasWithVersion(@Param("catalogMetas")
List<CatalogPO> catalogPOs);
@DeleteProvider(
type = CatalogMetaSQLProviderFactory.class,
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 c3a7954a25..9b3154fa9e 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
@@ -61,6 +61,11 @@ public class CatalogMetaSQLProviderFactory {
return getProvider().listCatalogPOsByMetalakeId(metalakeId);
}
+ /** Returns SQL that lists and locks all active catalogs in a metalake. */
+ public static String
listCatalogPOsByMetalakeIdForUpdate(@Param("metalakeId") Long metalakeId) {
+ return getProvider().listCatalogPOsByMetalakeIdForUpdate(metalakeId);
+ }
+
public static String listCatalogPOsByCatalogIds(@Param("catalogIds")
List<Long> catalogIds) {
return getProvider().listCatalogPOsByCatalogIds(catalogIds);
}
@@ -113,8 +118,10 @@ public class CatalogMetaSQLProviderFactory {
return getProvider().softDeleteCatalogMetasByCatalogId(catalogId);
}
- public static String softDeleteCatalogMetasByMetalakeId(@Param("metalakeId")
Long metalakeId) {
- return getProvider().softDeleteCatalogMetasByMetalakeId(metalakeId);
+ /** Returns SQL that soft-deletes catalogs using identifier-and-version
pairs. */
+ public static String softDeleteCatalogMetasWithVersion(
+ @Param("catalogMetas") List<CatalogPO> catalogPOs) {
+ return getProvider().softDeleteCatalogMetasWithVersion(catalogPOs);
}
public static String deleteCatalogMetasByLegacyTimeline(
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 f705c283ce..665ae02b79 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
@@ -47,6 +47,12 @@ public interface MetalakeMetaMapper {
@SelectProvider(type = MetalakeMetaSQLProviderFactory.class, method =
"selectMetalakeMetaById")
MetalakePO selectMetalakeMetaById(@Param("metalakeId") Long metalakeId);
+ /** Selects and locks an active metalake by ID for the current transaction.
*/
+ @SelectProvider(
+ type = MetalakeMetaSQLProviderFactory.class,
+ method = "selectMetalakeMetaByIdForUpdate")
+ MetalakePO selectMetalakeMetaByIdForUpdate(@Param("metalakeId") Long
metalakeId);
+
@SelectProvider(
type = MetalakeMetaSQLProviderFactory.class,
method = "listMetalakePOsByMetalakeIds")
@@ -65,15 +71,26 @@ public interface MetalakeMetaMapper {
method = "insertMetalakeMetaOnDuplicateKeyUpdate")
void insertMetalakeMetaOnDuplicateKeyUpdate(@Param("metalakeMeta")
MetalakePO metalakePO);
+ /**
+ * Updates an active metalake only when its OCC version still matches the
observed version.
+ *
+ * @return the number of updated rows
+ */
@UpdateProvider(type = MetalakeMetaSQLProviderFactory.class, method =
"updateMetalakeMeta")
Integer updateMetalakeMeta(
@Param("newMetalakeMeta") MetalakePO newMetalakePO,
@Param("oldMetalakeMeta") MetalakePO oldMetalakePO);
+ /**
+ * Soft-deletes an active metalake only when its OCC version still matches.
+ *
+ * @return the number of deleted rows
+ */
@UpdateProvider(
type = MetalakeMetaSQLProviderFactory.class,
method = "softDeleteMetalakeMetaByMetalakeId")
- Integer softDeleteMetalakeMetaByMetalakeId(@Param("metalakeId") Long
metalakeId);
+ Integer softDeleteMetalakeMetaByMetalakeId(
+ @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long
currentVersion);
@DeleteProvider(
type = MetalakeMetaSQLProviderFactory.class,
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 eba26f9e02..11f64ad662 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
@@ -65,6 +65,11 @@ public class MetalakeMetaSQLProviderFactory {
return getProvider().selectMetalakeMetaById(metalakeId);
}
+ /** Returns SQL that selects and locks an active metalake by ID. */
+ public static String selectMetalakeMetaByIdForUpdate(@Param("metalakeId")
Long metalakeId) {
+ return getProvider().selectMetalakeMetaByIdForUpdate(metalakeId);
+ }
+
public static String selectMetalakeIdMetaByName(@Param("metalakeName")
String metalakeName) {
return getProvider().selectMetalakeIdMetaByName(metalakeName);
}
@@ -88,8 +93,9 @@ public class MetalakeMetaSQLProviderFactory {
return getProvider().updateMetalakeMeta(newMetalakePO, oldMetalakePO);
}
- public static String softDeleteMetalakeMetaByMetalakeId(@Param("metalakeId")
Long metalakeId) {
- return getProvider().softDeleteMetalakeMetaByMetalakeId(metalakeId);
+ public static String softDeleteMetalakeMetaByMetalakeId(
+ @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long
currentVersion) {
+ return getProvider().softDeleteMetalakeMetaByMetalakeId(metalakeId,
currentVersion);
}
public static String deleteMetalakeMetasByLegacyTimeline(
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 1c9b5286b2..989f1ebd99 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
@@ -42,6 +42,10 @@ public interface SchemaMetaMapper {
@SelectProvider(type = SchemaMetaSQLProviderFactory.class, method =
"listSchemaPOsByCatalogId")
List<SchemaPO> listSchemaPOsByCatalogId(@Param("catalogId") Long catalogId);
+ /** Lists all active schemas in a metalake. */
+ @SelectProvider(type = SchemaMetaSQLProviderFactory.class, method =
"listSchemaPOsByMetalakeId")
+ List<SchemaPO> listSchemaPOsByMetalakeId(@Param("metalakeId") Long
metalakeId);
+
@SelectProvider(
type = SchemaMetaSQLProviderFactory.class,
method = "listSchemaPOsByFullQualifiedName")
@@ -109,13 +113,18 @@ public interface SchemaMetaMapper {
@UpdateProvider(
type = SchemaMetaSQLProviderFactory.class,
- method = "softDeleteSchemaMetasByMetalakeId")
- Integer softDeleteSchemaMetasByMetalakeId(@Param("metalakeId") Long
metalakeId);
+ method = "softDeleteSchemaMetasByCatalogId")
+ Integer softDeleteSchemaMetasByCatalogId(@Param("catalogId") Long catalogId);
+ /**
+ * Soft-deletes schemas whose identifiers and OCC versions still match.
+ *
+ * @return the number of deleted rows
+ */
@UpdateProvider(
type = SchemaMetaSQLProviderFactory.class,
- method = "softDeleteSchemaMetasByCatalogId")
- Integer softDeleteSchemaMetasByCatalogId(@Param("catalogId") Long catalogId);
+ method = "softDeleteSchemaMetasWithVersion")
+ Integer softDeleteSchemaMetasWithVersion(@Param("schemaMetas")
List<SchemaPO> schemaPOs);
@DeleteProvider(
type = SchemaMetaSQLProviderFactory.class,
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 acc2717026..30c2ee9dfc 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
@@ -72,6 +72,11 @@ public class SchemaMetaSQLProviderFactory {
return getProvider().listSchemaPOsByCatalogId(catalogId);
}
+ /** Returns SQL that lists all active schemas in a metalake. */
+ public static String listSchemaPOsByMetalakeId(@Param("metalakeId") Long
metalakeId) {
+ return getProvider().listSchemaPOsByMetalakeId(metalakeId);
+ }
+
public static String selectSchemaIdByCatalogIdAndName(
@Param("catalogId") Long catalogId, @Param("schemaName") String name) {
return getProvider().selectSchemaIdByCatalogIdAndName(catalogId, name);
@@ -120,14 +125,16 @@ public class SchemaMetaSQLProviderFactory {
return getProvider().softDeleteSchemaMetasBySchemaIds(schemaIds);
}
- public static String softDeleteSchemaMetasByMetalakeId(@Param("metalakeId")
Long metalakeId) {
- return getProvider().softDeleteSchemaMetasByMetalakeId(metalakeId);
- }
-
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) {
+ return getProvider().softDeleteSchemaMetasWithVersion(schemaPOs);
+ }
+
public static String deleteSchemaMetasByLegacyTimeline(
@Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
return getProvider().deleteSchemaMetasByLegacyTimeline(legacyTimeline,
limit);
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 be03900dc2..f50b9c203d 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
@@ -53,6 +53,11 @@ public class CatalogMetaBaseSQLProvider {
+ " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
}
+ /** Returns SQL that lists and locks all active catalogs in a metalake. */
+ public String listCatalogPOsByMetalakeIdForUpdate(@Param("metalakeId") Long
metalakeId) {
+ return listCatalogPOsByMetalakeId(metalakeId) + " FOR UPDATE";
+ }
+
public String listCatalogPOsByCatalogIds(@Param("catalogIds") List<Long>
catalogIds) {
return "<script>"
+ "SELECT catalog_id as catalogId, catalog_name as catalogName,"
@@ -227,12 +232,19 @@ public class CatalogMetaBaseSQLProvider {
+ " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
}
- public String softDeleteCatalogMetasByMetalakeId(@Param("metalakeId") Long
metalakeId) {
- return "UPDATE "
+ /** Returns SQL that soft-deletes catalogs using identifier-and-version
pairs. */
+ public String softDeleteCatalogMetasWithVersion(
+ @Param("catalogMetas") List<CatalogPO> catalogPOs) {
+ return "<script>"
+ + "UPDATE "
+ TABLE_NAME
+ " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+ " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
- + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
+ + " WHERE deleted_at = 0 AND "
+ + "<foreach collection='catalogMetas' item='item' separator=' OR '
open='(' close=')'>"
+ + "(catalog_id = #{item.catalogId} AND current_version =
#{item.currentVersion})"
+ + "</foreach>"
+ + "</script>";
}
public String deleteCatalogMetasByLegacyTimeline(
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 2524eda76f..a7a78f4b7b 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
@@ -59,6 +59,11 @@ public class MetalakeMetaBaseSQLProvider {
+ " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
}
+ /** Returns SQL that selects and locks an active metalake by ID. */
+ public String selectMetalakeMetaByIdForUpdate(@Param("metalakeId") Long
metalakeId) {
+ return selectMetalakeMetaById(metalakeId) + " FOR UPDATE";
+ }
+
public String selectMetalakeIdMetaByName(@Param("metalakeName") String
metalakeName) {
return "SELECT metalake_id as metalakeId"
+ " FROM "
@@ -143,23 +148,18 @@ public class MetalakeMetaBaseSQLProvider {
+ " current_version = #{newMetalakeMeta.currentVersion},"
+ " last_version = #{newMetalakeMeta.lastVersion}"
+ " WHERE metalake_id = #{oldMetalakeMeta.metalakeId}"
- + " AND metalake_name = #{oldMetalakeMeta.metalakeName}"
- + " AND (metalake_comment = #{oldMetalakeMeta.metalakeComment} "
- + " OR (metalake_comment IS NULL and
#{oldMetalakeMeta.metalakeComment} IS NULL))"
- + " AND properties = #{oldMetalakeMeta.properties}"
- + " AND audit_info = #{oldMetalakeMeta.auditInfo}"
- + " AND schema_version = #{oldMetalakeMeta.schemaVersion}"
+ " AND current_version = #{oldMetalakeMeta.currentVersion}"
- + " AND last_version = #{oldMetalakeMeta.lastVersion}"
+ " AND deleted_at = 0";
}
- public String softDeleteMetalakeMetaByMetalakeId(@Param("metalakeId") Long
metalakeId) {
+ public String softDeleteMetalakeMetaByMetalakeId(
+ @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long
currentVersion) {
return "UPDATE "
+ TABLE_NAME
+ " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+ " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
- + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
+ + " WHERE metalake_id = #{metalakeId}"
+ + " AND current_version = #{currentVersion} AND deleted_at = 0";
}
public String deleteMetalakeMetasByLegacyTimeline(
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 822ac3cf25..ee72e38324 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
@@ -38,6 +38,18 @@ public class SchemaMetaBaseSQLProvider {
+ " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
}
+ /** Returns SQL that lists all active schemas in a metalake. */
+ public String listSchemaPOsByMetalakeId(@Param("metalakeId") Long
metalakeId) {
+ return "SELECT schema_id as schemaId, schema_name as schemaName,"
+ + " metalake_id as metalakeId, catalog_id as catalogId,"
+ + " schema_comment as schemaComment, properties, audit_info as
auditInfo,"
+ + " current_version as currentVersion, last_version as lastVersion,"
+ + " deleted_at as deletedAt"
+ + " FROM "
+ + TABLE_NAME
+ + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
+ }
+
public String listSchemaPOsByFullQualifiedName(
@Param("metalakeName") String metalakeName, @Param("catalogName") String
catalogName) {
return """
@@ -299,20 +311,26 @@ public class SchemaMetaBaseSQLProvider {
+ "</script>";
}
- public String softDeleteSchemaMetasByMetalakeId(@Param("metalakeId") Long
metalakeId) {
+ 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 metalake_id = #{metalakeId} AND deleted_at = 0";
+ + " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
}
- public String softDeleteSchemaMetasByCatalogId(@Param("catalogId") Long
catalogId) {
- return "UPDATE "
+ /** Returns SQL that soft-deletes schemas using identifier-and-version
pairs. */
+ public String softDeleteSchemaMetasWithVersion(@Param("schemaMetas")
List<SchemaPO> schemaPOs) {
+ return "<script>"
+ + "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 deleted_at = 0 AND "
+ + "<foreach collection='schemaMetas' item='item' separator=' OR '
open='(' close=')'>"
+ + "(schema_id = #{item.schemaId} AND current_version =
#{item.currentVersion})"
+ + "</foreach>"
+ + "</script>";
}
public String deleteSchemaMetasByLegacyTimeline(
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 0482d9b330..9ff0849817 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
@@ -20,6 +20,7 @@ package
org.apache.gravitino.storage.relational.mapper.provider.postgresql;
import static
org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper.TABLE_NAME;
+import java.util.List;
import
org.apache.gravitino.storage.relational.mapper.provider.base.CatalogMetaBaseSQLProvider;
import org.apache.gravitino.storage.relational.po.CatalogPO;
import org.apache.ibatis.annotations.Param;
@@ -33,12 +34,18 @@ public class CatalogMetaPostgreSQLProvider extends
CatalogMetaBaseSQLProvider {
+ " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
}
+ /** {@inheritDoc} */
@Override
- public String softDeleteCatalogMetasByMetalakeId(Long metalakeId) {
- return "UPDATE "
+ public String softDeleteCatalogMetasWithVersion(List<CatalogPO> catalogPOs) {
+ return "<script>"
+ + "UPDATE "
+ TABLE_NAME
+ " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000
AS BIGINT)"
- + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
+ + " WHERE deleted_at = 0 AND "
+ + "<foreach collection='catalogMetas' item='item' separator=' OR '
open='(' close=')'>"
+ + "(catalog_id = #{item.catalogId} AND current_version =
#{item.currentVersion})"
+ + "</foreach>"
+ + "</script>";
}
@Override
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 5ce01e6715..20a92d1063 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
@@ -26,11 +26,12 @@ import org.apache.ibatis.annotations.Param;
public class MetalakeMetaPostgreSQLProvider extends
MetalakeMetaBaseSQLProvider {
@Override
- public String softDeleteMetalakeMetaByMetalakeId(Long metalakeId) {
+ public String softDeleteMetalakeMetaByMetalakeId(Long metalakeId, Long
currentVersion) {
return "UPDATE "
+ TABLE_NAME
+ " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000
AS BIGINT)"
- + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
+ + " WHERE metalake_id = #{metalakeId}"
+ + " AND current_version = #{currentVersion} AND deleted_at = 0";
}
@Override
@@ -75,15 +76,7 @@ public class MetalakeMetaPostgreSQLProvider extends
MetalakeMetaBaseSQLProvider
+ " current_version = #{newMetalakeMeta.currentVersion},"
+ " last_version = #{newMetalakeMeta.lastVersion}"
+ " WHERE metalake_id = #{oldMetalakeMeta.metalakeId}"
- + " AND metalake_name = #{oldMetalakeMeta.metalakeName}"
- + " AND (metalake_comment = #{oldMetalakeMeta.metalakeComment} "
- + " OR (CAST(metalake_comment AS VARCHAR) IS NULL AND "
- + " CAST(#{oldMetalakeMeta.metalakeComment} AS VARCHAR) IS NULL))"
- + " AND properties = #{oldMetalakeMeta.properties}"
- + " AND audit_info = #{oldMetalakeMeta.auditInfo}"
- + " AND schema_version = #{oldMetalakeMeta.schemaVersion}"
+ " AND current_version = #{oldMetalakeMeta.currentVersion}"
- + " AND last_version = #{oldMetalakeMeta.lastVersion}"
+ " AND deleted_at = 0";
}
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 ba2087aa61..d805e52c0b 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
@@ -126,19 +126,25 @@ public class SchemaMetaPostgreSQLProvider extends
SchemaMetaBaseSQLProvider {
}
@Override
- public String softDeleteSchemaMetasByMetalakeId(Long metalakeId) {
+ public String softDeleteSchemaMetasByCatalogId(Long catalogId) {
return "UPDATE "
+ TABLE_NAME
+ " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000
AS BIGINT)"
- + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
+ + " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
}
+ /** {@inheritDoc} */
@Override
- public String softDeleteSchemaMetasByCatalogId(Long catalogId) {
- return "UPDATE "
+ public String softDeleteSchemaMetasWithVersion(List<SchemaPO> schemaPOs) {
+ return "<script>"
+ + "UPDATE "
+ TABLE_NAME
+ " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000
AS BIGINT)"
- + " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
+ + " WHERE deleted_at = 0 AND "
+ + "<foreach collection='schemaMetas' item='item' separator=' OR '
open='(' close=')'>"
+ + "(schema_id = #{item.schemaId} AND current_version =
#{item.currentVersion})"
+ + "</foreach>"
+ + "</script>";
}
@Override
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 ba810a6130..cc42229f73 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
@@ -25,7 +25,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;
@@ -34,7 +33,6 @@ import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NonEmptyEntityException;
import org.apache.gravitino.meta.BaseMetalake;
-import org.apache.gravitino.meta.CatalogEntity;
import org.apache.gravitino.metrics.Monitored;
import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper;
@@ -64,12 +62,13 @@ import
org.apache.gravitino.storage.relational.mapper.TopicMetaMapper;
import org.apache.gravitino.storage.relational.mapper.UserMetaMapper;
import org.apache.gravitino.storage.relational.mapper.UserRoleRelMapper;
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;
import org.apache.gravitino.utils.NameIdentifierUtil;
-import org.apache.gravitino.utils.NamespaceUtil;
/**
* The service class for metalake metadata. It provides the basic database
operations for metalake.
@@ -175,25 +174,27 @@ public class MetalakeMetaService {
MetalakePO newMetalakePO =
POConverters.updateMetalakePOWithVersion(oldMetalakePO,
newMetalakeEntity);
- AtomicInteger updateResult = new AtomicInteger(0);
try {
SessionUtils.doMultipleWithCommit(
- () ->
- updateResult.set(
- SessionUtils.getWithoutCommit(
- MetalakeMetaMapper.class,
- mapper -> mapper.updateMetalakeMeta(newMetalakePO,
oldMetalakePO))));
+ () -> {
+ int updated =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class,
+ mapper -> mapper.updateMetalakeMeta(newMetalakePO,
oldMetalakePO));
+ if (updated == 0) {
+ // The row may have a new version, or it may have been deleted
or renamed. Re-read it
+ // to return the correct conflict or missing-entity error.
+ throw metalakeWriteFailure(
+ ident, oldMetalakePO.getMetalakeId(),
oldMetalakePO.getMetalakeName());
+ }
+ });
} catch (RuntimeException re) {
ExceptionUtils.checkSQLException(
re, Entity.EntityType.METALAKE,
newMetalakeEntity.nameIdentifier().toString());
throw re;
}
- if (updateResult.get() > 0) {
- return newMetalakeEntity;
- } else {
- throw new IOException("Failed to update the entity: " + ident);
- }
+ return newMetalakeEntity;
}
@Monitored(
@@ -201,22 +202,27 @@ public class MetalakeMetaService {
baseMetricName = "deleteMetalake")
public boolean deleteMetalake(NameIdentifier ident, boolean cascade) {
NameIdentifierUtil.checkMetalake(ident);
- Long metalakeId = getMetalakeIdByName(ident.name());
+ MetalakePO metalakePO =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByName(ident.name()));
+ if (metalakePO == null) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.METALAKE.name().toLowerCase(),
+ ident.toString());
+ }
+ Long metalakeId = metalakePO.getMetalakeId();
+ Long currentVersion = metalakePO.getCurrentVersion();
if (metalakeId != null) {
if (cascade) {
SessionUtils.doMultipleWithCommit(
- () ->
- SessionUtils.doWithoutCommit(
- MetalakeMetaMapper.class,
- mapper ->
mapper.softDeleteMetalakeMetaByMetalakeId(metalakeId)),
- () ->
- SessionUtils.doWithoutCommit(
- CatalogMetaMapper.class,
- mapper ->
mapper.softDeleteCatalogMetasByMetalakeId(metalakeId)),
- () ->
- SessionUtils.doWithoutCommit(
- SchemaMetaMapper.class,
- mapper ->
mapper.softDeleteSchemaMetasByMetalakeId(metalakeId)),
+ () -> {
+ // Take the parent lock before the child snapshot, so catalog
creation cannot add a
+ // child after the snapshot. A later failure rolls back this
soft delete as well.
+ deleteMetalakeWithVersion(ident, metalakeId, currentVersion);
+ deleteCatalogsWithVersions(ident, metalakeId);
+ deleteSchemasWithVersions(ident,
listSchemaPOsForCascade(metalakeId));
+ },
() ->
SessionUtils.doWithoutCommit(
TableMetaMapper.class,
@@ -318,18 +324,24 @@ public class MetalakeMetaService {
ViewMetaMapper.class,
mapper ->
mapper.softDeleteViewMetasByMetalakeId(metalakeId)));
} else {
- List<CatalogEntity> catalogEntities =
- CatalogMetaService.getInstance()
-
.listCatalogsByNamespace(NamespaceUtil.ofCatalog(ident.name()));
- if (!catalogEntities.isEmpty()) {
- throw new NonEmptyEntityException(
- "Entity %s has sub-entities, you should remove sub-entities
first", ident);
- }
SessionUtils.doMultipleWithCommit(
- () ->
- SessionUtils.doWithoutCommit(
- MetalakeMetaMapper.class,
- mapper ->
mapper.softDeleteMetalakeMetaByMetalakeId(metalakeId)),
+ () -> {
+ // Delete the metalake before checking its children. The UPDATE
takes an exclusive
+ // lock on the metalake row, and catalog creation locks the same
row before inserting.
+ // If creation gets its lock first, this delete waits and the
check below sees the new
+ // catalog. If this delete gets the lock first, creation waits
until this transaction
+ // commits or rolls back. Checking first would allow a catalog
to be inserted between
+ // the check and this delete. A non-empty result rolls back the
soft delete.
+ deleteMetalakeWithVersion(ident, metalakeId, currentVersion);
+ List<CatalogPO> catalogPOs =
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class,
+ mapper -> mapper.listCatalogPOsByMetalakeId(metalakeId));
+ if (!catalogPOs.isEmpty()) {
+ throw new NonEmptyEntityException(
+ "Entity %s has sub-entities, you should remove
sub-entities first", ident);
+ }
+ },
() ->
SessionUtils.doWithoutCommit(
UserRoleRelMapper.class,
@@ -383,6 +395,81 @@ public class MetalakeMetaService {
return true;
}
+ void deleteMetalakeWithVersion(NameIdentifier identifier, Long metalakeId,
Long currentVersion) {
+ int deleted =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class,
+ mapper -> mapper.softDeleteMetalakeMetaByMetalakeId(metalakeId,
currentVersion));
+ if (deleted == 0) {
+ throw metalakeWriteFailure(identifier, metalakeId, identifier.name());
+ }
+ }
+
+ 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.
+ MetalakePO currentMetalakePO =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByIdForUpdate(metalakeId));
+ if (currentMetalakePO == null
+ || !Objects.equals(currentMetalakePO.getMetalakeName(), observedName))
{
+ return new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.METALAKE.name().toLowerCase(),
+ identifier.name());
+ }
+ return ExceptionUtils.concurrentModification(Entity.EntityType.METALAKE,
identifier);
+ }
+
+ private void deleteCatalogsWithVersions(NameIdentifier metalakeIdentifier,
Long metalakeId) {
+ // Lock all catalog rows before taking the schema snapshot. Schema
creation and deletion lock
+ // their parent catalog, so they cannot add or remove a schema after this
point. A schema alter
+ // can still run, but the version check below detects it. Taking parent
locks before child rows
+ // also keeps the same lock order for every metalake cascade.
+ List<CatalogPO> catalogPOs =
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class,
+ mapper -> mapper.listCatalogPOsByMetalakeIdForUpdate(metalakeId));
+ if (catalogPOs.isEmpty()) {
+ return;
+ }
+ int deleted =
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class,
+ mapper -> mapper.softDeleteCatalogMetasWithVersion(catalogPOs));
+ // Never commit a partial cascade. A smaller count means that a catalog no
longer matches the
+ // ID and version read above, so the outer transaction must roll back all
deletes.
+ if (deleted != catalogPOs.size()) {
+ throw ExceptionUtils.concurrentChildModification(
+ Entity.EntityType.CATALOG, Entity.EntityType.METALAKE,
metalakeIdentifier);
+ }
+ }
+
+ List<SchemaPO> listSchemaPOsForCascade(Long metalakeId) {
+ return SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class, mapper ->
mapper.listSchemaPOsByMetalakeId(metalakeId));
+ }
+
+ private void deleteSchemasWithVersions(
+ NameIdentifier metalakeIdentifier, List<SchemaPO> schemaPOs) {
+ if (schemaPOs.isEmpty()) {
+ return;
+ }
+ int deleted =
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class, mapper ->
mapper.softDeleteSchemaMetasWithVersion(schemaPOs));
+ // The version check protects this snapshot from a schema alter that does
not use the parent
+ // catalog lock. Roll back the whole cascade instead of silently losing
that schema change.
+ if (deleted != schemaPOs.size()) {
+ throw ExceptionUtils.concurrentChildModification(
+ Entity.EntityType.SCHEMA, Entity.EntityType.METALAKE,
metalakeIdentifier);
+ }
+ }
+
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "deleteMetalakeMetasByLegacyTimeline")
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/utils/ExceptionUtils.java
b/core/src/main/java/org/apache/gravitino/storage/relational/utils/ExceptionUtils.java
index eb08cfd2e8..7da3008500 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/utils/ExceptionUtils.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/utils/ExceptionUtils.java
@@ -20,7 +20,10 @@ package org.apache.gravitino.storage.relational.utils;
import java.io.IOException;
import java.sql.SQLException;
+import java.util.Locale;
import org.apache.gravitino.Entity;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import
org.apache.gravitino.storage.relational.converters.SQLExceptionConverterFactory;
public class ExceptionUtils {
@@ -33,4 +36,37 @@ public class ExceptionUtils {
.toGravitinoException((SQLException) re.getCause(), type,
entityName);
}
}
+
+ /**
+ * Creates an {@link OptimisticLockException} for an entity that was
modified concurrently by
+ * another writer, which makes the version-guarded write fail.
+ *
+ * @param type The type of the entity that was modified concurrently.
+ * @param identifier The identifier of the entity that was modified
concurrently.
+ * @return The {@link OptimisticLockException} to throw.
+ */
+ public static OptimisticLockException concurrentModification(
+ Entity.EntityType type, NameIdentifier identifier) {
+ return new OptimisticLockException(
+ "The %s %s was modified concurrently; retry the operation",
+ type.name().toLowerCase(Locale.ROOT), identifier);
+ }
+
+ /**
+ * Creates an {@link OptimisticLockException} for a child entity that was
modified concurrently
+ * while its parent was being deleted in cascade mode.
+ *
+ * @param childType The type of the child entity that was modified
concurrently.
+ * @param parentType The type of the parent entity being operated on.
+ * @param parentIdentifier The identifier of the parent entity being
operated on.
+ * @return The {@link OptimisticLockException} to throw.
+ */
+ public static OptimisticLockException concurrentChildModification(
+ Entity.EntityType childType, Entity.EntityType parentType,
NameIdentifier parentIdentifier) {
+ return new OptimisticLockException(
+ "A %s under %s %s was modified concurrently; retry the operation",
+ childType.name().toLowerCase(Locale.ROOT),
+ parentType.name().toLowerCase(Locale.ROOT),
+ parentIdentifier);
+ }
}
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 d019800d1a..2a05888fd7 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
@@ -137,9 +137,9 @@ public class POConverters {
*/
public static MetalakePO updateMetalakePOWithVersion(
MetalakePO oldMetalakePO, BaseMetalake newMetalake) {
- Long lastVersion = oldMetalakePO.getLastVersion();
- // Will set the version to the last version + 1 when having some fields
need be multiple version
- Long nextVersion = lastVersion;
+ // Every metadata update advances the OCC token. Both version columns stay
aligned because
+ // metalakes do not retain independently addressable historical versions.
+ Long nextVersion = oldMetalakePO.getCurrentVersion() + 1;
try {
return MetalakePO.builder()
.withMetalakeId(newMetalake.id())
diff --git
a/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java
b/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java
index c5be0774ea..55084847f3 100644
--- a/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java
+++ b/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java
@@ -31,6 +31,7 @@ import java.util.Map;
import java.util.Set;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.Config;
+import org.apache.gravitino.Entity.EntityType;
import org.apache.gravitino.EntityStore;
import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.MetalakeChange;
@@ -39,11 +40,13 @@ import org.apache.gravitino.StringIdentifier;
import org.apache.gravitino.UserPrincipal;
import org.apache.gravitino.auth.AuthConstants;
import org.apache.gravitino.exceptions.MetalakeAlreadyExistsException;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchMetalakeException;
import org.apache.gravitino.lock.LockManager;
import org.apache.gravitino.meta.BaseMetalake;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.memory.TestMemoryEntityStore;
+import
org.apache.gravitino.storage.memory.TestMemoryEntityStore.InMemoryEntityStore;
import org.apache.gravitino.utils.PrincipalUtils;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
@@ -210,6 +213,23 @@ public class TestMetalakeManager {
Assertions.assertFalse(dropped1, "metalake should be non-existent");
}
+ @Test
+ public void testDropMetalakeReturnsFalseWhenConcurrentDeleteWins() throws
IOException {
+ InMemoryEntityStore store = Mockito.spy(new InMemoryEntityStore());
+ store.initialize(config);
+ MetalakeManager manager = new MetalakeManager(store, new
RandomIdGenerator());
+ NameIdentifier ident = NameIdentifier.of("concurrently_deleted_metalake");
+ manager.createMetalake(ident, "comment", ImmutableMap.of());
+ Mockito.doThrow(
+ new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, "metalake",
ident.toString()))
+ .when(store)
+ .delete(ident, EntityType.METALAKE, true);
+
+ Assertions.assertFalse(manager.dropMetalake(ident, true));
+ store.close();
+ }
+
@Test
public void testListInUseMetalakes() {
// Create some metalakes with different in-use status
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestMetalakeMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestMetalakeMetaService.java
index b1e6389a20..7948c75a9c 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestMetalakeMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestMetalakeMetaService.java
@@ -27,10 +27,17 @@ import java.time.Instant;
import java.util.List;
import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityAlreadyExistsException;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.meta.BaseMetalake;
import org.apache.gravitino.meta.SchemaVersion;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
+import org.apache.gravitino.storage.relational.po.MetalakePO;
+import org.apache.gravitino.storage.relational.utils.POConverters;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.TestTemplate;
@@ -92,6 +99,172 @@ public class TestMetalakeMetaService extends
TestJDBCBackend {
backend.delete(metalake.nameIdentifier(), Entity.EntityType.METALAKE,
false);
}
+ @TestTemplate
+ public void testAlterAndDeleteUseCurrentVersion() throws IOException {
+ BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME);
+ MetalakePO oldPO =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByName(metalake.name()));
+ BaseMetalake updatedMetalake =
+ BaseMetalake.builder()
+ .withId(metalake.id())
+ .withName(metalake.name())
+ .withAuditInfo(metalake.auditInfo())
+ .withComment("updated")
+ .withProperties(metalake.properties())
+ .withVersion(metalake.getVersion())
+ .build();
+ MetalakePO newPO = POConverters.updateMetalakePOWithVersion(oldPO,
updatedMetalake);
+
+ int updated =
+ SessionUtils.doWithCommitAndFetchResult(
+ MetalakeMetaMapper.class, mapper ->
mapper.updateMetalakeMeta(newPO, oldPO));
+ int staleUpdate =
+ SessionUtils.doWithCommitAndFetchResult(
+ MetalakeMetaMapper.class, mapper ->
mapper.updateMetalakeMeta(newPO, oldPO));
+ int staleDelete =
+ SessionUtils.doWithCommitAndFetchResult(
+ MetalakeMetaMapper.class,
+ mapper ->
+ mapper.softDeleteMetalakeMetaByMetalakeId(
+ metalake.id(), oldPO.getCurrentVersion()));
+ Assertions.assertEquals(1, updated);
+ Assertions.assertEquals(0, staleUpdate);
+ Assertions.assertEquals(0, staleDelete);
+ assertTrue(backend.exists(metalake.nameIdentifier(),
Entity.EntityType.METALAKE));
+ int deleted =
+ SessionUtils.doWithCommitAndFetchResult(
+ MetalakeMetaMapper.class,
+ mapper ->
+ mapper.softDeleteMetalakeMetaByMetalakeId(
+ metalake.id(), newPO.getCurrentVersion()));
+ Assertions.assertEquals(1, deleted);
+ }
+
+ @TestTemplate
+ public void testAlterReportsOptimisticLockConflict() throws IOException {
+ BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME);
+
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+ MetalakeMetaService.getInstance()
+ .updateMetalake(
+ metalake.nameIdentifier(),
+ entity -> {
+ BaseMetalake current = (BaseMetalake) entity;
+ MetalakePO currentPO =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class,
+ mapper ->
mapper.selectMetalakeMetaByName(current.name()));
+ BaseMetalake competingUpdate =
+ BaseMetalake.builder()
+ .withId(current.id())
+ .withName(current.name())
+ .withAuditInfo(current.auditInfo())
+ .withComment("competing update")
+ .withProperties(current.properties())
+ .withVersion(current.getVersion())
+ .build();
+ MetalakePO competingPO =
+ POConverters.updateMetalakePOWithVersion(currentPO,
competingUpdate);
+ SessionUtils.doWithCommitAndFetchResult(
+ MetalakeMetaMapper.class,
+ mapper -> mapper.updateMetalakeMeta(competingPO,
currentPO));
+ return BaseMetalake.builder()
+ .withId(current.id())
+ .withName(current.name())
+ .withAuditInfo(current.auditInfo())
+ .withComment("requested update")
+ .withProperties(current.properties())
+ .withVersion(current.getVersion())
+ .build();
+ }));
+ }
+
+ @TestTemplate
+ public void testAlterReportsNoSuchWhenMetalakeIsDeletedConcurrently() throws
IOException {
+ BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME);
+
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ MetalakeMetaService.getInstance()
+ .updateMetalake(
+ metalake.nameIdentifier(),
+ entity -> {
+ BaseMetalake current = (BaseMetalake) entity;
+ MetalakePO currentPO =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class,
+ mapper ->
mapper.selectMetalakeMetaById(current.id()));
+ SessionUtils.doWithCommitAndFetchResult(
+ MetalakeMetaMapper.class,
+ mapper ->
+ mapper.softDeleteMetalakeMetaByMetalakeId(
+ current.id(),
currentPO.getCurrentVersion()));
+ return BaseMetalake.builder()
+ .withId(current.id())
+ .withName(current.name())
+ .withAuditInfo(current.auditInfo())
+ .withComment("requested update")
+ .withProperties(current.properties())
+ .withVersion(current.getVersion())
+ .build();
+ }));
+ }
+
+ @TestTemplate
+ public void testDeleteReportsOptimisticLockConflict() throws IOException {
+ BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME);
+ MetalakePO stalePO =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByName(metalake.name()));
+ BaseMetalake competingUpdate =
+ BaseMetalake.builder()
+ .withId(metalake.id())
+ .withName(metalake.name())
+ .withAuditInfo(metalake.auditInfo())
+ .withComment("competing update")
+ .withProperties(metalake.properties())
+ .withVersion(metalake.getVersion())
+ .build();
+ MetalakePO competingPO = POConverters.updateMetalakePOWithVersion(stalePO,
competingUpdate);
+ SessionUtils.doWithCommitAndFetchResult(
+ MetalakeMetaMapper.class, mapper ->
mapper.updateMetalakeMeta(competingPO, stalePO));
+
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ MetalakeMetaService.getInstance()
+ .deleteMetalakeWithVersion(
+ metalake.nameIdentifier(),
+ metalake.id(),
+ stalePO.getCurrentVersion())));
+ assertTrue(backend.exists(metalake.nameIdentifier(),
Entity.EntityType.METALAKE));
+ }
+
+ @TestTemplate
+ public void testNonCascadeDeleteRollsBackMetalakeFence() throws IOException {
+ BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME);
+ createAndInsertCatalog(METALAKE_NAME, "catalog");
+ MetalakePO beforeDelete =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByName(metalake.name()));
+
+ assertThrows(
+ NonEmptyEntityException.class,
+ () ->
MetalakeMetaService.getInstance().deleteMetalake(metalake.nameIdentifier(),
false));
+
+ MetalakePO afterDelete =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByName(metalake.name()));
+ Assertions.assertEquals(beforeDelete.getCurrentVersion(),
afterDelete.getCurrentVersion());
+ assertTrue(backend.exists(metalake.nameIdentifier(),
Entity.EntityType.METALAKE));
+ }
+
@TestTemplate
public void testMetaLifeCycleFromCreationToDeletion() throws IOException {
// meta data creation
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestExceptionUtils.java
b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestExceptionUtils.java
new file mode 100644
index 0000000000..0d489be688
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestExceptionUtils.java
@@ -0,0 +1,50 @@
+/*
+ * 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.utils;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.junit.jupiter.api.Test;
+
+public class TestExceptionUtils {
+
+ @Test
+ public void testConcurrentModificationMessage() {
+ OptimisticLockException e =
+ ExceptionUtils.concurrentModification(
+ Entity.EntityType.CATALOG, NameIdentifier.of("m1", "c1"));
+
+ assertEquals(
+ "The catalog m1.c1 was modified concurrently; retry the operation",
e.getMessage());
+ }
+
+ @Test
+ public void testConcurrentChildModificationMessage() {
+ OptimisticLockException e =
+ ExceptionUtils.concurrentChildModification(
+ Entity.EntityType.SCHEMA, Entity.EntityType.CATALOG,
NameIdentifier.of("m1", "c1"));
+
+ assertEquals(
+ "A schema under catalog m1.c1 was modified concurrently; retry the
operation",
+ e.getMessage());
+ }
+}
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 9ce171d878..22a5389506 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
@@ -665,6 +665,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.getMetalakeComment());
}
