[#3346] feat(core,server): Supports to list roles operations (#5022)

### What changes were proposed in this pull request?
Supports to list roles operations

### Why are the changes needed?

Fix: #3346

### Does this PR introduce _any_ user-facing change?
Yes, will add the document later.

### How was this patch tested?
Add UTs.

Co-authored-by: roryqi <[email protected]>

api/src/main/java/org/apache/gravitino/authorization/Role.java

@@ -25,8 +25,7 @@
 
 /**
  * The interface of a role. The role is the entity which has kinds of privileges. One role can have
- * multiple privileges of one securable object. Gravitino chooses to bind one securable object to
- * one role to avoid granting too many privileges to one role.
+ * multiple privileges of multiple securable objects.
  */
 @Evolving
 public interface Role extends Auditable {

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

@@ -359,6 +359,16 @@ public void setOwner(MetadataObject object, String ownerName, Owner.Type ownerTy
     getMetalake().setOwner(object, ownerName, ownerType);
   }
 
+  /**
+   * Lists the role names.
+   *
+   * @return The role name list.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   */
+  public String[] listRoleNames() throws NoSuchMetalakeException {
+    return getMetalake().listRoleNames();
+  }
+
   /**
    * Creates a new builder for constructing a GravitinoClient.
    *

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

@@ -100,9 +100,9 @@ public class GravitinoMetalake extends MetalakeDTO implements SupportsCatalogs,
   private static final String API_METALAKES_GROUPS_PATH = "api/metalakes/%s/groups/%s";
   private static final String API_METALAKES_ROLES_PATH = "api/metalakes/%s/roles/%s";
   private static final String API_METALAKES_OWNERS_PATH = "api/metalakes/%s/owners/%s";
-  private static final String BLANK_PLACE_HOLDER = "";
 
   private static final String API_METALAKES_TAGS_PATH = "api/metalakes/%s/tags";
+  private static final String BLANK_PLACEHOLDER = "";
 
   private final RESTClient restClient;
 
@@ -464,7 +464,7 @@ public User addUser(String user) throws UserAlreadyExistsException, NoSuchMetala
 
     UserResponse resp =
         restClient.post(
-            String.format(API_METALAKES_USERS_PATH, this.name(), BLANK_PLACE_HOLDER),
+            String.format(API_METALAKES_USERS_PATH, this.name(), BLANK_PLACEHOLDER),
             req,
             UserResponse.class,
             Collections.emptyMap(),
@@ -528,7 +528,7 @@ public User[] listUsers() throws NoSuchMetalakeException {
 
     UserListResponse resp =
         restClient.get(
-            String.format(API_METALAKES_USERS_PATH, name(), BLANK_PLACE_HOLDER),
+            String.format(API_METALAKES_USERS_PATH, name(), BLANK_PLACEHOLDER),
             params,
             UserListResponse.class,
             Collections.emptyMap(),
@@ -547,7 +547,7 @@ public User[] listUsers() throws NoSuchMetalakeException {
   public String[] listUserNames() throws NoSuchMetalakeException {
     NameListResponse resp =
         restClient.get(
-            String.format(API_METALAKES_USERS_PATH, name(), BLANK_PLACE_HOLDER),
+            String.format(API_METALAKES_USERS_PATH, name(), BLANK_PLACEHOLDER),
             NameListResponse.class,
             Collections.emptyMap(),
             ErrorHandlers.userErrorHandler());
@@ -571,7 +571,7 @@ public Group addGroup(String group) throws GroupAlreadyExistsException, NoSuchMe
 
     GroupResponse resp =
         restClient.post(
-            String.format(API_METALAKES_GROUPS_PATH, this.name(), BLANK_PLACE_HOLDER),
+            String.format(API_METALAKES_GROUPS_PATH, this.name(), BLANK_PLACEHOLDER),
             req,
             GroupResponse.class,
             Collections.emptyMap(),
@@ -691,7 +691,7 @@ public Role createRole(
 
     RoleResponse resp =
         restClient.post(
-            String.format(API_METALAKES_ROLES_PATH, this.name(), BLANK_PLACE_HOLDER),
+            String.format(API_METALAKES_ROLES_PATH, this.name(), BLANK_PLACEHOLDER),
             req,
             RoleResponse.class,
             Collections.emptyMap(),
@@ -701,6 +701,24 @@ public Role createRole(
     return resp.getRole();
   }
 
+  /**
+   * Lists the role names.
+   *
+   * @return The role name list.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   */
+  public String[] listRoleNames() {
+    NameListResponse resp =
+        restClient.get(
+            String.format(API_METALAKES_ROLES_PATH, this.name(), BLANK_PLACEHOLDER),
+            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

@@ -40,6 +40,7 @@
 import org.apache.gravitino.dto.responses.DeleteResponse;
 import org.apache.gravitino.dto.responses.ErrorResponse;
 import org.apache.gravitino.dto.responses.MetalakeResponse;
+import org.apache.gravitino.dto.responses.NameListResponse;
 import org.apache.gravitino.dto.responses.RoleResponse;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
 import org.apache.gravitino.exceptions.NoSuchRoleException;
@@ -211,6 +212,29 @@ public void testDeleteRoles() throws Exception {
     Assertions.assertThrows(RuntimeException.class, () -> gravitinoClient.deleteRole(roleName));
   }
 
+  @Test
+  public void testListRoleNames() throws Exception {
+    String rolePath = withSlash(String.format(API_METALAKES_ROLES_PATH, metalakeName, ""));
+
+    NameListResponse listResponse = new NameListResponse(new String[] {"role1", "role2"});
+    buildMockResource(Method.GET, rolePath, null, listResponse, SC_OK);
+
+    Assertions.assertArrayEquals(new String[] {"role1", "role2"}, gravitinoClient.listRoleNames());
+
+    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.listRoleNames());
+    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.listRoleNames());
+  }
+
   private RoleDTO mockRoleDTO(String name) {
     SecurableObject securableObject =
         SecurableObjects.ofCatalog("catalog", Lists.newArrayList(Privileges.UseCatalog.allow()));

common/src/main/java/org/apache/gravitino/dto/authorization/RoleDTO.java

@@ -158,7 +158,7 @@ public S withProperties(Map<String, String> properties) {
     /**
      * Sets the securable objects of the role.
      *
-     * @param securableObjects The securableObjects of the role.
+     * @param securableObjects The securable objects of the role.
      * @return The builder instance.
      */
     public S withSecurableObjects(SecurableObjectDTO[] securableObjects) {
@@ -186,9 +186,7 @@ public S withAudit(AuditDTO audit) {
     public RoleDTO build() {
       Preconditions.checkArgument(StringUtils.isNotBlank(name), "name cannot be null or empty");
       Preconditions.checkArgument(audit != null, "audit cannot be null");
-      Preconditions.checkArgument(
-          securableObjects != null && securableObjects.length != 0,
-          "securable objects can't null or empty");
+      Preconditions.checkArgument(securableObjects != null, "securable objects can't null");
 
       return new RoleDTO(name, properties, securableObjects, audit);
     }

common/src/main/java/org/apache/gravitino/dto/responses/RoleResponse.java

@@ -65,7 +65,6 @@ public void validate() throws IllegalArgumentException {
         StringUtils.isNotBlank(role.name()), "role 'name' must not be null and empty");
     Preconditions.checkArgument(role.auditInfo() != null, "role 'auditInfo' must not be null");
     Preconditions.checkArgument(
-        role.securableObjects() != null && !role.securableObjects().isEmpty(),
-        "role 'securableObjects' can't null or empty");
+        role.securableObjects() != null, "role 'securable objects' can't null");
   }
 }

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

@@ -20,9 +20,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
-import java.util.Collections;
 import java.util.List;
-import java.util.Set;
 import java.util.function.Function;
 import org.apache.gravitino.Entity.EntityType;
 import org.apache.gravitino.exceptions.NoSuchEntityException;
@@ -66,7 +64,7 @@ public interface EntityStore extends Closeable {
    */
   default <E extends Entity & HasIdentifier> List<E> list(
       Namespace namespace, Class<E> type, EntityType entityType) throws IOException {
-    return list(namespace, type, entityType, Collections.emptySet());
+    return list(namespace, type, entityType, true /* allFields */);
   }
 
   /**
@@ -80,14 +78,15 @@ default <E extends Entity & HasIdentifier> List<E> list(
    * @param namespace the namespace of the entities
    * @param type the detailed type of the entity
    * @param entityType the general type of the entity
-   * @param skippingFields Some fields may have a relatively high acquisition cost, EntityStore
-   *     provides an optional setting to avoid fetching these high-cost fields to improve the
-   *     performance.
+   * @param allFields Some fields may have a relatively high acquisition cost, EntityStore provides
+   *     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 if the list operation fails
    */
   default <E extends Entity & HasIdentifier> List<E> list(
-      Namespace namespace, Class<E> type, EntityType entityType, Set<Field> skippingFields)
+      Namespace namespace, Class<E> type, EntityType entityType, boolean allFields)
       throws IOException {
     throw new UnsupportedOperationException("Don't support to skip fields");
   }

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

@@ -236,5 +236,14 @@ Role createRole(
    * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
    * @throws RuntimeException If deleting the Role encounters storage issues.
    */
-  public boolean deleteRole(String metalake, String role) throws NoSuchMetalakeException;
+  boolean deleteRole(String metalake, String role) throws NoSuchMetalakeException;
+
+  /**
+   * Lists the role names.
+   *
+   * @param metalake The Metalake of the Role.
+   * @return The role name list.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   */
+  String[] listRoleNames(String metalake) throws NoSuchMetalakeException;
 }

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

@@ -143,6 +143,11 @@ public boolean deleteRole(String metalake, String role) throws NoSuchMetalakeExc
     return roleManager.deleteRole(metalake, role);
   }
 
+  @Override
+  public String[] listRoleNames(String metalake) throws NoSuchMetalakeException {
+    return roleManager.listRoleNames(metalake);
+  }
+
   @VisibleForTesting
   RoleManager getRoleManager() {
     return roleManager;

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

@@ -19,6 +19,7 @@
 
 package org.apache.gravitino.authorization;
 
+import com.google.common.collect.Sets;
 import java.io.IOException;
 import java.time.Instant;
 import java.util.List;
@@ -27,14 +28,15 @@
 import org.apache.gravitino.EntityAlreadyExistsException;
 import org.apache.gravitino.EntityStore;
 import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
 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.PrincipalUtils;
-import org.glassfish.jersey.internal.guava.Sets;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -46,6 +48,7 @@
 class RoleManager {
 
   private static final Logger LOG = LoggerFactory.getLogger(RoleManager.class);
+  private static final String METALAKE_DOES_NOT_EXIST_MSG = "Metalake %s does not exist";
   private final EntityStore store;
   private final IdGenerator idGenerator;
 
@@ -129,6 +132,22 @@ boolean deleteRole(String metalake, String role) {
     }
   }
 
+  String[] listRoleNames(String metalake) {
+    try {
+      AuthorizationUtils.checkMetalakeExists(metalake);
+      Namespace namespace = AuthorizationUtils.ofRoleNamespace(metalake);
+      return store.list(namespace, RoleEntity.class, Entity.EntityType.ROLE).stream()
+          .map(Role::name)
+          .toArray(String[]::new);
+    } catch (NoSuchEntityException e) {
+      LOG.warn("Metalake {} does not exist", metalake, e);
+      throw new NoSuchMetalakeException(METALAKE_DOES_NOT_EXIST_MSG, metalake);
+    } catch (IOException ioe) {
+      LOG.error("Listing user under metalake {} failed due to storage issues", metalake, 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/authorization/UserGroupManager.java

@@ -23,11 +23,9 @@
 import java.time.Instant;
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.Set;
 import org.apache.gravitino.Entity;
 import org.apache.gravitino.EntityAlreadyExistsException;
 import org.apache.gravitino.EntityStore;
-import org.apache.gravitino.Field;
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.exceptions.GroupAlreadyExistsException;
 import org.apache.gravitino.exceptions.NoSuchEntityException;
@@ -40,7 +38,6 @@
 import org.apache.gravitino.meta.UserEntity;
 import org.apache.gravitino.storage.IdGenerator;
 import org.apache.gravitino.utils.PrincipalUtils;
-import org.glassfish.jersey.internal.guava.Sets;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -117,26 +114,23 @@ User getUser(String metalake, String user) throws NoSuchUserException {
   }
 
   String[] listUserNames(String metalake) {
-    Set<Field> skippingFields = Sets.newHashSet();
-    skippingFields.add(UserEntity.ROLE_NAMES);
-    skippingFields.add(UserEntity.ROLE_IDS);
 
-    return Arrays.stream(listUsersInternal(metalake, skippingFields))
+    return Arrays.stream(listUsersInternal(metalake, false /* allFields */))
         .map(User::name)
         .toArray(String[]::new);
   }
 
   User[] listUsers(String metalake) {
-    return listUsersInternal(metalake, Collections.emptySet());
+    return listUsersInternal(metalake, true /* allFields */);
   }
 
-  private User[] listUsersInternal(String metalake, Set<Field> skippingFields) {
+  private User[] listUsersInternal(String metalake, boolean allFields) {
     try {
       AuthorizationUtils.checkMetalakeExists(metalake);
 
       Namespace namespace = AuthorizationUtils.ofUserNamespace(metalake);
       return store
-          .list(namespace, UserEntity.class, Entity.EntityType.USER, skippingFields)
+          .list(namespace, UserEntity.class, Entity.EntityType.USER, allFields)
           .toArray(new User[0]);
     } catch (NoSuchEntityException e) {
       LOG.error("Metalake {} does not exist", metalake, e);

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

@@ -157,4 +157,9 @@ public Role getRole(String metalake, String role)
   public boolean deleteRole(String metalake, String role) throws NoSuchMetalakeException {
     return dispatcher.deleteRole(metalake, role);
   }
+
+  @Override
+  public String[] listRoleNames(String metalake) throws NoSuchMetalakeException {
+    return dispatcher.listRoleNames(metalake);
+  }
 }

core/src/main/java/org/apache/gravitino/meta/RoleEntity.java

@@ -46,8 +46,8 @@ public class RoleEntity implements Role, Entity, Auditable, HasIdentifier {
   public static final Field AUDIT_INFO =
       Field.required("audit_info", AuditInfo.class, "The audit details of the role entity.");
 
-  public static final Field SECURABLE_OBJECT =
-      Field.required("securable_objects", List.class, "The securable objects of the role entity.");
+  public static final Field SECURABLE_OBJECTS =
+      Field.optional("securable_objects", List.class, "The securable objects of the role entity.");
 
   private Long id;
   private String name;
@@ -91,15 +91,6 @@ public Map<String, String> properties() {
    */
   @Override
   public List<SecurableObject> securableObjects() {
-    // The securable object is a special kind of entities. Some entity types aren't the securable
-    // object, such as
-    // User, Role, etc.
-    // The securable object identifier must be unique.
-    // Gravitino assumes that the identifiers of the entities may be the same if they have different
-    // types.
-    // So one type of them can't be the securable object at least if there are the two same
-    // identifier
-    // entities .
     return securableObjects;
   }
 
@@ -115,7 +106,7 @@ public Map<Field, Object> fields() {
     fields.put(NAME, name);
     fields.put(AUDIT_INFO, auditInfo);
     fields.put(PROPERTIES, properties);
-    fields.put(SECURABLE_OBJECT, securableObjects);
+    fields.put(SECURABLE_OBJECTS, securableObjects);
 
     return Collections.unmodifiableMap(fields);
   }