Supports to list roles by object

clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClient.java

@@ -369,6 +369,10 @@ public String[] listRoleNames() throws NoSuchMetalakeException {
     return getMetalake().listRoleNames();
   }
 
+  public String[] listRoleNamesByObject(MetadataObject object) {
+    return getMetalake().listRoleNamesByObject(object);
+  }
+
   /**
    * Creates a new builder for constructing a GravitinoClient.
    *

clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoMetalake.java

@@ -719,6 +719,20 @@ public String[] listRoleNames() {
     return resp.getNames();
   }
 
+  public String[] listRoleNamesByObject(MetadataObject object) {
+    NameListResponse resp =
+        restClient.get(
+            String.format(
+                "api/metalakes/%s/objects/%s/%s/roles",
+                this.name(), object.type(), object.fullName()),
+            NameListResponse.class,
+            Collections.emptyMap(),
+            ErrorHandlers.roleErrorHandler());
+    resp.validate();
+
+    return resp.getNames();
+  }
+
   /**
    * Grant roles to a user.
    *

clients/client-java/src/test/java/org/apache/gravitino/client/TestRole.java

@@ -27,6 +27,8 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import java.time.Instant;
+import org.apache.gravitino.MetadataObject;
+import org.apache.gravitino.MetadataObjects;
 import org.apache.gravitino.authorization.Privileges;
 import org.apache.gravitino.authorization.Role;
 import org.apache.gravitino.authorization.SecurableObject;
@@ -235,6 +237,38 @@ public void testListRoleNames() throws Exception {
     Assertions.assertThrows(RuntimeException.class, () -> gravitinoClient.listRoleNames());
   }
 
+  @Test
+  public void testListRoleNamesByObject() throws Exception {
+    String rolePath =
+        withSlash(
+            String.format(
+                "api/metalakes/%s/objects/%s/%s/roles",
+                metalakeName, MetadataObject.Type.CATALOG.name(), "catalog"));
+
+    NameListResponse listResponse = new NameListResponse(new String[] {"role1", "role2"});
+    buildMockResource(Method.GET, rolePath, null, listResponse, SC_OK);
+    MetadataObject metadataObject =
+        MetadataObjects.of(null, "catalog", MetadataObject.Type.CATALOG);
+
+    Assertions.assertArrayEquals(
+        new String[] {"role1", "role2"}, gravitinoClient.listRoleNamesByObject(metadataObject));
+
+    ErrorResponse errRespNoMetalake =
+        ErrorResponse.notFound(NoSuchMetalakeException.class.getSimpleName(), "metalake not found");
+    buildMockResource(Method.GET, rolePath, null, errRespNoMetalake, SC_NOT_FOUND);
+    Exception ex =
+        Assertions.assertThrows(
+            NoSuchMetalakeException.class,
+            () -> gravitinoClient.listRoleNamesByObject(metadataObject));
+    Assertions.assertEquals("metalake not found", ex.getMessage());
+
+    // Test RuntimeException
+    ErrorResponse errResp = ErrorResponse.internalError("internal error");
+    buildMockResource(Method.GET, rolePath, null, errResp, SC_SERVER_ERROR);
+    Assertions.assertThrows(
+        RuntimeException.class, () -> gravitinoClient.listRoleNamesByObject(metadataObject));
+  }
+
   private RoleDTO mockRoleDTO(String name) {
     SecurableObject securableObject =
         SecurableObjects.ofCatalog("catalog", Lists.newArrayList(Privileges.UseCatalog.allow()));

clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/authorization/AccessControlIT.java

@@ -184,6 +184,12 @@ void testManageRoles() {
     String[] roleNames = metalake.listRoleNames();
     Arrays.sort(roleNames);
 
+    Assertions.assertEquals(
+        Lists.newArrayList(anotherRoleName, roleName), Arrays.asList(roleNames));
+
+    // List roles by the object
+    roleNames = metalake.listRoleNamesByObject(metalakeObject);
+    Arrays.sort(roleNames);
     Assertions.assertEquals(
         Lists.newArrayList(anotherRoleName, roleName), Arrays.asList(roleNames));
 

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

@@ -48,8 +48,26 @@ enum Type {
    * @return The list of entities
    * @throws IOException When occurs storage issues, it will throw IOException.
    */
