[#9535] feat(api): introduce FUNCTION metadata object type and function privileges

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

@@ -73,7 +73,9 @@ enum Type {
     /** A job represents a data processing task in Gravitino. */
     JOB,
     /** A job template represents a reusable template for creating jobs in Gravitino. */
-    JOB_TEMPLATE;
+    JOB_TEMPLATE,
+    /** A function represents a user-defined function registered in Gravitino. */
+    FUNCTION;
   }
 
   /**

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

@@ -59,7 +59,8 @@ public class MetadataObjects {
           MetadataObject.Type.TABLE,
           MetadataObject.Type.VIEW,
           MetadataObject.Type.TOPIC,
-          MetadataObject.Type.MODEL);
+          MetadataObject.Type.MODEL,
+          MetadataObject.Type.FUNCTION);
 
   private static final Set<MetadataObject.Type> VALID_FOUR_LEVEL_NAME_TYPES =
       Sets.newHashSet(MetadataObject.Type.COLUMN);
@@ -151,6 +152,7 @@ public static MetadataObject parent(MetadataObject object) {
       case FILESET:
       case TOPIC:
       case MODEL:
+      case FUNCTION:
         parentType = MetadataObject.Type.SCHEMA;
         break;
       case SCHEMA:

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

@@ -145,7 +145,13 @@ enum Name {
     /** The privilege to create a view. */
     CREATE_VIEW(0L, 1L << 28),
     /** The privilege to select data from a view. */
-    SELECT_VIEW(0L, 1L << 29);
+    SELECT_VIEW(0L, 1L << 29),
+    /** The privilege to register a function. */
+    REGISTER_FUNCTION(0L, 1L << 30),
+    /** The privilege to execute (invoke) a function. */
+    EXECUTE_FUNCTION(0L, 1L << 31),
+    /** The privilege to alter a function's metadata. */
+    MODIFY_FUNCTION(0L, 1L << 32);
 
     private final long highBits;
     private final long lowBits;

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

@@ -64,6 +64,13 @@ public class Privileges {
           MetadataObject.Type.SCHEMA,
           MetadataObject.Type.VIEW);
 
+  private static final Set<MetadataObject.Type> FUNCTION_SUPPORTED_TYPES =
+      Sets.immutableEnumSet(
+          MetadataObject.Type.METALAKE,
+          MetadataObject.Type.CATALOG,
+          MetadataObject.Type.SCHEMA,
+          MetadataObject.Type.FUNCTION);
+
   /**
    * Object types that {@link ManageGrants} can be bound to.
    *
@@ -79,7 +86,8 @@ public class Privileges {
           MetadataObject.Type.VIEW,
           MetadataObject.Type.TOPIC,
           MetadataObject.Type.FILESET,
-          MetadataObject.Type.MODEL);
+          MetadataObject.Type.MODEL,
+          MetadataObject.Type.FUNCTION);
 
   /**
    * Returns the Privilege with allow condition from the string representation.
@@ -190,6 +198,14 @@ public static Privilege allow(Privilege.Name name) {
       case SELECT_VIEW:
         return SelectView.allow();
 
+        // Function
+      case REGISTER_FUNCTION:
+        return RegisterFunction.allow();
+      case EXECUTE_FUNCTION:
+        return ExecuteFunction.allow();
+      case MODIFY_FUNCTION:
+        return ModifyFunction.allow();
+
       default:
         throw new IllegalArgumentException("Doesn't support the privilege: " + name);
     }
@@ -304,6 +320,14 @@ public static Privilege deny(Privilege.Name name) {
       case SELECT_VIEW:
         return SelectView.deny();
 
+        // Function
+      case REGISTER_FUNCTION:
+        return RegisterFunction.deny();
+      case EXECUTE_FUNCTION:
+        return ExecuteFunction.deny();
+      case MODIFY_FUNCTION:
+        return ModifyFunction.deny();
+
       default:
         throw new IllegalArgumentException("Doesn't support the privilege: " + name);
     }
@@ -1343,4 +1367,97 @@ public boolean canBindTo(MetadataObject.Type type) {
       return VIEW_SUPPORTED_TYPES.contains(type);
     }
   }
+
+  /** The privilege to register a function. */
+  public static class RegisterFunction extends GenericPrivilege<RegisterFunction> {
+    private static final RegisterFunction ALLOW_INSTANCE =
+        new RegisterFunction(Condition.ALLOW, Name.REGISTER_FUNCTION);
+    private static final RegisterFunction DENY_INSTANCE =
+        new RegisterFunction(Condition.DENY, Name.REGISTER_FUNCTION);
+
+    private RegisterFunction(Condition condition, Name name) {
+      super(condition, name);
+    }
+
+    /**
+     * @return The instance with allow condition of the privilege.
+     */
+    public static RegisterFunction allow() {
+      return ALLOW_INSTANCE;
+    }
+
+    /**
+     * @return The instance with deny condition of the privilege.
+     */
+    public static RegisterFunction deny() {
+      return DENY_INSTANCE;
+    }
+
+    @Override
+    public boolean canBindTo(MetadataObject.Type type) {
+      return SCHEMA_SUPPORTED_TYPES.contains(type);
+    }
+  }
+
+  /** The privilege to execute (invoke) a function and view its metadata. */
+  public static class ExecuteFunction extends GenericPrivilege<ExecuteFunction> {
+    private static final ExecuteFunction ALLOW_INSTANCE =
+        new ExecuteFunction(Condition.ALLOW, Name.EXECUTE_FUNCTION);
+    private static final ExecuteFunction DENY_INSTANCE =
+        new ExecuteFunction(Condition.DENY, Name.EXECUTE_FUNCTION);
+
+    private ExecuteFunction(Condition condition, Name name) {
+      super(condition, name);
+    }
+
+    /**
+     * @return The instance with allow condition of the privilege.
+     */
+    public static ExecuteFunction allow() {
+      return ALLOW_INSTANCE;
+    }
+
+    /**
+     * @return The instance with deny condition of the privilege.
+     */
+    public static ExecuteFunction deny() {
+      return DENY_INSTANCE;
+    }
+
+    @Override
+    public boolean canBindTo(MetadataObject.Type type) {
+      return FUNCTION_SUPPORTED_TYPES.contains(type);
+    }
+  }
+
+  /** The privilege to alter a function's metadata. */
+  public static class ModifyFunction extends GenericPrivilege<ModifyFunction> {
+    private static final ModifyFunction ALLOW_INSTANCE =
+        new ModifyFunction(Condition.ALLOW, Name.MODIFY_FUNCTION);
+    private static final ModifyFunction DENY_INSTANCE =
+        new ModifyFunction(Condition.DENY, Name.MODIFY_FUNCTION);
+
+    private ModifyFunction(Condition condition, Name name) {
+      super(condition, name);
+    }
+
+    /**
+     * @return The instance with allow condition of the privilege.
+     */
+    public static ModifyFunction allow() {
+      return ALLOW_INSTANCE;
+    }
+
+    /**
+     * @return The instance with deny condition of the privilege.
+     */
+    public static ModifyFunction deny() {
+      return DENY_INSTANCE;
+    }
+
+    @Override
+    public boolean canBindTo(MetadataObject.Type type) {
+      return FUNCTION_SUPPORTED_TYPES.contains(type);
+    }
+  }
 }

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

