yuqi1129 commented on code in PR #13737:
URL: https://github.com/apache/gravitino/pull/13737#discussion_r4237450298
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/ModelVersionMetaService.java:
##########
@@ -438,6 +380,129 @@ 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.
+ *
+ * @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())) {
Review Comment:
Fixed in 61ba1b9053 by keeping the aggregate token and documenting it. No
column today moves only on version writes (`model_latest_version` changes on
link only), so narrowing would need a new counter column, which is out of scope
here. The `readWithStableModel` Javadoc now says that model-level writes such
as a comment or property alter also advance `current_version` and can make a
concurrent read retry, or fail with `OptimisticLockException` if the model is
altered on every attempt. Added
`testReadWithStableModelRereadsAfterConcurrentModelAlter` for the retry, and
updated the PR description to match.
##########
core/src/main/java/org/apache/gravitino/storage/relational/service/ModelVersionMetaService.java:
##########
@@ -438,6 +380,129 @@ 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.
+ *
+ * @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);
+ }
+ LOG.debug(
+ "Model {} changed during attempt {} of a model-version read, reading
it again",
+ modelIdent,
+ attempt);
+ // A short random pause keeps readers of a busy model from re-reading in
lockstep with the
+ // writer stream that just invalidated them. A read through the entity
cache holds its cache
+ // segment lock here, so the pause is kept to a few milliseconds.
+ Uninterruptibles.sleepUninterruptibly(
Review Comment:
Good point, removed in 61ba1b9053. The pause ran under the cache segment
lock, and readers of the same model already serialize on that lock, so it
bought nothing. The read now re-runs immediately; the 3-attempt cap still
bounds the work.
##########
core/src/main/java/org/apache/gravitino/tag/TagManager.java:
##########
@@ -509,56 +441,45 @@ 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) {
+ throw new NoSuchMetadataObjectException(
+ e, "Failed to associate tags for metadata object %s due to not
found", metadataObject);
Review Comment:
Fixed in 61ba1b9053. The catch now matches the store message against the
added and removed tag names with `isMissingEntity` and throws
`NoSuchTagException` naming the tag; otherwise it still throws
`NoSuchMetadataObjectException`. A never-existing tag name is skipped by the
store, so this only fires when a tag is deleted before its row is locked.
`testConcurrentDeleteTagAndAssociateLeavesNoAssignment` now requires
`NoSuchTagException`, and
`testAssociateReportsMissingTagSeparatelyFromMissingObject` covers both cases
deterministically with a mocked store.
--
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]
