jerryshao commented on code in PR #12551:
URL: https://github.com/apache/gravitino/pull/12551#discussion_r3850762007
##########
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) {
Review Comment:
`operateOnEntity` now rethrows `OptimisticLockException` instead of
tolerating it (previously caught by the generic `catch (Exception e)` below and
just logged). For unmanaged/external tables, `alterTable` mutates the real
catalog (e.g. Hive) first and only afterward syncs Gravitino's internal entity
here. A lost CAS race on this internal sync now fails the whole `alterTable`
call even though the external system was already changed — and a client that
retries on error could re-apply a non-idempotent op (e.g. AddColumn) against
the external system. Was this behavior change intentional, or should this
particular sync failure still be tolerated for unmanaged catalogs?
##########
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:
`tableWriteFailure`'s zero-row-CAS classification (locking read +
identity-field comparison to distinguish `OptimisticLockException` from
`NoSuchEntityException`) duplicates `SchemaMetaService.schemaWriteFailure`. As
more entity types (catalog/fileset/topic/model/view) pick up OCC, might be
worth factoring this into a shared helper so future fixes to the classification
logic don't need to be re-applied in every service class.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java:
##########
@@ -139,15 +131,27 @@ public void insertTable(TableEntity tableEntity, boolean
overwrite) throws IOExc
SessionUtils.doWithoutCommit(
TableMetaMapper.class,
mapper -> {
- tablePORef.set(po);
ops.insertPO(mapper, po, overwrite);
+ if (overwrite) {
+ TablePO storedPO =
mapper.selectTableMetaByIdForUpdate(po.getTableId());
Review Comment:
`insertTable`'s overwrite branch re-selects the row by the caller's
freshly-generated `table_id` after the upsert. But MySQL's `INSERT ... ON
DUPLICATE KEY UPDATE` can instead conflict on the `(schema_id, table_name,
deleted_at)` unique key rather than the `table_id` PK, in which case the
existing row is updated in place and its `table_id` is left unchanged. The
follow-up `selectTableMetaByIdForUpdate(po.getTableId())` then finds nothing,
and `Preconditions.checkState(storedPO != null, ...)` throws
`IllegalStateException`. This looks reachable when two callers concurrently
import/upsert the same not-yet-tracked external table with different generated
ids. Worth selecting by the natural key instead of (or in addition to) the
generated id here?
Separately: this locking `SELECT` now runs unconditionally on every
overwrite insert, even on the common uncontested path (e.g. routine
external-table import/refresh), adding a round trip that wasn't there before —
is that acceptable, or could the version be derived without the extra read?
--
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]
