[apache#3343] feat(core): Supports multiple securable objects (apache…

…#3497)

### What changes were proposed in this pull request?
Refactor the underlying layout to support multiple securable objects.

### Why are the changes needed?

Fix: apache#3343 

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

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

---------

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

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

@@ -40,6 +40,18 @@ public interface Entity extends Serializable {
   /** The admin schema name in the authorization catalog of the system metalake. */
   String ADMIN_SCHEMA_NAME = "admin";
 
+  /**
+   * All metalakes are a virtual entity. It represents all the metalakes. We don't store it. We use
+   * a specific type to represent its entity type.
+   */
+  String ALL_METALAKES_ENTITY_TYPE = "ROOT";
+
+  /**
+   * All metalakes are a virtual entity. It represents all the metalakes. We don't store it. We use
+   * a specific id to represent its entity id.
+   */
+  long ALL_METALAKES_ENTITY_ID = 0;
+
   /** Enumeration defining the types of entities in the Gravitino framework. */
   @Getter
   enum EntityType {

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

@@ -249,17 +249,20 @@ public boolean isMetalakeAdmin(String user) {
    * @param metalake The Metalake of the Role.
    * @param role The name of the Role.
    * @param properties The properties of the Role.
-   * @param securableObject The securable object of the Role.
+   * @param securableObjects The securable objects of the Role.
    * @return The created Role instance.
    * @throws RoleAlreadyExistsException If a Role with the same name already exists.
    * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
    * @throws RuntimeException If creating the Role encounters storage issues.
    */
   public Role createRole(
-      String metalake, String role, Map<String, String> properties, SecurableObject securableObject)
+      String metalake,
+      String role,
+      Map<String, String> properties,
+      List<SecurableObject> securableObjects)
       throws RoleAlreadyExistsException, NoSuchMetalakeException {
     return doWithNonAdminLock(
-        () -> roleManager.createRole(metalake, role, properties, securableObject));
+        () -> roleManager.createRole(metalake, role, properties, securableObjects));
   }
 
   /**

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

@@ -72,15 +72,18 @@ class RoleManager {
   }
 
   RoleEntity createRole(
-      String metalake, String role, Map<String, String> properties, SecurableObject securableObject)
+      String metalake,
+      String role,
+      Map<String, String> properties,
+      List<SecurableObject> securableObjects)
       throws RoleAlreadyExistsException {
     AuthorizationUtils.checkMetalakeExists(metalake);
     RoleEntity roleEntity =
         RoleEntity.builder()
             .withId(idGenerator.nextId())
             .withName(role)
             .withProperties(properties)
-            .withSecurableObject(securableObject)
+            .withSecurableObjects(securableObjects)
             .withNamespace(AuthorizationUtils.ofRoleNamespace(metalake))
             .withAuditInfo(
                 AuditInfo.builder()

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

@@ -11,7 +11,7 @@
 import com.datastrato.gravitino.Namespace;
 import com.datastrato.gravitino.authorization.Role;
 import com.datastrato.gravitino.authorization.SecurableObject;
-import com.google.common.collect.Lists;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Maps;
 import java.util.Collections;
 import java.util.List;
@@ -33,15 +33,14 @@ public class RoleEntity implements Role, Entity, Auditable, HasIdentifier {
       Field.required("audit_info", AuditInfo.class, "The audit details of the role entity.");
 
   public static final Field SECURABLE_OBJECT =
-      Field.required(
-          "securable_object", SecurableObject.class, "The securable object of the role entity.");
+      Field.required("securable_objects", List.class, "The securable objects of the role entity.");
 
   private Long id;
   private String name;
   private Map<String, String> properties;
   private AuditInfo auditInfo;
   private Namespace namespace;
-  private SecurableObject securableObject;
+  private List<SecurableObject> securableObjects;
 
   /**
    * The name of the role.
@@ -87,7 +86,7 @@ public List<SecurableObject> securableObjects() {
     // So one type of them can't be the securable object at least if there are the two same
     // identifier
     // entities .
-    return Lists.newArrayList(securableObject);
+    return securableObjects;
   }
 
   /**
@@ -102,7 +101,7 @@ public Map<Field, Object> fields() {
     fields.put(NAME, name);
     fields.put(AUDIT_INFO, auditInfo);
     fields.put(PROPERTIES, properties);
-    fields.put(SECURABLE_OBJECT, securableObject);
+    fields.put(SECURABLE_OBJECT, securableObjects);
 
     return Collections.unmodifiableMap(fields);
   }
@@ -138,12 +137,12 @@ public boolean equals(Object o) {
         && Objects.equals(namespace, that.namespace)
         && Objects.equals(auditInfo, that.auditInfo)
         && Objects.equals(properties, that.properties)
-        && Objects.equals(securableObject, that.securableObject);
+        && Objects.equals(securableObjects, that.securableObjects);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(id, name, properties, auditInfo, securableObject);
+    return Objects.hash(id, name, properties, auditInfo, securableObjects);
   }
 
   /**
@@ -212,13 +211,13 @@ public Builder withAuditInfo(AuditInfo auditInfo) {
     }
 
     /**
-     * Sets the securable object of the role entity.
+     * Sets the securable objects of the role entity.
      *
-     * @param securableObject The securable object of the role entity.
+     * @param securableObjects The securable objects of the role entity.
      * @return The builder instance.
      */
-    public Builder withSecurableObject(SecurableObject securableObject) {
-      roleEntity.securableObject = securableObject;
+    public Builder withSecurableObjects(List<SecurableObject> securableObjects) {
+      roleEntity.securableObjects = ImmutableList.copyOf(securableObjects);
       return this;
     }
 

core/src/main/java/com/datastrato/gravitino/proto/RoleEntitySerDe.java

@@ -28,17 +28,25 @@ public Role serialize(RoleEntity roleEntity) {
         Role.newBuilder()
             .setId(roleEntity.id())
             .setName(roleEntity.name())
-            .setAuditInfo(new AuditInfoSerDe().serialize(roleEntity.auditInfo()))
-            .addAllPrivileges(
-                roleEntity.securableObjects().get(0).privileges().stream()
-                    .map(privilege -> privilege.name().toString())
-                    .collect(Collectors.toList()))
-            .addAllPrivilegeConditions(
-                roleEntity.securableObjects().get(0).privileges().stream()
-                    .map(privilege -> privilege.condition().toString())
-                    .collect(Collectors.toList()))
-            .setSecurableObjectFullName(roleEntity.securableObjects().get(0).fullName())
-            .setSecurableObjectType(roleEntity.securableObjects().get(0).type().name());
+            .setAuditInfo(new AuditInfoSerDe().serialize(roleEntity.auditInfo()));
+
+    for (SecurableObject securableObject : roleEntity.securableObjects()) {
+      builder.addSecurableObjects(
+          com.datastrato.gravitino.proto.SecurableObject.newBuilder()
+              .setFullName(securableObject.fullName())
+              .setType(securableObject.type().name())
+              .addAllPrivilegeConditions(
+                  securableObject.privileges().stream()
+                      .map(Privilege::condition)
+                      .map(Privilege.Condition::name)
+                      .collect(Collectors.toList()))
+              .addAllPrivilegeNames(
+                  securableObject.privileges().stream()
+                      .map(Privilege::name)
+                      .map(Privilege.Name::name)
+                      .collect(Collectors.toList()))
+              .build());
+    }
 
     if (roleEntity.properties() != null && !roleEntity.properties().isEmpty()) {
       builder.putAllProperties(roleEntity.properties());
@@ -55,31 +63,32 @@ public Role serialize(RoleEntity roleEntity) {
    */
   @Override
   public RoleEntity deserialize(Role role, Namespace namespace) {
-    List<Privilege> privileges = Lists.newArrayList();
-
-    if (!role.getPrivilegesList().isEmpty()) {
+    List<SecurableObject> securableObjects = Lists.newArrayList();
 
-      for (int index = 0; index < role.getPrivilegeConditionsCount(); index++) {
-        if (Privilege.Condition.ALLOW.name().equals(role.getPrivilegeConditions(index))) {
-          privileges.add(Privileges.allow(role.getPrivileges(index)));
+    for (int index = 0; index < role.getSecurableObjectsCount(); index++) {
+      List<Privilege> privileges = Lists.newArrayList();
+      com.datastrato.gravitino.proto.SecurableObject object = role.getSecurableObjects(index);
+      for (int privIndex = 0; privIndex < object.getPrivilegeConditionsCount(); privIndex++) {
+        if (Privilege.Condition.ALLOW.name().equals(object.getPrivilegeConditions(privIndex))) {
+          privileges.add(Privileges.allow(object.getPrivilegeNames(privIndex)));
         } else {
-          privileges.add(Privileges.deny(role.getPrivileges(index)));
+          privileges.add(Privileges.deny(object.getPrivilegeNames(privIndex)));
         }
       }
-    }
 
-    SecurableObject securableObject =
-        SecurableObjects.parse(
-            role.getSecurableObjectFullName(),
-            SecurableObject.Type.valueOf(role.getSecurableObjectType()),
-            privileges);
+      SecurableObject securableObject =
+          SecurableObjects.parse(
+              object.getFullName(), SecurableObject.Type.valueOf(object.getType()), privileges);
+
+      securableObjects.add(securableObject);
+    }
 
     RoleEntity.Builder builder =
         RoleEntity.builder()
             .withId(role.getId())
             .withName(role.getName())
             .withNamespace(namespace)
-            .withSecurableObject(securableObject)
+            .withSecurableObjects(securableObjects)
             .withAuditInfo(new AuditInfoSerDe().deserialize(role.getAuditInfo(), namespace));
 
     if (!role.getPropertiesMap().isEmpty()) {

core/src/main/java/com/datastrato/gravitino/storage/relational/mapper/CatalogMetaMapper.java

@@ -54,6 +54,17 @@ Long selectCatalogIdByMetalakeIdAndName(
   CatalogPO selectCatalogMetaByMetalakeIdAndName(
       @Param("metalakeId") Long metalakeId, @Param("catalogName") String name);
 
+  @Select(
+      "SELECT catalog_id as catalogId, catalog_name as catalogName,"
+          + " metalake_id as metalakeId, type, provider,"
+          + " catalog_comment as catalogComment, properties, audit_info as auditInfo,"
+          + " current_version as currentVersion, last_version as lastVersion,"
+          + " deleted_at as deletedAt"
+          + " FROM "
+          + TABLE_NAME
+          + " WHERE catalog_id = #{catalogId} AND deleted_at = 0")
+  CatalogPO selectCatalogMetaById(@Param("catalogId") Long catalogId);
+
   @Insert(
       "INSERT INTO "
           + TABLE_NAME

core/src/main/java/com/datastrato/gravitino/storage/relational/mapper/FilesetMetaMapper.java

@@ -112,6 +112,44 @@ Long selectFilesetIdBySchemaIdAndName(
   FilesetPO selectFilesetMetaBySchemaIdAndName(
       @Param("schemaId") Long schemaId, @Param("filesetName") String name);
 
+  @Select(
+      "SELECT fm.fileset_id, fm.fileset_name, fm.metalake_id, fm.catalog_id, fm.schema_id,"
+          + " fm.type, fm.audit_info, fm.current_version, fm.last_version, fm.deleted_at,"
+          + " vi.id, vi.metalake_id as version_metalake_id, vi.catalog_id as version_catalog_id,"
+          + " vi.schema_id as version_schema_id, vi.fileset_id as version_fileset_id,"
+          + " vi.version, vi.fileset_comment, vi.properties, vi.storage_location,"
+          + " vi.deleted_at as version_deleted_at"
+          + " FROM "
+          + META_TABLE_NAME
+          + " fm INNER JOIN "
+          + VERSION_TABLE_NAME
+          + " vi ON fm.fileset_id = vi.fileset_id AND fm.current_version = vi.version"
+          + " WHERE fm.fileset_id = #{filesetId}"
+          + " AND fm.deleted_at = 0 AND vi.deleted_at = 0")
+  @Results({
+    @Result(property = "filesetId", column = "fileset_id"),
+    @Result(property = "filesetName", column = "fileset_name"),
+    @Result(property = "metalakeId", column = "metalake_id"),
+    @Result(property = "catalogId", column = "catalog_id"),
+    @Result(property = "schemaId", column = "schema_id"),
+    @Result(property = "type", column = "type"),
+    @Result(property = "auditInfo", column = "audit_info"),
+    @Result(property = "currentVersion", column = "current_version"),
+    @Result(property = "lastVersion", column = "last_version"),
+    @Result(property = "deletedAt", column = "deleted_at"),
+    @Result(property = "filesetVersionPO.id", column = "id"),
+    @Result(property = "filesetVersionPO.metalakeId", column = "version_metalake_id"),
+    @Result(property = "filesetVersionPO.catalogId", column = "version_catalog_id"),
+    @Result(property = "filesetVersionPO.schemaId", column = "version_schema_id"),
+    @Result(property = "filesetVersionPO.filesetId", column = "version_fileset_id"),
+    @Result(property = "filesetVersionPO.version", column = "version"),
+    @Result(property = "filesetVersionPO.filesetComment", column = "fileset_comment"),
+    @Result(property = "filesetVersionPO.properties", column = "properties"),
+    @Result(property = "filesetVersionPO.storageLocation", column = "storage_location"),
+    @Result(property = "filesetVersionPO.deletedAt", column = "version_deleted_at")
+  })
+  FilesetPO selectFilesetMetaById(@Param("filesetId") Long filesetId);
+
   @Insert(
       "INSERT INTO "
           + META_TABLE_NAME

core/src/main/java/com/datastrato/gravitino/storage/relational/mapper/MetalakeMetaMapper.java

@@ -46,6 +46,17 @@ public interface MetalakeMetaMapper {
           + " WHERE metalake_name = #{metalakeName} and deleted_at = 0")
   MetalakePO selectMetalakeMetaByName(@Param("metalakeName") String name);
 
+  @Select(
+      "SELECT metalake_id as metalakeId, metalake_name as metalakeName,"
+          + " metalake_comment as metalakeComment, properties,"
+          + " audit_info as auditInfo, schema_version as schemaVersion,"
+          + " current_version as currentVersion, last_version as lastVersion,"
+          + " deleted_at as deletedAt"
+          + " FROM "
+          + TABLE_NAME
+          + " WHERE metalake_id = #{metalaId} and deleted_at = 0")
+  MetalakePO selectMetalakeMetaById(@Param("metalakeId") Long metalakeId);
+
   @Select(
       "SELECT metalake_id FROM "
           + TABLE_NAME

core/src/main/java/com/datastrato/gravitino/storage/relational/mapper/RoleMetaMapper.java

@@ -29,10 +29,6 @@ public interface RoleMetaMapper {
   @Select(
       "SELECT role_id as roleId, role_name as roleName,"
           + " metalake_id as metalakeId, properties as properties,"
-          + " securable_object_full_name as securableObjectFullName,"
-          + " securable_object_type as securableObjectType,"
-          + " privileges as privileges,"
-          + " privilege_conditions as privilegeConditions,"
           + " audit_info as auditInfo, current_version as currentVersion,"
           + " last_version as lastVersion, deleted_at as deletedAt"
           + " FROM "
@@ -53,10 +49,6 @@ Long selectRoleIdByMetalakeIdAndName(
   @Select(
       "SELECT ro.role_id as roleId, ro.role_name as roleName,"
           + " ro.metalake_id as metalakeId, ro.properties as properties,"
-          + " securable_object_full_name as securableObjectFullName,"
-          + " securable_object_type as securableObjectType,"
-          + " ro.privileges as privileges,"
-          + " ro.privilege_conditions as privilegeConditions,"
           + " ro.audit_info as auditInfo, ro.current_version as currentVersion,"
           + " ro.last_version as lastVersion, ro.deleted_at as deletedAt"
           + " FROM "
@@ -71,10 +63,6 @@ Long selectRoleIdByMetalakeIdAndName(
   @Select(
       "SELECT ro.role_id as roleId, ro.role_name as roleName,"
           + " ro.metalake_id as metalakeId, ro.properties as properties,"
-          + " ro.securable_object_full_name as securableObjectFullName,"
-          + " ro.securable_object_type as securableObjectType,"
-          + " ro.privileges as privileges,"
-          + " ro.privilege_conditions as privilegeConditions,"
           + " ro.audit_info as auditInfo, ro.current_version as currentVersion,"
           + " ro.last_version as lastVersion, ro.deleted_at as deletedAt"
           + " FROM "
@@ -91,20 +79,12 @@ Long selectRoleIdByMetalakeIdAndName(
           + ROLE_TABLE_NAME
           + "(role_id, role_name,"
           + " metalake_id, properties,"
-          + " securable_object_full_name,"
-          + " securable_object_type,"
-          + " privileges,"
-          + " privilege_conditions,"
           + " audit_info, current_version, last_version, deleted_at)"
           + " VALUES("
           + " #{roleMeta.roleId},"
           + " #{roleMeta.roleName},"
           + " #{roleMeta.metalakeId},"
           + " #{roleMeta.properties},"
-          + " #{roleMeta.securableObjectFullName},"
-          + " #{roleMeta.securableObjectType},"
-          + " #{roleMeta.privileges},"
-          + " #{roleMeta.privilegeConditions},"
           + " #{roleMeta.auditInfo},"
           + " #{roleMeta.currentVersion},"
           + " #{roleMeta.lastVersion},"
@@ -117,20 +97,12 @@ Long selectRoleIdByMetalakeIdAndName(
           + ROLE_TABLE_NAME
           + "(role_id, role_name,"
           + " metalake_id, properties,"
-          + " securable_object_full_name,"
-          + " securable_object_type,"
-          + " privileges,"
-          + " privilege_conditions,"
           + " audit_info, current_version, last_version, deleted_at)"
           + " VALUES("
           + " #{roleMeta.roleId},"
           + " #{roleMeta.roleName},"
           + " #{roleMeta.metalakeId},"
           + " #{roleMeta.properties},"
-          + " #{roleMeta.securableObjectFullName},"
-          + " #{roleMeta.securableObjectType},"
-          + " #{roleMeta.privileges},"
-          + " #{roleMeta.privilegeConditions},"
           + " #{roleMeta.auditInfo},"
           + " #{roleMeta.currentVersion},"
           + " #{roleMeta.lastVersion},"
@@ -139,10 +111,6 @@ Long selectRoleIdByMetalakeIdAndName(
           + " role_name = #{roleMeta.roleName},"
           + " metalake_id = #{roleMeta.metalakeId},"
           + " properties = #{roleMeta.properties},"
-          + " securable_object_full_name = #{roleMeta.securableObjectFullName},"
-          + " securable_object_type = #{roleMeta.securableObjectType},"
-          + " privileges = #{roleMeta.privileges},"
-          + " privilege_conditions = #{roleMeta.privilegeConditions},"
           + " audit_info = #{roleMeta.auditInfo},"
           + " current_version = #{roleMeta.currentVersion},"
           + " last_version = #{roleMeta.lastVersion},"