[#4886] feat(server,core): Supports to list roles by object

api/src/main/java/org/apache/gravitino/Catalog.java

@@ -21,6 +21,7 @@
 import java.util.Locale;
 import java.util.Map;
 import org.apache.gravitino.annotation.Evolving;
+import org.apache.gravitino.authorization.SupportsRoles;
 import org.apache.gravitino.file.FilesetCatalog;
 import org.apache.gravitino.messaging.TopicCatalog;
 import org.apache.gravitino.rel.TableCatalog;
@@ -181,4 +182,12 @@ default TopicCatalog asTopicCatalog() throws UnsupportedOperationException {
   default SupportsTags supportsTags() throws UnsupportedOperationException {
     throw new UnsupportedOperationException("Catalog does not support tag operations");
   }
+
+  /**
+   * @return the {@link SupportsRoles} if the catalog supports role operations.
+   * @throws UnsupportedOperationException if the catalog does not support role operations.
+   */
+  default SupportsRoles supportsRoles() throws UnsupportedOperationException {
+    throw new UnsupportedOperationException("Catalog does not support role operations");
+  }
 }

api/src/main/java/org/apache/gravitino/Metalake.java

@@ -20,6 +20,7 @@
 
 import java.util.Map;
 import org.apache.gravitino.annotation.Evolving;
+import org.apache.gravitino.authorization.SupportsRoles;
 
 /**
  * The interface of a metalake. The metalake is the top level entity in the Apache Gravitino system,
@@ -50,4 +51,12 @@ public interface Metalake extends Auditable {
    * @return The properties of the metalake.
    */
   Map<String, String> properties();
+
+  /**
+   * @return the {@link SupportsRoles} if the metalake supports role operations.
+   * @throws UnsupportedOperationException if the metalake does not support role operations.
+   */
+  default SupportsRoles supportsRoles() {
+    throw new UnsupportedOperationException("Metalake does not support role operations.");
+  }
 }

api/src/main/java/org/apache/gravitino/Schema.java

@@ -22,6 +22,7 @@
 import java.util.Map;
 import javax.annotation.Nullable;
 import org.apache.gravitino.annotation.Evolving;
+import org.apache.gravitino.authorization.SupportsRoles;
 import org.apache.gravitino.tag.SupportsTags;
 
 /**
@@ -56,4 +57,12 @@ default Map<String, String> properties() {
   default SupportsTags supportsTags() {
     throw new UnsupportedOperationException("Schema does not support tag operations.");
   }
+
+  /**
+   * @return the {@link SupportsRoles} if the schema supports role operations.
+   * @throws UnsupportedOperationException if the schema does not support role operations.
+   */
+  default SupportsRoles supportsRoles() {
+    throw new UnsupportedOperationException("Schema does not support role operations.");
+  }
 }

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

@@ -0,0 +1,38 @@
+/*
+ * 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.authorization;
+
+import org.apache.gravitino.annotation.Evolving;
+
+/**
+ * Interface for supporting list role names for objects. This interface will be mixed with metadata
+ * objects to provide listing role operations.
+ */
+@Evolving
+public interface SupportsRoles {
+
+  /**
+   * List all the role names associated with this metadata object. For the metalake metadata object,
+   * this method will only return the roles contains metalake object instead of all the roles in the
+   * metalake.
+   *
+   * @return The role name list associated with this metadata object.
+   */
+  String[] listBindingRoleNames();
+}

api/src/main/java/org/apache/gravitino/file/Fileset.java

@@ -24,6 +24,7 @@
 import org.apache.gravitino.Auditable;
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.annotation.Evolving;
+import org.apache.gravitino.authorization.SupportsRoles;
 import org.apache.gravitino.tag.SupportsTags;
 
 /**
@@ -114,4 +115,12 @@ default Map<String, String> properties() {
   default SupportsTags supportsTags() {
     throw new UnsupportedOperationException("Fileset does not support tag operations.");
   }
+
+  /**
+   * @return The {@link SupportsRoles} if the fileset supports role operations.
+   * @throws UnsupportedOperationException If the fileset does not support role operations.
+   */
+  default SupportsRoles supportsRoles() {
+    throw new UnsupportedOperationException("Fileset does not support role operations.");
+  }
 }

api/src/main/java/org/apache/gravitino/messaging/Topic.java

@@ -24,6 +24,7 @@
 import org.apache.gravitino.Auditable;
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.annotation.Evolving;
+import org.apache.gravitino.authorization.SupportsRoles;
 import org.apache.gravitino.tag.SupportsTags;
 
 /**
@@ -58,4 +59,12 @@ default Map<String, String> properties() {
   default SupportsTags supportsTags() {
     throw new UnsupportedOperationException("Topic does not support tag operations.");
   }
+
+  /**
+   * @return the {@link SupportsRoles} if the topic supports role operations.
+   * @throws UnsupportedOperationException if the topic does not support role operations.
+   */
+  default SupportsRoles supportsRoles() {
+    throw new UnsupportedOperationException("Topic does not support role operations.");
+  }
 }

