yuqi1129 commented on code in PR #12656:
URL: https://github.com/apache/gravitino/pull/12656#discussion_r3871906711
##########
core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/FilesetVersionBaseSQLProvider.java:
##########
@@ -116,6 +116,13 @@ public String deleteFilesetVersionsByLegacyTimeline(
+ " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT
#{limit}";
}
+ public String selectMaxFilesetVersion(@Param("filesetId") Long filesetId) {
Review Comment:
Addressed in the earlier follow-up. selectMaxFilesetVersion now has Javadoc
in FilesetVersionBaseSQLProvider, FilesetVersionSQLProviderFactory, and
FilesetVersionMapper; core:javadoc passes.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java:
##########
@@ -213,120 +235,84 @@ public <E extends Entity & HasIdentifier> FilesetEntity
updateFileset(
FilesetEntity newEntity = (FilesetEntity) updater.apply((E)
oldFilesetEntity);
Preconditions.checkArgument(
Objects.equals(oldFilesetEntity.id(), newEntity.id()),
- "The updated fileset entity id: %s should be same with the table
entity id before: %s",
+ "The updated fileset entity id: %s should be same with the fileset
entity id before: %s",
newEntity.id(),
oldFilesetEntity.id());
- Integer updateResult;
+ // A fileset written before the version reset was fixed can still own
snapshots newer than the
+ // version its metadata row records. The new snapshot has to be built
above all of them.
+ Long maxStoredVersion =
+ SessionUtils.getWithoutCommit(
+ FilesetVersionMapper.class,
+ mapper ->
mapper.selectMaxFilesetVersion(oldFilesetPO.getFilesetId()));
+
try {
- boolean checkNeedUpdateVersion =
- POConverters.checkFilesetVersionNeedUpdate(
- oldFilesetPO.getFilesetVersionPOs(), newEntity);
FilesetPO newFilesetPO =
- POConverters.updateFilesetPOWithVersion(oldFilesetPO, newEntity,
checkNeedUpdateVersion);
- if (checkNeedUpdateVersion) {
- // These operations are performed atomically within a single
transaction. The version
- // insert is protected by a unique constraint on `fileset_id + version
+ deleted_at`. If
- // the meta update affects 0 rows (concurrent modification), the
transaction is rolled
- // 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);
- }
- });
- updateResult = 1;
- } catch (RuntimeException re) {
- if (metaUpdateCountRef[0] == 0) {
- // The meta update matched no rows; the transaction was rolled
back,
- // including the version insert above.
- throw new IOException("Failed to update the entity: " +
identifier);
- } else {
- ExceptionUtils.checkSQLException(
- re, Entity.EntityType.FILESET,
newEntity.nameIdentifier().toString());
- throw re;
- }
- }
- } else {
- int[] metaUpdateCountRef = new int[1];
- SessionUtils.doMultipleWithCommit(
- () ->
- metaUpdateCountRef[0] =
- SessionUtils.getWithoutCommit(
- FilesetMetaMapper.class,
- mapper -> mapper.updateFilesetMeta(newFilesetPO,
oldFilesetPO)));
- updateResult = metaUpdateCountRef[0];
- }
+ POConverters.updateFilesetPOWithVersion(oldFilesetPO, newEntity,
maxStoredVersion);
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ // Decide the winner before writing fileset_version_info. Two
writers that read version
+ // N both prepare version N + 1, but only one can change the
metadata row. The loser
+ // stops here, so it cannot overwrite any storage-location row
written by the winner.
+ int updated =
+ SessionUtils.getWithoutCommit(
+ FilesetMetaMapper.class,
+ mapper -> mapper.updateFilesetMeta(newFilesetPO,
oldFilesetPO));
+ if (updated == 0) {
+ throw filesetWriteFailure(identifier, oldFilesetPO);
+ }
+ },
+ () -> {
+ // The metadata row now points to this complete snapshot. It stays
in the same
+ // transaction so a failed version insert also restores the
metadata version.
+ SessionUtils.doWithoutCommit(
+ FilesetVersionMapper.class,
+ mapper ->
mapper.insertFilesetVersions(newFilesetPO.getFilesetVersionPOs()));
+ });
} catch (RuntimeException re) {
ExceptionUtils.checkSQLException(
re, Entity.EntityType.FILESET,
newEntity.nameIdentifier().toString());
throw re;
}
- if (updateResult > 0) {
- return newEntity;
- } else {
- throw new IOException("Failed to update the entity: " + identifier);
- }
+ return newEntity;
}
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "deleteFileset")
public boolean deleteFileset(NameIdentifier identifier) {
FilesetPO filesetPO = getFilesetPOByIdentifier(identifier);
- Long filesetId = filesetPO.getFilesetId();
- // We should delete meta and version info
- AtomicInteger deleteResult = new AtomicInteger(0);
+ // Delete the root row first and only if it still has the version we read.
A stale drop stops
+ // before it can remove versions, tags, policies, or any other related
data.
SessionUtils.doMultipleWithCommit(
- () ->
- deleteResult.set(
- SessionUtils.getWithoutCommit(
- FilesetMetaMapper.class,
- mapper ->
mapper.softDeleteFilesetMetasByFilesetId(filesetId))),
- () -> {
- if (deleteResult.get() > 0) {
- SessionUtils.doWithoutCommit(
- FilesetVersionMapper.class,
- mapper ->
mapper.softDeleteFilesetVersionsByFilesetId(filesetId));
- SessionUtils.doWithoutCommit(
- OwnerMetaMapper.class,
- mapper ->
- mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
- filesetId, MetadataObject.Type.FILESET.name()));
- SessionUtils.doWithoutCommit(
- SecurableObjectMapper.class,
- mapper ->
- mapper.softDeleteObjectRelsByMetadataObject(
- filesetId, MetadataObject.Type.FILESET.name()));
- SessionUtils.doWithoutCommit(
- TagMetadataObjectRelMapper.class,
- mapper ->
- mapper.softDeleteTagMetadataObjectRelsByMetadataObject(
- filesetId, MetadataObject.Type.FILESET.name()));
- SessionUtils.doWithoutCommit(
- StatisticMetaMapper.class,
- mapper -> mapper.softDeleteStatisticsByEntityId(filesetId));
- SessionUtils.doWithoutCommit(
- PolicyMetadataObjectRelMapper.class,
- mapper ->
- mapper.softDeletePolicyMetadataObjectRelsByMetadataObject(
- filesetId, MetadataObject.Type.FILESET.name()));
- }
- });
+ () -> deleteFilesetWithVersion(identifier, filesetPO),
+ () -> deleteFilesetDependents(filesetPO.getFilesetId()));
+
+ return true;
+ }
- return deleteResult.get() > 0;
+ /**
+ * Soft-deletes the observed fileset metadata row without starting a
transaction.
+ *
+ * <p>The caller must run this method in the same transaction as dependent
cleanup. Package access
+ * also lets concurrency tests submit a deliberately stale snapshot without
duplicating the
+ * production CAS logic.
+ *
+ * @param identifier the fileset identity observed by the caller
+ * @param observedFilesetPO the fileset row and OCC version observed by the
caller
+ */
+ void deleteFilesetWithVersion(NameIdentifier identifier, FilesetPO
observedFilesetPO) {
+ int deleted =
Review Comment:
Addressed in the earlier follow-up. deleteFilesetWithVersion now sits in the
non-public helper section after all public methods, consistent with the
class-member-ordering convention.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java:
##########
@@ -480,4 +466,120 @@ public List<FilesetEntity>
batchGetFilesetByIdentifier(List<NameIdentifier> iden
return POConverters.fromFilesetPOs(filesetPOs,
firstIdent.namespace());
});
}
+
+ /**
+ * Rewrites the stored identifier property so that it names the fileset the
row is actually stored
+ * under. The overwrite keeps the fileset ID the database already had, while
the properties still
+ * carry the ID the caller generated, and a reader that trusts the property
would disagree with
+ * {@code fileset_meta}.
+ */
+ private String propertiesWithFilesetId(String serializedProperties, Long
filesetId) {
+ if (serializedProperties == null) {
+ return null;
+ }
+ try {
+ Map<String, String> properties =
+ JsonUtils.anyFieldMapper()
+ .readValue(serializedProperties, new TypeReference<Map<String,
String>>() {});
+ if (properties == null ||
!properties.containsKey(StringIdentifier.ID_KEY)) {
+ return serializedProperties;
+ }
+ Map<String, String> rewritten = new HashMap<>(properties);
+ rewritten.put(StringIdentifier.ID_KEY,
StringIdentifier.fromId(filesetId).toString());
+ return JsonUtils.anyFieldMapper().writeValueAsString(rewritten);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException("Failed to rewrite the fileset identifier
property", e);
+ }
+ }
+
+ private FilesetPO filesetPOWithPersistedIdentityAndVersion(
+ FilesetPO incomingPO, FilesetPO persistedPO) {
+ // The upsert chooses the version inside the database and may keep an
existing fileset ID. All
+ // storage-location rows must use those stored values or the metadata row
would point at a
+ // version snapshot that cannot be loaded.
+ List<FilesetVersionPO> persistedVersions =
+ incomingPO.getFilesetVersionPOs().stream()
+ .map(
+ versionPO ->
+ FilesetVersionPO.builder()
+ .withMetalakeId(persistedPO.getMetalakeId())
+ .withCatalogId(persistedPO.getCatalogId())
+ .withSchemaId(persistedPO.getSchemaId())
+ .withFilesetId(persistedPO.getFilesetId())
+ .withVersion(persistedPO.getCurrentVersion())
+ .withFilesetComment(versionPO.getFilesetComment())
+ .withProperties(
+ propertiesWithFilesetId(
+ versionPO.getProperties(),
persistedPO.getFilesetId()))
+ .withLocationName(versionPO.getLocationName())
+ .withStorageLocation(versionPO.getStorageLocation())
+ .withDeletedAt(versionPO.getDeletedAt())
+ .build())
+ .collect(Collectors.toList());
+ return FilesetPO.builder()
+ .withFilesetId(persistedPO.getFilesetId())
+ .withFilesetName(persistedPO.getFilesetName())
+ .withMetalakeId(persistedPO.getMetalakeId())
+ .withCatalogId(persistedPO.getCatalogId())
+ .withSchemaId(persistedPO.getSchemaId())
+ .withType(persistedPO.getType())
+ .withAuditInfo(persistedPO.getAuditInfo())
+ .withCurrentVersion(persistedPO.getCurrentVersion())
+ .withLastVersion(persistedPO.getLastVersion())
+ .withDeletedAt(persistedPO.getDeletedAt())
+ .withFilesetVersionPOs(persistedVersions)
+ .build();
+ }
+
+ private void deleteFilesetDependents(Long filesetId) {
+ // The fileset row has already passed its version check. All cleanup below
uses the same
+ // transaction, so either the root and every related row are deleted
together, or none are.
+ SessionUtils.doWithoutCommit(
+ FilesetVersionMapper.class,
+ mapper -> mapper.softDeleteFilesetVersionsByFilesetId(filesetId));
+ SessionUtils.doWithoutCommit(
+ OwnerMetaMapper.class,
+ mapper ->
+ mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
+ filesetId, MetadataObject.Type.FILESET.name()));
+ SessionUtils.doWithoutCommit(
+ SecurableObjectMapper.class,
+ mapper ->
+ mapper.softDeleteObjectRelsByMetadataObject(
+ filesetId, MetadataObject.Type.FILESET.name()));
+ SessionUtils.doWithoutCommit(
+ TagMetadataObjectRelMapper.class,
+ mapper ->
+ mapper.softDeleteTagMetadataObjectRelsByMetadataObject(
+ filesetId, MetadataObject.Type.FILESET.name()));
+ SessionUtils.doWithoutCommit(
+ StatisticMetaMapper.class, mapper ->
mapper.softDeleteStatisticsByEntityId(filesetId));
+ SessionUtils.doWithoutCommit(
+ PolicyMetadataObjectRelMapper.class,
+ mapper ->
+ mapper.softDeletePolicyMetadataObjectRelsByMetadataObject(
+ filesetId, MetadataObject.Type.FILESET.name()));
+ }
+
+ private RuntimeException filesetWriteFailure(
+ NameIdentifier identifier, FilesetPO observedFilesetPO) {
+ // A zero-row CAS means either another writer advanced this fileset, or
the fileset disappeared
+ // from the name the caller used. Locking the stable ID waits for an
in-flight writer to commit,
+ // so the result can be classified from committed identity data.
+ FilesetPO currentFilesetPO =
+ SessionUtils.getWithoutCommit(
+ FilesetMetaMapper.class,
+ mapper ->
mapper.selectFilesetMetaByIdForUpdate(observedFilesetPO.getFilesetId()));
+ if (currentFilesetPO == null
Review Comment:
Addressed in 35089fe7ed. The conflict classifier no longer issues SELECT FOR
UPDATE; it uses one non-locking natural-key lookup after the failed CAS and the
locking-by-ID mapper/provider path has been removed.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java:
##########
@@ -480,4 +466,120 @@ public List<FilesetEntity>
batchGetFilesetByIdentifier(List<NameIdentifier> iden
return POConverters.fromFilesetPOs(filesetPOs,
firstIdent.namespace());
});
}
+
+ /**
+ * Rewrites the stored identifier property so that it names the fileset the
row is actually stored
+ * under. The overwrite keeps the fileset ID the database already had, while
the properties still
+ * carry the ID the caller generated, and a reader that trusts the property
would disagree with
+ * {@code fileset_meta}.
+ */
+ private String propertiesWithFilesetId(String serializedProperties, Long
filesetId) {
+ if (serializedProperties == null) {
+ return null;
+ }
+ try {
+ Map<String, String> properties =
+ JsonUtils.anyFieldMapper()
+ .readValue(serializedProperties, new TypeReference<Map<String,
String>>() {});
+ if (properties == null ||
!properties.containsKey(StringIdentifier.ID_KEY)) {
+ return serializedProperties;
+ }
+ Map<String, String> rewritten = new HashMap<>(properties);
+ rewritten.put(StringIdentifier.ID_KEY,
StringIdentifier.fromId(filesetId).toString());
+ return JsonUtils.anyFieldMapper().writeValueAsString(rewritten);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException("Failed to rewrite the fileset identifier
property", e);
+ }
+ }
+
+ private FilesetPO filesetPOWithPersistedIdentityAndVersion(
+ FilesetPO incomingPO, FilesetPO persistedPO) {
Review Comment:
Addressed in 35089fe7ed. The overwrite path now locks and resolves the
existing natural-key row before constructing the replacement snapshot. It
builds directly with the persisted ID/version and removes the JSON round-trip
plus the post-read FilesetPO reconstruction helpers.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java:
##########
@@ -213,120 +235,84 @@ public <E extends Entity & HasIdentifier> FilesetEntity
updateFileset(
FilesetEntity newEntity = (FilesetEntity) updater.apply((E)
oldFilesetEntity);
Preconditions.checkArgument(
Objects.equals(oldFilesetEntity.id(), newEntity.id()),
- "The updated fileset entity id: %s should be same with the table
entity id before: %s",
+ "The updated fileset entity id: %s should be same with the fileset
entity id before: %s",
newEntity.id(),
oldFilesetEntity.id());
- Integer updateResult;
+ // A fileset written before the version reset was fixed can still own
snapshots newer than the
+ // version its metadata row records. The new snapshot has to be built
above all of them.
+ Long maxStoredVersion =
Review Comment:
Addressed in 35089fe7ed. Normal alters now use a single CAS whose predicate
also rejects an already occupied future snapshot version. MAX(version) is
queried only after that CAS detects the uncommon legacy-row shape, then the
write retries above the stored snapshots.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/FilesetMetaService.java:
##########
@@ -213,120 +235,84 @@ public <E extends Entity & HasIdentifier> FilesetEntity
updateFileset(
FilesetEntity newEntity = (FilesetEntity) updater.apply((E)
oldFilesetEntity);
Preconditions.checkArgument(
Objects.equals(oldFilesetEntity.id(), newEntity.id()),
- "The updated fileset entity id: %s should be same with the table
entity id before: %s",
+ "The updated fileset entity id: %s should be same with the fileset
entity id before: %s",
newEntity.id(),
oldFilesetEntity.id());
- Integer updateResult;
+ // A fileset written before the version reset was fixed can still own
snapshots newer than the
+ // version its metadata row records. The new snapshot has to be built
above all of them.
+ Long maxStoredVersion =
+ SessionUtils.getWithoutCommit(
+ FilesetVersionMapper.class,
+ mapper ->
mapper.selectMaxFilesetVersion(oldFilesetPO.getFilesetId()));
+
try {
- boolean checkNeedUpdateVersion =
- POConverters.checkFilesetVersionNeedUpdate(
- oldFilesetPO.getFilesetVersionPOs(), newEntity);
FilesetPO newFilesetPO =
- POConverters.updateFilesetPOWithVersion(oldFilesetPO, newEntity,
checkNeedUpdateVersion);
- if (checkNeedUpdateVersion) {
- // These operations are performed atomically within a single
transaction. The version
- // insert is protected by a unique constraint on `fileset_id + version
+ deleted_at`. If
- // the meta update affects 0 rows (concurrent modification), the
transaction is rolled
- // 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);
- }
- });
- updateResult = 1;
- } catch (RuntimeException re) {
- if (metaUpdateCountRef[0] == 0) {
- // The meta update matched no rows; the transaction was rolled
back,
- // including the version insert above.
- throw new IOException("Failed to update the entity: " +
identifier);
- } else {
- ExceptionUtils.checkSQLException(
- re, Entity.EntityType.FILESET,
newEntity.nameIdentifier().toString());
- throw re;
- }
- }
- } else {
- int[] metaUpdateCountRef = new int[1];
- SessionUtils.doMultipleWithCommit(
- () ->
- metaUpdateCountRef[0] =
- SessionUtils.getWithoutCommit(
- FilesetMetaMapper.class,
- mapper -> mapper.updateFilesetMeta(newFilesetPO,
oldFilesetPO)));
- updateResult = metaUpdateCountRef[0];
- }
+ POConverters.updateFilesetPOWithVersion(oldFilesetPO, newEntity,
maxStoredVersion);
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ // Decide the winner before writing fileset_version_info. Two
writers that read version
+ // N both prepare version N + 1, but only one can change the
metadata row. The loser
+ // stops here, so it cannot overwrite any storage-location row
written by the winner.
+ int updated =
+ SessionUtils.getWithoutCommit(
+ FilesetMetaMapper.class,
+ mapper -> mapper.updateFilesetMeta(newFilesetPO,
oldFilesetPO));
+ if (updated == 0) {
+ throw filesetWriteFailure(identifier, oldFilesetPO);
+ }
+ },
+ () -> {
+ // The metadata row now points to this complete snapshot. It stays
in the same
+ // transaction so a failed version insert also restores the
metadata version.
+ SessionUtils.doWithoutCommit(
+ FilesetVersionMapper.class,
+ mapper ->
mapper.insertFilesetVersions(newFilesetPO.getFilesetVersionPOs()));
+ });
} catch (RuntimeException re) {
ExceptionUtils.checkSQLException(
re, Entity.EntityType.FILESET,
newEntity.nameIdentifier().toString());
throw re;
}
- if (updateResult > 0) {
- return newEntity;
- } else {
- throw new IOException("Failed to update the entity: " + identifier);
- }
+ return newEntity;
}
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "deleteFileset")
public boolean deleteFileset(NameIdentifier identifier) {
FilesetPO filesetPO = getFilesetPOByIdentifier(identifier);
- Long filesetId = filesetPO.getFilesetId();
Review Comment:
Addressed in 9d4c822a41. deleteAndGet returns the exact snapshot protected
by the delete CAS, and the storage callback consumes that snapshot. A
concurrent alter makes the CAS fail before any stale location is removed;
testDeleteAndGetReturnsSnapshotProtectedByDeleteCas covers the handoff.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -712,6 +712,14 @@ public boolean dropFileset(NameIdentifier ident) {
FilesetEntity filesetEntity =
store.get(ident, Entity.EntityType.FILESET, FilesetEntity.class);
+ // Drop the metadata first. It is the only step that can still be
rejected, because the drop
+ // is refused when another writer altered the fileset after the read
above. Deleting the files
+ // first would leave the data gone while the rejected drop keeps the
fileset row pointing at
+ // storage locations that no longer exist.
+ if (!store.delete(ident, Entity.EntityType.FILESET)) {
Review Comment:
Addressed in the earlier follow-ups. Managed-storage cleanup now runs
through deleteAndGet before the metadata transaction commits, so an IOException
rolls the metadata delete back and the drop remains retryable.
testDropFilesetRollsBackMetadataWhenStorageDeletionFails covers this path.
--
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]
