Skip to content

[#12576] improvement(core): optimize schema write fencing #12577

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
yuqi1129 wants to merge 11 commits into
base: main
Choose a base branch
from
Open

[#12576] improvement(core): optimize schema write fencing #12577

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
791f5e9
[#12453] improvement(core): add OCC for schema writes
yuqi1129 Aug 13, 2026
0192405
[#12453] docs(core): say the H2 shared-lock fallback affects H2 backends
yuqi1129 Aug 17, 2026
7011c5d
[#12453] docs(core): explain the schema OCC and locking rules in code
yuqi1129 Aug 17, 2026
cc97151
[#12453] improvement(core): keep the schema OCC version monotonic on …
yuqi1129 Aug 19, 2026
ce9eb5d
[#12453] fix(core): lock schema for model version writes
yuqi1129 Aug 21, 2026
d8b75fe
[#12453] fix(core): close schema write race gaps
yuqi1129 Aug 24, 2026
10a17fe
[#12576] improvement(core): optimize schema write fencing
yuqi1129 Aug 24, 2026
0ef7195
Merge branch 'main' into improve/12576-schema-write-fencing
yuqi1129 Aug 24, 2026
7f9347a
Merge upstream/main into improve/12576-schema-write-fencing
yuqi1129 Aug 26, 2026
b265ad6
[#12576] refactor(core): Move schema child query to provider
yuqi1129 Aug 27, 2026
274868d
Merge upstream/main into improve/12576-schema-write-fencing
yuqi1129 Aug 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 31 additions & 0 deletions core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.ibatis.annotations.DeleteProvider;
import org.apache.ibatis.annotations.InsertProvider;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
import org.apache.ibatis.annotations.SelectProvider;
import org.apache.ibatis.annotations.UpdateProvider;

Expand Down Expand Up @@ -86,6 +87,36 @@ SchemaPO selectSchemaByFullQualifiedName(
@SelectProvider(type = SchemaMetaSQLProviderFactory.class, method = "selectSchemaMetaById")
SchemaPO selectSchemaMetaById(@Param("schemaId") Long schemaId);

/**
* Returns one when an active table, view, fileset, function, model, or topic exists in the
* schema, and {@code null} otherwise.
*
* <p>Only a literal is selected because callers need an existence answer, not complete child
* metadata. The final limit also lets the database stop as soon as it finds the first child.
*/
@Select({
"SELECT 1 FROM "
+ TableMetaMapper.TABLE_NAME
+ " WHERE schema_id = #{schemaId} AND deleted_at = 0",
"UNION ALL SELECT 1 FROM "
+ ViewMetaMapper.TABLE_NAME
+ " WHERE schema_id = #{schemaId} AND deleted_at = 0",
"UNION ALL SELECT 1 FROM "
+ FilesetMetaMapper.META_TABLE_NAME
+ " WHERE schema_id = #{schemaId} AND deleted_at = 0",
"UNION ALL SELECT 1 FROM "
+ FunctionMetaMapper.TABLE_NAME
+ " WHERE schema_id = #{schemaId} AND deleted_at = 0",
"UNION ALL SELECT 1 FROM "
+ ModelMetaMapper.TABLE_NAME
+ " WHERE schema_id = #{schemaId} AND deleted_at = 0",
"UNION ALL SELECT 1 FROM "
+ TopicMetaMapper.TABLE_NAME
+ " WHERE schema_id = #{schemaId} AND deleted_at = 0",
"LIMIT 1"
})

@jerryshao jerryshao Aug 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shall we use our convention to define the SQL string in the provider, rather than add a notation here directly?

@yuqi1129 yuqi1129 Aug 27, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated.

Integer selectActiveChildBySchemaId(@Param("schemaId") Long schemaId);

/** Selects and locks an active schema by ID for the current transaction. */
@SelectProvider(
type = SchemaMetaSQLProviderFactory.class,
Expand Down
113 changes: 63 additions & 50 deletions core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,36 +165,33 @@ public void insertFileset(FilesetEntity filesetEntity, boolean overwrite) throws
FilesetPO po = POConverters.initializeFilesetPOWithVersion(filesetEntity, builder);

// insert both fileset meta table and version table
SessionUtils.doMultipleWithCommit(
// Hold the parent schema row until this transaction ends, so the fileset cannot be
// written below a schema that is being dropped.
() ->
SchemaMetaService.getInstance()
.lockSchemaForEntityWrite(
filesetEntity.nameIdentifier(),
po.getSchemaId(),
po.getCatalogId(),
po.getMetalakeId()),
() ->
SessionUtils.doWithoutCommit(
FilesetMetaMapper.class,
mapper -> {
if (overwrite) {
mapper.insertFilesetMetaOnDuplicateKeyUpdate(po);
} else {
mapper.insertFilesetMeta(po);
}
}),
() ->
SessionUtils.doWithoutCommit(
FilesetVersionMapper.class,
mapper -> {
if (overwrite) {
mapper.insertFilesetVersionsOnDuplicateKeyUpdate(po.getFilesetVersionPOs());
} else {
mapper.insertFilesetVersions(po.getFilesetVersionPOs());
}
}));
SchemaMetaService.getInstance()
.doWithSchemaWriteLock(
filesetEntity.nameIdentifier(),
po.getSchemaId(),
po.getCatalogId(),
po.getMetalakeId(),
() ->
SessionUtils.doWithoutCommit(
FilesetMetaMapper.class,
mapper -> {
if (overwrite) {
mapper.insertFilesetMetaOnDuplicateKeyUpdate(po);
} else {
mapper.insertFilesetMeta(po);
}
}),
() ->
SessionUtils.doWithoutCommit(
FilesetVersionMapper.class,
mapper -> {
if (overwrite) {
mapper.insertFilesetVersionsOnDuplicateKeyUpdate(
po.getFilesetVersionPOs());
} else {
mapper.insertFilesetVersions(po.getFilesetVersionPOs());
}
}));
} catch (RuntimeException re) {
ExceptionUtils.checkSQLException(
re, Entity.EntityType.FILESET, filesetEntity.nameIdentifier().toString());
Expand Down Expand Up @@ -231,22 +228,33 @@ public <E extends Entity & HasIdentifier> FilesetEntity updateFileset(
// back — including the version insert — and the update is treated as a conflict.
int[] metaUpdateCountRef = new int[1];
try {
SessionUtils.doMultipleWithCommit(
() ->
SessionUtils.doWithoutCommit(
FilesetVersionMapper.class,
mapper -> mapper.insertFilesetVersions(newFilesetPO.getFilesetVersionPOs())),
() -> {
metaUpdateCountRef[0] =
SessionUtils.getWithoutCommit(
FilesetMetaMapper.class,
mapper -> mapper.updateFilesetMeta(newFilesetPO, oldFilesetPO));
if (metaUpdateCountRef[0] == 0) {
throw new RuntimeException("Failed to update the entity: " + identifier);
}
});
SchemaMetaService.getInstance()
.doWithSchemaWriteLock(
identifier,
oldFilesetPO.getSchemaId(),
oldFilesetPO.getCatalogId(),
oldFilesetPO.getMetalakeId(),
() ->
SessionUtils.doWithoutCommit(
FilesetVersionMapper.class,
mapper ->
mapper.insertFilesetVersions(newFilesetPO.getFilesetVersionPOs())),
() -> {
metaUpdateCountRef[0] =
SessionUtils.getWithoutCommit(
FilesetMetaMapper.class,
mapper -> mapper.updateFilesetMeta(newFilesetPO, oldFilesetPO));
if (metaUpdateCountRef[0] == 0) {
throw new RuntimeException("Failed to update the entity: " + identifier);
}
});
updateResult = 1;
} catch (RuntimeException re) {
// The schema fence runs before the fileset update. Keep its missing-schema error rather
// than turning it into a fileset write conflict merely because the update count is zero.
if (re instanceof NoSuchEntityException) {
throw re;
}
if (metaUpdateCountRef[0] == 0) {
// The meta update matched no rows; the transaction was rolled back,
// including the version insert above.
Expand All @@ -259,12 +267,17 @@ public <E extends Entity & HasIdentifier> FilesetEntity updateFileset(
}
} else {
int[] metaUpdateCountRef = new int[1];
SessionUtils.doMultipleWithCommit(
() ->
metaUpdateCountRef[0] =
SessionUtils.getWithoutCommit(
FilesetMetaMapper.class,
mapper -> mapper.updateFilesetMeta(newFilesetPO, oldFilesetPO)));
SchemaMetaService.getInstance()
.doWithSchemaWriteLock(
identifier,
oldFilesetPO.getSchemaId(),
oldFilesetPO.getCatalogId(),
oldFilesetPO.getMetalakeId(),
() ->
metaUpdateCountRef[0] =
SessionUtils.getWithoutCommit(
FilesetMetaMapper.class,
mapper -> mapper.updateFilesetMeta(newFilesetPO, oldFilesetPO)));
updateResult = metaUpdateCountRef[0];
}
} catch (RuntimeException re) {
Expand Down
93 changes: 43 additions & 50 deletions core/src/main/java/org/apache/gravitino/storage/relational/service/FunctionMetaService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,29 +114,26 @@ public void insertFunction(FunctionEntity functionEntity, boolean overwrite) thr
fillFunctionPOBuilderParentEntityId(builder, functionEntity.namespace());
FunctionPO po = initializeFunctionPO(functionEntity, builder);

SessionUtils.doMultipleWithCommit(
// Hold the parent schema row until this transaction ends, so the function cannot be
// written below a schema that is being dropped.
() ->
SchemaMetaService.getInstance()
.lockSchemaForEntityWrite(
functionEntity.nameIdentifier(),
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());
}
}));
SchemaMetaService.getInstance()
.doWithSchemaWriteLock(
functionEntity.nameIdentifier(),
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());
}
}));
} catch (RuntimeException re) {
ExceptionUtils.checkSQLException(
re, Entity.EntityType.FUNCTION, functionEntity.nameIdentifier().toString());
Expand Down Expand Up @@ -271,33 +268,29 @@ public <E extends Entity & HasIdentifier> FunctionEntity updateFunction(
try {
FunctionPO newFunctionPO = updateFunctionPO(oldFunctionPO, newEntity);
// Insert a new version and update function meta
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.
() ->
SchemaMetaService.getInstance()
.lockSchemaForEntityWrite(
identifier,
oldFunctionPO.schemaId(),
oldFunctionPO.catalogId(),
oldFunctionPO.metalakeId()),
() ->
SessionUtils.doWithoutCommit(
FunctionVersionMetaMapper.class,
mapper -> mapper.insertFunctionVersionMeta(newFunctionPO.functionVersionPO())),
() -> {
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);
}
});
SchemaMetaService.getInstance()
.doWithSchemaWriteLock(
identifier,
oldFunctionPO.schemaId(),
oldFunctionPO.catalogId(),
oldFunctionPO.metalakeId(),
() ->
SessionUtils.doWithoutCommit(
FunctionVersionMetaMapper.class,
mapper ->
mapper.insertFunctionVersionMeta(newFunctionPO.functionVersionPO())),
() -> {
int updated =
SessionUtils.getWithoutCommit(
FunctionMetaMapper.class,
mapper -> ops.updatePO(mapper, newFunctionPO, oldFunctionPO));
if (updated == 0) {
// The version was inserted above. Throwing here rolls it back instead of leaving
// an active version without function metadata.
throw ExceptionUtils.concurrentModification(
Entity.EntityType.FUNCTION, identifier);
}
});

return newEntity;
} catch (RuntimeException re) {
Expand Down
79 changes: 40 additions & 39 deletions core/src/main/java/org/apache/gravitino/storage/relational/service/ModelMetaService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,26 +98,22 @@ public void insertModel(ModelEntity modelEntity, boolean overwrite) throws IOExc
fillModelPOBuilderParentEntityId(builder, modelEntity.namespace());
ModelPO po = POConverters.initializeModelPO(modelEntity, builder);

SessionUtils.doMultipleWithCommit(
// Hold the parent schema row until this transaction ends, so the model cannot be
// written below a schema that is being dropped.
() ->
SchemaMetaService.getInstance()
.lockSchemaForEntityWrite(
modelEntity.nameIdentifier(),
po.getSchemaId(),
po.getCatalogId(),
po.getMetalakeId()),
() ->
SessionUtils.doWithoutCommit(
ModelMetaMapper.class,
mapper -> {
if (overwrite) {
mapper.insertModelMetaOnDuplicateKeyUpdate(po);
} else {
mapper.insertModelMeta(po);
}
}));
SchemaMetaService.getInstance()
.doWithSchemaWriteLock(
modelEntity.nameIdentifier(),
po.getSchemaId(),
po.getCatalogId(),
po.getMetalakeId(),
() ->
SessionUtils.doWithoutCommit(
ModelMetaMapper.class,
mapper -> {
if (overwrite) {
mapper.insertModelMetaOnDuplicateKeyUpdate(po);
} else {
mapper.insertModelMeta(po);
}
}));
} catch (RuntimeException re) {
ExceptionUtils.checkSQLException(
re, Entity.EntityType.MODEL, modelEntity.nameIdentifier().toString());
Expand Down Expand Up @@ -393,26 +389,31 @@ public <E extends Entity & HasIdentifier> ModelEntity updateModel(

AtomicInteger updateResult = new AtomicInteger(0);
try {
SessionUtils.doMultipleWithCommit(
() ->
updateResult.set(
SessionUtils.getWithoutCommit(
ModelMetaMapper.class,
SchemaMetaService.getInstance()
.doWithSchemaWriteLock(
identifier,
oldModelPO.getSchemaId(),
oldModelPO.getCatalogId(),
oldModelPO.getMetalakeId(),
() ->
updateResult.set(
SessionUtils.getWithoutCommit(
ModelMetaMapper.class,
mapper ->
mapper.updateModelMeta(
POConverters.updateModelPO(oldModelPO, newEntity), oldModelPO))),
() -> {
if (isRenamed && updateResult.get() > 0) {
SessionUtils.doWithoutCommit(
EntityChangeLogMapper.class,
mapper ->
mapper.updateModelMeta(
POConverters.updateModelPO(oldModelPO, newEntity), oldModelPO))),
() -> {
if (isRenamed && updateResult.get() > 0) {
SessionUtils.doWithoutCommit(
EntityChangeLogMapper.class,
mapper ->
mapper.insertEntityChange(
metalakeName,
Entity.EntityType.MODEL.name(),
oldFullName,
OperateType.ALTER));
}
});
mapper.insertEntityChange(
metalakeName,
Entity.EntityType.MODEL.name(),
oldFullName,
OperateType.ALTER));
}
});
} catch (RuntimeException re) {
ExceptionUtils.checkSQLException(
re, Entity.EntityType.MODEL, newEntity.nameIdentifier().toString());
Expand Down
Loading
Loading