api/src/main/java/org/apache/gravitino/rel/Table.java

@@ -24,6 +24,7 @@
 import org.apache.gravitino.Auditable;
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.annotation.Evolving;
+import org.apache.gravitino.authorization.SupportsRoles;
 import org.apache.gravitino.rel.expressions.distributions.Distribution;
 import org.apache.gravitino.rel.expressions.distributions.Distributions;
 import org.apache.gravitino.rel.expressions.sorts.SortOrder;
@@ -103,4 +104,12 @@ default SupportsPartitions supportPartitions() throws UnsupportedOperationExcept
   default SupportsTags supportsTags() {
     throw new UnsupportedOperationException("Table does not support tag operations.");
   }
+
+  /**
+   * @return The {@link SupportsRoles} if the table supports role operations.
+   * @throws UnsupportedOperationException If the table does not support role operations.
+   */
+  default SupportsRoles supportsRoles() {
+    throw new UnsupportedOperationException("Table does not support role operations.");
+  }
 }

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

@@ -31,6 +31,7 @@
 import org.apache.gravitino.Schema;
 import org.apache.gravitino.SchemaChange;
 import org.apache.gravitino.SupportsSchemas;
+import org.apache.gravitino.authorization.SupportsRoles;
 import org.apache.gravitino.dto.AuditDTO;
 import org.apache.gravitino.dto.CatalogDTO;
 import org.apache.gravitino.dto.requests.SchemaCreateRequest;
@@ -53,14 +54,15 @@
  * create, load, alter and drop a schema with specified identifier.
  */
 abstract class BaseSchemaCatalog extends CatalogDTO
