[#3348] feat(core,server): Add the list operation of the user

catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogOperations.java

@@ -149,7 +149,8 @@ public NameIdentifier[] listFilesets(Namespace namespace) throws NoSuchSchemaExc
       }
 
       List<FilesetEntity> filesets =
-          store.list(namespace, FilesetEntity.class, Entity.EntityType.FILESET);
+          store.list(
+              namespace, FilesetEntity.class, Entity.EntityType.FILESET, Collections.emptyList());
       return filesets.stream()
           .map(f -> NameIdentifier.of(namespace, f.name()))
           .toArray(NameIdentifier[]::new);
@@ -387,7 +388,8 @@ public String getFileLocation(NameIdentifier ident, String subPath)
   public NameIdentifier[] listSchemas(Namespace namespace) throws NoSuchCatalogException {
     try {
       List<SchemaEntity> schemas =
-          store.list(namespace, SchemaEntity.class, Entity.EntityType.SCHEMA);
+          store.list(
+              namespace, SchemaEntity.class, Entity.EntityType.SCHEMA, Collections.emptyList());
       return schemas.stream()
           .map(s -> NameIdentifier.of(namespace, s.name()))
           .toArray(NameIdentifier[]::new);

catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalogOperations.java

@@ -391,7 +391,8 @@ public boolean dropTopic(NameIdentifier ident) {
   public NameIdentifier[] listSchemas(Namespace namespace) throws NoSuchCatalogException {
     try {
       List<SchemaEntity> schemas =
-          store.list(namespace, SchemaEntity.class, Entity.EntityType.SCHEMA);
+          store.list(
+              namespace, SchemaEntity.class, Entity.EntityType.SCHEMA, Collections.emptyList());
       return schemas.stream()
           .map(s -> NameIdentifier.of(namespace, s.name()))
           .toArray(NameIdentifier[]::new);

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

@@ -168,6 +168,26 @@ public User getUser(String user) throws NoSuchUserException, NoSuchMetalakeExcep
     return getMetalake().getUser(user);
   }
 
+  /**
+   * Lists the users.
+   *
+   * @return The User list.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   */
+  public User[] listUsers() {
+    return getMetalake().listUsers();
+  }
+
+  /**
+   * Lists the usernames.
+   *
+   * @return The username list.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   */
+  public String[] listUserNames() {
+    return getMetalake().listUserNames();
+  }
+
   /**
    * Adds a new Group.
    *

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

@@ -68,6 +68,7 @@
 import org.apache.gravitino.dto.responses.SetResponse;
 import org.apache.gravitino.dto.responses.TagListResponse;
 import org.apache.gravitino.dto.responses.TagResponse;
+import org.apache.gravitino.dto.responses.UserListResponse;
 import org.apache.gravitino.dto.responses.UserResponse;
 import org.apache.gravitino.exceptions.CatalogAlreadyExistsException;
 import org.apache.gravitino.exceptions.GroupAlreadyExistsException;
@@ -515,6 +516,46 @@ public User getUser(String user) throws NoSuchUserException, NoSuchMetalakeExcep
     return resp.getUser();
   }
 
+  /**
+   * Lists the users.
+   *
+   * @return The User list.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   */
+  public User[] listUsers() throws NoSuchMetalakeException {
+    Map<String, String> params = new HashMap<>();
+    params.put("details", "true");
+
+    UserListResponse resp =
+        restClient.get(
+            String.format(API_METALAKES_USERS_PATH, name(), BLANK_PLACE_HOLDER),
+            params,
+            UserListResponse.class,
+            Collections.emptyMap(),
+            ErrorHandlers.userErrorHandler());
+    resp.validate();
+
+    return resp.getUsers();
+  }
+
+  /**
+   * Lists the usernames.
+   *
+   * @return The username list.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   */
+  public String[] listUserNames() throws NoSuchMetalakeException {
+    NameListResponse resp =
+        restClient.get(
+            String.format(API_METALAKES_USERS_PATH, name(), BLANK_PLACE_HOLDER),
+            NameListResponse.class,
+            Collections.emptyMap(),
+            ErrorHandlers.userErrorHandler());
+    resp.validate();
+
+    return resp.getNames();
+  }
+
   /**
    * Adds a new Group.
    *

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

@@ -24,6 +24,8 @@
 import static org.apache.hc.core5.http.HttpStatus.SC_SERVER_ERROR;
 
 import java.time.Instant;
+import java.util.Collections;
+import java.util.Map;
 import org.apache.gravitino.authorization.Group;
 import org.apache.gravitino.authorization.User;
 import org.apache.gravitino.dto.AuditDTO;
@@ -35,7 +37,9 @@
 import org.apache.gravitino.dto.responses.ErrorResponse;
 import org.apache.gravitino.dto.responses.GroupResponse;
 import org.apache.gravitino.dto.responses.MetalakeResponse;
+import org.apache.gravitino.dto.responses.NameListResponse;
 import org.apache.gravitino.dto.responses.RemoveResponse;
+import org.apache.gravitino.dto.responses.UserListResponse;
 import org.apache.gravitino.dto.responses.UserResponse;
 import org.apache.gravitino.exceptions.GroupAlreadyExistsException;
 import org.apache.gravitino.exceptions.NoSuchGroupException;
@@ -175,6 +179,56 @@ public void testRemoveUsers() throws Exception {
     Assertions.assertThrows(RuntimeException.class, () -> gravitinoClient.removeUser(username));
   }
 
+  @Test
+  public void testListUserNames() throws Exception {
+    String userPath = withSlash(String.format(API_METALAKES_USERS_PATH, metalakeName, ""));
+
+    NameListResponse listResponse = new NameListResponse(new String[] {"user1", "user2"});
+    buildMockResource(Method.GET, userPath, null, listResponse, SC_OK);
+
+    Assertions.assertArrayEquals(new String[] {"user1", "user2"}, gravitinoClient.listUserNames());
+
+    ErrorResponse errRespNoMetalake =
+        ErrorResponse.notFound(NoSuchMetalakeException.class.getSimpleName(), "metalake not found");
+    buildMockResource(Method.GET, userPath, null, errRespNoMetalake, SC_NOT_FOUND);
+    Exception ex =
+        Assertions.assertThrows(
+            NoSuchMetalakeException.class, () -> gravitinoClient.listUserNames());
+    Assertions.assertEquals("metalake not found", ex.getMessage());
+
+    // Test RuntimeException
+    ErrorResponse errResp = ErrorResponse.internalError("internal error");
+    buildMockResource(Method.GET, userPath, null, errResp, SC_SERVER_ERROR);
+    Assertions.assertThrows(RuntimeException.class, () -> gravitinoClient.listUserNames());
+  }
+
+  @Test
+  public void testListUsers() throws Exception {
+    String userPath = withSlash(String.format(API_METALAKES_USERS_PATH, metalakeName, ""));
+    UserDTO user1 = mockUserDTO("user1");
+    UserDTO user2 = mockUserDTO("user2");
+    Map<String, String> params = Collections.singletonMap("details", "true");
+    UserListResponse listResponse = new UserListResponse(new UserDTO[] {user1, user2});
+    buildMockResource(Method.GET, userPath, params, null, listResponse, SC_OK);
+
+    User[] users = gravitinoClient.listUsers();
+    Assertions.assertEquals(2, users.length);
+    assertUser(user1, users[0]);
+    assertUser(user2, users[1]);
+
+    ErrorResponse errRespNoMetalake =
+        ErrorResponse.notFound(NoSuchMetalakeException.class.getSimpleName(), "metalake not found");
+    buildMockResource(Method.GET, userPath, params, null, errRespNoMetalake, SC_NOT_FOUND);
+    Exception ex =
+        Assertions.assertThrows(NoSuchMetalakeException.class, () -> gravitinoClient.listUsers());
+    Assertions.assertEquals("metalake not found", ex.getMessage());
+
+    // Test RuntimeException
+    ErrorResponse errResp = ErrorResponse.internalError("internal error");
+    buildMockResource(Method.GET, userPath, params, null, errResp, SC_SERVER_ERROR);
+    Assertions.assertThrows(RuntimeException.class, () -> gravitinoClient.listUsers());
+  }
+
   @Test
   public void testAddGroups() throws Exception {
     String groupName = "group";

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

@@ -20,8 +20,10 @@
 
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.Map;
+import java.util.stream.Collectors;
 import org.apache.gravitino.Configs;
 import org.apache.gravitino.auth.AuthConstants;
 import org.apache.gravitino.authorization.Group;
@@ -73,11 +75,31 @@ void testManageUsers() {
     Assertions.assertEquals(username, user.name());
     Assertions.assertTrue(user.roles().isEmpty());
 
+    // List users
+    String anotherUser = "another-user";
+    metalake.addUser(anotherUser);
+    String[] usernames = metalake.listUserNames();
+    Arrays.sort(usernames);
+    Assertions.assertEquals(
+        Lists.newArrayList(AuthConstants.ANONYMOUS_USER, anotherUser, username),
+        Arrays.asList(usernames));
+    User[] users = metalake.listUsers();
+    Assertions.assertEquals(
+        Lists.newArrayList(AuthConstants.ANONYMOUS_USER, anotherUser, username),
+        Arrays.stream(users)
+            .map(User::name)
+            .sorted(String::compareTo)
+            .collect(Collectors.toList()));
+
     // Get a not-existed user
     Assertions.assertThrows(NoSuchUserException.class, () -> metalake.getUser("not-existed"));
 
     Assertions.assertTrue(metalake.removeUser(username));
+
     Assertions.assertFalse(metalake.removeUser(username));
+
+    // clean up
+    metalake.removeUser(anotherUser);
   }
 
   @Test

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

@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.dto.responses;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+import org.apache.gravitino.dto.authorization.UserDTO;
+
+/** Represents a response containing a list of users. */
+@Getter
+@ToString
+@EqualsAndHashCode(callSuper = true)
+public class UserListResponse extends BaseResponse {
+
+  @JsonProperty("users")
+  private final UserDTO[] users;
+
+  /**
+   * Constructor for UserListResponse.
+   *
+   * @param users The array of users.
+   */
+  public UserListResponse(UserDTO[] users) {
+    super(0);
+    this.users = users;
+  }
+
+  /**
+   * This is the constructor that is used by Jackson deserializer to create an instance of
+   * UserListResponse.
+   */
+  public UserListResponse() {
+    super(0);
+    this.users = null;
+  }
+
+  /**
+   * Validates the response data.
+   *
+   * @throws IllegalArgumentException if users are not set.
+   */
+  @Override
+  public void validate() throws IllegalArgumentException {
+    super.validate();
+    Preconditions.checkArgument(users != null, "users must not be null");
+  }
+}

common/src/main/java/org/apache/gravitino/dto/util/DTOConverters.java

@@ -678,6 +678,19 @@ public static CatalogDTO[] toDTOs(Catalog[] catalogs) {
     return Arrays.stream(catalogs).map(DTOConverters::toDTO).toArray(CatalogDTO[]::new);
   }
 
+  /**
+   * Converts an array of Users to an array of UserDTOs.
+   *
+   * @param users The users to be converted.
+   * @return The array of UserDTOs.
+   */
+  public static UserDTO[] toDTOs(User[] users) {
+    if (ArrayUtils.isEmpty(users)) {
+      return new UserDTO[0];
+    }
+    return Arrays.stream(users).map(DTOConverters::toDTO).toArray(UserDTO[]::new);
+  }
+
   /**
    * Converts a DistributionDTO to a Distribution.
    *

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

@@ -55,15 +55,19 @@ public interface EntityStore extends Closeable {
    * <p>Note. Depends on the isolation levels provided by the underlying storage, the returned list
    * may not be consistent.
    *
-   * @param namespace the namespace of the entities
    * @param <E> class of the entity
+   * @param namespace the namespace of the entities
    * @param type the detailed type of the entity
    * @param entityType the general type of the entity
-   * @throws IOException if the list operation fails
+   * @param allowMissingFields 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.
    * @return the list of entities
+   * @throws IOException if the list operation fails
    */
   <E extends Entity & HasIdentifier> List<E> list(
-      Namespace namespace, Class<E> type, EntityType entityType) throws IOException;
+      Namespace namespace, Class<E> type, EntityType entityType, List<Field> allowMissingFields)
+      throws IOException;
 
   /**
    * Check if the entity with the specified {@link org.apache.gravitino.NameIdentifier} exists.

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

@@ -71,6 +71,24 @@ User addUser(String metalake, String user)
    */
   User getUser(String metalake, String user) throws NoSuchUserException, NoSuchMetalakeException;
 
+  /**
+   * Lists the users.
+   *
+   * @param metalake The Metalake of the User.
+   * @return The User list.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   */
+  User[] listUsers(String metalake) throws NoSuchMetalakeException;
+
+  /**
+   * Lists the usernames.
+   *
+   * @param metalake The Metalake of the User.
+   * @return The username list.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   */
+  String[] listUserNames(String metalake) throws NoSuchMetalakeException;
+
   /**
    * Adds a new Group.
    *

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

@@ -69,6 +69,15 @@ public User getUser(String metalake, String user)
   }
 
   @Override
+  public String[] listUserNames(String metalake) throws NoSuchMetalakeException {
+    return userGroupManager.listUserNames(metalake);
+  }
+
+  @Override
+  public User[] listUsers(String metalake) throws NoSuchMetalakeException {
+    return userGroupManager.listUsers(metalake);
+  }
+
   public Group addGroup(String metalake, String group)
       throws GroupAlreadyExistsException, NoSuchMetalakeException {
     return userGroupManager.addGroup(metalake, group);
@@ -130,16 +139,6 @@ public Role getRole(String metalake, String role)
     return roleManager.getRole(metalake, role);
   }
 
-  /**
-   * Deletes a Role.
-   *
-   * @param metalake The Metalake of the Role.
-   * @param role The name of the Role.
-   * @return True if the Role was successfully deleted, false only when there's no such role,
-   *     otherwise it will throw an exception.
-   * @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 {
     return roleManager.deleteRole(metalake, role);
   }