[#12179] feat(core): Support policy-tag association

core/src/main/java/org/apache/gravitino/SupportsRelationOperations.java

@@ -42,6 +42,8 @@ enum Type {
     POLICY_METADATA_OBJECT_REL,
     /** Metadata object to tag relationship */
     TAG_METADATA_OBJECT_REL,
+    /** Policy and tag relationship */
+    POLICY_TAG_REL,
   }
 
   /**

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

@@ -105,7 +105,8 @@ public ReverseIndexCache getReverseIndex() {
           SupportsRelationOperations.Type.ROLE_USER_REL,
           SupportsRelationOperations.Type.ROLE_GROUP_REL,
           SupportsRelationOperations.Type.POLICY_METADATA_OBJECT_REL,
-          SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
+          SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
+          SupportsRelationOperations.Type.POLICY_TAG_REL);
 
   /**
    * Constructs a new {@link CaffeineEntityCache}.

core/src/main/java/org/apache/gravitino/hook/PolicyHookDispatcher.java

@@ -29,6 +29,7 @@
 import org.apache.gravitino.policy.PolicyChange;
 import org.apache.gravitino.policy.PolicyContent;
 import org.apache.gravitino.policy.PolicyDispatcher;
+import org.apache.gravitino.tag.Tag;
 import org.apache.gravitino.utils.NameIdentifierUtil;
 import org.apache.gravitino.utils.PrincipalUtils;
 
@@ -104,6 +105,16 @@ public MetadataObject[] listMetadataObjectsForPolicy(String metalake, String pol
     return dispatcher.listMetadataObjectsForPolicy(metalake, policyName);
   }
 
+  @Override
+  public String[] listTagsForPolicy(String metalake, String policyName) {
+    return dispatcher.listTagsForPolicy(metalake, policyName);
+  }
+
+  @Override
+  public Tag[] listTagInfosForPolicy(String metalake, String policyName) {
+    return dispatcher.listTagInfosForPolicy(metalake, policyName);
+  }
+
   @Override
   public PolicyEntity[] listPolicyInfosForMetadataObject(
       String metalake, MetadataObject metadataObject) {

core/src/main/java/org/apache/gravitino/hook/TagHookDispatcher.java

@@ -25,6 +25,7 @@
 import org.apache.gravitino.authorization.OwnerDispatcher;
 import org.apache.gravitino.exceptions.NoSuchTagException;
 import org.apache.gravitino.exceptions.TagAlreadyExistsException;
+import org.apache.gravitino.meta.PolicyEntity;
 import org.apache.gravitino.tag.Tag;
 import org.apache.gravitino.tag.TagChange;
 import org.apache.gravitino.tag.TagDispatcher;
@@ -89,6 +90,22 @@ public MetadataObject[] listMetadataObjectsForTag(String metalake, String name)
     return dispatcher.listMetadataObjectsForTag(metalake, name);
   }
 
+  @Override
+  public String[] listPoliciesForTag(String metalake, String name) {
+    return dispatcher.listPoliciesForTag(metalake, name);
+  }
+
+  @Override
+  public PolicyEntity[] listPolicyInfosForTag(String metalake, String name) {
+    return dispatcher.listPolicyInfosForTag(metalake, name);
+  }
+
+  @Override
+  public String[] associatePoliciesForTag(
+      String metalake, String name, String[] policiesToAdd, String[] policiesToRemove) {
+    return dispatcher.associatePoliciesForTag(metalake, name, policiesToAdd, policiesToRemove);
+  }
+
   @Override
   public String[] listTagsForMetadataObject(String metalake, MetadataObject metadataObject) {
     return dispatcher.listTagsForMetadataObject(metalake, metadataObject);

core/src/main/java/org/apache/gravitino/listener/PolicyEventDispatcher.java

@@ -63,6 +63,7 @@
 import org.apache.gravitino.policy.PolicyChange;
 import org.apache.gravitino.policy.PolicyContent;
 import org.apache.gravitino.policy.PolicyDispatcher;
+import org.apache.gravitino.tag.Tag;
 import org.apache.gravitino.utils.NameIdentifierUtil;
 import org.apache.gravitino.utils.PrincipalUtils;
 
@@ -288,6 +289,16 @@ public MetadataObject[] listMetadataObjectsForPolicy(String metalake, String pol
     }
   }
 
+  @Override
+  public String[] listTagsForPolicy(String metalake, String policyName) {
+    return dispatcher.listTagsForPolicy(metalake, policyName);
+  }
+
+  @Override
+  public Tag[] listTagInfosForPolicy(String metalake, String policyName) {
+    return dispatcher.listTagInfosForPolicy(metalake, policyName);
+  }
+
   @Override
   public PolicyEntity[] listPolicyInfosForMetadataObject(
       String metalake, MetadataObject metadataObject) {

core/src/main/java/org/apache/gravitino/listener/TagEventDispatcher.java

@@ -56,6 +56,7 @@
 import org.apache.gravitino.listener.api.event.ListTagsInfoPreEvent;
 import org.apache.gravitino.listener.api.event.ListTagsPreEvent;
 import org.apache.gravitino.listener.api.info.TagInfo;
+import org.apache.gravitino.meta.PolicyEntity;
 import org.apache.gravitino.tag.Tag;
 import org.apache.gravitino.tag.TagChange;
 import org.apache.gravitino.tag.TagDispatcher;
@@ -210,6 +211,22 @@ public MetadataObject[] listMetadataObjectsForTag(String metalake, String name)
     }
   }
 
+  @Override
+  public String[] listPoliciesForTag(String metalake, String name) {
+    return dispatcher.listPoliciesForTag(metalake, name);
+  }
+
+  @Override
+  public PolicyEntity[] listPolicyInfosForTag(String metalake, String name) {
+    return dispatcher.listPolicyInfosForTag(metalake, name);
+  }
+
+  @Override
+  public String[] associatePoliciesForTag(
+      String metalake, String name, String[] policiesToAdd, String[] policiesToRemove) {
+    return dispatcher.associatePoliciesForTag(metalake, name, policiesToAdd, policiesToRemove);
+  }
+
   @Override
   public String[] listTagsForMetadataObject(String metalake, MetadataObject metadataObject) {
     eventBus.dispatchEvent(

core/src/main/java/org/apache/gravitino/policy/PolicyDispatcher.java

@@ -25,6 +25,7 @@
 import org.apache.gravitino.exceptions.NoSuchPolicyException;
 import org.apache.gravitino.exceptions.PolicyAlreadyExistsException;
 import org.apache.gravitino.meta.PolicyEntity;
+import org.apache.gravitino.tag.Tag;
 
 /**
  * The interface provides functionalities for managing policies within a metalake. It includes a
@@ -126,6 +127,28 @@ PolicyEntity createPolicy(
    */
   MetadataObject[] listMetadataObjectsForPolicy(String metalake, String policyName);
 
+  /**
+   * List all tag names associated with the specified policy under a metalake.
+   *
+   * @param metalake the name of the metalake
+   * @param policyName the name of the policy
+   * @return The array of tag names associated with the specified policy.
+   */
+  default String[] listTagsForPolicy(String metalake, String policyName) {
+    return Arrays.stream(listTagInfosForPolicy(metalake, policyName))
+        .map(Tag::name)
+        .toArray(String[]::new);
+  }
+
+  /**
+   * List detailed information for all tags associated with the specified policy under a metalake.
+   *
+   * @param metalake the name of the metalake
+   * @param policyName the name of the policy
+   * @return The array of tags associated with the specified policy.
+   */
+  Tag[] listTagInfosForPolicy(String metalake, String policyName);
+
   /**
    * List all the policy names associated with a metadata object under a metalake.
    *

core/src/main/java/org/apache/gravitino/policy/PolicyManager.java

@@ -46,6 +46,7 @@
 import org.apache.gravitino.meta.PolicyEntity;
 import org.apache.gravitino.storage.IdGenerator;
 import org.apache.gravitino.storage.relational.service.MetadataObjectService;
+import org.apache.gravitino.tag.Tag;
 import org.apache.gravitino.utils.MetadataObjectUtil;
 import org.apache.gravitino.utils.NameIdentifierUtil;
 import org.apache.gravitino.utils.NamespaceUtil;
@@ -260,6 +261,37 @@ public MetadataObject[] listMetadataObjectsForPolicy(String metalake, String pol
         });
   }
 
+  @Override
+  public Tag[] listTagInfosForPolicy(String metalake, String policyName) {
+    NameIdentifier policyIdent = NameIdentifierUtil.ofPolicy(metalake, policyName);
+    checkMetalake(NameIdentifier.of(metalake), entityStore);
+
+    return TreeLockUtils.doWithTreeLock(
+        policyIdent,
+        LockType.READ,
+        () -> {
+          try {
+            return entityStore
+                .relationOperations()
+                .listEntitiesByRelation(
+                    SupportsRelationOperations.Type.POLICY_TAG_REL,
+                    policyIdent,
+                    Entity.EntityType.POLICY,
+                    true /* allFields */)
+                .stream()
+                .map(entity -> (Tag) entity)
+                .toArray(Tag[]::new);
+          } catch (NoSuchEntityException e) {
+            throw new NoSuchPolicyException(
+                e, "Policy with name %s under metalake %s does not exist", policyName, metalake);
+          } catch (IOException e) {
+            LOG.error(
+                "Failed to list tags for policy {} under metalake {}", policyName, metalake, e);
+            throw new RuntimeException(e);
+          }
+        });
+  }
+
   @Override
   public PolicyEntity[] listPolicyInfosForMetadataObject(
       String metalake, MetadataObject metadataObject) {

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

@@ -732,6 +732,15 @@ public <E extends Entity & HasIdentifier> List<E> listEntitiesByRelation(
           return (List<E>)
               TagMetaService.getInstance().listTagsForMetadataObject(nameIdentifier, identType);
         }
+      case POLICY_TAG_REL:
+        if (identType == Entity.EntityType.TAG) {
+          return (List<E>) PolicyMetaService.getInstance().listPoliciesForTag(nameIdentifier);
+        } else if (identType == Entity.EntityType.POLICY) {
+          return (List<E>) PolicyMetaService.getInstance().listTagsForPolicy(nameIdentifier);
+        } else {
+          throw new IllegalArgumentException(
+              String.format("POLICY_TAG_REL doesn't support type %s", identType.name()));
+        }
       default:
         throw new IllegalArgumentException(
             String.format("Doesn't support the relation type %s", relType));
@@ -811,6 +820,15 @@ public <E extends Entity & HasIdentifier> List<E> updateEntityRelations(
             TagMetaService.getInstance()
                 .associateTagsWithMetadataObject(
                     srcEntityIdent, srcEntityType, destEntitiesToAdd, destEntitiesToRemove);
+      case POLICY_TAG_REL:
+        if (srcEntityType == Entity.EntityType.TAG) {
+          return (List<E>)
+              PolicyMetaService.getInstance()
+                  .associatePoliciesWithTag(
+                      srcEntityIdent, destEntitiesToAdd, destEntitiesToRemove);
+        }
+        throw new IllegalArgumentException(
+            String.format("POLICY_TAG_REL doesn't support type %s", srcEntityType.name()));
       default:
         throw new IllegalArgumentException(
             String.format("Doesn't support the relation type %s", relType));
@@ -833,6 +851,13 @@ public <E extends Entity & HasIdentifier> E getEntityByRelation(
         return (E)
             TagMetaService.getInstance()
                 .getTagForMetadataObject(srcIdentifier, srcType, destEntityIdent);
+      case POLICY_TAG_REL:
+        if (srcType == Entity.EntityType.TAG) {
+          return (E)
+              PolicyMetaService.getInstance().getPolicyForTag(srcIdentifier, destEntityIdent);
+        }
+        throw new IllegalArgumentException(
+            String.format("POLICY_TAG_REL doesn't support type %s", srcType.name()));
       default:
         throw new IllegalArgumentException(
             String.format("Doesn't support the relation type %s", relType));

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

@@ -440,12 +440,13 @@ public <E extends Entity & HasIdentifier> List<E> updateEntityRelations(
             relType, srcEntityIdent, srcEntityType, destEntitiesToAdd, destEntitiesToRemove);
 
     cache.invalidate(srcEntityIdent, srcEntityType, relType);
+    Entity.EntityType destEntityType = relationDestinationType(relType, srcEntityType);
     for (NameIdentifier destToAdd : destEntitiesToAdd) {
-      cache.invalidate(destToAdd, srcEntityType, relType);
+      cache.invalidate(destToAdd, destEntityType, relType);
     }
 
     for (NameIdentifier destToRemove : destEntitiesToRemove) {
-      cache.invalidate(destToRemove, srcEntityType, relType);
+      cache.invalidate(destToRemove, destEntityType, relType);
     }
 
     return result;
@@ -504,6 +505,27 @@ private <E extends Entity & HasIdentifier> void batchPopulateRelationCache(
     }
   }
 
+  private Entity.EntityType relationDestinationType(
+      SupportsRelationOperations.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;
+      case POLICY_TAG_REL:
+        if (srcEntityType == Entity.EntityType.TAG) {
+          return Entity.EntityType.POLICY;
+        } else if (srcEntityType == Entity.EntityType.POLICY) {
+          return Entity.EntityType.TAG;
+        }
+        break;
+      default:
+        break;
+    }
+    throw new IllegalArgumentException(
+        String.format("Doesn't support the relation type %s", relType));
+  }
+
   /**
    * Invalidates the relation cache entries keyed by the counterpart of a role-aggregating entity
    * after that entity is written, so that reverse lookups reflect the change immediately.