[#2234] feat(core): Add the support of User entity (#2481)

### What changes were proposed in this pull request?

Add the UserEntity. Gravitino doesn't manage users, just sets up the
relationship between the metalake and the user. So we don't bring too
many fields in the User entity. More user information should be managed
by external user system.

### Why are the changes needed?

Fix: #2234

### Does this PR introduce _any_ user-facing change?
No.

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

---------

Co-authored-by: Heng Qin <qqtt@123.com>

api/src/main/java/com/datastrato/gravitino/authorization/User.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.authorization;
+
+import com.datastrato.gravitino.Auditable;
+import com.datastrato.gravitino.annotation.Evolving;
+import java.util.List;
+
+/** The interface of a user. The user is the entity which executes every operation. */
+@Evolving
+public interface User extends Auditable {
+
+  /**
+   * The name of the user. It's the identifier of User. It must be unique. Usually the name comes
+   * from a external user management system like LDAP, IAM and so on.
+   *
+   * @return The name of the user.
+   */
+  String name();
+
+  /**
+   * The roles of the user. A user can have multiple roles. Every role binds several privileges.
+   *
+   * @return The roles of the user.
+   */
+  List<String> roles();
+}

api/src/main/java/com/datastrato/gravitino/exceptions/NoSuchUserException.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.exceptions;
+
+import com.google.errorprone.annotations.FormatMethod;
+import com.google.errorprone.annotations.FormatString;
+
+/** An exception thrown when a user is not found. */
+public class NoSuchUserException extends NotFoundException {
+
+  /**
+   * Constructs a new exception with the specified detail message.
+   *
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public NoSuchUserException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+
+  /**
+   * Constructs a new exception with the specified detail message and cause.
+   *
+   * @param cause the cause.
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public NoSuchUserException(Throwable cause, String message, Object... args) {
+    super(cause, message, args);
+  }
+}

api/src/main/java/com/datastrato/gravitino/exceptions/UserAlreadyExistsException.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.exceptions;
+
+import com.google.errorprone.annotations.FormatMethod;
+import com.google.errorprone.annotations.FormatString;
+
+/** An exception thrown when a user already exists. */
+public class UserAlreadyExistsException extends AlreadyExistsException {
+
+  /**
+   * Constructs a new exception with the specified detail message.
+   *
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public UserAlreadyExistsException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+
+  /**
+   * Constructs a new exception with the specified detail message and cause.
+   *
+   * @param cause the cause.
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public UserAlreadyExistsException(Throwable cause, String message, Object... args) {
+    super(cause, message, args);
+  }
+}

core/src/main/java/com/datastrato/gravitino/Entity.java

@@ -21,6 +21,7 @@ enum EntityType {
     COLUMN("co", 4),
     FILESET("fi", 5),
     TOPIC("to", 6),
+    USER("us", 7),
 
     AUDIT("au", 65534);
 

core/src/main/java/com/datastrato/gravitino/GravitinoEnv.java

@@ -4,6 +4,7 @@
  */
 package com.datastrato.gravitino;
 
+import com.datastrato.gravitino.authorization.AccessControlManager;
 import com.datastrato.gravitino.auxiliary.AuxiliaryServiceManager;
 import com.datastrato.gravitino.catalog.CatalogManager;
 import com.datastrato.gravitino.catalog.CatalogOperationDispatcher;
@@ -35,6 +36,8 @@ public class GravitinoEnv {
 
   private MetalakeManager metalakeManager;
 
+  private AccessControlManager accessControlManager;
+
   private IdGenerator idGenerator;
 
   private AuxiliaryServiceManager auxServiceManager;
@@ -97,6 +100,9 @@ public void initialize(Config config) {
     this.catalogOperationDispatcher =
         new CatalogOperationDispatcher(catalogManager, entityStore, idGenerator);
 
+    // Create and initialize access control related modules
+    this.accessControlManager = new AccessControlManager(entityStore, idGenerator);
+
     this.auxServiceManager = new AuxiliaryServiceManager();
     this.auxServiceManager.serviceInit(
         config.getConfigsWithPrefix(AuxiliaryServiceManager.GRAVITINO_AUX_SERVICE_PREFIX));
@@ -174,6 +180,15 @@ public LockManager getLockManager() {
     return lockManager;
   }
 
+  /**
+   * Get the AccessControlManager associated with the Gravitino environment.
+   *
+   * @return The AccessControlManager instance.
+   */
+  public AccessControlManager accessControlManager() {
+    return accessControlManager;
+  }
+
   public void start() {
     auxServiceManager.serviceStart();
     metricsSystem.start();

core/src/main/java/com/datastrato/gravitino/authorization/AccessControlManager.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.authorization;
+
+import com.datastrato.gravitino.EntityStore;
+import com.datastrato.gravitino.exceptions.NoSuchUserException;
+import com.datastrato.gravitino.exceptions.UserAlreadyExistsException;
+import com.datastrato.gravitino.storage.IdGenerator;
+
+/**
+ * AccessControlManager is used for manage users, roles, grant information, this class is an
+ * entrance class for tenant management.
+ */
+public class AccessControlManager {
+
+  private final UserManager userManager;
+
+  public AccessControlManager(EntityStore store, IdGenerator idGenerator) {
+    this.userManager = new UserManager(store, idGenerator);
+  }
+
+  /**
+   * Adds a new User.
+   *
+   * @param metalake The Metalake of the User.
+   * @param name The name of the User.
+   * @return The added User instance.
+   * @throws UserAlreadyExistsException If a User with the same identifier already exists.
+   * @throws RuntimeException If adding the User encounters storage issues.
+   */
+  public User addUser(String metalake, String name) throws UserAlreadyExistsException {
+    return userManager.addUser(metalake, name);
+  }
+
+  /**
+   * Removes a User.
+   *
+   * @param metalake The Metalake of the User.
+   * @param user THe name of the User.
+   * @return `true` if the User was successfully removed, `false` otherwise.
+   * @throws RuntimeException If removing the User encounters storage issues.
+   */
+  public boolean removeUser(String metalake, String user) {
+    return userManager.removeUser(metalake, user);
+  }
+
+  /**
+   * Gets a User.
+   *
+   * @param metalake The Metalake of the User.
+   * @param user The name of the User.
+   * @return The getting User instance.
+   * @throws NoSuchUserException If the User with the given identifier does not exist.
+   * @throws RuntimeException If getting the User encounters storage issues.
+   */
+  public User getUser(String metalake, String user) throws NoSuchUserException {
+    return userManager.getUser(metalake, user);
+  }
+}

core/src/main/java/com/datastrato/gravitino/authorization/UserManager.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.authorization;
+
+import com.datastrato.gravitino.Entity;
+import com.datastrato.gravitino.EntityAlreadyExistsException;
+import com.datastrato.gravitino.EntityStore;
+import com.datastrato.gravitino.NameIdentifier;
+import com.datastrato.gravitino.Namespace;
+import com.datastrato.gravitino.exceptions.NoSuchEntityException;
+import com.datastrato.gravitino.exceptions.NoSuchUserException;
+import com.datastrato.gravitino.exceptions.UserAlreadyExistsException;
+import com.datastrato.gravitino.meta.AuditInfo;
+import com.datastrato.gravitino.meta.CatalogEntity;
+import com.datastrato.gravitino.meta.UserEntity;
+import com.datastrato.gravitino.storage.IdGenerator;
+import com.datastrato.gravitino.utils.PrincipalUtils;
+import com.google.common.collect.Lists;
+import java.io.IOException;
+import java.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * UserManager is used for add, remove and get users from one metalake. UserManager doesn't manage
+ * users, just sets up the relationship between the metalake and the user. Metalake is like a
+ * concept of the organization. `AddUser` means that a user enter an organization.
+ */
+public class UserManager {
+
+  private static final Logger LOG = LoggerFactory.getLogger(UserManager.class);
+  private static final String USER_DOES_NOT_EXIST_MSG = "User %s does not exist in th metalake %s";
+
+  private final EntityStore store;
+  private final IdGenerator idGenerator;
+
+  public UserManager(EntityStore store, IdGenerator idGenerator) {
+    this.store = store;
+    this.idGenerator = idGenerator;
+  }
+
+  /**
+   * Adds a new User.
+   *
+   * @param metalake The Metalake of the User.
+   * @param name The name of the User.
+   * @return The added User instance.
+   * @throws UserAlreadyExistsException If a User with the same identifier already exists.
+   * @throws RuntimeException If adding the User encounters storage issues.
+   */
+  public User addUser(String metalake, String name) throws UserAlreadyExistsException {
+    UserEntity userEntity =
+        UserEntity.builder()
+            .withId(idGenerator.nextId())
+            .withName(name)
+            .withNamespace(
+                Namespace.of(
+                    metalake,
+                    CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME,
+                    UserEntity.USER_SCHEMA_NAME))
+            .withRoles(Lists.newArrayList())
+            .withAuditInfo(
+                AuditInfo.builder()
+                    .withCreator(PrincipalUtils.getCurrentPrincipal().getName())
+                    .withCreateTime(Instant.now())
+                    .build())
+            .build();
+    try {
+      store.put(userEntity, false /* overwritten */);
+      return userEntity;
+    } catch (EntityAlreadyExistsException e) {
+      LOG.warn("User {} in the metalake {} already exists", name, metalake, e);
+      throw new UserAlreadyExistsException(
+          "User %s in the metalake %s already exists", name, metalake);
+    } catch (IOException ioe) {
+      LOG.error(
+          "Adding user {} failed in the metalake {} due to storage issues", name, metalake, ioe);
+      throw new RuntimeException(ioe);
+    }
+  }
+
+  /**
+   * Removes a User.
+   *
+   * @param metalake The Metalake of the User.
+   * @param user THe name of the User.
+   * @return `true` if the User was successfully removed, `false` otherwise.
+   * @throws RuntimeException If removing the User encounters storage issues.
+   */
+  public boolean removeUser(String metalake, String user) {
+
+    try {
+      return store.delete(ofUser(metalake, user), Entity.EntityType.USER);
+    } catch (IOException ioe) {
+      LOG.error(
+          "Removing user {} in the metalake {} failed due to storage issues", user, metalake, ioe);
+      throw new RuntimeException(ioe);
+    }
+  }
+
+  /**
+   * Gets a User.
+   *
+   * @param metalake The Metalake of the User.
+   * @param user The name of the User.
+   * @return The getting User instance.
+   * @throws NoSuchUserException If the User with the given identifier does not exist.
+   * @throws RuntimeException If getting the User encounters storage issues.
+   */
+  public User getUser(String metalake, String user) throws NoSuchUserException {
+    try {
+      return store.get(ofUser(metalake, user), Entity.EntityType.USER, UserEntity.class);
+    } catch (NoSuchEntityException e) {
+      LOG.warn("User {} does not exist in the metalake {}", user, metalake, e);
+      throw new NoSuchUserException(USER_DOES_NOT_EXIST_MSG, user, metalake);
+    } catch (IOException ioe) {
+      LOG.error("Getting user {} failed due to storage issues", user, ioe);
+      throw new RuntimeException(ioe);
+    }
+  }
+
+  private NameIdentifier ofUser(String metalake, String user) {
+    return NameIdentifier.of(
+        metalake, CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME, UserEntity.USER_SCHEMA_NAME, user);
+  }
+}

core/src/main/java/com/datastrato/gravitino/catalog/CatalogManager.java

@@ -272,6 +272,10 @@ public Catalog createCatalog(
       Map<String, String> properties)
       throws NoSuchMetalakeException, CatalogAlreadyExistsException {
 
+    if (CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME.equals(ident.name())) {
+      throw new IllegalArgumentException("Can't create a catalog with with reserved name `system`");
+    }
+
     // load catalog-related configuration from catalog-specific configuration file
     Map<String, String> catalogSpecificConfig = loadCatalogSpecificConfig(properties, provider);
     Map<String, String> mergedConfig = mergeConf(properties, catalogSpecificConfig);
@@ -658,6 +662,13 @@ private CatalogEntity.Builder updateEntity(
     for (CatalogChange change : changes) {
       if (change instanceof CatalogChange.RenameCatalog) {
         CatalogChange.RenameCatalog rename = (CatalogChange.RenameCatalog) change;
+
+        if (CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME.equals(
+            ((CatalogChange.RenameCatalog) change).getNewName())) {
+          throw new IllegalArgumentException(
+              "Can't rename a catalog with with reserved name `system`");
+        }
+
         builder.withName(rename.getNewName());
 
       } else if (change instanceof CatalogChange.UpdateCatalogComment) {

core/src/main/java/com/datastrato/gravitino/meta/CatalogEntity.java

@@ -24,6 +24,8 @@
 @ToString
 public class CatalogEntity implements Entity, Auditable, HasIdentifier {
 
+  public static final String SYSTEM_CATALOG_RESERVED_NAME = "system";
+
   public static final Field ID =
       Field.required("id", Long.class, "The catalog's unique identifier");
   public static final Field NAME = Field.required("name", String.class, "The catalog's name");