+  default <E extends Entity & HasIdentifier> List<E> listEntitiesByRelation(
+      Type relType, NameIdentifier nameIdentifier, Entity.EntityType identType) throws IOException {
+    return listEntitiesByRelation(relType, nameIdentifier, identType, true /* allFields*/);
+  }
+
+  /**
+   * List the entities according to a give entity in a specific relation.
+   *
+   * @param relType The type of relation.
+   * @param nameIdentifier The given entity identifier
+   * @param identType The given entity type.
+   * @param allFields Some fields may have a relatively high acquisition cost, EntityStore provide
+   *     an optional setting to avoid fetching these high-cost fields to improve the performance. If
+   *     true, we skip the fields. Otherwise, we don't skip the fields.
+   * @return The list of entities
+   * @throws IOException When occurs storage issues, it will throw IOException.
+   */
   <E extends Entity & HasIdentifier> List<E> listEntitiesByRelation(
-      Type relType, NameIdentifier nameIdentifier, Entity.EntityType identType) throws IOException;
+      Type relType, NameIdentifier nameIdentifier, Entity.EntityType identType, boolean allFields)
+      throws IOException;
 
   /**
    * insert a relation between two entities

core/src/main/java/org/apache/gravitino/authorization/AccessControlDispatcher.java

@@ -20,6 +20,7 @@
 
 import java.util.List;
 import java.util.Map;
+import org.apache.gravitino.MetadataObject;
 import org.apache.gravitino.exceptions.GroupAlreadyExistsException;
 import org.apache.gravitino.exceptions.NoSuchGroupException;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
@@ -246,4 +247,13 @@ Role createRole(
    * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
    */
   String[] listRoleNames(String metalake) throws NoSuchMetalakeException;
+
+  /**
+   * Lists the role names contains the metadata object.
+   *
+   * @param metalake The Metalake of the Role.
+   * @return The role list.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   */
+  String[] listRoleNamesByObject(String metalake, MetadataObject object);
 }

core/src/main/java/org/apache/gravitino/authorization/AccessControlManager.java

@@ -24,6 +24,7 @@
 import org.apache.gravitino.Config;
 import org.apache.gravitino.Configs;
 import org.apache.gravitino.EntityStore;
+import org.apache.gravitino.MetadataObject;
 import org.apache.gravitino.exceptions.GroupAlreadyExistsException;
 import org.apache.gravitino.exceptions.NoSuchGroupException;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
@@ -148,6 +149,11 @@ public String[] listRoleNames(String metalake) throws NoSuchMetalakeException {
     return roleManager.listRoleNames(metalake);
   }
 