@@ -154,6 +154,22 @@ public static SecurableObject ofModel(
     return of(MetadataObject.Type.MODEL, names, privileges);
   }
 
+  /**
+   * Create the function {@link SecurableObject} with the given securable schema object, function
+   * name and privileges.
+   *
+   * @param schema The schema securable object
+   * @param function The function name
+   * @param privileges The privileges of the function
+   * @return The created function {@link SecurableObject}
+   */
+  public static SecurableObject ofFunction(
+      SecurableObject schema, String function, List<Privilege> privileges) {
+    List<String> names = Lists.newArrayList(DOT_SPLITTER.splitToList(schema.fullName()));
+    names.add(function);
+    return of(MetadataObject.Type.FUNCTION, names, privileges);
+  }
+
   /**
    * Create the tag {@link SecurableObject} with the given tag name and privileges.
    *

api/src/test/java/org/apache/gravitino/TestMetadataObjects.java

@@ -179,4 +179,47 @@ public void testViewObject() {
         () ->
             MetadataObjects.of(Lists.newArrayList("catalog", "schema"), MetadataObject.Type.VIEW));
   }
+
+  @Test
+  public void testFunctionObject() {
+    MetadataObject functionObject =
+        MetadataObjects.of("catalog.schema", "func1", MetadataObject.Type.FUNCTION);
+    Assertions.assertEquals("catalog.schema", functionObject.parent());
+    Assertions.assertEquals("func1", functionObject.name());
+    Assertions.assertEquals(MetadataObject.Type.FUNCTION, functionObject.type());
+    Assertions.assertEquals("catalog.schema.func1", functionObject.fullName());
+
+    MetadataObject functionObject2 =
+        MetadataObjects.of(
+            Lists.newArrayList("catalog", "schema", "func2"), MetadataObject.Type.FUNCTION);
+    Assertions.assertEquals("catalog.schema", functionObject2.parent());
+    Assertions.assertEquals("func2", functionObject2.name());
+    Assertions.assertEquals(MetadataObject.Type.FUNCTION, functionObject2.type());
+    Assertions.assertEquals("catalog.schema.func2", functionObject2.fullName());
+
+    MetadataObject functionObject3 =
+        MetadataObjects.parse("catalog.schema.func3", MetadataObject.Type.FUNCTION);
+    Assertions.assertEquals("catalog.schema", functionObject3.parent());
+    Assertions.assertEquals("func3", functionObject3.name());
+    Assertions.assertEquals(MetadataObject.Type.FUNCTION, functionObject3.type());
+    Assertions.assertEquals("catalog.schema.func3", functionObject3.fullName());
+
+    // Test parent
+    MetadataObject parent = MetadataObjects.parent(functionObject);
+    Assertions.assertEquals("catalog.schema", parent.fullName());
+    Assertions.assertEquals("catalog", parent.parent());
+    Assertions.assertEquals("schema", parent.name());
+    Assertions.assertEquals(MetadataObject.Type.SCHEMA, parent.type());
+
+    // Test incomplete name
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () -> MetadataObjects.parse("func1", MetadataObject.Type.FUNCTION));
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () -> MetadataObjects.parse("catalog", MetadataObject.Type.FUNCTION));
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () -> MetadataObjects.parse("catalog.schema", MetadataObject.Type.FUNCTION));
+  }
 }

api/src/test/java/org/apache/gravitino/authorization/TestSecurableObjects.java

@@ -211,6 +211,9 @@ public void testPrivileges() {
     Privilege useJobTemplate = Privileges.UseJobTemplate.allow();
     Privilege createView = Privileges.CreateView.allow();
     Privilege selectView = Privileges.SelectView.allow();
+    Privilege registerFunction = Privileges.RegisterFunction.allow();
+    Privilege executeFunction = Privileges.ExecuteFunction.allow();
+    Privilege modifyFunction = Privileges.ModifyFunction.allow();
 
     // Test create catalog
     Assertions.assertTrue(createCatalog.canBindTo(MetadataObject.Type.METALAKE));
@@ -381,6 +384,7 @@ public void testPrivileges() {
     Assertions.assertTrue(manageGrants.canBindTo(MetadataObject.Type.FILESET));
     Assertions.assertTrue(manageGrants.canBindTo(MetadataObject.Type.VIEW));
     Assertions.assertTrue(manageGrants.canBindTo(MetadataObject.Type.MODEL));
+    Assertions.assertTrue(manageGrants.canBindTo(MetadataObject.Type.FUNCTION));
     Assertions.assertFalse(manageGrants.canBindTo(MetadataObject.Type.ROLE));
     Assertions.assertFalse(manageGrants.canBindTo(MetadataObject.Type.COLUMN));
 
@@ -504,5 +508,38 @@ public void testPrivileges() {
     Assertions.assertFalse(selectView.canBindTo(MetadataObject.Type.ROLE));
     Assertions.assertFalse(selectView.canBindTo(MetadataObject.Type.COLUMN));
     Assertions.assertTrue(selectView.canBindTo(MetadataObject.Type.VIEW));
+
+    // Test register function
+    Assertions.assertTrue(registerFunction.canBindTo(MetadataObject.Type.METALAKE));
+    Assertions.assertTrue(registerFunction.canBindTo(MetadataObject.Type.CATALOG));
+    Assertions.assertTrue(registerFunction.canBindTo(MetadataObject.Type.SCHEMA));
+    Assertions.assertFalse(registerFunction.canBindTo(MetadataObject.Type.TABLE));
+    Assertions.assertFalse(registerFunction.canBindTo(MetadataObject.Type.TOPIC));
+    Assertions.assertFalse(registerFunction.canBindTo(MetadataObject.Type.FILESET));
+    Assertions.assertFalse(registerFunction.canBindTo(MetadataObject.Type.ROLE));
+    Assertions.assertFalse(registerFunction.canBindTo(MetadataObject.Type.COLUMN));
+    Assertions.assertFalse(registerFunction.canBindTo(MetadataObject.Type.FUNCTION));
+
+    // Test execute function
+    Assertions.assertTrue(executeFunction.canBindTo(MetadataObject.Type.METALAKE));
+    Assertions.assertTrue(executeFunction.canBindTo(MetadataObject.Type.CATALOG));
+    Assertions.assertTrue(executeFunction.canBindTo(MetadataObject.Type.SCHEMA));
+    Assertions.assertTrue(executeFunction.canBindTo(MetadataObject.Type.FUNCTION));
+    Assertions.assertFalse(executeFunction.canBindTo(MetadataObject.Type.TABLE));
+    Assertions.assertFalse(executeFunction.canBindTo(MetadataObject.Type.TOPIC));
+    Assertions.assertFalse(executeFunction.canBindTo(MetadataObject.Type.FILESET));
+    Assertions.assertFalse(executeFunction.canBindTo(MetadataObject.Type.ROLE));
+    Assertions.assertFalse(executeFunction.canBindTo(MetadataObject.Type.COLUMN));
+
+    // Test modify function
+    Assertions.assertTrue(modifyFunction.canBindTo(MetadataObject.Type.METALAKE));
+    Assertions.assertTrue(modifyFunction.canBindTo(MetadataObject.Type.CATALOG));
+    Assertions.assertTrue(modifyFunction.canBindTo(MetadataObject.Type.SCHEMA));
+    Assertions.assertTrue(modifyFunction.canBindTo(MetadataObject.Type.FUNCTION));
+    Assertions.assertFalse(modifyFunction.canBindTo(MetadataObject.Type.TABLE));
+    Assertions.assertFalse(modifyFunction.canBindTo(MetadataObject.Type.TOPIC));
+    Assertions.assertFalse(modifyFunction.canBindTo(MetadataObject.Type.FILESET));
+    Assertions.assertFalse(modifyFunction.canBindTo(MetadataObject.Type.ROLE));
+    Assertions.assertFalse(modifyFunction.canBindTo(MetadataObject.Type.COLUMN));
   }
 }