[#12451] improvement(core): add OCC for metalake writes - #12454
Open
yuqi1129 wants to merge 2 commits into
Open
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
Introduces database-backed optimistic concurrency control (OCC) for metalake writes in the relational store, ensuring version-guarded alters/deletes and transactional boundaries for cascade operations so concurrent writers are surfaced as conflicts instead of silently overwriting metadata.
Changes:
- Advance metalake OCC version on every update and guard metalake update/delete with version compare-and-set (CAS) semantics.
- Add shared
ExceptionUtilsfactories for consistentOptimisticLockExceptionmessages and update metalake manager to preserve idempotent drop behavior when a concurrent delete wins. - Add CAS-based cascade deletion helpers that lock catalog rows and soft-delete catalogs/schemas using identifier-and-version pairs, plus new unit tests.
Reviewed changes
Copilot reviewed 20 out of 20 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java | Treat concurrent underlying store delete (NoSuchEntityException) as an idempotent false drop result. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaMapper.java | Add mapper APIs for locking catalog rows and CAS soft-deleting catalogs by (id, version). |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaSQLProviderFactory.java | Expose SQL provider hooks for catalog row locking and CAS delete-by-version. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaMapper.java | Add selectMetalakeMetaByIdForUpdate and versioned soft-delete signature for metalakes. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaSQLProviderFactory.java | Wire provider factory methods for FOR UPDATE metalake select and versioned soft delete. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaMapper.java | Add list-by-metalake API and CAS soft-delete-by-version for schemas; remove metalake-wide schema soft-delete. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.java | Add provider factory methods for list-by-metalake and CAS soft-delete-by-version. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java | Add FOR UPDATE catalog listing and CAS delete-by-version SQL generation. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/MetalakeMetaBaseSQLProvider.java | Add FOR UPDATE metalake select; update soft-delete to require matching current_version. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java | Add list schemas by metalake and CAS delete-by-version SQL generation; adjust catalog-wide soft delete. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java | Implement PostgreSQL-specific CAS delete-by-version for catalogs. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/MetalakeMetaPostgreSQLProvider.java | Update metalake soft-delete to be version-guarded; update update statement to version-based CAS. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SchemaMetaPostgreSQLProvider.java | Implement PostgreSQL-specific CAS delete-by-version for schemas; adjust catalog-wide soft delete SQL. |
| core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java | Enforce metalake update/delete CAS behavior, transactional fencing, and CAS-based cascade deletion with locking. |
| core/src/main/java/org/apache/gravitino/storage/relational/utils/ExceptionUtils.java | Add shared OptimisticLockException factory helpers for concurrent modification scenarios. |
| core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java | Make metalake updates always advance OCC token (current/last versions aligned). |
| core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java | Add test verifying drop returns false when a concurrent delete wins (store throws NoSuchEntityException). |
| core/src/test/java/org/apache/gravitino/storage/relational/service/TestMetalakeMetaService.java | Add tests for version-based CAS behavior and conflict/missing classification paths. |
| core/src/test/java/org/apache/gravitino/storage/relational/utils/TestExceptionUtils.java | Add unit tests for the new shared optimistic-lock exception messages. |
| core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java | Update converter tests to assert metalake version increments on update. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
yuqi1129
force-pushed
the
feat/12342-occ-metalake
branch
from
August 13, 2026 07:41
e7224b4 to
b13af59
Compare
Advance the metalake OCC version on every alter and guard alter and delete with a compare-and-set on the observed version, classifying a failed CAS as either a stale conflict or a missing entity. Keep the metalake root CAS and the non-empty check or the cascade cleanup inside one database transaction. A cascade locks the catalog rows first, then compare-and-set deletes descendant catalogs and schemas with their observed identifier-and-version pairs, so a concurrent child write is reported instead of silently dropped. Also add the shared OptimisticLockException factories used by the follow-up catalog and schema changes.
yuqi1129
force-pushed
the
feat/12342-occ-metalake
branch
from
August 13, 2026 08:07
b13af59 to
29db0ce
Compare
Contributor
Author
|
Thanks for the review. Responses to the three Copilot comments: 1. The lock is load-bearing for correctness, not incidental. Under MySQL REPEATABLE READ a plain It also does not add waiting in practice: the compare-and-set is an 2 & 3. Row-value
Pushed the clarifying comment for (1); the same note is applied to the equivalent paths in #12455 and #12456. |
Contributor
Author
|
@jerryshao |
Code Coverage Report
Files
|
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 20 out of 20 changed files in this pull request and generated no new comments.
Suppressed comments (2)
core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java:449
- Passing every schema in the metalake to one generated
UPDATEuses two bind parameters per row. PostgreSQL and MySQL reject prepared statements beyond 65,535 parameters, so 32,768 schemas make cascade deletion fail, with the largeORtree potentially failing sooner. Process this snapshot in bounded chunks in the existing transaction and compare the total affected rows with the snapshot size.
int deleted =
SessionUtils.getWithoutCommit(
SchemaMetaMapper.class, mapper -> mapper.softDeleteSchemaMetasWithVersion(schemaPOs));
core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java:430
- This sends the entire catalog snapshot as one generated
UPDATE, with two bind parameters per catalog. PostgreSQL and MySQL cap prepared statements at 65,535 parameters, so a metalake with 32,768 catalogs cannot be cascade-deleted; the longORexpression may hit parser or packet limits even earlier. Execute the CAS deletes in bounded chunks within this same transaction and sum the affected-row counts before committing.
This issue also appears on line 447 of the same file.
int deleted =
SessionUtils.getWithoutCommit(
CatalogMetaMapper.class,
mapper -> mapper.softDeleteCatalogMetasWithVersion(catalogPOs));
jerryshao
reviewed
Aug 14, 2026
Comment on lines
+324
to
+333
| () -> { | ||
| deleteMetalakeWithVersion(ident, metalakeId, currentVersion); | ||
| List<CatalogPO> catalogPOs = | ||
| SessionUtils.getWithoutCommit( | ||
| CatalogMetaMapper.class, | ||
| mapper -> mapper.listCatalogPOsByMetalakeId(metalakeId)); | ||
| if (!catalogPOs.isEmpty()) { | ||
| throw new NonEmptyEntityException( | ||
| "Entity %s has sub-entities, you should remove sub-entities first", ident); | ||
| } |
Contributor
There was a problem hiding this comment.
Why do we delete the metalake first, then check if the catalog is empty or not?
jerryshao
reviewed
Aug 14, 2026
| Entity.EntityType.SCHEMA, Entity.EntityType.METALAKE, metalakeIdentifier); | ||
| } | ||
| } | ||
|
|
Contributor
There was a problem hiding this comment.
I don't what's the performance now after changing to OCC. My feeling is that we add more db operations compared to before.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Add database-backed optimistic concurrency control and transaction boundaries for metalake writes.
JDBCBackendmutation boundary after [#12151] improvement(core): complete cached entity change-log coverage #12374, so this PR no longer touches it.OptimisticLockExceptionfactories used by the follow-up catalog and schema PRs.Rebased on current
main(on top of #12374). This is the first of three PRs that replace #12350, which reviewers found too large. The stack is metalake -> catalog -> schema; each PR is independently green. Two cross-entity tests inTestMetalakeMetaService(concurrent schema alter during a metalake cascade, and metalake cascade racing a schema create) land with the schema PR, because the behaviour they assert only exists once schema writes take the catalog row lock and bump the schema version.Why are the changes needed?
Managed metalake operations previously consisted of multiple independent reads and writes. Concurrent alter and delete requests could overwrite newer metadata, and a cascade delete could run partial cleanup while another writer was still modifying descendants.
Fix: #12451
Does this PR introduce any user-facing change?
Concurrent metalake version conflicts are reported as HTTP 409. If the observed entity was deleted or renamed away, alter reports not found and drop preserves its idempotent false result.
How was this patch tested?
./gradlew :core:test :core:javadoc :catalogs:catalog-fileset:test :catalogs:catalog-kafka:test -PskipITs(H2)TestMetalakeMetaService,TestMetalakeManager,TestExceptionUtils,TestPOConverters.<foreach>CAS delete and the PostgreSQL lock syntax is left to CI (-PskipDockerTests=false).