+  @Override
+  public String[] listRoleNamesByObject(String metalake, MetadataObject object) {
+    return roleManager.listRoleNamesByObject(metalake, object);
+  }
+
   @VisibleForTesting
   RoleManager getRoleManager() {
     return roleManager;

core/src/main/java/org/apache/gravitino/authorization/FutureGrantManager.java

@@ -20,6 +20,7 @@
 
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
@@ -37,7 +38,6 @@
 import org.apache.gravitino.meta.GroupEntity;
 import org.apache.gravitino.meta.RoleEntity;
 import org.apache.gravitino.meta.UserEntity;
-import org.glassfish.jersey.internal.guava.Sets;
 
 /**
  * FutureGrantManager is responsible for granting privileges to future object. When you grant a

core/src/main/java/org/apache/gravitino/authorization/RoleManager.java

@@ -27,15 +27,18 @@
 import org.apache.gravitino.Entity;
 import org.apache.gravitino.EntityAlreadyExistsException;
 import org.apache.gravitino.EntityStore;
+import org.apache.gravitino.MetadataObject;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
+import org.apache.gravitino.SupportsRelationOperations;
 import org.apache.gravitino.exceptions.NoSuchEntityException;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
 import org.apache.gravitino.exceptions.NoSuchRoleException;
 import org.apache.gravitino.exceptions.RoleAlreadyExistsException;
 import org.apache.gravitino.meta.AuditInfo;
 import org.apache.gravitino.meta.RoleEntity;
 import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.utils.MetadataObjectUtil;
 import org.apache.gravitino.utils.PrincipalUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -148,6 +151,29 @@ String[] listRoleNames(String metalake) {
     }
   }
 
+  String[] listRoleNamesByObject(String metalake, MetadataObject object) {
+    try {
+      return store.relationOperations()
+          .listEntitiesByRelation(
+              SupportsRelationOperations.Type.METADATA_OBJECT_ROLE_REL,
+              MetadataObjectUtil.toEntityIdent(metalake, object),
+              MetadataObjectUtil.toEntityType(object),
+              false /* allFields */)
+          .stream()
+          .map(entity -> ((RoleEntity) entity).name())
+          .toArray(String[]::new);
+
+    } catch (IOException ioe) {
+      LOG.error(
+          "Listing roles under metalake {} by object full name {} and type {} failed due to storage issues",
+          metalake,
+          object.fullName(),
+          object.type(),
+          ioe);
+      throw new RuntimeException(ioe);
+    }
+  }
+
   private RoleEntity getRoleEntity(NameIdentifier identifier) {
     try {
       return store.get(identifier, Entity.EntityType.ROLE, RoleEntity.class);

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

@@ -22,6 +22,7 @@
 import java.util.Map;
 import org.apache.gravitino.Entity;
 import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.MetadataObject;
 import org.apache.gravitino.authorization.AccessControlDispatcher;
 import org.apache.gravitino.authorization.AuthorizationUtils;
 import org.apache.gravitino.authorization.Group;
@@ -162,4 +163,9 @@ public boolean deleteRole(String metalake, String role) throws NoSuchMetalakeExc
   public String[] listRoleNames(String metalake) throws NoSuchMetalakeException {
     return dispatcher.listRoleNames(metalake);
   }
+
+  @Override
+  public String[] listRoleNamesByObject(String metalake, MetadataObject object) {
+    return dispatcher.listRoleNamesByObject(metalake, object);
+  }
 }

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

@@ -369,9 +369,7 @@ public List<TagEntity> associateTagsWithMetadataObject(
 
   @Override
   public <E extends Entity & HasIdentifier> List<E> listEntitiesByRelation(
-      SupportsRelationOperations.Type relType,
-      NameIdentifier nameIdentifier,
-      Entity.EntityType identType) {
+      Type relType, NameIdentifier nameIdentifier, Entity.EntityType identType, boolean allFields) {
     switch (relType) {
       case OWNER_REL:
         List<E> list = Lists.newArrayList();
@@ -382,20 +380,25 @@ public <E extends Entity & HasIdentifier> List<E> listEntitiesByRelation(
       case METADATA_OBJECT_ROLE_REL:
         return (List<E>)
             RoleMetaService.getInstance()
-                .listRolesByMetadataObjectIdentAndType(nameIdentifier, identType);
+                .listRolesByMetadataObjectIdentAndType(nameIdentifier, identType, allFields);
       case ROLE_GROUP_REL:
         if (identType == Entity.EntityType.ROLE) {
-          return (List<E>) GroupMetaService.getInstance().listGroupsByRoleIdent(nameIdentifier);
+          return (List<E>)
+              GroupMetaService.getInstance().listGroupsByRoleIdent(nameIdentifier, allFields);
         } else {
           throw new IllegalArgumentException(
-              String.format("ROLE_GROUP_REL doesn't support type %s", identType.name()));
+              String.format(
+                  "ROLE_GROUP_REL doesn't support type %s or loading all fields",
+                  identType.name()));
         }
       case ROLE_USER_REL:
         if (identType == Entity.EntityType.ROLE) {
-          return (List<E>) UserMetaService.getInstance().listUsersByRoleIdent(nameIdentifier);
+          return (List<E>)
+              UserMetaService.getInstance().listUsersByRoleIdent(nameIdentifier, allFields);
         } else {
           throw new IllegalArgumentException(
-              String.format("ROLE_USER_REL doesn't support type %s", identType.name()));
+              String.format(
+                  "ROLE_USER_REL doesn't support type %s or loading all fields", identType.name()));
         }
       default:
         throw new IllegalArgumentException(

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

@@ -188,11 +188,9 @@ public List<TagEntity> associateTagsWithMetadataObject(
 
   @Override
   public <E extends Entity & HasIdentifier> List<E> listEntitiesByRelation(
-      SupportsRelationOperations.Type relType,
-      NameIdentifier nameIdentifier,
-      Entity.EntityType identType)
+      Type relType, NameIdentifier nameIdentifier, Entity.EntityType identType, boolean allFields)
       throws IOException {
-    return backend.listEntitiesByRelation(relType, nameIdentifier, identType);
+    return backend.listEntitiesByRelation(relType, nameIdentifier, identType, allFields);
   }
 
   @Override

core/src/main/java/org/apache/gravitino/storage/relational/service/GroupMetaService.java

@@ -97,7 +97,11 @@ public GroupEntity getGroupByIdentifier(NameIdentifier identifier) {
     return POConverters.fromGroupPO(groupPO, rolePOs, identifier.namespace());
   }
 
-  public List<GroupEntity> listGroupsByRoleIdent(NameIdentifier roleIdent) {
+  public List<GroupEntity> listGroupsByRoleIdent(NameIdentifier roleIdent, boolean allFields) {
+    if (allFields) {
+      throw new IllegalArgumentException("Don't support list all fields groups for a role");
+    }
+
     RoleEntity roleEntity = RoleMetaService.getInstance().getRoleByIdentifier(roleIdent);
     List<GroupPO> groupPOs =
         SessionUtils.getWithoutCommit(