This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new cbf0198fd5 [#12233] refactor(core): Support relation targets (#12234)
cbf0198fd5 is described below
commit cbf0198fd5b2e72b4c063c3b3d6b7555f20897b8
Author: roryqi <[email protected]>
AuthorDate: Tue Aug 4 09:24:57 2026 +0800
[#12233] refactor(core): Support relation targets (#12234)
### What changes were proposed in this pull request?
Introduce object-based relation query and update requests for
relation-edge metadata.
This PR adds `RelationQuery`, `RelationUpdate`, and `RelationEdgeTarget`
as the canonical generic relation APIs. Existing primitive
`SupportsRelationOperations` methods are kept as compatibility adapters,
while relation-value reads and value-carrying updates now evolve through
the request/update objects instead of tag-specific or value-specific
overloads.
`RelationalEntityStore` delegates the object APIs, validates relation
target entity types before backend writes, and invalidates relation
caches after backend writes using the same derived target type.
This is the first split from the original tag assignment value storage
PR. The schema and tag storage implementation are kept in
`tag-assignment-values-storage-next` for the follow-up PR.
### Why are the changes needed?
Tag assignment values need relation-level metadata, but the store
contract should stay relation-generic instead of adding tag-specific
methods.
Fix: #12233
### Does this PR introduce _any_ user-facing change?
No. This is internal core/store infrastructure only.
### How was this patch tested?
`./gradlew :core:spotlessApply`
`./gradlew :core:test --tests
org.apache.gravitino.TestSupportsRelationOperations --tests
org.apache.gravitino.storage.relational.TestRelationalEntityStore
-PskipITs -PskipDockerTests=false`
---
.../org/apache/gravitino/RelationEdgeTarget.java | 81 +++++++++
.../java/org/apache/gravitino/RelationQuery.java | 116 +++++++++++++
.../java/org/apache/gravitino/RelationUpdate.java | 134 +++++++++++++++
.../gravitino/SupportsRelationOperations.java | 87 +++++++++-
.../storage/relational/RelationalEntityStore.java | 116 +++++++++++--
.../gravitino/TestSupportsRelationOperations.java | 186 +++++++++++++++++++++
.../relational/TestRelationalEntityStore.java | 120 ++++++++++++-
7 files changed, 814 insertions(+), 26 deletions(-)
diff --git a/core/src/main/java/org/apache/gravitino/RelationEdgeTarget.java
b/core/src/main/java/org/apache/gravitino/RelationEdgeTarget.java
new file mode 100644
index 0000000000..e654670c2b
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/RelationEdgeTarget.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino;
+
+import com.google.common.base.Preconditions;
+import java.util.Optional;
+import javax.annotation.Nullable;
+
+/**
+ * Represents the immutable target endpoint of a relation edge in a relation
update. The source
+ * endpoint is supplied by {@link RelationUpdate}.
+ *
+ * <p>The optional relation value is metadata stored on the edge itself rather
than on either
+ * endpoint entity. For example, a tag assignment value belongs to the
tag-to-metadata-object
+ * relation, not to the tag definition or metadata object.
+ */
+public final class RelationEdgeTarget {
+
+ private final NameIdentifier nameIdentifier;
+ private final Entity.EntityType entityType;
+ @Nullable private final String relationValue;
+
+ private RelationEdgeTarget(
+ NameIdentifier nameIdentifier, Entity.EntityType entityType, @Nullable
String relationValue) {
+ this.nameIdentifier = nameIdentifier;
+ this.entityType = entityType;
+ this.relationValue = relationValue;
+ }
+
+ /**
+ * Creates a relation edge target.
+ *
+ * @param nameIdentifier The target entity identifier.
+ * @param entityType The target entity type.
+ * @param relationValue Optional string value carried by the relation edge.
+ * @return A relation edge target.
+ */
+ public static RelationEdgeTarget of(
+ NameIdentifier nameIdentifier, Entity.EntityType entityType, @Nullable
String relationValue) {
+ Preconditions.checkArgument(nameIdentifier != null, "nameIdentifier must
not be null");
+ Preconditions.checkArgument(entityType != null, "entityType must not be
null");
+ return new RelationEdgeTarget(nameIdentifier, entityType, relationValue);
+ }
+
+ /**
+ * @return The target entity identifier.
+ */
+ public NameIdentifier nameIdentifier() {
+ return nameIdentifier;
+ }
+
+ /**
+ * @return The target entity type.
+ */
+ public Entity.EntityType entityType() {
+ return entityType;
+ }
+
+ /**
+ * @return The optional string value carried by the relation edge.
+ */
+ public Optional<String> relationValue() {
+ return Optional.ofNullable(relationValue);
+ }
+}
diff --git a/core/src/main/java/org/apache/gravitino/RelationQuery.java
b/core/src/main/java/org/apache/gravitino/RelationQuery.java
new file mode 100644
index 0000000000..2947a2ab76
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/RelationQuery.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino;
+
+import com.google.common.base.Preconditions;
+import java.util.Optional;
+import javax.annotation.Nullable;
+
+/**
+ * Describes a relation lookup from an anchor entity. The anchor may be either
endpoint of the
+ * relation edge, and the returned entities come from the opposite endpoint.
+ *
+ * <p>For reverse lookups, callers still use the same relation type. For
example, querying all
+ * metadata objects that have a specific policy uses {@link
+ * SupportsRelationOperations.Type#POLICY_METADATA_OBJECT_REL} with the policy
as the anchor entity,
+ * and querying all metadata objects that have a specific tag uses {@link
+ * SupportsRelationOperations.Type#TAG_METADATA_OBJECT_REL} with the tag as
the anchor entity.
+ *
+ * <p>The optional relation value is an exact string value carried by the
relation edge, such as a
+ * tag assignment value. A null value means the query should not filter by
relation value.
+ */
+public final class RelationQuery {
+
+ private final SupportsRelationOperations.Type relationType;
+ private final NameIdentifier anchorIdentifier;
+ private final Entity.EntityType anchorEntityType;
+ private final boolean allFields;
+ @Nullable private final String relationValue;
+
+ private RelationQuery(
+ SupportsRelationOperations.Type relationType,
+ NameIdentifier anchorIdentifier,
+ Entity.EntityType anchorEntityType,
+ boolean allFields,
+ @Nullable String relationValue) {
+ this.relationType = relationType;
+ this.anchorIdentifier = anchorIdentifier;
+ this.anchorEntityType = anchorEntityType;
+ this.allFields = allFields;
+ this.relationValue = relationValue;
+ }
+
+ /**
+ * Creates a relation query.
+ *
+ * @param relationType The type of relation.
+ * @param anchorIdentifier The anchor entity identifier.
+ * @param anchorEntityType The entity type that {@code anchorIdentifier}
represents.
+ * @param allFields Whether to fetch all fields.
+ * @param relationValue Optional exact string value carried by the relation
edge.
+ * @return A relation query.
+ */
+ public static RelationQuery of(
+ SupportsRelationOperations.Type relationType,
+ NameIdentifier anchorIdentifier,
+ Entity.EntityType anchorEntityType,
+ boolean allFields,
+ @Nullable String relationValue) {
+ Preconditions.checkArgument(relationType != null, "relationType must not
be null");
+ Preconditions.checkArgument(anchorIdentifier != null, "anchorIdentifier
must not be null");
+ Preconditions.checkArgument(anchorEntityType != null, "anchorEntityType
must not be null");
+ return new RelationQuery(
+ relationType, anchorIdentifier, anchorEntityType, allFields,
relationValue);
+ }
+
+ /**
+ * @return The type of relation.
+ */
+ public SupportsRelationOperations.Type relationType() {
+ return relationType;
+ }
+
+ /**
+ * @return The anchor entity identifier.
+ */
+ public NameIdentifier anchorIdentifier() {
+ return anchorIdentifier;
+ }
+
+ /**
+ * @return The entity type that {@link #anchorIdentifier()} represents.
+ */
+ public Entity.EntityType anchorEntityType() {
+ return anchorEntityType;
+ }
+
+ /**
+ * @return Whether to fetch all fields.
+ */
+ public boolean allFields() {
+ return allFields;
+ }
+
+ /**
+ * @return The optional exact string value carried by the relation edge.
+ */
+ public Optional<String> relationValue() {
+ return Optional.ofNullable(relationValue);
+ }
+}
diff --git a/core/src/main/java/org/apache/gravitino/RelationUpdate.java
b/core/src/main/java/org/apache/gravitino/RelationUpdate.java
new file mode 100644
index 0000000000..5a847cb62e
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/RelationUpdate.java
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino;
+
+import com.google.common.base.Preconditions;
+import java.util.Arrays;
+
+/**
+ * Describes an immutable relation update from a source entity to target
endpoints. Target endpoint
+ * identity and relation-edge attributes are carried by immutable {@link
RelationEdgeTarget}
+ * instances.
+ *
+ * <p>The target endpoint arrays are copied when the update is created and
when the targets are
+ * returned. The copy protects the array container from external mutation; the
elements themselves
+ * are immutable.
+ */
+public final class RelationUpdate {
+
+ private static final RelationEdgeTarget[] EMPTY_TARGETS = new
RelationEdgeTarget[0];
+
+ private final SupportsRelationOperations.Type relationType;
+ private final NameIdentifier sourceIdentifier;
+ private final Entity.EntityType sourceEntityType;
+ private final RelationEdgeTarget[] targetsToAdd;
+ private final RelationEdgeTarget[] targetsToRemove;
+
+ private RelationUpdate(
+ SupportsRelationOperations.Type relationType,
+ NameIdentifier sourceIdentifier,
+ Entity.EntityType sourceEntityType,
+ RelationEdgeTarget[] targetsToAdd,
+ RelationEdgeTarget[] targetsToRemove) {
+ this.relationType = relationType;
+ this.sourceIdentifier = sourceIdentifier;
+ this.sourceEntityType = sourceEntityType;
+ this.targetsToAdd = copyTargets(targetsToAdd, "targetsToAdd");
+ this.targetsToRemove = copyTargets(targetsToRemove, "targetsToRemove");
+ }
+
+ /**
+ * Creates a relation update.
+ *
+ * @param relationType The type of relation.
+ * @param sourceIdentifier The identifier of the source entity whose
relations are being updated.
+ * @param sourceEntityType The source entity type.
+ * @param targetsToAdd Target endpoints to associate with the source entity.
+ * @param targetsToRemove Target endpoints to disassociate from the source
entity.
+ * @return A relation update.
+ */
+ public static RelationUpdate of(
+ SupportsRelationOperations.Type relationType,
+ NameIdentifier sourceIdentifier,
+ Entity.EntityType sourceEntityType,
+ RelationEdgeTarget[] targetsToAdd,
+ RelationEdgeTarget[] targetsToRemove) {
+ Preconditions.checkArgument(relationType != null, "relationType must not
be null");
+ Preconditions.checkArgument(sourceIdentifier != null, "sourceIdentifier
must not be null");
+ Preconditions.checkArgument(sourceEntityType != null, "sourceEntityType
must not be null");
+ return new RelationUpdate(
+ relationType, sourceIdentifier, sourceEntityType, targetsToAdd,
targetsToRemove);
+ }
+
+ /**
+ * @return The type of relation.
+ */
+ public SupportsRelationOperations.Type relationType() {
+ return relationType;
+ }
+
+ /**
+ * @return The source entity identifier.
+ */
+ public NameIdentifier sourceIdentifier() {
+ return sourceIdentifier;
+ }
+
+ /**
+ * @return The source entity type.
+ */
+ public Entity.EntityType sourceEntityType() {
+ return sourceEntityType;
+ }
+
+ /**
+ * @return A copy of the target endpoint array to associate with the source
entity.
+ */
+ public RelationEdgeTarget[] targetsToAdd() {
+ return targetsToAdd.clone();
+ }
+
+ /**
+ * @return A copy of the target endpoint array to disassociate from the
source entity.
+ */
+ public RelationEdgeTarget[] targetsToRemove() {
+ return targetsToRemove.clone();
+ }
+
+ /**
+ * @return Whether any target endpoint carries relation values.
+ */
+ public boolean hasRelationValues() {
+ return Arrays.stream(targetsToAdd).anyMatch(target ->
target.relationValue().isPresent())
+ || Arrays.stream(targetsToRemove).anyMatch(target ->
target.relationValue().isPresent());
+ }
+
+ private static RelationEdgeTarget[] copyTargets(
+ RelationEdgeTarget[] targets, String parameterName) {
+ if (targets == null) {
+ return EMPTY_TARGETS;
+ }
+
+ for (RelationEdgeTarget target : targets) {
+ Preconditions.checkArgument(target != null, "%s must not contain null",
parameterName);
+ }
+
+ return targets.clone();
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/SupportsRelationOperations.java
b/core/src/main/java/org/apache/gravitino/SupportsRelationOperations.java
index 25c4fc60b1..ca1ba1c230 100644
--- a/core/src/main/java/org/apache/gravitino/SupportsRelationOperations.java
+++ b/core/src/main/java/org/apache/gravitino/SupportsRelationOperations.java
@@ -19,6 +19,7 @@
package org.apache.gravitino;
import java.io.IOException;
+import java.util.Arrays;
import java.util.List;
import org.apache.gravitino.exceptions.NoSuchEntityException;
@@ -60,7 +61,8 @@ public interface SupportsRelationOperations {
}
/**
- * List the entities according to a given entity in a specific relation.
+ * List the entities according to a given entity in a specific relation.
Compatibility adapter for
+ * {@link #listEntitiesByRelation(RelationQuery)}.
*
* @param <E> the type of entities returned.
* @param relType The type of relation.
@@ -88,9 +90,27 @@ public interface SupportsRelationOperations {
* The end entity type is determined by the relType parameter and start
entity.
* </pre>
*/
- <E extends Entity & HasIdentifier> List<E> listEntitiesByRelation(
+ default <E extends Entity & HasIdentifier> List<E> listEntitiesByRelation(
Type relType, NameIdentifier nameIdentifier, Entity.EntityType
identType, boolean allFields)
- throws IOException;
+ throws IOException {
+ return listEntitiesByRelation(
+ RelationQuery.of(relType, nameIdentifier, identType, allFields, null));
+ }
+
+ /**
+ * Canonical relation query API. Implementations should add future
relation-edge query attributes
+ * to {@link RelationQuery} instead of introducing new relation-specific
overloads.
+ *
+ * @param <E> The type of entities returned.
+ * @param query The relation query.
+ * @return The list of entities.
+ * @throws IOException When occurs storage issues, it will throw IOException.
+ */
+ default <E extends Entity & HasIdentifier> List<E>
listEntitiesByRelation(RelationQuery query)
+ throws IOException {
+ throw new UnsupportedOperationException(
+ "listEntitiesByRelation with RelationQuery is not supported by this
implementation");
+ }
/**
* Retrieves the relations for a batch of source entities in a single call.
@@ -194,8 +214,15 @@ public interface SupportsRelationOperations {
}
/**
- * Updates the relations for a given entity by adding a set of new relations
and removing another
- * set of relations.
+ * Updates the relations for a given source entity by adding a set of new
relation targets and
+ * removing another set of relation targets. Compatibility adapter for {@link
+ * #updateEntityRelations(RelationUpdate)}.
+ *
+ * <p>For {@link Type#POLICY_METADATA_OBJECT_REL} and {@link
Type#TAG_METADATA_OBJECT_REL}, this
+ * update adapter treats the source entity as the metadata object and the
destination identifiers
+ * as policies or tags. Reverse traversal is done by {@link
+ * #listEntitiesByRelation(RelationQuery)} using the same relation type with
the policy or tag as
+ * the query anchor.
*
* @param <E> The type of the entity returned in the list, which represents
the final state of
* related entities.
@@ -217,7 +244,55 @@ public interface SupportsRelationOperations {
NameIdentifier[] destEntitiesToAdd,
NameIdentifier[] destEntitiesToRemove)
throws IOException, NoSuchEntityException, EntityAlreadyExistsException {
+ return updateEntityRelations(
+ RelationUpdate.of(
+ relType,
+ srcEntityIdent,
+ srcEntityType,
+ toRelationEdgeTargets(relType, srcEntityType, destEntitiesToAdd),
+ toRelationEdgeTargets(relType, srcEntityType,
destEntitiesToRemove)));
+ }
+
+ /**
+ * Canonical relation update API. Implementations should add future
relation-edge update
+ * attributes to {@link RelationUpdate} or {@link RelationEdgeTarget}
instead of introducing new
+ * relation-specific overloads.
+ *
+ * @param <E> The type of the entity returned in the list, which represents
the final state of
+ * related entities.
+ * @param update The relation update.
+ * @return A list of entities that are related to the given entity after the
update.
+ * @throws IOException If a storage-related error occurs.
+ * @throws NoSuchEntityException If any of the specified entities does not
exist.
+ * @throws EntityAlreadyExistsException If a relation to be added already
exists.
+ */
+ default <E extends Entity & HasIdentifier> List<E>
updateEntityRelations(RelationUpdate update)
+ throws IOException, NoSuchEntityException, EntityAlreadyExistsException {
throw new UnsupportedOperationException(
- "updateEntityRelations is not supported by this implementation");
+ "updateEntityRelations with RelationUpdate is not supported by this
implementation");
+ }
+
+ private static RelationEdgeTarget[] toRelationEdgeTargets(
+ Type relType, Entity.EntityType srcEntityType, NameIdentifier[]
nameIdentifiers) {
+ if (nameIdentifiers == null) {
+ return null;
+ }
+
+ Entity.EntityType targetEntityType = relationUpdateTargetType(relType,
srcEntityType);
+ return Arrays.stream(nameIdentifiers)
+ .map(nameIdentifier -> RelationEdgeTarget.of(nameIdentifier,
targetEntityType, null))
+ .toArray(RelationEdgeTarget[]::new);
+ }
+
+ private static Entity.EntityType relationUpdateTargetType(
+ Type relType, Entity.EntityType srcEntityType) {
+ switch (relType) {
+ case POLICY_METADATA_OBJECT_REL:
+ return Entity.EntityType.POLICY;
+ case TAG_METADATA_OBJECT_REL:
+ return Entity.EntityType.TAG;
+ default:
+ return srcEntityType;
+ }
}
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
index f14f41c9a1..cd303eb0a9 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
@@ -21,9 +21,11 @@ package org.apache.gravitino.storage.relational;
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_STORE;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
@@ -41,6 +43,9 @@ import org.apache.gravitino.EntityStore;
import org.apache.gravitino.HasIdentifier;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
+import org.apache.gravitino.RelationEdgeTarget;
+import org.apache.gravitino.RelationQuery;
+import org.apache.gravitino.RelationUpdate;
import org.apache.gravitino.RelationalEntity;
import org.apache.gravitino.SupportsExternalIdOperations;
import org.apache.gravitino.SupportsRelationOperations;
@@ -471,22 +476,55 @@ public class RelationalEntityStore
NameIdentifier[] destEntitiesToAdd,
NameIdentifier[] destEntitiesToRemove)
throws IOException, NoSuchEntityException, EntityAlreadyExistsException {
+ return updateEntityRelations(
+ RelationUpdate.of(
+ relType,
+ srcEntityIdent,
+ srcEntityType,
+ toRelationEdgeTargets(relType, srcEntityType, destEntitiesToAdd),
+ toRelationEdgeTargets(relType, srcEntityType,
destEntitiesToRemove)));
+ }
- // Invalidate after the backend write, not before. Invalidating before
creates a window where
- // a concurrent read can repopulate the cache with stale pre-commit data.
- List<E> result =
- backend.updateEntityRelations(
- relType, srcEntityIdent, srcEntityType, destEntitiesToAdd,
destEntitiesToRemove);
-
- cache.invalidate(srcEntityIdent, srcEntityType, relType);
- for (NameIdentifier destToAdd : destEntitiesToAdd) {
- cache.invalidate(destToAdd, srcEntityType, relType);
+ @Override
+ public <E extends Entity & HasIdentifier> List<E>
listEntitiesByRelation(RelationQuery query)
+ throws IOException {
+ if (query.relationValue().isPresent()) {
+ return backend.listEntitiesByRelation(query);
}
- for (NameIdentifier destToRemove : destEntitiesToRemove) {
- cache.invalidate(destToRemove, srcEntityType, relType);
+ return listEntitiesByRelation(
+ query.relationType(),
+ query.anchorIdentifier(),
+ query.anchorEntityType(),
+ query.allFields());
+ }
+
+ @Override
+ public <E extends Entity & HasIdentifier> List<E>
updateEntityRelations(RelationUpdate update)
+ throws IOException, NoSuchEntityException, EntityAlreadyExistsException {
+ validateRelationTargetTypes(update);
+
+ RelationEdgeTarget[] targetsToAdd = update.targetsToAdd();
+ RelationEdgeTarget[] targetsToRemove = update.targetsToRemove();
+ List<E> result;
+ if (update.hasRelationValues()) {
+ result = backend.updateEntityRelations(update);
+ } else {
+ result =
+ backend.updateEntityRelations(
+ update.relationType(),
+ update.sourceIdentifier(),
+ update.sourceEntityType(),
+ toNameIdentifiers(targetsToAdd),
+ toNameIdentifiers(targetsToRemove));
}
+ Entity.EntityType targetEntityType =
+ relationUpdateTargetType(update.relationType(),
update.sourceEntityType());
+ cache.invalidate(update.sourceIdentifier(), update.sourceEntityType(),
update.relationType());
+ invalidateRelationTargetCache(update.relationType(), targetEntityType,
targetsToAdd);
+ invalidateRelationTargetCache(update.relationType(), targetEntityType,
targetsToRemove);
+
return result;
}
@@ -503,6 +541,62 @@ public class RelationalEntityStore
backend.batchPut(entities, overwritten);
}
+ private void invalidateRelationTargetCache(
+ Type relType, Entity.EntityType targetEntityType, RelationEdgeTarget[]
relationTargets) {
+ for (RelationEdgeTarget relationTarget : relationTargets) {
+ cache.invalidate(relationTarget.nameIdentifier(), targetEntityType,
relType);
+ }
+ }
+
+ private static void validateRelationTargetTypes(RelationUpdate update) {
+ Entity.EntityType targetEntityType =
+ relationUpdateTargetType(update.relationType(),
update.sourceEntityType());
+ validateRelationTargetTypes(update.relationType(), targetEntityType,
update.targetsToAdd());
+ validateRelationTargetTypes(update.relationType(), targetEntityType,
update.targetsToRemove());
+ }
+
+ private static void validateRelationTargetTypes(
+ Type relType, Entity.EntityType targetEntityType, RelationEdgeTarget[]
relationTargets) {
+ for (RelationEdgeTarget relationTarget : relationTargets) {
+ Preconditions.checkArgument(
+ relationTarget.entityType() == targetEntityType,
+ "Relation target type %s does not match expected destination type %s
for relation type %s",
+ relationTarget.entityType(),
+ targetEntityType,
+ relType);
+ }
+ }
+
+ private static RelationEdgeTarget[] toRelationEdgeTargets(
+ Type relType, Entity.EntityType srcEntityType, NameIdentifier[]
nameIdentifiers) {
+ if (nameIdentifiers == null) {
+ return new RelationEdgeTarget[0];
+ }
+
+ Entity.EntityType targetEntityType = relationUpdateTargetType(relType,
srcEntityType);
+ return Arrays.stream(nameIdentifiers)
+ .map(nameIdentifier -> RelationEdgeTarget.of(nameIdentifier,
targetEntityType, null))
+ .toArray(RelationEdgeTarget[]::new);
+ }
+
+ private static NameIdentifier[] toNameIdentifiers(RelationEdgeTarget[]
relationTargets) {
+ return Arrays.stream(relationTargets)
+ .map(RelationEdgeTarget::nameIdentifier)
+ .toArray(NameIdentifier[]::new);
+ }
+
+ private static Entity.EntityType relationUpdateTargetType(
+ Type relType, Entity.EntityType srcEntityType) {
+ switch (relType) {
+ case POLICY_METADATA_OBJECT_REL:
+ return Entity.EntityType.POLICY;
+ case TAG_METADATA_OBJECT_REL:
+ return Entity.EntityType.TAG;
+ default:
+ return srcEntityType;
+ }
+ }
+
private <E extends Entity & HasIdentifier>
Optional<List<RelationalEntity<?>>> getCachedRelations(
SupportsRelationOperations.Type relType,
NameIdentifier nameIdentifier,
diff --git
a/core/src/test/java/org/apache/gravitino/TestSupportsRelationOperations.java
b/core/src/test/java/org/apache/gravitino/TestSupportsRelationOperations.java
new file mode 100644
index 0000000000..85e621a222
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/TestSupportsRelationOperations.java
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino;
+
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestSupportsRelationOperations {
+
+ @Test
+ public void testRelationOperationsDoesNotExposeTagSpecificMethods() {
+ Set<String> methodNames =
+ Arrays.stream(SupportsRelationOperations.class.getDeclaredMethods())
+ .map(Method::getName)
+ .collect(Collectors.toSet());
+
+ Assertions.assertFalse(methodNames.contains("listMetadataObjectsForTag"));
+ Assertions.assertFalse(methodNames.contains("updateTagRelations"));
+ }
+
+ @Test
+ public void testRelationEdgeTargetCarriesOptionalRelationValue() {
+ NameIdentifier targetIdent = NameIdentifier.of("metalake", "tag");
+
+ RelationEdgeTarget targetWithValue =
+ RelationEdgeTarget.of(targetIdent, Entity.EntityType.TAG, "finance");
+ Assertions.assertEquals(Entity.EntityType.TAG,
targetWithValue.entityType());
+ Assertions.assertEquals(targetIdent, targetWithValue.nameIdentifier());
+ Assertions.assertTrue(targetWithValue.relationValue().isPresent());
+ Assertions.assertEquals("finance", targetWithValue.relationValue().get());
+
+ RelationEdgeTarget targetWithNoValue =
+ RelationEdgeTarget.of(targetIdent, Entity.EntityType.TAG, null);
+ Assertions.assertEquals(Entity.EntityType.TAG,
targetWithNoValue.entityType());
+ Assertions.assertEquals(targetIdent, targetWithNoValue.nameIdentifier());
+ Assertions.assertFalse(targetWithNoValue.relationValue().isPresent());
+ }
+
+ @Test
+ public void testPrimitiveListDelegatesToRelationQuery() throws IOException {
+ RecordingRelationOperations relationOperations = new
RecordingRelationOperations();
+ NameIdentifier tagIdent = NameIdentifier.of("metalake", "tag");
+
+ relationOperations.listEntitiesByRelation(
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
+ tagIdent,
+ Entity.EntityType.TAG,
+ false);
+
+ RelationQuery query = relationOperations.relationQuery;
+ Assertions.assertEquals(
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
query.relationType());
+ Assertions.assertEquals(tagIdent, query.anchorIdentifier());
+ Assertions.assertEquals(Entity.EntityType.TAG, query.anchorEntityType());
+ Assertions.assertFalse(query.allFields());
+ Assertions.assertFalse(query.relationValue().isPresent());
+ }
+
+ @Test
+ public void testPrimitiveUpdateDelegatesToRelationUpdate()
+ throws IOException, NoSuchEntityException, EntityAlreadyExistsException {
+ RecordingRelationOperations relationOperations = new
RecordingRelationOperations();
+ NameIdentifier srcIdent = NameIdentifier.of("metalake", "catalog",
"schema", "table");
+ NameIdentifier targetIdent = NameIdentifier.of("metalake", "tag");
+ NameIdentifier[] targetsToAdd = new NameIdentifier[] {targetIdent};
+ NameIdentifier[] targetsToRemove = new NameIdentifier[0];
+
+ relationOperations.updateEntityRelations(
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
+ srcIdent,
+ Entity.EntityType.TABLE,
+ targetsToAdd,
+ targetsToRemove);
+
+ RelationUpdate update = relationOperations.relationUpdate;
+ Assertions.assertEquals(
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
update.relationType());
+ Assertions.assertEquals(srcIdent, update.sourceIdentifier());
+ Assertions.assertEquals(Entity.EntityType.TABLE,
update.sourceEntityType());
+ Assertions.assertEquals(1, update.targetsToAdd().length);
+ Assertions.assertEquals(targetIdent,
update.targetsToAdd()[0].nameIdentifier());
+ Assertions.assertEquals(Entity.EntityType.TAG,
update.targetsToAdd()[0].entityType());
+
Assertions.assertFalse(update.targetsToAdd()[0].relationValue().isPresent());
+ Assertions.assertEquals(0, update.targetsToRemove().length);
+ }
+
+ @Test
+ public void testRelationUpdateCopiesTargetArrays() {
+ NameIdentifier srcIdent = NameIdentifier.of("metalake", "catalog",
"schema", "table");
+ RelationEdgeTarget originalTarget =
+ RelationEdgeTarget.of(NameIdentifier.of("metalake", "tag"),
Entity.EntityType.TAG, "dev");
+ RelationEdgeTarget replacementTarget =
+ RelationEdgeTarget.of(NameIdentifier.of("metalake", "tag2"),
Entity.EntityType.TAG, "prod");
+ RelationEdgeTarget[] targetsToAdd = new RelationEdgeTarget[]
{originalTarget};
+ RelationEdgeTarget[] targetsToRemove = new RelationEdgeTarget[]
{originalTarget};
+
+ RelationUpdate update =
+ RelationUpdate.of(
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
+ srcIdent,
+ Entity.EntityType.TABLE,
+ targetsToAdd,
+ targetsToRemove);
+
+ targetsToAdd[0] = replacementTarget;
+ targetsToRemove[0] = replacementTarget;
+ Assertions.assertSame(originalTarget, update.targetsToAdd()[0]);
+ Assertions.assertSame(originalTarget, update.targetsToRemove()[0]);
+
+ RelationEdgeTarget[] returnedTargetsToAdd = update.targetsToAdd();
+ RelationEdgeTarget[] returnedTargetsToRemove = update.targetsToRemove();
+ returnedTargetsToAdd[0] = replacementTarget;
+ returnedTargetsToRemove[0] = replacementTarget;
+ Assertions.assertSame(originalTarget, update.targetsToAdd()[0]);
+ Assertions.assertSame(originalTarget, update.targetsToRemove()[0]);
+ }
+
+ private static class RecordingRelationOperations implements
SupportsRelationOperations {
+ private RelationQuery relationQuery;
+ private RelationUpdate relationUpdate;
+
+ @Override
+ public <E extends Entity & HasIdentifier> List<E>
listEntitiesByRelation(RelationQuery query)
+ throws IOException {
+ this.relationQuery = query;
+ return List.of();
+ }
+
+ @Override
+ public List<RelationalEntity<?>> batchListEntitiesByRelation(
+ Type relType, List<NameIdentifier> nameIdentifiers, Entity.EntityType
identType)
+ throws IOException {
+ return List.of();
+ }
+
+ @Override
+ public <E extends Entity & HasIdentifier> E getEntityByRelation(
+ Type relType,
+ NameIdentifier srcIdentifier,
+ Entity.EntityType srcType,
+ NameIdentifier destEntityIdent)
+ throws IOException, NoSuchEntityException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void insertRelation(
+ Type relType,
+ NameIdentifier srcIdentifier,
+ Entity.EntityType srcType,
+ NameIdentifier dstIdentifier,
+ Entity.EntityType dstType,
+ boolean override)
+ throws IOException {}
+
+ @Override
+ public <E extends Entity & HasIdentifier> List<E>
updateEntityRelations(RelationUpdate update)
+ throws IOException, NoSuchEntityException,
EntityAlreadyExistsException {
+ this.relationUpdate = update;
+ return List.of();
+ }
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStore.java
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStore.java
index 042c29681b..e9451c5895 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStore.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStore.java
@@ -30,6 +30,9 @@ import org.apache.gravitino.Configs;
import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityAlreadyExistsException;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.RelationEdgeTarget;
+import org.apache.gravitino.RelationQuery;
+import org.apache.gravitino.RelationUpdate;
import org.apache.gravitino.SupportsRelationOperations;
import org.apache.gravitino.cache.NoOpsCache;
import org.apache.gravitino.exceptions.NoSuchEntityException;
@@ -178,22 +181,22 @@ public class TestRelationalEntityStore {
Mockito.verify(cache, Mockito.never())
.invalidate(
destToAdd,
- Entity.EntityType.TABLE,
+ Entity.EntityType.TAG,
SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
Mockito.verify(cache, Mockito.never())
.invalidate(
destToRemove,
- Entity.EntityType.TABLE,
+ Entity.EntityType.TAG,
SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
return List.of();
})
.when(backend)
.updateEntityRelations(
- SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
- src,
- Entity.EntityType.TABLE,
- destEntitiesToAdd,
- destEntitiesToRemove);
+ eq(SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL),
+ eq(src),
+ eq(Entity.EntityType.TABLE),
+ any(NameIdentifier[].class),
+ any(NameIdentifier[].class));
store.updateEntityRelations(
SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
@@ -206,11 +209,93 @@ public class TestRelationalEntityStore {
inOrder
.verify(backend)
.updateEntityRelations(
+ eq(SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL),
+ eq(src),
+ eq(Entity.EntityType.TABLE),
+ any(NameIdentifier[].class),
+ any(NameIdentifier[].class));
+ inOrder
+ .verify(cache)
+ .invalidate(
+ src, Entity.EntityType.TABLE,
SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
+ inOrder
+ .verify(cache)
+ .invalidate(
+ destToAdd,
+ Entity.EntityType.TAG,
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
+ inOrder
+ .verify(cache)
+ .invalidate(
+ destToRemove,
+ Entity.EntityType.TAG,
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
+ }
+
+ @Test
+ void testListEntitiesByRelationWithRelationValueDelegatesToBackend() throws
IOException {
+ NameIdentifier tag = NameIdentifier.of("metalake", "tag1");
+ RelationQuery query =
+ RelationQuery.of(
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
+ tag,
+ Entity.EntityType.TAG,
+ true,
+ "dev");
+
+ store.listEntitiesByRelation(query);
+
+ Mockito.verify(backend).listEntitiesByRelation(query);
+ }
+
+ @Test
+ void testUpdateRelationWithValuesInvalidatesCacheAfterBackendUpdate()
+ throws IOException, NoSuchEntityException, EntityAlreadyExistsException,
+ IllegalAccessException {
+ NameIdentifier src = NameIdentifier.of("metalake", "catalog", "schema",
"table1");
+ NameIdentifier destToAdd = NameIdentifier.of("metalake", "tag1");
+ NameIdentifier destToRemove = NameIdentifier.of("metalake", "tag2");
+ RelationEdgeTarget[] destEntitiesToAdd =
+ new RelationEdgeTarget[] {RelationEdgeTarget.of(destToAdd,
Entity.EntityType.TAG, "dev")};
+ RelationEdgeTarget[] destEntitiesToRemove =
+ new RelationEdgeTarget[] {
+ RelationEdgeTarget.of(destToRemove, Entity.EntityType.TAG, "prod")
+ };
+ RelationUpdate update =
+ RelationUpdate.of(
SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
src,
Entity.EntityType.TABLE,
destEntitiesToAdd,
destEntitiesToRemove);
+ NoOpsCache cache = (NoOpsCache) FieldUtils.readField(store, "cache", true);
+
+ Mockito.doAnswer(
+ invocation -> {
+ Mockito.verify(cache, Mockito.never())
+ .invalidate(
+ src,
+ Entity.EntityType.TABLE,
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
+ Mockito.verify(cache, Mockito.never())
+ .invalidate(
+ destToAdd,
+ Entity.EntityType.TAG,
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
+ Mockito.verify(cache, Mockito.never())
+ .invalidate(
+ destToRemove,
+ Entity.EntityType.TAG,
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
+ return List.of();
+ })
+ .when(backend)
+ .updateEntityRelations(update);
+
+ store.updateEntityRelations(update);
+
+ InOrder inOrder = Mockito.inOrder(backend, cache);
+ inOrder.verify(backend).updateEntityRelations(update);
inOrder
.verify(cache)
.invalidate(
@@ -219,13 +304,30 @@ public class TestRelationalEntityStore {
.verify(cache)
.invalidate(
destToAdd,
- Entity.EntityType.TABLE,
+ Entity.EntityType.TAG,
SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
inOrder
.verify(cache)
.invalidate(
destToRemove,
- Entity.EntityType.TABLE,
+ Entity.EntityType.TAG,
SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
}
+
+ @Test
+ void testUpdateRelationRejectsMismatchedTargetTypeBeforeBackendUpdate() {
+ NameIdentifier src = NameIdentifier.of("metalake", "catalog", "schema",
"table1");
+ NameIdentifier tag = NameIdentifier.of("metalake", "tag1");
+ RelationUpdate update =
+ RelationUpdate.of(
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
+ src,
+ Entity.EntityType.TABLE,
+ new RelationEdgeTarget[] {RelationEdgeTarget.of(tag,
Entity.EntityType.TABLE, "dev")},
+ new RelationEdgeTarget[0]);
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
store.updateEntityRelations(update));
+ Mockito.verifyNoInteractions(backend);
+ }
}
