jerryshao commented on code in PR #13737:
URL: https://github.com/apache/gravitino/pull/13737#discussion_r4245145892
##########
core/src/main/java/org/apache/gravitino/tag/TagManager.java:
##########
@@ -509,56 +443,59 @@ private String[] associateTagValuesForMetadataObject(
TagValue[] tagValuesToAdd = tagsToAddSet.toArray(new TagValue[0]);
TagValue[] tagValuesToRemove = tagsToRemoveSet.toArray(new TagValue[0]);
- return TreeLockUtils.doWithTreeLock(
- entityIdent,
- LockType.READ,
- () ->
- TreeLockUtils.doWithTreeLock(
- NameIdentifier.of(NamespaceUtil.ofTag(metalake).levels()),
- LockType.WRITE,
- () -> {
- try {
- List<TagEntity> tags;
- if (mode == TagAssociationMode.TAG_VALUES) {
- tags =
- entityStore
- .relationOperations()
- .updateEntityRelations(
- RelationUpdate.of(
-
SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
- entityIdent,
- entityType,
- toRelationEdgeTargets(metalake,
tagValuesToAdd),
- toRelationEdgeTargets(metalake,
tagValuesToRemove)));
- } else {
- tags =
- entityStore
- .relationOperations()
- .updateEntityRelations(
-
SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
- entityIdent,
- entityType,
- toNameIdentifiers(metalake, tagValuesToAdd),
- toNameIdentifiers(metalake,
tagValuesToRemove));
- }
- return
tags.stream().map(Tag::name).distinct().toArray(String[]::new);
- } catch (NoSuchEntityException e) {
- throw new NoSuchMetadataObjectException(
- e,
- "Failed to associate tags for metadata object %s due
to not found",
- metadataObject);
- } catch (EntityAlreadyExistsException e) {
- throw new TagAlreadyAssociatedException(
- e,
- "Failed to associate tags for metadata object due to
some tag values %s already "
- + "associated to the metadata object %s",
- Arrays.toString(tagsToAdd),
- metadataObject);
- } catch (IOException e) {
- LOG.error("Failed to associate tags for metadata object
{}", metadataObject, e);
- throw new RuntimeException(e);
- }
- }));
+ try {
+ List<TagEntity> tags;
+ if (mode == TagAssociationMode.TAG_VALUES) {
+ tags =
+ entityStore
+ .relationOperations()
+ .updateEntityRelations(
+ RelationUpdate.of(
+
SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
+ entityIdent,
+ entityType,
+ toRelationEdgeTargets(metalake, tagValuesToAdd),
+ toRelationEdgeTargets(metalake, tagValuesToRemove)));
+ } else {
+ tags =
+ entityStore
+ .relationOperations()
+ .updateEntityRelations(
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
+ entityIdent,
+ entityType,
+ toNameIdentifiers(metalake, tagValuesToAdd),
+ toNameIdentifiers(metalake, tagValuesToRemove));
+ }
+ return tags.stream().map(Tag::name).distinct().toArray(String[]::new);
+ } catch (NoSuchEntityException e) {
+ // A tag deleted concurrently is reported by the store when its row is
locked; tell it apart
+ // from a missing metadata object so the caller is not told the object
is gone.
+ Optional<String> missingTag =
+ Stream.concat(Arrays.stream(tagValuesToAdd),
Arrays.stream(tagValuesToRemove))
+ .map(TagValue::name)
+ .filter(tagName -> isMissingEntity(e, Entity.EntityType.TAG,
tagName))
+ .findFirst();
+ if (missingTag.isPresent()) {
+ throw new NoSuchTagException(
Review Comment:
[Nit] Worth one line in the user-facing change section: this changes the
error identity of the association endpoint, not just its message.
`TagExceptionHandler` routes every `NotFoundException` through
`Utils.notFound(errorMsg, e)` (`ExceptionHandlers.java:840-841`), so the 404
body's `type` switches from `NoSuchMetadataObjectException` to
`NoSuchTagException` for this race. The Java client then throws
`NoSuchTagException` where it previously fell through to the generic
`NotFoundException`
(`clients/client-java/src/main/java/org/apache/gravitino/client/ErrorHandlers.java:1100-1106`).
Both are `NotFoundException` subclasses (`NoSuchTagException.java:25`,
`NoSuchMetadataObjectException.java:25`), so no caller catching
`NotFoundException` breaks - but anyone switching on the error `type` string,
or catching `NoSuchMetadataObjectException` specifically, sees a new value. The
PR body's user-facing section lists the 409 and idempotent-delete changes but
not this one.
Verified by: reading the new catch block here, then
`TagExceptionHandler.handle` and the client `TagErrorHandler` switch, and both
exception class declarations.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/ModelVersionMetaService.java:
##########
@@ -438,6 +374,130 @@ private boolean isModelVersionAliasUpdated(
return !oldAliases.equals(newAliases);
}
+ private List<ModelVersionEntity> listModelVersions(NameIdentifier
modelIdent, long modelId) {
+ List<ModelVersionPO> modelVersionPOs =
+ SessionUtils.getWithoutCommit(
+ ModelVersionMetaMapper.class, mapper ->
mapper.listModelVersionMetasByModelId(modelId));
+
+ if (modelVersionPOs.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ // Get the aliases for all the model versions.
+ List<ModelVersionAliasRelPO> aliasRelPOs =
+ SessionUtils.getWithoutCommit(
+ ModelVersionAliasRelMapper.class,
+ mapper -> mapper.selectModelVersionAliasRelsByModelId(modelId));
+ Multimap<Integer, ModelVersionAliasRelPO> aliasRelPOsByModelVersion =
+ ArrayListMultimap.create();
+ aliasRelPOs.forEach(r ->
aliasRelPOsByModelVersion.put(r.getModelVersion(), r));
+
+ return ImmutableList.copyOf(
+ modelVersionPOs.stream()
+ .collect(
+ Collectors.groupingBy(
+ ModelVersionPO::getModelVersion,
+ Collectors.collectingAndThen(
+ Collectors.<ModelVersionPO>toList(),
+ m -> {
+ List<ModelVersionAliasRelPO> versionAliasRelPOs =
+ Lists.newArrayList(
+
aliasRelPOsByModelVersion.get(m.get(0).getModelVersion()));
+ return POConverters.fromModelVersionPO(modelIdent,
m, versionAliasRelPOs);
+ })))
+ .values());
+ }
+
+ private ModelVersionEntity getModelVersion(
+ NameIdentifier ident, NameIdentifier modelIdent, long modelId) {
+ boolean isVersionNumber = NumberUtils.isCreatable(ident.name());
+
+ List<ModelVersionPO> modelVersionPOs =
+ SessionUtils.getWithoutCommit(
+ ModelVersionMetaMapper.class,
+ mapper -> {
+ if (isVersionNumber) {
+ return mapper.selectModelVersionMeta(modelId,
Integer.valueOf(ident.name()));
+ } else {
+ return mapper.selectModelVersionMetaByAlias(modelId,
ident.name());
+ }
+ });
+
+ if (modelVersionPOs.isEmpty()) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.MODEL_VERSION.name().toLowerCase(Locale.ROOT),
+ ident.toString());
+ }
+
+ List<ModelVersionAliasRelPO> aliasRelPOs =
+ SessionUtils.getWithoutCommit(
+ ModelVersionAliasRelMapper.class,
+ mapper -> {
+ if (isVersionNumber) {
+ return mapper.selectModelVersionAliasRelsByModelIdAndVersion(
+ modelId, Integer.valueOf(ident.name()));
+ } else {
+ return
mapper.selectModelVersionAliasRelsByModelIdAndAlias(modelId, ident.name());
+ }
+ });
+
+ return POConverters.fromModelVersionPO(modelIdent, modelVersionPOs,
aliasRelPOs);
+ }
+
+ /**
+ * Runs a read of a model's version and alias rows and returns it only if
the model did not change
+ * while the rows were read.
+ *
+ * <p>The version rows and the alias rows are read by separate statements,
so a concurrent link,
+ * alter or delete could otherwise be seen half applied, for example with
the old URIs and the new
+ * aliases. Every such write advances the model row's {@code
current_version} in its own
+ * transaction, so an unchanged version after the read proves the rows
belong to one committed
+ * state. A changed version re-runs the read, up to {@link
#MAX_STABLE_READ_ATTEMPTS} times.
+ *
+ * <p>{@code current_version} is the model's aggregate concurrency token, so
model-level writes
+ * such as a comment or property change through {@code alterModel} also
advance it. Those writes
+ * cannot tear the version and alias rows, but they still make a concurrent
read retry, and a
+ * model altered on every attempt fails the read with {@link
OptimisticLockException}.
+ *
+ * @param modelIdent the identifier of the model whose rows are read
+ * @param read the read to run against the observed model row
+ * @return the result of a read that saw no concurrent model change
+ * @throws NoSuchEntityException if the model does not exist, or was dropped
during the read
+ * @throws OptimisticLockException if the model kept changing
+ */
+ @VisibleForTesting
+ <T> T readWithStableModel(NameIdentifier modelIdent, Function<ModelPO, T>
read) {
+ for (int attempt = 1; ; attempt++) {
+ // Will throw a NoSuchEntityException if the model does not exist.
+ ModelPO observedModelPO =
ModelMetaService.getInstance().getModelPOByIdentifier(modelIdent);
+ T result = read.apply(observedModelPO);
+
+ ModelPO currentModelPO =
+ SessionUtils.getWithoutCommit(
+ ModelMetaMapper.class,
+ mapper ->
mapper.selectModelMetaByModelId(observedModelPO.getModelId()));
+ if (currentModelPO == null) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.MODEL.name().toLowerCase(Locale.ROOT),
+ modelIdent.toString());
+ }
+ if (Objects.equals(currentModelPO.getCurrentVersion(),
observedModelPO.getCurrentVersion())) {
+ return result;
+ }
+ if (attempt >= MAX_STABLE_READ_ATTEMPTS) {
+ throw ExceptionUtils.concurrentModification(Entity.EntityType.MODEL,
modelIdent);
+ }
+ // Re-read immediately: a read through the entity cache holds its cache
segment lock here,
+ // so pausing would also stall unrelated entities hashed to the same
segment.
Review Comment:
[Question] With the pause gone, the three attempts now run back to back with
only the two SQL round-trips between them, so the span a contended read
tolerates shrinks from roughly 20 ms to microseconds. Against a sustained
writer stream on one model - the alter loop from the previous round - that
makes exhausting `MAX_STABLE_READ_ATTEMPTS` and returning 409 somewhat more
likely than before, not less.
That is a fine trade if it is deliberate; I raise it only because the reason
recorded in the comment (the cache segment lock) argues for moving the retry
loop above that lock rather than for having no spacing at all -
`RelationalEntityStore#get` takes `withCacheLock` around the whole backend read
(`RelationalEntityStore.java:242-253`), so a loop outside it could both pause
and not block unrelated keys. If lifting the loop is left to the #13181
follow-up, saying so in a half-sentence here would stop a future reader from
re-adding a sleep inside the lock.
Verified by: reading the loop in full at `:469-499` after the change, and
re-reading `RelationalEntityStore#get` on this head to confirm the lock still
wraps the whole 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]
