yuqi1129 commented on code in PR #12551:
URL: https://github.com/apache/gravitino/pull/12551#discussion_r3853975123
##########
core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java:
##########
@@ -416,6 +417,8 @@ public boolean dropTable(NameIdentifier ident) {
if (droppedFromCatalog) {
try {
store.delete(ident, TABLE);
+ } catch (OptimisticLockException e) {
Review 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.
##########
core/src/main/java/org/apache/gravitino/catalog/OperationDispatcher.java:
##########
@@ -219,6 +220,9 @@ protected <R extends HasIdentifier> R operateOnEntity(
R ret = null;
try {
ret = fn.apply(ident);
+ } catch (OptimisticLockException e) {
+ // A version conflict is actionable: the caller must retry against the
latest entity.
+ throw e;
Review Comment:
On this path the entity is external and the catalog change already happened,
so we log and keep the stale Gravitino copy instead of failing. Failing here
would push the client to retry the whole alter and apply the external change
twice - `AddColumn` is not idempotent - which is worse than a stale copy, and
the copy is repaired on the next load.
Managed entities never reach this helper: `alterTable`, `alterSchema` and
`alterView` return earlier when the entity is managed, and no catalog reports
managed storage for topics (`KafkaCatalogCapability`). A managed alter writes
to the store directly, so its conflict still reaches the caller. I rewrote the
comment to name those checks so the claim can be verified in place.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java:
##########
@@ -215,24 +219,35 @@ public <E extends Entity & HasIdentifier> TableEntity
updateTable(
oldTablePO.getMetalakeId());
}
},
- () ->
- updateResult.set(
- SessionUtils.getWithoutCommit(
- TableMetaMapper.class,
- mapper -> ops.updatePO(mapper, newTablePO, oldTablePO))),
- () ->
- SessionUtils.doWithoutCommit(
- TableVersionMapper.class,
- mapper -> {
- mapper.softDeleteTableVersionByTableIdAndVersion(
- oldTablePO.getTableId(),
oldTablePO.getCurrentVersion());
- mapper.insertTableVersionOnDuplicateKeyUpdate(newTablePO);
- }),
() -> {
- if (updateResult.get() > 0) {
- TableColumnMetaService.getInstance()
- .updateColumnPOsFromTableDiff(oldTableEntity,
newTableEntity, newTablePO);
+ // This update is the decision point for the whole transaction.
current_version is the
+ // table's OCC token: if another writer changed the table after we
read it, that writer
+ // has already increased the token and this UPDATE changes zero
rows. Throwing here
+ // rolls back the transaction before it can touch the version
history or columns.
+ int updated =
+ SessionUtils.getWithoutCommit(
+ TableMetaMapper.class, mapper -> ops.updatePO(mapper,
newTablePO, oldTablePO));
+ if (updated == 0) {
+ throw tableWriteFailure(identifier, oldTablePO);
}
+ },
+ () -> {
+ // The table details live in table_version_info, while table_meta
points to the current
+ // version. These two rows must move together. This step runs only
after the table_meta
+ // CAS above succeeds, so a losing writer cannot overwrite the
winner's version row.
Review Comment:
`table_meta` holds the identity and the current version; the details live in
`table_version_info`, keyed by `(table_id, version)`. An alter moves both, and
the upsert that writes the version row has no version predicate of its own - it
overwrites whatever sits under that key.
If two writers both read version 5 and both build version 6, their version
rows carry the same key, so the second one to run would silently replace the
first one's details. Doing the `table_meta` CAS first is what prevents it: the
loser matches no row there, throws, and the transaction rolls back before
reaching the version row. Only the winner ever writes version 6. I expanded the
comment with this example.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java:
##########
@@ -374,4 +381,76 @@ private void
fillTablePOBuilderParentEntityId(TablePO.Builder builder, Namespace
builder.withCatalogId(namespacedEntityId.namespaceIds()[1]);
builder.withSchemaId(namespacedEntityId.entityId());
}
+
+ private TablePO tablePOWithPersistedIdentityAndVersions(TablePO incomingPO,
TablePO persistedPO) {
+ // The upsert derives the version inside the database and may preserve an
existing table ID, so
+ // its dependent rows must carry the identity and versions the database
ended up with.
+ return TablePO.builder(incomingPO)
+ .withTableId(persistedPO.getTableId())
+ .withCurrentVersion(persistedPO.getCurrentVersion())
+ .withLastVersion(persistedPO.getLastVersion())
+ .build();
+ }
+
+ private void deleteTableDependents(TablePO tablePO) {
+ // The table row has already passed its version check. All cleanup below
uses the same database
+ // transaction, so either the table and every related row are deleted
together, or none are.
+ SessionUtils.doWithoutCommit(
+ OwnerMetaMapper.class,
+ mapper ->
+ mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
+ tablePO.getTableId(), MetadataObject.Type.TABLE.name()));
+
TableColumnMetaService.getInstance().deleteColumnsByTableId(tablePO.getTableId());
+ SessionUtils.doWithoutCommit(
+ SecurableObjectMapper.class,
+ mapper ->
+ mapper.softDeleteObjectRelsByMetadataObject(
+ tablePO.getTableId(), MetadataObject.Type.TABLE.name()));
+ SessionUtils.doWithoutCommit(
+ TagMetadataObjectRelMapper.class,
+ mapper ->
+ mapper.softDeleteTagMetadataObjectRelsByMetadataObject(
+ tablePO.getTableId(), MetadataObject.Type.TABLE.name()));
+ SessionUtils.doWithoutCommit(
+ TagMetadataObjectRelMapper.class,
+ mapper ->
mapper.softDeleteTagMetadataObjectRelsByTableId(tablePO.getTableId()));
+ SessionUtils.doWithoutCommit(
+ StatisticMetaMapper.class,
+ mapper -> mapper.softDeleteStatisticsByEntityId(tablePO.getTableId()));
+ SessionUtils.doWithoutCommit(
+ PolicyMetadataObjectRelMapper.class,
+ mapper ->
mapper.softDeletePolicyMetadataObjectRelsByTableId(tablePO.getTableId()));
+ SessionUtils.doWithoutCommit(
+ TableVersionMapper.class,
+ mapper ->
+ mapper.softDeleteTableVersionByTableIdAndVersion(
+ tablePO.getTableId(), tablePO.getCurrentVersion()));
+ }
+
+ private RuntimeException tableWriteFailure(NameIdentifier identifier,
TablePO observedTablePO) {
Review 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.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java:
##########
@@ -374,4 +370,87 @@ private void
fillTablePOBuilderParentEntityId(TablePO.Builder builder, Namespace
builder.withCatalogId(namespacedEntityId.namespaceIds()[1]);
builder.withSchemaId(namespacedEntityId.entityId());
}
+
+ private TablePO tablePOWithPersistedVersions(TablePO incomingPO, TablePO
persistedPO) {
+ return TablePO.builder()
+ .withTableId(incomingPO.getTableId())
+ .withTableName(incomingPO.getTableName())
+ .withMetalakeId(incomingPO.getMetalakeId())
+ .withCatalogId(incomingPO.getCatalogId())
+ .withSchemaId(incomingPO.getSchemaId())
+ .withAuditInfo(incomingPO.getAuditInfo())
+ .withCurrentVersion(persistedPO.getCurrentVersion())
+ .withLastVersion(persistedPO.getLastVersion())
+ .withDeletedAt(incomingPO.getDeletedAt())
+ .withFormat(incomingPO.getFormat())
+ .withProperties(incomingPO.getProperties())
+ .withPartitions(incomingPO.getPartitions())
+ .withSortOrders(incomingPO.getSortOrders())
+ .withDistribution(incomingPO.getDistribution())
+ .withIndexes(incomingPO.getIndexes())
+ .withComment(incomingPO.getComment())
+ .build();
+ }
+
+ private void deleteTableDependents(TablePO tablePO) {
+ // The table row has already passed its version check. All cleanup below
uses the same database
+ // transaction, so either the table and every related row are deleted
together, or none are.
+ SessionUtils.doWithoutCommit(
+ OwnerMetaMapper.class,
+ mapper ->
+ mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
+ tablePO.getTableId(), MetadataObject.Type.TABLE.name()));
+
TableColumnMetaService.getInstance().deleteColumnsByTableId(tablePO.getTableId());
+ SessionUtils.doWithoutCommit(
+ SecurableObjectMapper.class,
+ mapper ->
+ mapper.softDeleteObjectRelsByMetadataObject(
+ tablePO.getTableId(), MetadataObject.Type.TABLE.name()));
+ SessionUtils.doWithoutCommit(
+ TagMetadataObjectRelMapper.class,
+ mapper ->
+ mapper.softDeleteTagMetadataObjectRelsByMetadataObject(
+ tablePO.getTableId(), MetadataObject.Type.TABLE.name()));
+ SessionUtils.doWithoutCommit(
+ TagMetadataObjectRelMapper.class,
+ mapper ->
mapper.softDeleteTagMetadataObjectRelsByTableId(tablePO.getTableId()));
+ SessionUtils.doWithoutCommit(
+ StatisticMetaMapper.class,
+ mapper -> mapper.softDeleteStatisticsByEntityId(tablePO.getTableId()));
+ SessionUtils.doWithoutCommit(
+ PolicyMetadataObjectRelMapper.class,
+ mapper ->
mapper.softDeletePolicyMetadataObjectRelsByTableId(tablePO.getTableId()));
+ SessionUtils.doWithoutCommit(
+ TableVersionMapper.class,
+ mapper ->
+ mapper.softDeleteTableVersionByTableIdAndVersion(
+ tablePO.getTableId(), tablePO.getCurrentVersion()));
+ }
+
+ private RuntimeException tableWriteFailure(NameIdentifier identifier,
TablePO observedTablePO) {
Review Comment:
Answered on the newer thread for the same method: the extraction is deferred
to the end of the OCC series on purpose, and I will say so in the PR
description.
##########
core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TableMetaBaseSQLProvider.java:
##########
@@ -178,6 +178,29 @@ public String selectTableMetaById(@Param("tableId") Long
tableId) {
+ " WHERE tm.table_id = #{tableId} AND tm.deleted_at = 0";
}
+ /**
+ * Returns the active table metadata row and holds it exclusively for the
current transaction.
+ *
+ * <p>Unlike the metalake, catalog and schema providers, this cannot be
written as {@code
+ * selectTableMetaById(id) + " FOR UPDATE"}: that select LEFT JOINs {@code
table_version_info},
+ * and locking the nullable side of an outer join is rejected by PostgreSQL
and locks the wrong
+ * rows on MySQL. The projection is therefore spelled out for {@code
table_meta} alone, and the
+ * returned row carries only the identity and version columns its callers
read.
+ *
+ * @param tableId the table ID
+ * @return the locking select SQL
+ */
+ public String selectTableMetaByIdForUpdate(@Param("tableId") Long tableId) {
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