-    implements Catalog, SupportsSchemas, SupportsTags {
+    implements Catalog, SupportsSchemas, SupportsTags, SupportsRoles {
   /** The REST client to send the requests. */
   protected final RESTClient restClient;
 
   /** The namespace of current catalog, which is the metalake name. */
   private final Namespace catalogNamespace;
 
   private final MetadataObjectTagOperations objectTagOperations;
+  private final MetadataObjectRoleOperations objectRoleOperations;
 
   BaseSchemaCatalog(
       Namespace catalogNamespace,
@@ -84,6 +86,8 @@ abstract class BaseSchemaCatalog extends CatalogDTO
         MetadataObjects.of(null, this.name(), MetadataObject.Type.CATALOG);
     this.objectTagOperations =
         new MetadataObjectTagOperations(catalogNamespace.level(0), metadataObject, restClient);
+    this.objectRoleOperations =
+        new MetadataObjectRoleOperations(catalogNamespace.level(0), metadataObject, restClient);
   }
 
   @Override
@@ -96,6 +100,11 @@ public SupportsTags supportsTags() throws UnsupportedOperationException {
     return this;
   }
 
+  @Override
+  public SupportsRoles supportsRoles() throws UnsupportedOperationException {
+    return this;
+  }
+
   /**
    * List all the schemas under the given catalog namespace.
    *
@@ -239,6 +248,11 @@ public String[] associateTags(String[] tagsToAdd, String[] tagsToRemove) {
     return objectTagOperations.associateTags(tagsToAdd, tagsToRemove);
   }
 
+  @Override
+  public String[] listBindingRoleNames() {
+    return objectRoleOperations.listBindingRoleNames();
+  }
+
   /**
    * Get the namespace of the current catalog, which is "metalake".
    *

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

@@ -26,18 +26,20 @@
 import org.apache.gravitino.MetadataObject;
 import org.apache.gravitino.MetadataObjects;
 import org.apache.gravitino.Namespace;
+import org.apache.gravitino.authorization.SupportsRoles;
 import org.apache.gravitino.dto.file.FilesetDTO;
 import org.apache.gravitino.exceptions.NoSuchTagException;
 import org.apache.gravitino.file.Fileset;
 import org.apache.gravitino.tag.SupportsTags;
 import org.apache.gravitino.tag.Tag;
 
 /** Represents a generic fileset. */
-class GenericFileset implements Fileset, SupportsTags {
+class GenericFileset implements Fileset, SupportsTags, SupportsRoles {
 
   private final FilesetDTO filesetDTO;
 
   private final MetadataObjectTagOperations objectTagOperations;
+  private final MetadataObjectRoleOperations objectRoleOperations;
 
   GenericFileset(FilesetDTO filesetDTO, RESTClient restClient, Namespace filesetNs) {
     this.filesetDTO = filesetDTO;
@@ -46,6 +48,8 @@ class GenericFileset implements Fileset, SupportsTags {
     MetadataObject filesetObject = MetadataObjects.of(filesetFullName, MetadataObject.Type.FILESET);
     this.objectTagOperations =
         new MetadataObjectTagOperations(filesetNs.level(0), filesetObject, restClient);
+    this.objectRoleOperations =
+        new MetadataObjectRoleOperations(filesetNs.level(0), filesetObject, restClient);
   }
 
   @Override
@@ -84,6 +88,11 @@ public SupportsTags supportsTags() {
     return this;
   }
 
+  @Override
+  public SupportsRoles supportsRoles() {
+    return this;
+  }
+
   @Override
   public String[] listTags() {
     return objectTagOperations.listTags();
@@ -104,6 +113,11 @@ public String[] associateTags(String[] tagsToAdd, String[] tagsToRemove) {
     return objectTagOperations.associateTags(tagsToAdd, tagsToRemove);
   }
 
+  @Override
+  public String[] listBindingRoleNames() {
+    return objectRoleOperations.listBindingRoleNames();
+  }
+
   @Override
   public boolean equals(Object obj) {
     if (this == obj) {

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

@@ -23,30 +23,39 @@
 import org.apache.gravitino.MetadataObject;
 import org.apache.gravitino.MetadataObjects;
 import org.apache.gravitino.Schema;
+import org.apache.gravitino.authorization.SupportsRoles;
 import org.apache.gravitino.dto.SchemaDTO;
 import org.apache.gravitino.exceptions.NoSuchTagException;
 import org.apache.gravitino.tag.SupportsTags;
 import org.apache.gravitino.tag.Tag;
 
 /** Represents a generic schema. */
-class GenericSchema implements Schema, SupportsTags {
+class GenericSchema implements Schema, SupportsTags, SupportsRoles {
 
   private final SchemaDTO schemaDTO;
 
   private final MetadataObjectTagOperations objectTagOperations;
+  private final MetadataObjectRoleOperations objectRoleOperations;
 
   GenericSchema(SchemaDTO schemaDTO, RESTClient restClient, String metalake, String catalog) {
     this.schemaDTO = schemaDTO;
     MetadataObject schemaObject =
         MetadataObjects.of(catalog, schemaDTO.name(), MetadataObject.Type.SCHEMA);
     this.objectTagOperations = new MetadataObjectTagOperations(metalake, schemaObject, restClient);
+    this.objectRoleOperations =
+        new MetadataObjectRoleOperations(metalake, schemaObject, restClient);
   }
 
   @Override
   public SupportsTags supportsTags() {
     return this;
   }
 
+  @Override
+  public SupportsRoles supportsRoles() {
+    return this;
+  }
+
   @Override
   public String name() {
     return schemaDTO.name();
@@ -87,6 +96,11 @@ public String[] associateTags(String[] tagsToAdd, String[] tagsToRemove) {
     return objectTagOperations.associateTags(tagsToAdd, tagsToRemove);
   }
 
+  @Override
+  public String[] listBindingRoleNames() {
+    return objectRoleOperations.listBindingRoleNames();
+  }
+
   @Override
   public boolean equals(Object obj) {
     if (this == obj) {

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

@@ -25,18 +25,20 @@
 import org.apache.gravitino.MetadataObject;
 import org.apache.gravitino.MetadataObjects;
 import org.apache.gravitino.Namespace;
+import org.apache.gravitino.authorization.SupportsRoles;
 import org.apache.gravitino.dto.messaging.TopicDTO;
 import org.apache.gravitino.exceptions.NoSuchTagException;
 import org.apache.gravitino.messaging.Topic;
 import org.apache.gravitino.tag.SupportsTags;
 import org.apache.gravitino.tag.Tag;
 
 /** Represents a generic topic. */
-class GenericTopic implements Topic, SupportsTags {
+class GenericTopic implements Topic, SupportsTags, SupportsRoles {
 
   private final TopicDTO topicDTO;
 
   private final MetadataObjectTagOperations objectTagOperations;
+  private final MetadataObjectRoleOperations objectRoleOperations;
 
   GenericTopic(TopicDTO topicDTO, RESTClient restClient, Namespace topicNs) {
     this.topicDTO = topicDTO;
@@ -45,6 +47,8 @@ class GenericTopic implements Topic, SupportsTags {
     MetadataObject topicObject = MetadataObjects.of(topicFullName, MetadataObject.Type.TOPIC);
     this.objectTagOperations =
         new MetadataObjectTagOperations(topicNs.level(0), topicObject, restClient);
+    this.objectRoleOperations =
+        new MetadataObjectRoleOperations(topicNs.level(0), topicObject, restClient);
   }
 
   @Override
@@ -72,6 +76,11 @@ public SupportsTags supportsTags() {
     return this;
   }
 
+  @Override
+  public SupportsRoles supportsRoles() {
+    return this;
+  }
+
   @Override
   public String[] listTags() {
     return objectTagOperations.listTags();
@@ -92,6 +101,11 @@ public String[] associateTags(String[] tagsToAdd, String[] tagsToRemove) {
     return objectTagOperations.associateTags(tagsToAdd, tagsToRemove);
   }
 
+  @Override
+  public String[] listBindingRoleNames() {
+    return objectRoleOperations.listBindingRoleNames();
+  }
+
   @Override
   public boolean equals(Object obj) {
     if (this == obj) {