[#12345] improvement(core): add OCC for table writes - #12551
Conversation
Advance the schema OCC version on every alter and guard alter and drop with a compare-and-set on the observed version, classifying a failed CAS as either a stale conflict or a missing entity. Make managed schema creation insert-only so a concurrent same-name create returns SchemaAlreadyExistsException instead of overwriting the winner, and take a shared lock on the parent catalog row so a schema cannot be created below a catalog that is being dropped. Serialize hierarchical ancestor materialization and schema drops through the catalog row so overlapping cascades share one lock order. Lock the parent schema row before writing a table, view, fileset, function, model, or topic, and check views and functions before a non-cascade schema drop. Accepted tradeoff: a hierarchical schema create that materializes implicit ancestors takes an exclusive lock on the catalog row, because two concurrent creates can both find the same ancestor missing and both insert it, and a shared lock does not prevent that under MySQL REPEATABLE READ.
H2 is also the default embedded backend, not only a test backend. Spell out that falling back to an exclusive lock serializes schema creations under one catalog there and can surface as an H2 lock timeout.
Review feedback: the concurrency-critical parts need comments so a reader can follow why the statements are ordered the way they are. - Say what the catalog row lock buys on a schema create, and why a nested name has to take it exclusively while a plain name does not. - Say why both drop paths delete the schema row before looking at its children, and why every drop takes catalog before schema. - Say what the shared schema lock in front of a table, view, fileset, function, model, or topic write is for, and that only a cross-schema rename needs it. - Say why the alter UPDATE compares only the version, what zero affected rows can mean, and why a partial cascade must roll back. - Say why managed schema creation is insert-only now. - Correct the schemaWriteFailure comment: sessions run at READ_COMMITTED, so the locking read is there to wait out an in-flight writer.
…overwrite Carry the fix that #12455 already made for catalogs over to schemas, so both sides of the hierarchy follow the same rule. - Advance current_version on all four schema upsert paths (single and batch, on MySQL/H2 and PostgreSQL) instead of writing the initial version back, which would let a writer holding an older version still pass its own version check. - Name the table on the PostgreSQL assignments: a bare column on that side of ON CONFLICT is ambiguous there, which is how the previous CI run broke. - Add TestSchemaMetaPostgreSQLProvider to pin both rules without a database, so they are checked on every run and not only in the Docker-backed CI job. - Cover the race this PR is meant to close: a catalog cascade that holds the catalog row makes a concurrent schema create wait and then report the catalog as missing, leaving no orphan behind. Reading the cascade snapshot moved into a package-private method so the test can pause exactly at that point, the same seam MetalakeMetaService already offers. - Say that the H2 shared-lock fallback affects H2 backends, not just tests.
Code Coverage Report
Files
|
jerryshao
left a comment
jerryshao
left a comment
There was a problem hiding this comment.
Automated review (Claude Code) of the OCC-for-table-writes change. Left inline notes on a few spots worth a look before merge.
One more that doesn't map to a changed line (so can't be left inline): TableOperationDispatcher.dropTable/purgeTable call store.delete() inside their own generic catch (Exception e) { throw new RuntimeException(e); }, which would swallow the new OptimisticLockException that TableMetaService.deleteTable can now throw (e.g. a concurrent alterTable bumping current_version between the read and the delete transaction). The alter path was explicitly wired to propagate this signal — should drop/purge do the same instead of flattening it into a plain RuntimeException?
Cleanup pass over the OCC change, no behavior change. - Give TablePO a copy builder so tablePOWithPersistedVersions swaps the two version columns instead of restating all sixteen. Restating them meant a column added later would be silently blanked on the overwrite path, with no compile error to catch it. - Say why selectTableMetaByIdForUpdate cannot be written as selectTableMetaById(id) + " FOR UPDATE" like the metalake, catalog and schema providers: that select LEFT JOINs table_version_info, and locking the nullable side of an outer join is rejected by PostgreSQL and locks the wrong rows on MySQL. - Drop a redundant snapshot read and a single-statement lambda block. - Build the PostgreSQL SQL assertions from TableMetaMapper.TABLE_NAME and add the unqualified-reference guard, matching the catalog and schema tests. - Collapse the repeated test scaffolding: one column() factory, one copyTableWithColumns, one assertion helper for the two stale-alter cases, and one runWhileSchemaDeleteUncommitted for the two schema-delete races.
…lter paths operateOnEntity is a best-effort helper: all nine call sites treat a null result as "the store write did not land, carry on". Letting an OptimisticLockException out of it turned that into a hard failure for every caller, including the read paths. That is too blunt. On the load and import paths the external catalog is the source of truth, so failing to write the Gravitino copy is allowed and the next load repairs it. The worst case was updateColumnsIfNecessaryWhenLoad, reached from loadTable: two concurrent loads of a table whose columns drifted made the loser fail a plain read, which is the very race the Lance repair retry exists to absorb. Keep operateOnEntity best-effort and add operateOnEntityAndPropagateConflict for the four alter paths, where the store write is the operation the user asked for and reporting success on a lost race would hide the loss. The conflict is now logged on the best-effort paths instead of being lost among generic failures, and a new entity type added later has to opt in to hard failure.
The old comment said the two rows must move together and that ordering keeps a losing writer from overwriting the winner's version row, but not why the overwrite would happen in the first place, which made it easy to read as if the version row were protected on its own. It is not: the upsert is keyed by (table_id, version) and has no version predicate. Name the key, say the upsert is unguarded, and walk through two writers racing from version 5 to 6 so the reason the CAS has to come first is on the page.
|
Also addressed the non-inline review note in 1d874db: TableOperationDispatcher.dropTable and purgeTable now rethrow OptimisticLockException before the generic exception wrapper, preserving the conflict contract. Both paths have dispatcher regression coverage. |
| if (droppedFromCatalog) { | ||
| try { | ||
| store.delete(ident, TABLE); | ||
| } catch (OptimisticLockException e) { |
There was a problem hiding this comment.
Correctness: concurrent-alter-vs-drop race can permanently orphan the internal table_meta row.
Sequence: (1) doWithCatalog(...).dropTable(ident) succeeds against the external catalog (droppedFromCatalog = true, external data is now gone); (2) store.delete(ident, TABLE) loses its new version CAS (softDeleteTableMetasByTableId now requires current_version = #{currentVersion}) because another writer concurrently altered the table between the initial read and this delete, so it throws OptimisticLockException; (3) that exception now propagates straight out of dropTable instead of being retried.
Because the external table is already gone, a client retry of the same dropTable call will have doWithCatalog(...).dropTable(ident) report "not found" (see e.g. HiveCatalogOperations.dropTable, which returns false when the table is already absent), so droppedFromCatalog becomes false on retry and the if (droppedFromCatalog) { store.delete(...) } block — the only place that ever calls store.delete — is skipped entirely. The internal Gravitino table_meta row is then permanently orphaned; no later drop attempt, and no existing reconciliation job (OrphanedSchemaCleanup only targets schemas, not tables), will ever remove it.
Contrast with alterTable, which deliberately swallows the same kind of conflict via OperationDispatcher.operateOnEntity with an explicit retry-safety rationale ("failing the request would encourage a retry that could apply the external change twice"). Drop/purge propagate instead, but retry-safety is actually worse here — it silently blocks all future cleanup of the internal entity. The same issue applies to purgeTable below (line 476).
There was a problem hiding this comment.
The sequence is real. I opened #12597 to handle it for every entity type instead of patching this one path.
Two things pushed me that way. Removing the catch would not fix it: without it the conflict falls into catch (Exception e) and is rethrown as RuntimeException, so the request still fails and the retry still no-ops. The orphan comes from the CAS being able to fail at all. And the same shape is in dropSchema and dropView; schema already has OCC, so dropSchema can hit this today, and it has no OptimisticLockException catch, so a schema conflict surfaces as a generic failure rather than a conflict. dropTopic differs again - its store.delete is not gated on droppedFromCatalog, so a retry does re-attempt it.
There is also a design question underneath: retrying the delete until it wins is close to not checking the version on that path at all, since delete is idempotent and a conflict only means the row moved on. #12597 lists that alongside a shared retry and an orphan-cleanup job. Note the block already documents that an out-of-band drop can leave a stale registration needing separate cleanup, so this is a new trigger for an accepted outcome rather than a new class of outcome.
| tablePO.getTableId(), tablePO.getCurrentVersion())); | ||
| } | ||
|
|
||
| private RuntimeException tableWriteFailure(NameIdentifier identifier, TablePO observedTablePO) { |
There was a problem hiding this comment.
Reuse: this classification algorithm is now duplicated a third time.
tableWriteFailure (lock-and-reread via selectTableMetaByIdForUpdate, compare natural-key fields, return NoSuchEntityException or ExceptionUtils.concurrentModification) is essentially a line-for-line copy of SchemaMetaService.schemaWriteFailure and CatalogMetaService.catalogWriteFailure (added by the earlier schema-OCC PR). No shared helper was extracted, so a future fix to the classification logic (e.g. adding a field to compare, or changing lock semantics) has to be made identically in three places — and it already wasn't kept in sync once (the schema/catalog versions and this one already differ slightly in structure). Consider extracting one generic writeFailure(NameIdentifier, T observed, Function<Long,T> lockingLookup, BiPredicate<T,T> sameParent, EntityType) helper that all three services call.
There was a problem hiding this comment.
Agreed on the duplication. I would rather extract it when the series is done than in this PR: what genuinely differs per entity is the mapper and its locking select, which identity columns to compare (metalake has no parent, table compares name plus schema, catalog and metalake ids), and schema's physicalToLogicalSchemaPO conversion before comparing. Doing it now means refactoring three already-merged services from inside a fourth. I will note the deferral in the PR description so the fifth copy does not land by default.
There was a problem hiding this comment.
Can you create an issue for this?
| R ret = null; | ||
| try { | ||
| ret = fn.apply(ident); | ||
| } catch (OptimisticLockException e) { |
There was a problem hiding this comment.
Altitude: this special case depends on an unenforced fact about callers.
The comment concedes operateOnEntity is safe to swallow OptimisticLockException in only because "managed operations do not use this best-effort helper" — a fact about callers, not something this generic dispatcher-level helper can verify or enforce. If any future managed-table code path (or a copy/paste in a new dispatcher) is ever routed through operateOnEntity, an OCC conflict on that path would silently be downgraded to a log warning instead of surfacing — exactly the class of bug this PR is fixing everywhere else. Consider making the strict-vs-best-effort choice explicit at each call site (e.g. two named helpers, operateOnEntityBestEffort/operateOnEntityStrict, or a boolean parameter) rather than relying on this dispatcher-wide catch ordering plus a code comment to keep the invariant true.
| * @param tableId the table ID | ||
| * @return the locking select SQL | ||
| */ | ||
| public String selectTableMetaByIdForUpdate(@Param("tableId") Long tableId) { |
There was a problem hiding this comment.
Reuse (minor): this locking-select pattern is now written a third time.
selectTableMetaByIdForUpdate (single-table "SELECT ... WHERE id = #{id} AND deleted_at = 0 FOR UPDATE" projection) mirrors selectSchemaMetaByIdForUpdate and selectCatalogMetaByIdForUpdate added by the earlier schema/catalog OCC work — same shape modulo column/table names, with no shared SQL-fragment helper across the three providers. Not blocking, just flagging alongside the similar duplication in TableMetaService.tableWriteFailure for a possible follow-up cleanup.
There was a problem hiding this comment.
Same call as for tableWriteFailure: deferred to a follow-up cleanup once the series is done, and noted in the PR description rather than left silent.
|
|
||
| // Copies every column of the source row. When a field is added to TablePO above, add it here | ||
| // too, otherwise callers that copy a row would silently blank it. | ||
| private Builder(TablePO source) { |
There was a problem hiding this comment.
Simplification/maintenance risk: hand-rolled full-field copy constructor.
This copy constructor duplicates every TablePO field by hand, guarded only by the comment "When a field is added to TablePO above, add it here too, otherwise callers that copy a row would silently blank it." That's a maintenance trap enforced by convention rather than the compiler — a future field addition that misses this constructor fails silently (no compile error, no test failure unless the new field happens to be asserted). No shared "copy a PO" utility exists elsewhere in the codebase (e.g. SchemaPO's equivalent need is met via a narrower copySchemaPOWithName-style static builder), so this exact hazard is likely to recur for the next PO that needs copy semantics.
There was a problem hiding this comment.
The hand-written copy is still there, but it is no longer guarded only by a comment: TestTablePO#testCopyBuilderCarriesEveryField compares every declared field of source and copy reflectively, and the fixture asserts each field was set, so a field added later without a matching line in the builder fails the test rather than silently blanking on copy.
Removing the hand-written copy altogether would mean changing how TablePO is written, which I would rather not do inside this PR.
| // winner's ID. The upsert already holds that row until commit, so read the | ||
| // database-derived identity and version back through the same natural key. | ||
| TablePO storedPO = | ||
| mapper.selectTableMetaBySchemaIdAndName( |
There was a problem hiding this comment.
Efficiency/simplification: unconditional extra round trip + unexplained guard on the overwrite path.
Two related points in this overwrite branch of insertTable:
selectTableMetaBySchemaIdAndNamenow runs on everyinsertTable(..., overwrite=true)call, not only when the natural key actually collided with a different table_id. It's needed becausecurrent_version/last_versionare now computed inside the UPDATE (current_version + 1) rather than passed in, so Java has to read the result back — this adds a fixed extra round trip to the common overwrite path that didn't exist before this PR (previouslyinsertTableVersionOnDuplicateKeyUpdate(po)used the pre-computedpodirectly).- The
storedPO.getCurrentVersion() > POConverters.INIT_VERSIONguard a few lines below (before soft-deleting the prior version row) isn't explained by a comment — it reads as if it exists to avoid an unnecessary no-op delete for the "brand new row via INSERT branch" case, but that's worth confirming/documenting explicitly, since a reader has to reconstruct why the guard is only sometimes needed.
There was a problem hiding this comment.
Added a comment for the guard: there is an earlier version row to retire only when the upsert updated an existing table, which moves the version from N to N+1; a fresh insert leaves it at the initial version with no earlier row.
On the extra round trip, I could not find a way to avoid it. The version is now derived inside the statement (current_version + 1), so the resulting version has to be read back before table_version_info can be keyed by it. Deriving it with a subquery would need three different spellings for MySQL, PostgreSQL and H2. The read itself is a single-table select, cheaper than selectTableMetaById, which joins table_version_info. Happy to revisit if you see a portable way to fold it in.
…le in the code - Spell out why swallowing a conflict in operateOnEntity is safe. The old comment asserted that managed operations do not use the helper, which a reader cannot check from there. Name the checks instead: alterTable, alterSchema and alterView return earlier for managed entities, and no catalog reports managed storage for topics. - Say what the INIT_VERSION guard on the overwrite path is for: there is an earlier version row to retire only when the upsert updated an existing table. - Pin TablePO's copy builder with a reflective test. The builder lists the fields by hand, so a field added later without a matching line there would be blanked on every copy with nothing to catch it. The test compares every declared field and refuses to pass if the fixture leaves one unset.
What changes were proposed in this pull request?
current_versionas the OCC token for table alter and direct delete.OperationDispatcher.operateOnEntityhelper keeps swallowing them: only external entities reach it, and there the catalog was already changed, so failing would invite a retry that re-applies a non-idempotent change.This PR depends on #12456.
Two things are deliberately left out. The conflict classification (
tableWriteFailure) and the locking select now exist once per entity type; extracting them is a follow-up once the series is done, because what differs per entity is the mapper, which identity columns to compare, and schema's physical-to-logical conversion, so doing it now means refactoring three merged services from inside a fourth. And a version conflict while dropping is tracked in #12597, which covers every entity type rather than this path alone.Why are the changes needed?
Concurrent table writes could overwrite the winning version metadata, while a stale delete could remove data belonging to a newer table version. Some optimistic-lock conflicts were also swallowed or treated as generic IO failures.
Fix: #12345
Does this PR introduce any user-facing change?
Stale managed-table writes now fail with the existing optimistic-lock conflict response instead of silently overwriting newer metadata. Reads and the import path are unchanged: loading an entity still repairs the Gravitino copy on a best-effort basis, so a load that loses a version race keeps returning the entity. No API or property is added.
How was this patch tested?
TableMetaServicecases across H2, MySQL, and PostgreSQL.TestTableOperationDispatchertests.git diff --check.