Supports to list roles by object

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

@@ -369,6 +369,20 @@ public String[] listRoleNames() throws NoSuchMetalakeException {
     return getMetalake().listRoleNames();
   }
 
+  /**
+   * Lists the role names associated with a metadata object.
+   *
+   * @param object The object associated with the role.
+   * @return The role name list.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   * @throws NoSuchMetadataObjectException If the Metadata object with the given name does not
+   *     exist.
+   */
+  public String[] listRoleNamesByObject(MetadataObject object)
+      throws NoSuchMetalakeException, NoSuchMetadataObjectException {
+    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,29 @@ public String[] listRoleNames() {
     return resp.getNames();
   }
 
+  /**
+   * Lists the role names associated with a metadata object.
+   *
+   * @param object The object associated with the role.
+   * @return The role name list.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   * @throws NoSuchMetadataObjectException If the Metadata object with the given name does not
+   *     exist.
+   */
+  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,27 @@ 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, the method will fetch all the fields, Otherwise, the method will fetch all the fields
+   *     except for high-cost 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,8 +20,10 @@
 
 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.NoSuchMetadataObjectException;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
 import org.apache.gravitino.exceptions.NoSuchRoleException;
 import org.apache.gravitino.exceptions.NoSuchUserException;
@@ -246,4 +248,16 @@ Role createRole(
    * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
    */
   String[] listRoleNames(String metalake) throws NoSuchMetalakeException;
+
+  /**
+   * Lists the role names associated 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.
+   * @throws NoSuchMetadataObjectException If the Metadata object with the given name does not
+   *     exist.
+   */
+  String[] listRoleNamesByObject(String metalake, MetadataObject object)
+      throws NoSuchMetalakeException, NoSuchMetadataObjectException;
 }

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

@@ -18,14 +18,15 @@
  */
 package org.apache.gravitino.authorization;
 
-import com.google.common.annotations.VisibleForTesting;
 import java.util.List;
 import java.util.Map;
 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.NoSuchMetadataObjectException;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
 import org.apache.gravitino.exceptions.NoSuchRoleException;
 import org.apache.gravitino.exceptions.NoSuchUserException;
@@ -148,8 +149,9 @@ public String[] listRoleNames(String metalake) throws NoSuchMetalakeException {
     return roleManager.listRoleNames(metalake);
   }
 
-  @VisibleForTesting
-  RoleManager getRoleManager() {
-    return roleManager;
+  @Override
+  public String[] listRoleNamesByObject(String metalake, MetadataObject object)
+      throws NoSuchMetalakeException, NoSuchMetadataObjectException {
+    return roleManager.listRoleNamesByObject(metalake, object);
   }
 }

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,19 @@
 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.NoSuchMetadataObjectException;
 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 +152,35 @@ String[] listRoleNames(String metalake) {
     }
   }
 
+  String[] listRoleNamesByObject(String metalake, MetadataObject object) {
+    try {
+      AuthorizationUtils.checkMetalakeExists(metalake);
+
+      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 (NoSuchEntityException nse) {
+      LOG.error("Metadata object {} (type {}) doesn't exist", object.fullName(), object.type());
+      throw new NoSuchMetadataObjectException(
+          "Metadata object %s (type %s) doesn't exist", object.fullName(), object.type());
+    } 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;
@@ -32,6 +33,7 @@
 import org.apache.gravitino.authorization.User;
 import org.apache.gravitino.exceptions.GroupAlreadyExistsException;
 import org.apache.gravitino.exceptions.NoSuchGroupException;
+import org.apache.gravitino.exceptions.NoSuchMetadataObjectException;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
 import org.apache.gravitino.exceptions.NoSuchRoleException;
 import org.apache.gravitino.exceptions.NoSuchUserException;
@@ -162,4 +164,10 @@ 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)
+      throws NoSuchMetalakeException, NoSuchMetadataObjectException {
+    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,23 @@ 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);
         } 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);
         } 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