This is an automated email from the ASF dual-hosted git repository.
yuqi1129 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 a7b8096318 [#12652] improvement(core): add OCC for function writes
(#12823)
a7b8096318 is described below
commit a7b809631877fc17d811ee28d846087562f37e16
Author: Qi Yu <[email protected]>
AuthorDate: Tue Sep 8 18:06:17 2026 +0800
[#12652] improvement(core): add OCC for function writes (#12823)
### What changes were proposed in this pull request?
- Add version-CAS updates and deletes for managed functions.
- Execute root metadata CAS before writing function versions and
dependent rows.
- Classify stale-version conflicts and identity changes through
stable-ID locking reads.
- Use a dedicated root-only locking query for overwrite while keeping
normal reads on the strict current-version INNER JOIN.
- Preserve stored function IDs and monotonic versions for natural-key
overwrite on H2, MySQL, and PostgreSQL.
- Fence parent schemas during create and the correct target ancestry
during cross-catalog moves.
- Add real two-transaction coverage for parent deletion,
overwrite/rename, and target-schema deletion races.
### Why are the changes needed?
Concurrent writers could create competing function-version rows, stale
deletes could remove relationships belonging to newer functions, and
overwrite or move operations could violate identity, version, or parent
invariants.
Fix: #12652
### Does this PR introduce _any_ user-facing change?
No API or configuration changes. Concurrent function writes now report
optimistic-lock or not-found errors consistently.
### How was this patch tested?
- `./gradlew :core:spotlessApply :core:compileTestJava :core:javadoc`
- Ran `TestFunctionMetaService`, `TestFunctionMetaBaseSQLProvider`, and
`TestFunctionMetaPostgreSQLProvider`.
- Service tests ran against H2, MySQL, and PostgreSQL.
---------
Co-authored-by: Jerry Shao <[email protected]>
---
.../relational/mapper/FunctionMetaMapper.java | 52 +-
.../mapper/FunctionMetaSQLProviderFactory.java | 33 +-
.../mapper/FunctionVersionMetaMapper.java | 6 -
.../FunctionVersionMetaSQLProviderFactory.java | 5 -
.../provider/base/FunctionMetaBaseSQLProvider.java | 93 ++-
.../base/FunctionVersionMetaBaseSQLProvider.java | 33 +-
.../postgresql/FunctionMetaPostgreSQLProvider.java | 38 +-
.../FunctionVersionMetaPostgreSQLProvider.java | 33 +-
.../relational/service/FunctionMetaService.java | 302 ++++++--
.../relational/service/FunctionPOStorageOps.java | 17 +-
.../base/TestFunctionMetaBaseSQLProvider.java | 74 ++
.../TestFunctionMetaPostgreSQLProvider.java | 43 ++
.../service/TestFunctionMetaService.java | 790 ++++++++++++++++++++-
13 files changed, 1295 insertions(+), 224 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionMetaMapper.java
index 100c2683c3..92f3cea359 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionMetaMapper.java
@@ -57,11 +57,6 @@ public interface FunctionMetaMapper {
@InsertProvider(type = FunctionMetaSQLProviderFactory.class, method =
"insertFunctionMeta")
void insertFunctionMeta(@Param("functionMeta") FunctionPO functionPO);
- @InsertProvider(
- type = FunctionMetaSQLProviderFactory.class,
- method = "insertFunctionMetaOnDuplicateKeyUpdate")
- void insertFunctionMetaOnDuplicateKeyUpdate(@Param("functionMeta")
FunctionPO functionPO);
-
@Results({
@Result(property = "functionId", column = "function_id", id = true),
@Result(property = "functionName", column = "function_name"),
@@ -142,16 +137,61 @@ public interface FunctionMetaMapper {
FunctionPO selectFunctionMetaBySchemaIdAndName(
@Param("schemaId") Long schemaId, @Param("functionName") String
functionName);
+ /**
+ * Selects and exclusively locks an active function by its natural key
without joining a version
+ * row.
+ *
+ * @param schemaId the schema ID
+ * @param functionName the function name
+ * @return the locked function root, or {@code null} when it does not exist
+ */
+ @SelectProvider(
+ type = FunctionMetaSQLProviderFactory.class,
+ method = "selectFunctionMetaBySchemaIdAndNameForUpdate")
+ FunctionPO selectFunctionMetaBySchemaIdAndNameForUpdate(
+ @Param("schemaId") Long schemaId, @Param("functionName") String
functionName);
+
+ /**
+ * Selects and exclusively locks an active function metadata row.
+ *
+ * @param functionId the function ID
+ * @return the active function metadata, or {@code null} when it no longer
exists
+ */
+ @SelectProvider(
+ type = FunctionMetaSQLProviderFactory.class,
+ method = "selectFunctionMetaByIdForUpdate")
+ FunctionPO selectFunctionMetaByIdForUpdate(@Param("functionId") Long
functionId);
+
+ /**
+ * Checks whether a soft-deleted function still owns the requested primary
key.
+ *
+ * @param functionId the function ID
+ * @return one if a deleted row reserves the ID, otherwise zero
+ */
+ @Select(
+ "SELECT COUNT(*) FROM "
+ + TABLE_NAME
+ + " WHERE function_id = #{functionId} AND deleted_at > 0")
+ int countDeletedFunctionMetasById(@Param("functionId") Long functionId);
+
@SelectProvider(
type = FunctionMetaSQLProviderFactory.class,
method = "selectFunctionIdBySchemaIdAndFunctionName")
Long selectFunctionIdBySchemaIdAndFunctionName(
@Param("schemaId") Long schemaId, @Param("functionName") String
functionName);
+ /**
+ * Soft-deletes a function only if its version has not changed since the
caller read it.
+ *
+ * @param functionId the function ID
+ * @param currentVersion the version observed by the caller
+ * @return the number of deleted rows; zero means the function changed or
disappeared
+ */
@UpdateProvider(
type = FunctionMetaSQLProviderFactory.class,
method = "softDeleteFunctionMetaByFunctionId")
- Integer softDeleteFunctionMetaByFunctionId(@Param("functionId") Long
functionId);
+ Integer softDeleteFunctionMetaByFunctionId(
+ @Param("functionId") Long functionId, @Param("currentVersion") Integer
currentVersion);
@UpdateProvider(
type = FunctionMetaSQLProviderFactory.class,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionMetaSQLProviderFactory.java
index 2e2ad59129..421ca32cb2 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionMetaSQLProviderFactory.java
@@ -56,11 +56,6 @@ public class FunctionMetaSQLProviderFactory {
return getProvider().insertFunctionMeta(functionPO);
}
- public static String insertFunctionMetaOnDuplicateKeyUpdate(
- @Param("functionMeta") FunctionPO functionPO) {
- return getProvider().insertFunctionMetaOnDuplicateKeyUpdate(functionPO);
- }
-
public static String listFunctionPOsBySchemaId(@Param("schemaId") Long
schemaId) {
return getProvider().listFunctionPOsBySchemaId(schemaId);
}
@@ -90,13 +85,37 @@ public class FunctionMetaSQLProviderFactory {
return getProvider().selectFunctionMetaBySchemaIdAndName(schemaId,
functionName);
}
+ /** Delegates a root-only locking lookup by function natural key. */
+ public static String selectFunctionMetaBySchemaIdAndNameForUpdate(
+ @Param("schemaId") Long schemaId, @Param("functionName") String
functionName) {
+ return
getProvider().selectFunctionMetaBySchemaIdAndNameForUpdate(schemaId,
functionName);
+ }
+
+ /**
+ * Returns SQL that selects and exclusively locks an active function
metadata row.
+ *
+ * @param functionId the function ID
+ * @return the locking select SQL
+ */
+ public static String selectFunctionMetaByIdForUpdate(@Param("functionId")
Long functionId) {
+ return getProvider().selectFunctionMetaByIdForUpdate(functionId);
+ }
+
public static String selectFunctionIdBySchemaIdAndFunctionName(
@Param("schemaId") Long schemaId, @Param("functionName") String
functionName) {
return getProvider().selectFunctionIdBySchemaIdAndFunctionName(schemaId,
functionName);
}
- public static String softDeleteFunctionMetaByFunctionId(@Param("functionId")
Long functionId) {
- return getProvider().softDeleteFunctionMetaByFunctionId(functionId);
+ /**
+ * Returns SQL that soft-deletes a function with a version check.
+ *
+ * @param functionId the function ID
+ * @param currentVersion the version observed by the caller
+ * @return the version-checked delete SQL
+ */
+ public static String softDeleteFunctionMetaByFunctionId(
+ @Param("functionId") Long functionId, @Param("currentVersion") Integer
currentVersion) {
+ return getProvider().softDeleteFunctionMetaByFunctionId(functionId,
currentVersion);
}
public static String softDeleteFunctionMetasByCatalogId(@Param("catalogId")
Long catalogId) {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionVersionMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionVersionMetaMapper.java
index 9a40934ca0..7b6dde6c09 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionVersionMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionVersionMetaMapper.java
@@ -37,12 +37,6 @@ public interface FunctionVersionMetaMapper {
method = "insertFunctionVersionMeta")
void insertFunctionVersionMeta(@Param("functionVersionMeta")
FunctionVersionPO functionVersionPO);
- @InsertProvider(
- type = FunctionVersionMetaSQLProviderFactory.class,
- method = "insertFunctionVersionMetaOnDuplicateKeyUpdate")
- void insertFunctionVersionMetaOnDuplicateKeyUpdate(
- @Param("functionVersionMeta") FunctionVersionPO functionVersionPO);
-
@UpdateProvider(
type = FunctionVersionMetaSQLProviderFactory.class,
method = "softDeleteFunctionVersionMetasBySchemaIds")
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionVersionMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionVersionMetaSQLProviderFactory.java
index a5f846798b..db8e917dfc 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionVersionMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/FunctionVersionMetaSQLProviderFactory.java
@@ -57,11 +57,6 @@ public class FunctionVersionMetaSQLProviderFactory {
return getProvider().insertFunctionVersionMeta(functionVersionPO);
}
- public static String insertFunctionVersionMetaOnDuplicateKeyUpdate(
- @Param("functionVersionMeta") FunctionVersionPO functionVersionPO) {
- return
getProvider().insertFunctionVersionMetaOnDuplicateKeyUpdate(functionVersionPO);
- }
-
public static String softDeleteFunctionVersionMetasBySchemaIds(
@Param("schemaIds") List<Long> schemaIds) {
return getProvider().softDeleteFunctionVersionMetasBySchemaIds(schemaIds);
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FunctionMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FunctionMetaBaseSQLProvider.java
index 4d01ddd63d..b7b4b7ad51 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FunctionMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FunctionMetaBaseSQLProvider.java
@@ -101,32 +101,6 @@ public class FunctionMetaBaseSQLProvider {
+ " #{functionMeta.deletedAt})";
}
- public String insertFunctionMetaOnDuplicateKeyUpdate(
- @Param("functionMeta") FunctionPO functionPO) {
- return "INSERT INTO "
- + TABLE_NAME
- + " (function_id, function_name, metalake_id, catalog_id, schema_id,"
- + " function_type, `deterministic`,"
- + " function_current_version, function_latest_version, audit_info,
deleted_at)"
- + " VALUES (#{functionMeta.functionId}, #{functionMeta.functionName},"
- + " #{functionMeta.metalakeId}, #{functionMeta.catalogId},
#{functionMeta.schemaId},"
- + " #{functionMeta.functionType}, #{functionMeta.deterministic},"
- + " #{functionMeta.functionCurrentVersion},"
- + " #{functionMeta.functionLatestVersion}, #{functionMeta.auditInfo},"
- + " #{functionMeta.deletedAt})"
- + " ON DUPLICATE KEY UPDATE"
- + " function_name = #{functionMeta.functionName},"
- + " metalake_id = #{functionMeta.metalakeId},"
- + " catalog_id = #{functionMeta.catalogId},"
- + " schema_id = #{functionMeta.schemaId},"
- + " function_type = #{functionMeta.functionType},"
- + " `deterministic` = #{functionMeta.deterministic},"
- + " function_current_version = #{functionMeta.functionCurrentVersion},"
- + " function_latest_version = #{functionMeta.functionLatestVersion},"
- + " audit_info = #{functionMeta.auditInfo},"
- + " deleted_at = #{functionMeta.deletedAt}";
- }
-
public String selectFunctionMetaByFullQualifiedName(
@Param("metalakeName") String metalakeName,
@Param("catalogName") String catalogName,
@@ -202,6 +176,26 @@ public class FunctionMetaBaseSQLProvider {
+ " WHERE fm.schema_id = #{schemaId} AND fm.deleted_at = 0 AND
vi.deleted_at = 0";
}
+ /**
+ * Returns the active function metadata row and holds it exclusively for the
transaction.
+ *
+ * <p>The version table is deliberately not joined: PostgreSQL rejects
locking the nullable side
+ * of an outer join, and conflict classification only needs the root row's
identity and version.
+ *
+ * @param functionId the function ID
+ * @return the locking select SQL
+ */
+ public String selectFunctionMetaByIdForUpdate(@Param("functionId") Long
functionId) {
+ return "SELECT function_id as functionId, function_name as functionName,"
+ + " metalake_id as metalakeId, catalog_id as catalogId, schema_id as
schemaId,"
+ + " function_current_version as functionCurrentVersion,"
+ + " function_latest_version as functionLatestVersion,"
+ + " audit_info as auditInfo, deleted_at as deletedAt"
+ + " FROM "
+ + TABLE_NAME
+ + " WHERE function_id = #{functionId} AND deleted_at = 0 FOR UPDATE";
+ }
+
public String listFunctionPOsByFunctionIds(@Param("functionIds") List<Long>
functionIds) {
return "<script>"
+ " SELECT function_id, function_name, schema_id"
@@ -235,6 +229,26 @@ public class FunctionMetaBaseSQLProvider {
+ " AND fm.deleted_at = 0 AND vi.deleted_at = 0";
}
+ /**
+ * Returns SQL that locks an active function by natural key without joining
its version row.
+ *
+ * <p>This query is reserved for overwrite decisions. Normal reads keep
using the inner-joined
+ * query above so a broken current-version invariant is reported as missing
instead of producing a
+ * partially populated {@code FunctionPO}.
+ */
+ public String selectFunctionMetaBySchemaIdAndNameForUpdate(
+ @Param("schemaId") Long schemaId, @Param("functionName") String
functionName) {
+ return "SELECT function_id as functionId, function_name as functionName,"
+ + " metalake_id as metalakeId, catalog_id as catalogId, schema_id as
schemaId,"
+ + " function_current_version as functionCurrentVersion,"
+ + " function_latest_version as functionLatestVersion,"
+ + " deleted_at as deletedAt"
+ + " FROM "
+ + TABLE_NAME
+ + " WHERE schema_id = #{schemaId} AND function_name = #{functionName}"
+ + " AND deleted_at = 0 FOR UPDATE";
+ }
+
public String selectFunctionIdBySchemaIdAndFunctionName(
@Param("schemaId") Long schemaId, @Param("functionName") String
functionName) {
return "SELECT function_id"
@@ -243,12 +257,21 @@ public class FunctionMetaBaseSQLProvider {
+ " WHERE schema_id = #{schemaId} AND function_name = #{functionName}
AND deleted_at = 0";
}
- public String softDeleteFunctionMetaByFunctionId(@Param("functionId") Long
functionId) {
+ /**
+ * Returns SQL that deletes only the function version observed by the caller.
+ *
+ * @param functionId the function ID
+ * @param currentVersion the version observed by the caller
+ * @return the version-checked delete SQL
+ */
+ public String softDeleteFunctionMetaByFunctionId(
+ @Param("functionId") Long functionId, @Param("currentVersion") Integer
currentVersion) {
return "UPDATE "
+ TABLE_NAME
+ " SET deleted_at = "
+ DatabaseTimeSQL.MYSQL
- + " WHERE function_id = #{functionId} AND deleted_at = 0";
+ + " WHERE function_id = #{functionId}"
+ + " AND function_current_version = #{currentVersion} AND deleted_at =
0";
}
public String softDeleteFunctionMetasByCatalogId(@Param("catalogId") Long
catalogId) {
@@ -288,6 +311,13 @@ public class FunctionMetaBaseSQLProvider {
+ " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT
#{limit}";
}
+ /**
+ * Returns SQL that updates a function only while its OCC version is
unchanged.
+ *
+ * @param newFunctionPO the function values to write
+ * @param oldFunctionPO the function values and OCC version read by the
caller
+ * @return the version-checked update SQL
+ */
public String updateFunctionMeta(
@Param("newFunctionMeta") FunctionPO newFunctionPO,
@Param("oldFunctionMeta") FunctionPO oldFunctionPO) {
@@ -304,14 +334,7 @@ public class FunctionMetaBaseSQLProvider {
+ " audit_info = #{newFunctionMeta.auditInfo},"
+ " deleted_at = #{newFunctionMeta.deletedAt}"
+ " WHERE function_id = #{oldFunctionMeta.functionId}"
- + " AND function_name = #{oldFunctionMeta.functionName}"
- + " AND metalake_id = #{oldFunctionMeta.metalakeId}"
- + " AND catalog_id = #{oldFunctionMeta.catalogId}"
- + " AND schema_id = #{oldFunctionMeta.schemaId}"
- + " AND function_type = #{oldFunctionMeta.functionType}"
+ " AND function_current_version =
#{oldFunctionMeta.functionCurrentVersion}"
- + " AND function_latest_version =
#{oldFunctionMeta.functionLatestVersion}"
- + " AND audit_info = #{oldFunctionMeta.auditInfo}"
+ " AND deleted_at = 0";
}
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FunctionVersionMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FunctionVersionMetaBaseSQLProvider.java
index 961c8f1ef3..aee3a42bb7 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FunctionVersionMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FunctionVersionMetaBaseSQLProvider.java
@@ -19,6 +19,7 @@
package org.apache.gravitino.storage.relational.mapper.provider.base;
import java.util.List;
+import org.apache.gravitino.storage.relational.mapper.FunctionMetaMapper;
import
org.apache.gravitino.storage.relational.mapper.FunctionVersionMetaMapper;
import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL;
import org.apache.gravitino.storage.relational.po.FunctionVersionPO;
@@ -39,24 +40,6 @@ public class FunctionVersionMetaBaseSQLProvider {
+ " #{functionVersionMeta.deletedAt})";
}
- public String insertFunctionVersionMetaOnDuplicateKeyUpdate(
- @Param("functionVersionMeta") FunctionVersionPO functionVersionPO) {
- return "INSERT INTO "
- + FunctionVersionMetaMapper.TABLE_NAME
- + " (metalake_id, catalog_id, schema_id, function_id, version,"
- + " function_comment, definitions, audit_info, deleted_at)"
- + " VALUES (#{functionVersionMeta.metalakeId},
#{functionVersionMeta.catalogId},"
- + " #{functionVersionMeta.schemaId},
#{functionVersionMeta.functionId},"
- + " #{functionVersionMeta.functionVersion},
#{functionVersionMeta.functionComment},"
- + " #{functionVersionMeta.definitions},
#{functionVersionMeta.auditInfo},"
- + " #{functionVersionMeta.deletedAt})"
- + " ON DUPLICATE KEY UPDATE"
- + " function_comment = #{functionVersionMeta.functionComment},"
- + " definitions = #{functionVersionMeta.definitions},"
- + " audit_info = #{functionVersionMeta.auditInfo},"
- + " deleted_at = #{functionVersionMeta.deletedAt}";
- }
-
public String softDeleteFunctionVersionMetasBySchemaIds(
@Param("schemaIds") List<Long> schemaIds) {
return "<script>"
@@ -64,11 +47,15 @@ public class FunctionVersionMetaBaseSQLProvider {
+ FunctionVersionMetaMapper.TABLE_NAME
+ " SET deleted_at = "
+ DatabaseTimeSQL.MYSQL
+ // History follows the stable entity ID, not the parent recorded in
each snapshot.
+ // Include deleted roots: cascade cleanup soft-deletes roots before
their versions.
+ + " WHERE function_id IN (SELECT function_id FROM "
+ + FunctionMetaMapper.TABLE_NAME
+ " WHERE schema_id IN ("
+ "<foreach collection='schemaIds' item='schemaId' separator=','>"
+ "#{schemaId}"
+ "</foreach>"
- + ") AND deleted_at = 0"
+ + ")) AND deleted_at = 0"
+ "</script>";
}
@@ -77,7 +64,9 @@ public class FunctionVersionMetaBaseSQLProvider {
+ FunctionVersionMetaMapper.TABLE_NAME
+ " SET deleted_at = "
+ DatabaseTimeSQL.MYSQL
- + " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
+ + " WHERE function_id IN (SELECT function_id FROM "
+ + FunctionMetaMapper.TABLE_NAME
+ + " WHERE catalog_id = #{catalogId}) AND deleted_at = 0";
}
public String
softDeleteFunctionVersionMetasByMetalakeId(@Param("metalakeId") Long
metalakeId) {
@@ -85,7 +74,9 @@ public class FunctionVersionMetaBaseSQLProvider {
+ FunctionVersionMetaMapper.TABLE_NAME
+ " SET deleted_at = "
+ DatabaseTimeSQL.MYSQL
- + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
+ + " WHERE function_id IN (SELECT function_id FROM "
+ + FunctionMetaMapper.TABLE_NAME
+ + " WHERE metalake_id = #{metalakeId}) AND deleted_at = 0";
}
public String deleteFunctionVersionMetasByLegacyTimeline(
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/FunctionMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/FunctionMetaPostgreSQLProvider.java
index df8f83f012..8da7a70402 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/FunctionMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/FunctionMetaPostgreSQLProvider.java
@@ -40,31 +40,6 @@ public class FunctionMetaPostgreSQLProvider extends
FunctionMetaBaseSQLProvider
+ " #{functionMeta.deletedAt})";
}
- @Override
- public String insertFunctionMetaOnDuplicateKeyUpdate(
- @Param("functionMeta") FunctionPO functionPO) {
- return "INSERT INTO "
- + FunctionMetaMapper.TABLE_NAME
- + " (function_id, function_name, metalake_id, catalog_id, schema_id,"
- + " function_type, \"deterministic\", function_current_version,
function_latest_version, audit_info, deleted_at)"
- + " VALUES (#{functionMeta.functionId}, #{functionMeta.functionName},
#{functionMeta.metalakeId},"
- + " #{functionMeta.catalogId}, #{functionMeta.schemaId},
#{functionMeta.functionType},"
- + " #{functionMeta.deterministic},"
- + " #{functionMeta.functionCurrentVersion},
#{functionMeta.functionLatestVersion}, #{functionMeta.auditInfo},"
- + " #{functionMeta.deletedAt})"
- + " ON CONFLICT (function_id) DO UPDATE SET"
- + " function_name = #{functionMeta.functionName},"
- + " metalake_id = #{functionMeta.metalakeId},"
- + " catalog_id = #{functionMeta.catalogId},"
- + " schema_id = #{functionMeta.schemaId},"
- + " function_type = #{functionMeta.functionType},"
- + " \"deterministic\" = #{functionMeta.deterministic},"
- + " function_current_version = #{functionMeta.functionCurrentVersion},"
- + " function_latest_version = #{functionMeta.functionLatestVersion},"
- + " audit_info = #{functionMeta.auditInfo},"
- + " deleted_at = #{functionMeta.deletedAt}";
- }
-
@Override
public String listFunctionPOsBySchemaId(@Param("schemaId") Long schemaId) {
return "SELECT fm.function_id, fm.function_name, fm.metalake_id,
fm.catalog_id, fm.schema_id,"
@@ -102,12 +77,14 @@ public class FunctionMetaPostgreSQLProvider extends
FunctionMetaBaseSQLProvider
}
@Override
- public String softDeleteFunctionMetaByFunctionId(@Param("functionId") Long
functionId) {
+ public String softDeleteFunctionMetaByFunctionId(
+ @Param("functionId") Long functionId, @Param("currentVersion") Integer
currentVersion) {
return "UPDATE "
+ FunctionMetaMapper.TABLE_NAME
+ " SET deleted_at = "
+ DatabaseTimeSQL.POSTGRESQL
- + " WHERE function_id = #{functionId} AND deleted_at = 0";
+ + " WHERE function_id = #{functionId}"
+ + " AND function_current_version = #{currentVersion} AND deleted_at =
0";
}
@Override
@@ -170,14 +147,7 @@ public class FunctionMetaPostgreSQLProvider extends
FunctionMetaBaseSQLProvider
+ " audit_info = #{newFunctionMeta.auditInfo},"
+ " deleted_at = #{newFunctionMeta.deletedAt}"
+ " WHERE function_id = #{oldFunctionMeta.functionId}"
- + " AND function_name = #{oldFunctionMeta.functionName}"
- + " AND metalake_id = #{oldFunctionMeta.metalakeId}"
- + " AND catalog_id = #{oldFunctionMeta.catalogId}"
- + " AND schema_id = #{oldFunctionMeta.schemaId}"
- + " AND function_type = #{oldFunctionMeta.functionType}"
+ " AND function_current_version =
#{oldFunctionMeta.functionCurrentVersion}"
- + " AND function_latest_version =
#{oldFunctionMeta.functionLatestVersion}"
- + " AND audit_info = #{oldFunctionMeta.auditInfo}"
+ " AND deleted_at = 0";
}
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/FunctionVersionMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/FunctionVersionMetaPostgreSQLProvider.java
index 2132332947..ae8bb55c72 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/FunctionVersionMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/FunctionVersionMetaPostgreSQLProvider.java
@@ -19,33 +19,14 @@
package org.apache.gravitino.storage.relational.mapper.provider.postgresql;
import java.util.List;
+import org.apache.gravitino.storage.relational.mapper.FunctionMetaMapper;
import
org.apache.gravitino.storage.relational.mapper.FunctionVersionMetaMapper;
import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL;
import
org.apache.gravitino.storage.relational.mapper.provider.base.FunctionVersionMetaBaseSQLProvider;
-import org.apache.gravitino.storage.relational.po.FunctionVersionPO;
import org.apache.ibatis.annotations.Param;
public class FunctionVersionMetaPostgreSQLProvider extends
FunctionVersionMetaBaseSQLProvider {
- @Override
- public String insertFunctionVersionMetaOnDuplicateKeyUpdate(
- @Param("functionVersionMeta") FunctionVersionPO functionVersionPO) {
- return "INSERT INTO "
- + FunctionVersionMetaMapper.TABLE_NAME
- + " (metalake_id, catalog_id, schema_id, function_id, version,"
- + " function_comment, definitions, audit_info, deleted_at)"
- + " VALUES (#{functionVersionMeta.metalakeId},
#{functionVersionMeta.catalogId},"
- + " #{functionVersionMeta.schemaId},
#{functionVersionMeta.functionId},"
- + " #{functionVersionMeta.functionVersion},
#{functionVersionMeta.functionComment},"
- + " #{functionVersionMeta.definitions},
#{functionVersionMeta.auditInfo},"
- + " #{functionVersionMeta.deletedAt})"
- + " ON CONFLICT (function_id, version, deleted_at) DO UPDATE SET"
- + " function_comment = #{functionVersionMeta.functionComment},"
- + " definitions = #{functionVersionMeta.definitions},"
- + " audit_info = #{functionVersionMeta.auditInfo},"
- + " deleted_at = #{functionVersionMeta.deletedAt}";
- }
-
@Override
public String softDeleteFunctionVersionMetasBySchemaIds(
@Param("schemaIds") List<Long> schemaIds) {
@@ -54,11 +35,13 @@ public class FunctionVersionMetaPostgreSQLProvider extends
FunctionVersionMetaBa
+ FunctionVersionMetaMapper.TABLE_NAME
+ " SET deleted_at = "
+ DatabaseTimeSQL.POSTGRESQL
+ + " WHERE function_id IN (SELECT function_id FROM "
+ + FunctionMetaMapper.TABLE_NAME
+ " WHERE schema_id IN ("
+ "<foreach collection='schemaIds' item='schemaId' separator=','>"
+ "#{schemaId}"
+ "</foreach>"
- + ") AND deleted_at = 0"
+ + ")) AND deleted_at = 0"
+ "</script>";
}
@@ -68,7 +51,9 @@ public class FunctionVersionMetaPostgreSQLProvider extends
FunctionVersionMetaBa
+ FunctionVersionMetaMapper.TABLE_NAME
+ " SET deleted_at = "
+ DatabaseTimeSQL.POSTGRESQL
- + " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
+ + " WHERE function_id IN (SELECT function_id FROM "
+ + FunctionMetaMapper.TABLE_NAME
+ + " WHERE catalog_id = #{catalogId}) AND deleted_at = 0";
}
@Override
@@ -77,7 +62,9 @@ public class FunctionVersionMetaPostgreSQLProvider extends
FunctionVersionMetaBa
+ FunctionVersionMetaMapper.TABLE_NAME
+ " SET deleted_at = "
+ DatabaseTimeSQL.POSTGRESQL
- + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
+ + " WHERE function_id IN (SELECT function_id FROM "
+ + FunctionMetaMapper.TABLE_NAME
+ + " WHERE metalake_id = #{metalakeId}) AND deleted_at = 0";
}
@Override
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/FunctionMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/FunctionMetaService.java
index 2d6fcfe444..31fe233741 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/FunctionMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/FunctionMetaService.java
@@ -29,10 +29,10 @@ import java.io.IOException;
import java.util.List;
import java.util.Locale;
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;
+import org.apache.gravitino.EntityAlreadyExistsException;
import org.apache.gravitino.HasIdentifier;
import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.NameIdentifier;
@@ -49,6 +49,7 @@ import
org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
import
org.apache.gravitino.storage.relational.mapper.TagMetadataObjectRelMapper;
import org.apache.gravitino.storage.relational.po.FunctionMaxVersionPO;
import org.apache.gravitino.storage.relational.po.FunctionPO;
+import org.apache.gravitino.storage.relational.po.FunctionVersionPO;
import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
import org.apache.gravitino.utils.NameIdentifierUtil;
@@ -125,22 +126,20 @@ public class FunctionMetaService {
po.schemaId(),
po.catalogId(),
po.metalakeId()),
- () ->
- SessionUtils.doWithoutCommit(
- FunctionMetaMapper.class, mapper -> ops.insertPO(mapper, po,
overwrite)),
- () ->
- SessionUtils.doWithoutCommit(
- FunctionVersionMetaMapper.class,
- mapper -> {
- if (overwrite) {
-
mapper.insertFunctionVersionMetaOnDuplicateKeyUpdate(po.functionVersionPO());
- } else {
- mapper.insertFunctionVersionMeta(po.functionVersionPO());
- }
- }));
+ () -> insertFunctionWithoutCommit(functionEntity, po, overwrite));
} catch (RuntimeException re) {
- ExceptionUtils.checkSQLException(
- re, Entity.EntityType.FUNCTION,
functionEntity.nameIdentifier().toString());
+ try {
+ ExceptionUtils.checkSQLException(
+ re, Entity.EntityType.FUNCTION,
functionEntity.nameIdentifier().toString());
+ } catch (EntityAlreadyExistsException duplicate) {
+ if (overwrite) {
+ // A missing-row locking read cannot fence a concurrent insert at
READ_COMMITTED.
+ // Propagate the conflict so the whole transaction is rolled back
before retrying.
+ throw ExceptionUtils.concurrentModification(
+ Entity.EntityType.FUNCTION, functionEntity.nameIdentifier());
+ }
+ throw duplicate;
+ }
throw re;
}
}
@@ -150,47 +149,31 @@ public class FunctionMetaService {
baseMetricName = "deleteFunction")
public boolean deleteFunction(NameIdentifier ident) {
FunctionPO functionPO = getFunctionPOByIdentifier(ident);
- Long functionId = functionPO.functionId();
- AtomicInteger functionDeletedCount = new AtomicInteger();
+ deleteFunctionWithVersion(ident, functionPO);
+ return true;
+ }
+
+ /**
+ * Deletes the observed function and its dependent rows in one transaction.
+ *
+ * <p>Package-private access lets concurrency tests submit a deliberately
stale snapshot while
+ * exercising the same root-first ordering as the public delete path.
+ */
+ void deleteFunctionWithVersion(NameIdentifier identifier, FunctionPO
observedFunctionPO) {
SessionUtils.doMultipleWithCommit(
- // delete function meta
+ // Check the root version before touching relationships. A stale drop
stops here.
() ->
- functionDeletedCount.set(
- SessionUtils.getWithoutCommit(
- FunctionMetaMapper.class,
- mapper ->
mapper.softDeleteFunctionMetaByFunctionId(functionId))),
-
- // delete function versions, owner rels, and securable object rels
after meta deletion
- () -> {
- if (functionDeletedCount.get() > 0) {
- SessionUtils.doWithoutCommit(
- FunctionVersionMetaMapper.class,
- mapper ->
mapper.softDeleteFunctionVersionsByFunctionId(functionId));
- SessionUtils.doWithoutCommit(
- OwnerMetaMapper.class,
- mapper ->
- mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
- functionId, MetadataObject.Type.FUNCTION.name()));
- SessionUtils.doWithoutCommit(
- SecurableObjectMapper.class,
- mapper ->
- mapper.softDeleteObjectRelsByMetadataObject(
- functionId, MetadataObject.Type.FUNCTION.name()));
- SessionUtils.doWithoutCommit(
- TagMetadataObjectRelMapper.class,
- mapper ->
- mapper.softDeleteTagMetadataObjectRelsByMetadataObject(
- functionId, MetadataObject.Type.FUNCTION.name()));
- SessionUtils.doWithoutCommit(
- PolicyMetadataObjectRelMapper.class,
- mapper ->
- mapper.softDeletePolicyMetadataObjectRelsByMetadataObject(
- functionId, MetadataObject.Type.FUNCTION.name()));
- }
- });
-
- return functionDeletedCount.get() > 0;
+ OccWriteSupport.deleteWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ FunctionMetaMapper.class,
+ mapper ->
+ mapper.softDeleteFunctionMetaByFunctionId(
+ observedFunctionPO.functionId(),
+ observedFunctionPO.functionCurrentVersion())),
+ () -> functionWriteFailure(identifier, observedFunctionPO)),
+ () -> deleteFunctionDependents(observedFunctionPO.functionId()));
}
@Monitored(
@@ -274,36 +257,44 @@ public class FunctionMetaService {
newEntity.id(),
oldFunctionEntity.id());
+ boolean isSchemaChanged =
!newEntity.namespace().equals(oldFunctionEntity.namespace());
+ NamespacedEntityId targetSchemaIds =
+ isSchemaChanged
+ ? EntityIdService.getEntityIds(
+ NameIdentifier.of(newEntity.namespace().levels()),
Entity.EntityType.SCHEMA)
+ : null;
+ Long newSchemaId = isSchemaChanged ? targetSchemaIds.entityId() :
oldFunctionPO.schemaId();
+ Long newCatalogId =
+ isSchemaChanged ? targetSchemaIds.namespaceIds()[1] :
oldFunctionPO.catalogId();
+ Long newMetalakeId =
+ isSchemaChanged ? targetSchemaIds.namespaceIds()[0] :
oldFunctionPO.metalakeId();
+
try {
- FunctionPO newFunctionPO = updateFunctionPO(oldFunctionPO, newEntity);
- // Insert a new version and update function meta
+ FunctionPO newFunctionPO =
+ updateFunctionPO(oldFunctionPO, newEntity, newSchemaId,
newCatalogId, newMetalakeId);
SessionUtils.doMultipleWithCommit(
- // The function was read before this transaction started. Lock its
observed parent again
- // before writing, so a schema drop cannot finish its function
cleanup and then let this
- // update add a new version below the deleted schema.
- () ->
+ () -> {
+ if (isSchemaChanged) {
SchemaMetaService.getInstance()
.lockSchemaForEntityWrite(
- identifier,
- oldFunctionPO.schemaId(),
- oldFunctionPO.catalogId(),
- oldFunctionPO.metalakeId()),
- () ->
- SessionUtils.doWithoutCommit(
- FunctionVersionMetaMapper.class,
- mapper ->
mapper.insertFunctionVersionMeta(newFunctionPO.functionVersionPO())),
+ newEntity.nameIdentifier(), newSchemaId, newCatalogId,
newMetalakeId);
+ }
+ },
() -> {
+ // function_current_version is the sole OCC token. The root CAS is
the transaction's
+ // decision point and must run before the unguarded version-row
insert below.
int updated =
SessionUtils.getWithoutCommit(
FunctionMetaMapper.class,
mapper -> ops.updatePO(mapper, newFunctionPO,
oldFunctionPO));
if (updated == 0) {
- // The version row was inserted earlier in this transaction.
Throwing here rolls the
- // whole transaction back instead of leaving that version
without an active function
- // metadata row.
- throw
ExceptionUtils.concurrentModification(Entity.EntityType.FUNCTION, identifier);
+ throw functionWriteFailure(identifier, oldFunctionPO);
}
- });
+ },
+ () ->
+ SessionUtils.doWithoutCommit(
+ FunctionVersionMetaMapper.class,
+ mapper ->
mapper.insertFunctionVersionMeta(newFunctionPO.functionVersionPO())));
return newEntity;
} catch (RuntimeException re) {
@@ -333,15 +324,170 @@ public class FunctionMetaService {
builder.withSchemaId(namespacedEntityId.entityId());
}
- private FunctionPO updateFunctionPO(FunctionPO oldFunctionPO, FunctionEntity
newFunction) {
- Integer newVersion = oldFunctionPO.functionLatestVersion() + 1;
+ private FunctionPO updateFunctionPO(
+ FunctionPO oldFunctionPO,
+ FunctionEntity newFunction,
+ Long newSchemaId,
+ Long newCatalogId,
+ Long newMetalakeId) {
+ // Both version columns always advance together, but deriving the next
version from
+ // the higher of the two keeps it above every version row that exists,
even if a
+ // future path ever leaves the current version behind the latest one.
+ Integer newVersion = nextVersion(oldFunctionPO);
FunctionPO.FunctionPOBuilder builder =
FunctionPO.builder()
- .withMetalakeId(oldFunctionPO.metalakeId())
- .withCatalogId(oldFunctionPO.catalogId())
- .withSchemaId(oldFunctionPO.schemaId())
+ .withMetalakeId(newMetalakeId)
+ .withCatalogId(newCatalogId)
+ .withSchemaId(newSchemaId)
.withFunctionLatestVersion(newVersion)
.withFunctionCurrentVersion(newVersion);
return buildFunctionPO(newFunction, builder, newVersion);
}
+
+ /**
+ * Writes a new function or replaces the active function selected by natural
key or stable ID.
+ * Locking the root before choosing the next version makes overwrite
behavior identical across
+ * databases and keeps standard reads strict about the matching version-row
invariant.
+ */
+ private void insertFunctionWithoutCommit(
+ FunctionEntity functionEntity, FunctionPO initializedFunctionPO, boolean
overwrite) {
+ if (!overwrite) {
+ insertNewFunctionWithoutCommit(initializedFunctionPO);
+ return;
+ }
+
+ FunctionPO existingFunctionPO =
findAndLockFunctionForOverwrite(initializedFunctionPO);
+ if (existingFunctionPO == null) {
+ if (SessionUtils.getWithoutCommit(
+ FunctionMetaMapper.class,
+ mapper ->
mapper.countDeletedFunctionMetasById(initializedFunctionPO.functionId()))
+ > 0) {
+ throw new EntityAlreadyExistsException(
+ "The function ID %s is reserved by a deleted function; use a new
ID",
+ initializedFunctionPO.functionId());
+ }
+ insertNewFunctionWithoutCommit(initializedFunctionPO);
+ return;
+ }
+
+ FunctionPO replacementPO = functionPOForOverwrite(initializedFunctionPO,
existingFunctionPO);
+ int updated =
+ SessionUtils.getWithoutCommit(
+ FunctionMetaMapper.class,
+ mapper -> ops.updatePO(mapper, replacementPO, existingFunctionPO));
+ if (updated == 0) {
+ throw functionWriteFailure(functionEntity.nameIdentifier(),
existingFunctionPO);
+ }
+ SessionUtils.doWithoutCommit(
+ FunctionVersionMetaMapper.class,
+ mapper ->
mapper.insertFunctionVersionMeta(replacementPO.functionVersionPO()));
+ }
+
+ private void insertNewFunctionWithoutCommit(FunctionPO functionPO) {
+ SessionUtils.doWithoutCommit(
+ FunctionMetaMapper.class, mapper -> ops.insertPO(mapper, functionPO,
false));
+ SessionUtils.doWithoutCommit(
+ FunctionVersionMetaMapper.class,
+ mapper ->
mapper.insertFunctionVersionMeta(functionPO.functionVersionPO()));
+ }
+
+ private FunctionPO findAndLockFunctionForOverwrite(FunctionPO
initializedFunctionPO) {
+ return OccWriteSupport.findAndLockForOverwrite(
+ () ->
+ SessionUtils.getWithoutCommit(
+ FunctionMetaMapper.class,
+ mapper ->
+ mapper.selectFunctionMetaBySchemaIdAndNameForUpdate(
+ initializedFunctionPO.schemaId(),
initializedFunctionPO.functionName())),
+ () ->
+ SessionUtils.getWithoutCommit(
+ FunctionMetaMapper.class,
+ mapper ->
+
mapper.selectFunctionMetaByIdForUpdate(initializedFunctionPO.functionId())),
+ current -> Objects.equals(current.schemaId(),
initializedFunctionPO.schemaId()));
+ }
+
+ /**
+ * Returns the version to write next, one above every version this function
has ever had.
+ *
+ * @param functionPO the function row observed by the caller
+ * @return the next monotonic version
+ */
+ private static int nextVersion(FunctionPO functionPO) {
+ return Math.max(functionPO.functionCurrentVersion(),
functionPO.functionLatestVersion()) + 1;
+ }
+
+ private FunctionPO functionPOForOverwrite(FunctionPO incomingPO, FunctionPO
persistedPO) {
+ int nextVersion = nextVersion(persistedPO);
+ FunctionVersionPO incomingVersionPO = incomingPO.functionVersionPO();
+ FunctionVersionPO persistedVersionPO =
+ FunctionVersionPO.builder()
+ .withFunctionId(persistedPO.functionId())
+ .withMetalakeId(incomingVersionPO.metalakeId())
+ .withCatalogId(incomingVersionPO.catalogId())
+ .withSchemaId(incomingVersionPO.schemaId())
+ .withFunctionVersion(nextVersion)
+ .withFunctionComment(incomingVersionPO.functionComment())
+ .withDefinitions(incomingVersionPO.definitions())
+ .withAuditInfo(incomingVersionPO.auditInfo())
+ .withDeletedAt(incomingVersionPO.deletedAt())
+ .build();
+ return FunctionPO.builder()
+ .withFunctionId(persistedPO.functionId())
+ .withFunctionName(incomingPO.functionName())
+ .withMetalakeId(incomingPO.metalakeId())
+ .withCatalogId(incomingPO.catalogId())
+ .withSchemaId(incomingPO.schemaId())
+ .withFunctionType(incomingPO.functionType())
+ .withDeterministic(incomingPO.deterministic())
+ .withFunctionLatestVersion(nextVersion)
+ .withFunctionCurrentVersion(nextVersion)
+ .withAuditInfo(incomingPO.auditInfo())
+ .withDeletedAt(incomingPO.deletedAt())
+ .withFunctionVersionPO(persistedVersionPO)
+ .build();
+ }
+
+ private void deleteFunctionDependents(Long functionId) {
+ SessionUtils.doWithoutCommit(
+ FunctionVersionMetaMapper.class,
+ mapper -> mapper.softDeleteFunctionVersionsByFunctionId(functionId));
+ SessionUtils.doWithoutCommit(
+ OwnerMetaMapper.class,
+ mapper ->
+ mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
+ functionId, MetadataObject.Type.FUNCTION.name()));
+ SessionUtils.doWithoutCommit(
+ SecurableObjectMapper.class,
+ mapper ->
+ mapper.softDeleteObjectRelsByMetadataObject(
+ functionId, MetadataObject.Type.FUNCTION.name()));
+ SessionUtils.doWithoutCommit(
+ TagMetadataObjectRelMapper.class,
+ mapper ->
+ mapper.softDeleteTagMetadataObjectRelsByMetadataObject(
+ functionId, MetadataObject.Type.FUNCTION.name()));
+ SessionUtils.doWithoutCommit(
+ PolicyMetadataObjectRelMapper.class,
+ mapper ->
+ mapper.softDeletePolicyMetadataObjectRelsByMetadataObject(
+ functionId, MetadataObject.Type.FUNCTION.name()));
+ }
+
+ private RuntimeException functionWriteFailure(
+ NameIdentifier identifier, FunctionPO observedFunctionPO) {
+ return OccWriteSupport.writeFailure(
+ identifier,
+ Entity.EntityType.FUNCTION,
+ () ->
+ SessionUtils.getWithoutCommit(
+ FunctionMetaMapper.class,
+ mapper ->
mapper.selectFunctionMetaByIdForUpdate(observedFunctionPO.functionId())),
+ null,
+ current ->
+ Objects.equals(current.functionName(),
observedFunctionPO.functionName())
+ && Objects.equals(current.schemaId(),
observedFunctionPO.schemaId())
+ && Objects.equals(current.catalogId(),
observedFunctionPO.catalogId())
+ && Objects.equals(current.metalakeId(),
observedFunctionPO.metalakeId()));
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/FunctionPOStorageOps.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/FunctionPOStorageOps.java
index 45f22a6b3e..ef3155cce9 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/FunctionPOStorageOps.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/FunctionPOStorageOps.java
@@ -18,6 +18,7 @@
*/
package org.apache.gravitino.storage.relational.service;
+import com.google.common.base.Preconditions;
import java.util.List;
import java.util.stream.Collectors;
import org.apache.gravitino.Entity;
@@ -31,13 +32,19 @@ public class FunctionPOStorageOps extends
BasePOStorageOps<FunctionPO, FunctionM
public FunctionPOStorageOps() {}
+ /**
+ * {@inheritDoc}
+ *
+ * <p>Overwrite is not supported here. {@link FunctionMetaService} replaces
an existing function
+ * by locking its root row and advancing it with a version compare-and-set,
which keeps the stored
+ * ID and the version sequence identical on every database. A
database-specific upsert would
+ * bypass that and is rejected instead of silently taking a second code path.
+ */
@Override
public void insertPO(FunctionMetaMapper mapper, FunctionPO functionPO,
boolean overwrite) {
- if (overwrite) {
- mapper.insertFunctionMetaOnDuplicateKeyUpdate(functionPO);
- } else {
- mapper.insertFunctionMeta(functionPO);
- }
+ Preconditions.checkArgument(
+ !overwrite, "Function overwrite is handled by FunctionMetaService, not
by an upsert");
+ mapper.insertFunctionMeta(functionPO);
}
@Override
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestFunctionMetaBaseSQLProvider.java
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestFunctionMetaBaseSQLProvider.java
new file mode 100644
index 0000000000..0e0aeb452b
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestFunctionMetaBaseSQLProvider.java
@@ -0,0 +1,74 @@
+/*
+ * 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.base;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestFunctionMetaBaseSQLProvider {
+
+ private static final FunctionMetaBaseSQLProvider PROVIDER = new
FunctionMetaBaseSQLProvider();
+
+ @Test
+ void testUpdateUsesOnlyIdVersionAndActiveStateForCas() {
+ String sql = PROVIDER.updateFunctionMeta(null, null);
+ String whereClause = sql.substring(sql.indexOf(" WHERE"));
+
+ Assertions.assertEquals(
+ " WHERE function_id = #{oldFunctionMeta.functionId}"
+ + " AND function_current_version =
#{oldFunctionMeta.functionCurrentVersion}"
+ + " AND deleted_at = 0",
+ whereClause);
+ }
+
+ @Test
+ void testDirectDeleteUsesVersionCas() {
+ String sql = PROVIDER.softDeleteFunctionMetaByFunctionId(null, null);
+
+ Assertions.assertTrue(sql.contains("AND function_current_version =
#{currentVersion}"));
+ Assertions.assertTrue(sql.endsWith("AND deleted_at = 0"));
+ }
+
+ @Test
+ void testConflictReadLocksActiveRow() {
+ String sql = PROVIDER.selectFunctionMetaByIdForUpdate(null);
+
+ Assertions.assertTrue(sql.contains("function_name as functionName"));
+ Assertions.assertTrue(sql.contains("function_current_version as
functionCurrentVersion"));
+ Assertions.assertTrue(sql.contains("WHERE function_id = #{functionId} AND
deleted_at = 0"));
+ Assertions.assertTrue(sql.endsWith("FOR UPDATE"));
+ }
+
+ @Test
+ void testNormalReadRequiresCurrentVersionRow() {
+ String sql = PROVIDER.selectFunctionMetaBySchemaIdAndName(null, null);
+
+ Assertions.assertTrue(sql.contains("fm INNER JOIN function_version_info
vi"));
+ Assertions.assertTrue(sql.endsWith("AND fm.deleted_at = 0 AND
vi.deleted_at = 0"));
+ }
+
+ @Test
+ void testOverwriteLookupLocksOnlyFunctionRoot() {
+ String sql = PROVIDER.selectFunctionMetaBySchemaIdAndNameForUpdate(null,
null);
+
+ Assertions.assertFalse(sql.contains("function_version_info"));
+ Assertions.assertTrue(sql.contains("function_current_version as
functionCurrentVersion"));
+ Assertions.assertTrue(sql.endsWith("AND deleted_at = 0 FOR UPDATE"));
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestFunctionMetaPostgreSQLProvider.java
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestFunctionMetaPostgreSQLProvider.java
new file mode 100644
index 0000000000..a8e4518fd4
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestFunctionMetaPostgreSQLProvider.java
@@ -0,0 +1,43 @@
+/*
+ * 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.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestFunctionMetaPostgreSQLProvider {
+
+ @Test
+ void testDirectDeleteUsesVersionCas() {
+ String sql =
+ new
FunctionMetaPostgreSQLProvider().softDeleteFunctionMetaByFunctionId(null, null);
+
+ Assertions.assertTrue(sql.contains("AND function_current_version =
#{currentVersion}"));
+ Assertions.assertTrue(sql.endsWith("AND deleted_at = 0"));
+ }
+
+ @Test
+ void testNormalReadRequiresCurrentVersionRow() {
+ String sql =
+ new
FunctionMetaPostgreSQLProvider().selectFunctionMetaBySchemaIdAndName(null,
null);
+
+ Assertions.assertTrue(sql.contains("fm INNER JOIN function_version_info
vi"));
+ Assertions.assertTrue(sql.endsWith("AND fm.deleted_at = 0 AND
vi.deleted_at = 0"));
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestFunctionMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestFunctionMetaService.java
index 9cc3e039d0..4bc84df448 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestFunctionMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestFunctionMetaService.java
@@ -21,6 +21,7 @@ package org.apache.gravitino.storage.relational.service;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -35,6 +36,12 @@ import java.time.Instant;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+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 java.util.concurrent.TimeoutException;
import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityAlreadyExistsException;
import org.apache.gravitino.NameIdentifier;
@@ -52,7 +59,13 @@ import org.apache.gravitino.meta.TagEntity;
import org.apache.gravitino.meta.UserEntity;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.mapper.FunctionMetaMapper;
+import
org.apache.gravitino.storage.relational.mapper.FunctionVersionMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
+import org.apache.gravitino.storage.relational.po.FunctionPO;
+import org.apache.gravitino.storage.relational.po.SchemaPO;
import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.NamespaceUtil;
import org.apache.ibatis.session.SqlSession;
@@ -71,6 +84,22 @@ public class TestFunctionMetaService extends TestJDBCBackend
{
createAndInsertSchema(metalakeName, catalogName, schemaName);
}
+ /** Dropping the old parent after a move must preserve every historical
version. */
+ @TestTemplate
+ public void testMovedFunctionHistorySurvivesSourceCascade() throws
IOException {
+ for (String parent : new String[] {"schema", "catalog", "metalake"}) {
+ assertMovedFunctionHistoryCascade(parent, true);
+ }
+ }
+
+ /** Dropping the current parent must delete versions created under previous
parents too. */
+ @TestTemplate
+ public void testMovedFunctionHistoryIsDeletedWithDestination() throws
IOException {
+ for (String parent : new String[] {"schema", "catalog", "metalake"}) {
+ assertMovedFunctionHistoryCascade(parent, false);
+ }
+ }
+
@TestTemplate
public void testInsertAlreadyExistsException() throws IOException {
FunctionEntity function =
@@ -242,6 +271,79 @@ public class TestFunctionMetaService extends
TestJDBCBackend {
assertTrue(versions.containsKey(2));
}
+ @TestTemplate
+ public void testCreateFunctionWaitsForConcurrentSchemaDelete() throws
Exception {
+ SchemaPO observedSchemaPO =
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class,
+ mapper ->
+ mapper.selectSchemaByFullQualifiedName(metalakeName,
catalogName, schemaName));
+ Namespace namespace = NamespaceUtil.ofFunction(metalakeName, catalogName,
schemaName);
+ FunctionEntity function =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ namespace,
+ GravitinoITUtils.genRandomName("function_parent_delete_race"),
+ AUDIT_INFO);
+
+ CountDownLatch deleteWritten = new CountDownLatch(1);
+ CountDownLatch allowDeleteCommit = new CountDownLatch(1);
+ CountDownLatch createStarted = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> deleteResult =
+ executor.submit(
+ () -> {
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ assertEquals(
+ Integer.valueOf(1),
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class,
+ mapper ->
+
mapper.softDeleteSchemaMetaBySchemaIdAndVersion(
+ observedSchemaPO.getSchemaId(),
+
observedSchemaPO.getCurrentVersion()))),
+ () -> {
+ deleteWritten.countDown();
+ await(allowDeleteCommit);
+ });
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+
+ try {
+ assertTrue(deleteWritten.await(30, TimeUnit.SECONDS));
+ Future<Throwable> createResult =
+ executor.submit(
+ () -> {
+ createStarted.countDown();
+ try {
+ FunctionMetaService.getInstance().insertFunction(function,
false);
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+ assertTrue(createStarted.await(30, TimeUnit.SECONDS));
+ assertThrows(TimeoutException.class, () -> createResult.get(500,
TimeUnit.MILLISECONDS));
+
+ allowDeleteCommit.countDown();
+ assertNull(deleteResult.get(30, TimeUnit.SECONDS));
+ Throwable createFailure = createResult.get(30, TimeUnit.SECONDS);
+ assertTrue(createFailure instanceof NoSuchEntityException,
String.valueOf(createFailure));
+ } finally {
+ allowDeleteCommit.countDown();
+ executor.shutdownNow();
+ }
+
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
FunctionMetaService.getInstance().getFunctionByIdentifier(function.nameIdentifier()));
+ }
+
@TestTemplate
public void testUpdateFunctionFailsWhenSchemaIsDeletedConcurrently() throws
IOException {
String functionName = GravitinoITUtils.genRandomName("test_function");
@@ -276,7 +378,7 @@ public class TestFunctionMetaService extends
TestJDBCBackend {
}
@TestTemplate
- public void testUpdateFunctionRollsBackNewVersionAfterConcurrentDelete()
throws IOException {
+ public void testUpdateFunctionReportsNoSuchAfterConcurrentDelete() throws
IOException {
String functionName = GravitinoITUtils.genRandomName("test_function");
Namespace namespace = NamespaceUtil.ofFunction(metalakeName, catalogName,
schemaName);
FunctionEntity function =
@@ -289,7 +391,7 @@ public class TestFunctionMetaService extends
TestJDBCBackend {
FunctionEntity updatedFunction = copyFunctionWithComment(function,
"updated comment");
assertThrows(
- OptimisticLockException.class,
+ NoSuchEntityException.class,
() ->
FunctionMetaService.getInstance()
.updateFunction(
@@ -308,6 +410,333 @@ public class TestFunctionMetaService extends
TestJDBCBackend {
assertFalse(versions.containsKey(2));
}
+ @TestTemplate
+ public void testAlterReportsOptimisticLockConflictAndKeepsWinnerVersion()
throws IOException {
+ String functionName =
GravitinoITUtils.genRandomName("function_alter_conflict");
+ Namespace namespace = NamespaceUtil.ofFunction(metalakeName, catalogName,
schemaName);
+ FunctionEntity function =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace, functionName,
AUDIT_INFO);
+ FunctionMetaService.getInstance().insertFunction(function, false);
+
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+ FunctionMetaService.getInstance()
+ .updateFunction(
+ function.nameIdentifier(),
+ entity -> {
+ try {
+ FunctionMetaService.getInstance()
+ .updateFunction(
+ function.nameIdentifier(),
+ competing ->
+ copyFunctionWithComment(
+ (FunctionEntity) competing, "competing
update"));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return copyFunctionWithComment((FunctionEntity) entity,
"requested update");
+ }));
+
+ FunctionEntity current =
+
FunctionMetaService.getInstance().getFunctionByIdentifier(function.nameIdentifier());
+ assertEquals("competing update", current.comment());
+ Map<Integer, Long> versions = listFunctionVersions(function.id());
+ assertEquals(2, versions.size());
+ assertTrue(versions.containsKey(2));
+ }
+
+ @TestTemplate
+ public void testAlterRollsBackRootCasWhenVersionInsertFails() throws
IOException {
+ String functionName =
GravitinoITUtils.genRandomName("function_version_insert_failure");
+ Namespace namespace = NamespaceUtil.ofFunction(metalakeName, catalogName,
schemaName);
+ FunctionEntity function =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace, functionName,
AUDIT_INFO);
+ FunctionMetaService.getInstance().insertFunction(function, false);
+ FunctionPO originalPO =
+
FunctionMetaService.getInstance().getFunctionPOByIdentifier(function.nameIdentifier());
+ FunctionEntity conflictingVersion = copyFunctionWithComment(function,
"conflicting version");
+ FunctionPO conflictingPO =
+ FunctionPO.buildFunctionPO(
+ conflictingVersion,
+ FunctionPO.builder()
+ .withMetalakeId(originalPO.metalakeId())
+ .withCatalogId(originalPO.catalogId())
+ .withSchemaId(originalPO.schemaId())
+ .withFunctionLatestVersion(2)
+ .withFunctionCurrentVersion(2),
+ 2);
+ SessionUtils.doWithCommit(
+ FunctionVersionMetaMapper.class,
+ mapper ->
mapper.insertFunctionVersionMeta(conflictingPO.functionVersionPO()));
+
+ assertThrows(
+ EntityAlreadyExistsException.class,
+ () ->
+ FunctionMetaService.getInstance()
+ .updateFunction(
+ function.nameIdentifier(),
+ entity -> copyFunctionWithComment((FunctionEntity) entity,
"must roll back")));
+
+ FunctionPO currentPO =
+
FunctionMetaService.getInstance().getFunctionPOByIdentifier(function.nameIdentifier());
+ assertEquals(1, currentPO.functionCurrentVersion());
+ assertEquals(1, currentPO.functionLatestVersion());
+ assertEquals(
+ function.comment(),
+ FunctionMetaService.getInstance()
+ .getFunctionByIdentifier(function.nameIdentifier())
+ .comment());
+ }
+
+ @TestTemplate
+ public void testAlterReportsNoSuchWhenRenamedConcurrently() throws
IOException {
+ String functionName =
GravitinoITUtils.genRandomName("function_alter_renamed");
+ Namespace namespace = NamespaceUtil.ofFunction(metalakeName, catalogName,
schemaName);
+ FunctionEntity function =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace, functionName,
AUDIT_INFO);
+ FunctionMetaService.getInstance().insertFunction(function, false);
+ NameIdentifier renamedIdentifier = NameIdentifier.of(namespace,
functionName + "_winner");
+
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ FunctionMetaService.getInstance()
+ .updateFunction(
+ function.nameIdentifier(),
+ entity -> {
+ try {
+ FunctionMetaService.getInstance()
+ .updateFunction(
+ function.nameIdentifier(),
+ competing ->
+ copyFunction(
+ (FunctionEntity) competing,
+ renamedIdentifier.name(),
+ namespace,
+ "renamed winner"));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return copyFunctionWithComment((FunctionEntity) entity,
"stale update");
+ }));
+
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
FunctionMetaService.getInstance().getFunctionByIdentifier(function.nameIdentifier()));
+ assertEquals(
+ "renamed winner",
+
FunctionMetaService.getInstance().getFunctionByIdentifier(renamedIdentifier).comment());
+ }
+
+ @TestTemplate
+ public void testAlterReportsNoSuchWhenMovedConcurrently() throws IOException
{
+ String targetCatalogName =
GravitinoITUtils.genRandomName("function_target_catalog");
+ String targetSchemaName =
GravitinoITUtils.genRandomName("function_target_schema");
+ createAndInsertCatalog(metalakeName, targetCatalogName);
+ createAndInsertSchema(metalakeName, targetCatalogName, targetSchemaName);
+ String functionName =
GravitinoITUtils.genRandomName("function_alter_moved");
+ Namespace namespace = NamespaceUtil.ofFunction(metalakeName, catalogName,
schemaName);
+ Namespace movedNamespace =
+ NamespaceUtil.ofFunction(metalakeName, targetCatalogName,
targetSchemaName);
+ FunctionEntity function =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace, functionName,
AUDIT_INFO);
+ FunctionMetaService.getInstance().insertFunction(function, false);
+ NameIdentifier movedIdentifier = NameIdentifier.of(movedNamespace,
functionName);
+
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ FunctionMetaService.getInstance()
+ .updateFunction(
+ function.nameIdentifier(),
+ entity -> {
+ try {
+ FunctionMetaService.getInstance()
+ .updateFunction(
+ function.nameIdentifier(),
+ competing ->
+ copyFunction(
+ (FunctionEntity) competing,
+ functionName,
+ movedNamespace,
+ "moved winner"));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return copyFunctionWithComment((FunctionEntity) entity,
"stale update");
+ }));
+
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
FunctionMetaService.getInstance().getFunctionByIdentifier(function.nameIdentifier()));
+ assertEquals(
+ "moved winner",
+
FunctionMetaService.getInstance().getFunctionByIdentifier(movedIdentifier).comment());
+ SchemaPO targetSchemaPO =
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class,
+ mapper ->
+ mapper.selectSchemaByFullQualifiedName(
+ metalakeName, targetCatalogName, targetSchemaName));
+ FunctionPO movedPO =
+
FunctionMetaService.getInstance().getFunctionPOByIdentifier(movedIdentifier);
+ assertEquals(targetSchemaPO.getSchemaId(), movedPO.schemaId());
+ assertEquals(targetSchemaPO.getCatalogId(), movedPO.catalogId());
+ assertEquals(targetSchemaPO.getMetalakeId(), movedPO.metalakeId());
+ }
+
+ @TestTemplate
+ public void testMoveWaitsForConcurrentTargetSchemaDelete() throws Exception {
+ String targetCatalogName =
GravitinoITUtils.genRandomName("function_deleted_target_catalog");
+ String targetSchemaName =
GravitinoITUtils.genRandomName("function_deleted_target_schema");
+ createAndInsertCatalog(metalakeName, targetCatalogName);
+ createAndInsertSchema(metalakeName, targetCatalogName, targetSchemaName);
+ SchemaPO targetSchemaPO =
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class,
+ mapper ->
+ mapper.selectSchemaByFullQualifiedName(
+ metalakeName, targetCatalogName, targetSchemaName));
+ String functionName =
GravitinoITUtils.genRandomName("function_target_delete_race");
+ Namespace sourceNamespace = NamespaceUtil.ofFunction(metalakeName,
catalogName, schemaName);
+ Namespace targetNamespace =
+ NamespaceUtil.ofFunction(metalakeName, targetCatalogName,
targetSchemaName);
+ FunctionEntity function =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), sourceNamespace,
functionName, AUDIT_INFO);
+ FunctionMetaService.getInstance().insertFunction(function, false);
+ FunctionEntity moved = copyFunction(function, functionName,
targetNamespace, "must not move");
+
+ CountDownLatch deleteWritten = new CountDownLatch(1);
+ CountDownLatch allowDeleteCommit = new CountDownLatch(1);
+ CountDownLatch moveStarted = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> deleteResult =
+ executor.submit(
+ () -> {
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ assertEquals(
+ Integer.valueOf(1),
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class,
+ mapper ->
+
mapper.softDeleteSchemaMetaBySchemaIdAndVersion(
+ targetSchemaPO.getSchemaId(),
+ targetSchemaPO.getCurrentVersion()))),
+ () -> {
+ deleteWritten.countDown();
+ await(allowDeleteCommit);
+ });
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+
+ try {
+ assertTrue(deleteWritten.await(30, TimeUnit.SECONDS));
+ Future<Throwable> moveResult =
+ executor.submit(
+ () -> {
+ moveStarted.countDown();
+ try {
+ FunctionMetaService.getInstance()
+ .updateFunction(function.nameIdentifier(), ignored ->
moved);
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+ assertTrue(moveStarted.await(30, TimeUnit.SECONDS));
+ assertThrows(TimeoutException.class, () -> moveResult.get(500,
TimeUnit.MILLISECONDS));
+
+ allowDeleteCommit.countDown();
+ assertNull(deleteResult.get(30, TimeUnit.SECONDS));
+ Throwable moveFailure = moveResult.get(30, TimeUnit.SECONDS);
+ assertTrue(moveFailure instanceof NoSuchEntityException,
String.valueOf(moveFailure));
+ } finally {
+ allowDeleteCommit.countDown();
+ executor.shutdownNow();
+ }
+
+ assertEquals(
+ function.comment(),
+ FunctionMetaService.getInstance()
+ .getFunctionByIdentifier(function.nameIdentifier())
+ .comment());
+ assertEquals(1, listFunctionVersions(function.id()).size());
+ }
+
+ @TestTemplate
+ public void testDeleteRejectsStaleVersion() throws IOException {
+ String functionName =
GravitinoITUtils.genRandomName("function_stale_delete");
+ Namespace namespace = NamespaceUtil.ofFunction(metalakeName, catalogName,
schemaName);
+ FunctionEntity function =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace, functionName,
AUDIT_INFO);
+ FunctionMetaService.getInstance().insertFunction(function, false);
+ TagEntity tag =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("function_occ_tag")
+ .withNamespace(NamespaceUtil.ofTag(metalakeName))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TagMetaService.getInstance().insertTag(tag, false);
+ TagMetaService.getInstance()
+ .associateTagsWithMetadataObject(
+ function.nameIdentifier(),
+ function.type(),
+ new NameIdentifier[] {tag.nameIdentifier()},
+ new NameIdentifier[0]);
+ FunctionPO stalePO =
+
FunctionMetaService.getInstance().getFunctionPOByIdentifier(function.nameIdentifier());
+
+ FunctionMetaService.getInstance()
+ .updateFunction(
+ function.nameIdentifier(),
+ entity -> copyFunctionWithComment((FunctionEntity) entity,
"winning update"));
+
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+ FunctionMetaService.getInstance()
+ .deleteFunctionWithVersion(function.nameIdentifier(),
stalePO));
+ assertEquals(
+ "winning update",
+ FunctionMetaService.getInstance()
+ .getFunctionByIdentifier(function.nameIdentifier())
+ .comment());
+ assertEquals(1, countActiveTagRelForMetadataObject(function.id(),
"FUNCTION"));
+ }
+
+ @TestTemplate
+ public void testDeleteReportsNoSuchWhenDeletedConcurrently() throws
IOException {
+ String functionName =
GravitinoITUtils.genRandomName("function_delete_deleted");
+ Namespace namespace = NamespaceUtil.ofFunction(metalakeName, catalogName,
schemaName);
+ FunctionEntity function =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace, functionName,
AUDIT_INFO);
+ FunctionMetaService.getInstance().insertFunction(function, false);
+ FunctionPO stalePO =
+
FunctionMetaService.getInstance().getFunctionPOByIdentifier(function.nameIdentifier());
+
+
FunctionMetaService.getInstance().deleteFunction(function.nameIdentifier());
+
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ FunctionMetaService.getInstance()
+ .deleteFunctionWithVersion(function.nameIdentifier(),
stalePO));
+ }
+
@TestTemplate
public void testDeleteFunction() throws IOException {
String functionName = GravitinoITUtils.genRandomName("test_function");
@@ -612,6 +1041,280 @@ public class TestFunctionMetaService extends
TestJDBCBackend {
FunctionMetaService.getInstance().getFunctionByIdentifier(functionIdent);
assertEquals("overwritten comment", loadedFunction.comment());
assertTrue(loadedFunction.deterministic());
+ FunctionPO overwrittenPO =
+
FunctionMetaService.getInstance().getFunctionPOByIdentifier(functionIdent);
+ assertEquals(2, overwrittenPO.functionCurrentVersion());
+ assertEquals(2, overwrittenPO.functionLatestVersion());
+ assertEquals(2, listFunctionVersions(function.id()).size());
+ }
+
+ @TestTemplate
+ public void testOverwriteDoesNotAdoptFunctionFromAnotherSchema() throws
IOException {
+ String otherSchemaName =
GravitinoITUtils.genRandomName("tst_fn_schema_other");
+ createAndInsertSchema(metalakeName, catalogName, otherSchemaName);
+
+ String functionName =
GravitinoITUtils.genRandomName("function_cross_schema_overwrite");
+ Namespace otherNamespace = NamespaceUtil.ofFunction(metalakeName,
catalogName, otherSchemaName);
+ FunctionEntity foreign =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), otherNamespace, functionName,
AUDIT_INFO);
+ FunctionMetaService.getInstance().insertFunction(foreign, false);
+
+ // The overwrite targets a schema that holds no function with this name,
but it carries the ID
+ // of a function stored under another schema. Adopting that row would move
it out of its own
+ // schema without ever fencing that schema, so the write is rejected
instead.
+ Namespace namespace = NamespaceUtil.ofFunction(metalakeName, catalogName,
schemaName);
+ FunctionEntity replacement = copyFunction(foreign, functionName,
namespace, "replacement");
+ assertThrows(
+ EntityAlreadyExistsException.class,
+ () -> FunctionMetaService.getInstance().insertFunction(replacement,
true));
+
+ FunctionEntity stored =
+
FunctionMetaService.getInstance().getFunctionByIdentifier(foreign.nameIdentifier());
+ FunctionPO storedPO =
+
FunctionMetaService.getInstance().getFunctionPOByIdentifier(foreign.nameIdentifier());
+ assertEquals(otherNamespace, stored.namespace());
+ assertEquals(foreign.comment(), stored.comment());
+ assertEquals(1, storedPO.functionCurrentVersion());
+ assertEquals(1, listFunctionVersions(foreign.id()).size());
+ }
+
+ @TestTemplate
+ public void testNaturalKeyOverwriteUsesPersistedFunctionId() throws
IOException {
+ String functionName =
GravitinoITUtils.genRandomName("function_natural_key_overwrite");
+ Namespace namespace = NamespaceUtil.ofFunction(metalakeName, catalogName,
schemaName);
+ FunctionEntity original =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace, functionName,
AUDIT_INFO);
+ FunctionMetaService.getInstance().insertFunction(original, false);
+ FunctionEntity replacement =
+ copyFunction(
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace, functionName,
AUDIT_INFO),
+ functionName,
+ namespace,
+ "replacement");
+
+ FunctionMetaService.getInstance().insertFunction(replacement, true);
+
+ FunctionEntity stored =
+
FunctionMetaService.getInstance().getFunctionByIdentifier(original.nameIdentifier());
+ FunctionPO storedPO =
+
FunctionMetaService.getInstance().getFunctionPOByIdentifier(original.nameIdentifier());
+ assertEquals(original.id(), stored.id());
+ assertEquals("replacement", stored.comment());
+ assertEquals(2, storedPO.functionCurrentVersion());
+ assertEquals(2, listFunctionVersions(original.id()).size());
+ assertTrue(listFunctionVersions(replacement.id()).isEmpty());
+ }
+
+ @TestTemplate
+ public void testNormalReadRequiresCurrentVersionRow() throws IOException {
+ String functionName =
GravitinoITUtils.genRandomName("function_missing_current_version");
+ Namespace namespace = NamespaceUtil.ofFunction(metalakeName, catalogName,
schemaName);
+ FunctionEntity function =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace, functionName,
AUDIT_INFO);
+ FunctionMetaService.getInstance().insertFunction(function, false);
+
+ SessionUtils.doWithCommit(
+ FunctionVersionMetaMapper.class,
+ mapper ->
mapper.softDeleteFunctionVersionsByFunctionId(function.id()));
+
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
FunctionMetaService.getInstance().getFunctionByIdentifier(function.nameIdentifier()));
+ }
+
+ /** A deleted ID is a permanent conflict until GC, rather than a retryable
concurrent write. */
+ @TestTemplate
+ public void testOverwriteRejectsDeletedFunctionId() throws IOException {
+ Namespace ns = NamespaceUtil.ofFunction(metalakeName, catalogName,
schemaName);
+ FunctionEntity function =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), ns, "deleted_function_id",
AUDIT_INFO);
+ FunctionMetaService service = FunctionMetaService.getInstance();
+ service.insertFunction(function, false);
+ assertTrue(service.deleteFunction(function.nameIdentifier()));
+
+ EntityAlreadyExistsException failure =
+ assertThrows(
+ EntityAlreadyExistsException.class, () ->
service.insertFunction(function, true));
+ assertTrue(failure.getMessage().contains("use a new ID"));
+ assertThrows(
+ NoSuchEntityException.class,
+ () -> service.getFunctionByIdentifier(function.nameIdentifier()));
+ listFunctionVersions(function.id()).values().forEach(deletedAt ->
assertTrue(deletedAt > 0));
+
+ FunctionEntity replacement =
+ createFunctionEntity(RandomIdGenerator.INSTANCE.nextId(), ns,
function.name(), AUDIT_INFO);
+ service.insertFunction(replacement, true);
+ assertEquals(replacement.id(),
service.getFunctionByIdentifier(function.nameIdentifier()).id());
+ }
+
+ /** Verifies first-time overwrites either serialize or report a retryable
insert conflict. */
+ @TestTemplate
+ public void testConcurrentOverwriteOfMissingFunction() throws Exception {
+ String name = GravitinoITUtils.genRandomName("function_first_overwrite");
+ Namespace ns = NamespaceUtil.ofFunction(metalakeName, catalogName,
schemaName);
+ FunctionEntity first =
+ createFunctionEntity(RandomIdGenerator.INSTANCE.nextId(), ns, name,
AUDIT_INFO);
+ FunctionEntity second =
+ copyFunction(
+ createFunctionEntity(RandomIdGenerator.INSTANCE.nextId(), ns,
name, AUDIT_INFO),
+ name,
+ ns,
+ "second overwrite");
+ CountDownLatch firstWritten = new CountDownLatch(1);
+ CountDownLatch allowCommit = new CountDownLatch(1);
+ CountDownLatch secondStarted = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> firstResult =
+ executor.submit(
+ () -> {
+ SessionUtils.beginTransaction();
+ try {
+ FunctionMetaService.getInstance().insertFunction(first, true);
+ firstWritten.countDown();
+ await(allowCommit);
+ SessionUtils.commitTransaction();
+ return null;
+ } catch (Throwable failure) {
+ SessionUtils.rollbackTransaction();
+ return failure;
+ }
+ });
+ try {
+ assertTrue(firstWritten.await(30, TimeUnit.SECONDS));
+ Future<Throwable> secondResult =
+ executor.submit(
+ () -> {
+ secondStarted.countDown();
+ try {
+ FunctionMetaService.getInstance().insertFunction(second,
true);
+ return null;
+ } catch (Throwable failure) {
+ return failure;
+ }
+ });
+ assertTrue(secondStarted.await(30, TimeUnit.SECONDS));
+ assertThrows(TimeoutException.class, () -> secondResult.get(500,
TimeUnit.MILLISECONDS));
+ allowCommit.countDown();
+ assertNull(firstResult.get(30, TimeUnit.SECONDS));
+ Throwable failure = secondResult.get(30, TimeUnit.SECONDS);
+ // H2 may serialize at the parent lock; MySQL/PostgreSQL can reach the
missing-row INSERT.
+ if (failure != null) {
+ assertTrue(
+ failure instanceof OptimisticLockException, () -> "Unexpected
failure: " + failure);
+ assertEquals(1, listFunctionVersions(first.id()).size());
+ assertTrue(listFunctionVersions(second.id()).isEmpty());
+ FunctionMetaService.getInstance().insertFunction(second, true);
+ }
+ FunctionEntity stored =
+
FunctionMetaService.getInstance().getFunctionByIdentifier(first.nameIdentifier());
+ assertEquals(first.id(), stored.id());
+ assertEquals("second overwrite", stored.comment());
+ assertEquals(
+ 2,
+ FunctionMetaService.getInstance()
+ .getFunctionPOByIdentifier(first.nameIdentifier())
+ .functionCurrentVersion());
+ assertEquals(2, listFunctionVersions(first.id()).size());
+ assertTrue(listFunctionVersions(second.id()).isEmpty());
+ } finally {
+ allowCommit.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @TestTemplate
+ public void testNaturalKeyOverwriteWaitsForConcurrentRename() throws
Exception {
+ String functionName =
GravitinoITUtils.genRandomName("function_overwrite_rename_race");
+ Namespace namespace = NamespaceUtil.ofFunction(metalakeName, catalogName,
schemaName);
+ FunctionEntity original =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace, functionName,
AUDIT_INFO);
+ FunctionMetaService.getInstance().insertFunction(original, false);
+ FunctionPO observedPO =
+
FunctionMetaService.getInstance().getFunctionPOByIdentifier(original.nameIdentifier());
+ FunctionEntity renamed =
+ copyFunction(original, functionName + "_winner", namespace, "rename
winner");
+ FunctionPO renamedPO =
+ FunctionPO.buildFunctionPO(
+ renamed,
+ FunctionPO.builder()
+ .withMetalakeId(observedPO.metalakeId())
+ .withCatalogId(observedPO.catalogId())
+ .withSchemaId(observedPO.schemaId())
+ .withFunctionLatestVersion(2)
+ .withFunctionCurrentVersion(2),
+ 2);
+ FunctionEntity replacement =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace, functionName,
AUDIT_INFO);
+
+ CountDownLatch renameWritten = new CountDownLatch(1);
+ CountDownLatch allowRenameCommit = new CountDownLatch(1);
+ CountDownLatch overwriteStarted = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> renameResult =
+ executor.submit(
+ () -> {
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ assertEquals(
+ Integer.valueOf(1),
+ SessionUtils.getWithoutCommit(
+ FunctionMetaMapper.class,
+ mapper -> mapper.updateFunctionMeta(renamedPO,
observedPO))),
+ () ->
+ SessionUtils.doWithoutCommit(
+ FunctionVersionMetaMapper.class,
+ mapper ->
+
mapper.insertFunctionVersionMeta(renamedPO.functionVersionPO())),
+ () -> {
+ renameWritten.countDown();
+ await(allowRenameCommit);
+ });
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+
+ try {
+ assertTrue(renameWritten.await(30, TimeUnit.SECONDS));
+ Future<Throwable> overwriteResult =
+ executor.submit(
+ () -> {
+ overwriteStarted.countDown();
+ try {
+
FunctionMetaService.getInstance().insertFunction(replacement, true);
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+ assertTrue(overwriteStarted.await(30, TimeUnit.SECONDS));
+ assertThrows(TimeoutException.class, () -> overwriteResult.get(500,
TimeUnit.MILLISECONDS));
+
+ allowRenameCommit.countDown();
+ assertNull(renameResult.get(30, TimeUnit.SECONDS));
+ assertNull(overwriteResult.get(30, TimeUnit.SECONDS));
+ } finally {
+ allowRenameCommit.countDown();
+ executor.shutdownNow();
+ }
+
+ assertEquals(
+ original.id(),
+
FunctionMetaService.getInstance().getFunctionByIdentifier(renamed.nameIdentifier()).id());
+ assertEquals(
+ replacement.id(),
+ FunctionMetaService.getInstance()
+ .getFunctionByIdentifier(replacement.nameIdentifier())
+ .id());
}
private int countActiveOwnerRelForMetadataObject(
@@ -677,6 +1380,71 @@ public class TestFunctionMetaService extends
TestJDBCBackend {
}
}
+ private void assertMovedFunctionHistoryCascade(String parent, boolean
sourceFirst)
+ throws IOException {
+ String sourceMetalake = "source_function_" + parent + "_" + sourceFirst;
+ String destinationMetalake = "destination_function_" + parent + "_" +
sourceFirst;
+ String catalog = "history_catalog";
+ String schema = "history_schema";
+ for (String metalake : new String[] {sourceMetalake, destinationMetalake})
{
+ createAndInsertMakeLake(metalake);
+ createAndInsertCatalog(metalake, catalog);
+ createAndInsertSchema(metalake, catalog, schema);
+ }
+ FunctionMetaService service = FunctionMetaService.getInstance();
+ FunctionEntity original =
+ createFunctionEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofFunction(sourceMetalake, catalog, schema),
+ "moved_history",
+ AUDIT_INFO);
+ service.insertFunction(original, false);
+ FunctionEntity moved =
+ copyFunction(
+ original,
+ original.name(),
+ NamespaceUtil.ofFunction(destinationMetalake, catalog, schema),
+ "moved");
+ service.updateFunction(original.nameIdentifier(), ignored -> moved);
+ assertEquals(2, listFunctionVersions(original.id()).size());
+
+ if (sourceFirst) {
+ deleteHistoryParent(parent, sourceMetalake, catalog, schema);
+ assertEquals(original.id(),
service.getFunctionByIdentifier(moved.nameIdentifier()).id());
+ assertEquals(2, listFunctionVersions(original.id()).size());
+ listFunctionVersions(original.id())
+ .values()
+ .forEach(deletedAt -> assertEquals(0L, deletedAt.longValue()));
+ }
+
+ deleteHistoryParent(parent, destinationMetalake, catalog, schema);
+ assertThrows(
+ NoSuchEntityException.class, () ->
service.getFunctionByIdentifier(moved.nameIdentifier()));
+ assertEquals(2, listFunctionVersions(original.id()).size());
+ listFunctionVersions(original.id()).values().forEach(deletedAt ->
assertTrue(deletedAt > 0));
+ }
+
+ private void deleteHistoryParent(String parent, String metalake, String
catalog, String schema) {
+ switch (parent) {
+ case "schema":
+ assertTrue(
+ SchemaMetaService.getInstance()
+ .deleteSchema(NameIdentifier.of(metalake, catalog, schema),
true));
+ break;
+ case "catalog":
+ assertTrue(
+ CatalogMetaService.getInstance()
+ .deleteCatalog(NameIdentifier.of(metalake, catalog), true));
+ break;
+ case "metalake":
+ assertTrue(
+
MetalakeMetaService.getInstance().deleteMetalake(NameIdentifier.of(metalake),
true));
+ break;
+ default:
+ throw new AssertionError("Unexpected parent: " + parent);
+ }
+ }
+
private Map<Integer, Long> listFunctionVersions(Long functionId) {
Map<Integer, Long> versionDeletedTime = new HashMap<>();
try (SqlSession sqlSession =
@@ -698,10 +1466,15 @@ public class TestFunctionMetaService extends
TestJDBCBackend {
}
private FunctionEntity copyFunctionWithComment(FunctionEntity function,
String comment) {
+ return copyFunction(function, function.name(), function.namespace(),
comment);
+ }
+
+ private FunctionEntity copyFunction(
+ FunctionEntity function, String name, Namespace namespace, String
comment) {
return FunctionEntity.builder()
.withId(function.id())
- .withName(function.name())
- .withNamespace(function.namespace())
+ .withName(name)
+ .withNamespace(namespace)
.withComment(comment)
.withFunctionType(function.functionType())
.withDeterministic(function.deterministic())
@@ -710,6 +1483,15 @@ public class TestFunctionMetaService extends
TestJDBCBackend {
.build();
}
+ private void await(CountDownLatch latch) {
+ try {
+ assertTrue(latch.await(30, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
private void assertVersionActive(Map<Integer, Long> versionDeletedMap, int
version) {
assertTrue(versionDeletedMap.containsKey(version));
assertEquals(0L, versionDeletedMap.get(version));
