[TEST]

core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java

@@ -195,6 +195,23 @@ public boolean invalidate(
         });
   }
 
+  /** {@inheritDoc} */
+  @Override
+  public boolean invalidateRelationCache(
+      NameIdentifier ident, Entity.EntityType type, SupportsRelationOperations.Type relType) {
+    checkArguments(ident, type, relType);
+
+    EntityCacheRelationKey entityCacheKey = EntityCacheRelationKey.of(ident, type, relType);
+    return segmentedLock.withLock(
+        entityCacheKey,
+        () -> {
+          cacheData.invalidate(entityCacheKey);
+          cacheIndex.remove(entityCacheKey.toString());
+          reverseIndex.remove(entityCacheKey);
+          return true;
+        });
+  }
+
   /** {@inheritDoc} */
   @Override
   public boolean invalidate(NameIdentifier ident, Entity.EntityType type) {

core/src/main/java/org/apache/gravitino/cache/NoOpsCache.java

@@ -125,6 +125,13 @@ public boolean invalidate(
     return false;
   }
 
+  /** {@inheritDoc} */
+  @Override
+  public boolean invalidateRelationCache(
+      NameIdentifier ident, Entity.EntityType type, SupportsRelationOperations.Type relType) {
+    return false;
+  }
+
   /** {@inheritDoc} */
   @Override
   public boolean contains(

core/src/main/java/org/apache/gravitino/cache/SupportsRelationEntityCache.java

@@ -56,6 +56,17 @@ <E extends Entity & HasIdentifier> Optional<List<E>> getIfPresent(
   boolean invalidate(
       NameIdentifier ident, Entity.EntityType type, SupportsRelationOperations.Type relType);
 
+  /**
+   * Invalidates only the exact cached relation for the given entity and relation type.
+   *
+   * @param ident the name identifier
+   * @param type the entity type
+   * @param relType the relation type
+   * @return true if the cache entry was removed
+   */
+  boolean invalidateRelationCache(
+      NameIdentifier ident, Entity.EntityType type, SupportsRelationOperations.Type relType);
+
   /**
    * Checks whether an entity with the given name identifier, type, and relation type is present in
    * the cache.

core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java

@@ -41,15 +41,21 @@
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.RelationalEntity;
 import org.apache.gravitino.SupportsRelationOperations;
+import org.apache.gravitino.authorization.SecurableObject;
 import org.apache.gravitino.cache.CacheFactory;
 import org.apache.gravitino.cache.CachedEntityIdResolver;
 import org.apache.gravitino.cache.EntityCache;
 import org.apache.gravitino.cache.EntityCacheKey;
 import org.apache.gravitino.cache.EntityCacheRelationKey;
 import org.apache.gravitino.cache.NoOpsCache;
 import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.meta.GroupEntity;
+import org.apache.gravitino.meta.RoleEntity;
+import org.apache.gravitino.meta.UserEntity;
 import org.apache.gravitino.storage.relational.service.EntityIdService;
 import org.apache.gravitino.utils.Executable;
+import org.apache.gravitino.utils.MetadataObjectUtil;
+import org.apache.gravitino.utils.NameIdentifierUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -143,6 +149,7 @@ public boolean exists(NameIdentifier ident, Entity.EntityType entityType) throws
   public <E extends Entity & HasIdentifier> void put(E e, boolean overwritten)
       throws IOException, EntityAlreadyExistsException {
     backend.insert(e, overwritten);
+    invalidateDerivedRelationCaches(e);
     cache.put(e);
   }
 
@@ -152,6 +159,7 @@ public <E extends Entity & HasIdentifier> E update(
       throws IOException, NoSuchEntityException, EntityAlreadyExistsException {
     E updatedEntity = backend.update(ident, entityType, updater);
     cache.invalidate(ident, entityType);
+    invalidateDerivedRelationCaches(updatedEntity);
     return updatedEntity;
   }
 
@@ -412,6 +420,58 @@ public <E extends Entity & HasIdentifier> void batchPut(List<E> entities, boolea
     backend.batchPut(entities, overwritten);
   }
 
+  private void invalidateDerivedRelationCaches(Entity entity) {
+    if (entity instanceof RoleEntity) {
+      invalidateMetadataObjectRoleRelCache((RoleEntity) entity);
+    } else if (entity instanceof UserEntity) {
+      invalidateRoleUserRelCache((UserEntity) entity);
+    } else if (entity instanceof GroupEntity) {
+      invalidateRoleGroupRelCache((GroupEntity) entity);
+    }
+  }
+
+  private void invalidateMetadataObjectRoleRelCache(RoleEntity roleEntity) {
+    if (roleEntity.securableObjects() == null) {
+      return;
+    }
+
+    String metalakeName = roleEntity.namespace().level(0);
+    for (SecurableObject securableObject : roleEntity.securableObjects()) {
+      NameIdentifier identifier = MetadataObjectUtil.toEntityIdent(metalakeName, securableObject);
+      Entity.EntityType entityType = MetadataObjectUtil.toEntityType(securableObject.type());
+      cache.invalidateRelationCache(
+          identifier, entityType, SupportsRelationOperations.Type.METADATA_OBJECT_ROLE_REL);
+    }
+  }
+
+  private void invalidateRoleUserRelCache(UserEntity userEntity) {
+    if (userEntity.roleNames() == null) {
+      return;
+    }
+
+    String metalakeName = userEntity.namespace().level(0);
+    for (String roleName : userEntity.roleNames()) {
+      cache.invalidateRelationCache(
+          NameIdentifierUtil.ofRole(metalakeName, roleName),
+          Entity.EntityType.ROLE,
+          SupportsRelationOperations.Type.ROLE_USER_REL);
+    }
+  }
+
+  private void invalidateRoleGroupRelCache(GroupEntity groupEntity) {
+    if (groupEntity.roleNames() == null) {
+      return;
+    }
+
+    String metalakeName = groupEntity.namespace().level(0);
+    for (String roleName : groupEntity.roleNames()) {
+      cache.invalidateRelationCache(
+          NameIdentifierUtil.ofRole(metalakeName, roleName),
+          Entity.EntityType.ROLE,
+          SupportsRelationOperations.Type.ROLE_GROUP_REL);
+    }
+  }
+
   private <E extends Entity & HasIdentifier> Optional<List<RelationalEntity<?>>> getCachedRelations(
       SupportsRelationOperations.Type relType,
       NameIdentifier nameIdentifier,