jerryshao commented on code in PR #13737:
URL: https://github.com/apache/gravitino/pull/13737#discussion_r4237388893
##########
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:
[Question] Is this pause worth what it holds? When the cache is on, this
sleep runs inside a cache *segment* lock, not a per-key lock.
`RelationalEntityStore#get` wraps the whole backend read in
`cache.withCacheLock(EntityCacheKey.of(ident, entityType), ...)`
(`RelationalEntityStore.java:242-253`), and `CaffeineEntityCache` implements
that with `SegmentedLock` (`CaffeineEntityCache.java:103-150`, `:284-289`), so
the key hashes to one of `CACHE_LOCK_SEGMENTS` locks shared with unrelated
entities. Each contended read can then block that whole segment for up to 2 x
10 ms on top of the SQL time.
The stated rationale is also weakened by that same lock: two readers of the
same model already serialize on the segment, so they cannot re-read "in
lockstep" with each other. If the goal is only to avoid hammering a hot writer
stream, a shorter cap (1-2 ms) or no pause at all would cost less;
alternatively the retry loop could sit above the cache lock. Happy to be told
the 20 ms worst case is acceptable - just want it to be a deliberate choice.
Verified by: following `getModelVersionByIdentifier` back through
`JDBCBackend.java:276` to `RelationalEntityStore#get`, then reading
`SegmentedLock` usage in `CaffeineEntityCache`.
##########
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:
[Nit] This now-reachable path reports the wrong entity. Without the
tag-namespace WRITE lock, an associate that loses to a concurrent `deleteTag`
sees the tag row gone and the store raises `NoSuchEntityException` for the
*tag*, which this catch turns into `NoSuchMetadataObjectException` naming the
metadata object - so the caller gets a 404 saying the table was not found when
the table is fine and the tag was dropped.
`testConcurrentDeleteTagAndAssociateLeavesNoAssignment` accepts any
`NotFoundException`, so it does not pin which one.
`getTagForMetadataObject` in this same class already solves exactly this
with `isMissingEntity(e, Entity.EntityType.TAG, name)`; reusing it here would
let the loser surface `NoSuchTagException` instead.
Verified by: reading the new catch block and the `isMissingEntity` branch at
`TagManager.java:373-381`, plus the race test's assertion in `TestTagManager`.
##########
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:
[Important] The stability check is wider than the tearing risk it guards, so
model-version reads retry (and can end in HTTP 409) on writes that cannot tear
them.
`current_version` is a single aggregate OCC token on the model row, and
`alterModel` advances it too: `POConverters.updateModelPO` sets
`withCurrentVersion(nextModelVersion)` (`POConverters.java:1741`) and
`updateModelMeta` writes `current_version = #{newModelMeta.currentVersion}`
(`ModelMetaBaseSQLProvider.java:338`). A comment-only or property-only
`alterModel` therefore invalidates a concurrent
`getModelVersion`/`listModelVersions`, even though neither the version rows nor
the alias rows can have changed. With `MAX_STABLE_READ_ATTEMPTS = 3` and a 1-10
ms pause, a client that alters a model in a loop (a tagging or promotion job)
can make reads of that model's versions fail with `OptimisticLockException` ->
409 while nothing about the versions moved.
Two options: narrow the token the read compares to one that only version
writes move (`model_latest_version` plus a version-scoped check, or a dedicated
counter bumped only by `bumpModelVersion`), or keep the current token and say
so in the Javadoc above - the sentence "Every link, alter and delete of a
version advances the model row's `current_version`" reads as if those were the
only writes that do, which is what makes the extra 409s surprising.
Verified by: reading `readWithStableModel` in full, then
`POConverters.updateModelPO` and the `updateModelMeta` SQL to confirm
`alterModel` bumps the same column the read compares;
`ExceptionHandlers.java:1235` confirms the 409.
--
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]
