yuqi1129 commented on code in PR #13737:
URL: https://github.com/apache/gravitino/pull/13737#discussion_r4245246851
##########
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:
Added to the user-facing section of the PR description: a concurrently
deleted tag now returns `NoSuchTagException` (HTTP 404) instead of
`NoSuchMetadataObjectException`, and both extend `NotFoundException`.
##########
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:
Deliberate trade-off: a sustained writer exhausts the attempts with or
without spacing, and a sparse one is absorbed by the first retry, so the pause
only helped short bursts. Lifting the loop above the cache lock would
restructure `RelationalEntityStore#get` for one entity type, so I left it,
together with a single-statement read that cannot tear, to #13181. ff3e692c37
records this in the comment, including not to add a sleep inside the loop.
--
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]
