Merge branch 'main' of github.com:apache/gravitino into minor

authorizations/authorization-ranger/src/main/java/org/apache/gravitino/authorization/ranger/RangerClientExtension.java

@@ -100,12 +100,14 @@ public RangerClientExtension(String hostName, String authType, String username,
     }
   }
 
+  @Override
   public RangerPolicy createPolicy(RangerPolicy policy) throws RangerServiceException {
     Preconditions.checkArgument(
         policy.getResources().size() > 0, "Ranger policy resources can not be empty!");
     return super.createPolicy(policy);
   }
 
+  @Override
   public RangerPolicy updatePolicy(long policyId, RangerPolicy policy)
       throws RangerServiceException {
     Preconditions.checkArgument(

authorizations/authorization-ranger/src/main/java/org/apache/gravitino/authorization/ranger/reference/VXGroup.java

@@ -60,6 +60,7 @@ public VXGroup() {
    *
    * @return formatedStr
    */
+  @Override
   public String toString() {
     String str = "VXGroup={";
     str += super.toString();

authorizations/authorization-ranger/src/main/java/org/apache/gravitino/authorization/ranger/reference/VXUser.java

@@ -75,6 +75,7 @@ public String getName() {
    *
    * @return formatedStr
    */
+  @Override
   public String toString() {
     String str = "VXUser={";
     str += super.toString();

catalogs/catalog-jdbc-oceanbase/src/main/java/org/apache/gravitino/catalog/oceanbase/operation/OceanBaseTableOperations.java

@@ -185,6 +185,7 @@ protected Map<String, String> getTableProperties(Connection connection, String t
     }
   }
 
+  @Override
   protected void correctJdbcTableFields(
       Connection connection, String databaseName, String tableName, JdbcTable.Builder tableBuilder)
       throws SQLException {

core/src/main/java/org/apache/gravitino/connector/BaseCatalog.java

@@ -208,8 +208,15 @@ public void initAuthorizationPluginInstance(IsolatedClassLoader classLoader) {
           try {
             BaseAuthorization<?> authorization =
                 BaseAuthorization.createAuthorization(classLoader, authorizationProvider);
+
+            // Load the authorization plugin with the class loader of the catalog.
+            // Because the JDBC authorization plugin may load JDBC driver using the class loader.
             authorizationPlugin =
-                authorization.newPlugin(entity.namespace().level(0), provider(), this.conf);
+                classLoader.withClassLoader(
+                    cl ->
+                        authorization.newPlugin(
+                            entity.namespace().level(0), provider(), this.conf));
+
           } catch (Exception e) {
             LOG.error("Failed to load authorization with class loader", e);
             throw new RuntimeException(e);

core/src/main/java/org/apache/gravitino/hook/MetalakeHookDispatcher.java

@@ -116,6 +116,7 @@ public void disableMetalake(NameIdentifier ident) throws NoSuchMetalakeException
     dispatcher.disableMetalake(ident);
   }
 
+  @Override
   public boolean dropMetalake(NameIdentifier ident) {
     // For metalake, we don't clear all the privileges of catalog authorization plugin.
     // we just remove metalake.

core/src/main/java/org/apache/gravitino/listener/api/event/CreateTablePreEvent.java

@@ -43,6 +43,7 @@ public TableInfo createTableRequest() {
     return createTableRequest;
   }
 
+  @Override
   public OperationType operationType() {
     return OperationType.CREATE_TABLE;
   }

core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java

@@ -76,6 +76,7 @@ public String insertCatalogMetaOnDuplicateKeyUpdate(CatalogPO catalogPO) {
         + " deleted_at = #{catalogMeta.deletedAt}";
   }
 
+  @Override
   public String updateCatalogMeta(
       @Param("newCatalogMeta") CatalogPO newCatalogPO,
       @Param("oldCatalogMeta") CatalogPO oldCatalogPO) {

core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/MetalakeMetaPostgreSQLProvider.java

@@ -62,6 +62,7 @@ public String insertMetalakeMetaOnDuplicateKeyUpdate(MetalakePO metalakePO) {
         + " deleted_at = #{metalakeMeta.deletedAt}";
   }
 
+  @Override
   public String updateMetalakeMeta(
       @Param("newMetalakeMeta") MetalakePO newMetalakePO,
       @Param("oldMetalakeMeta") MetalakePO oldMetalakePO) {

core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SecurableObjectPostgreSQLProvider.java

@@ -32,6 +32,7 @@
 import org.apache.ibatis.annotations.Param;
 
 public class SecurableObjectPostgreSQLProvider extends SecurableObjectBaseSQLProvider {
+  @Override
   public String batchSoftDeleteSecurableObjects(
       @Param("securableObjects") List<SecurableObjectPO> securableObjectPOs) {
     return "<script>"

core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TagMetaPostgreSQLProvider.java

@@ -77,6 +77,7 @@ public String insertTagMetaOnDuplicateKeyUpdate(TagPO tagPO) {
         + " deleted_at = #{tagMeta.deletedAt}";
   }
 
+  @Override
   public String updateTagMeta(
       @Param("newTagMeta") TagPO newTagPO, @Param("oldTagMeta") TagPO oldTagPO) {
     return "UPDATE "

docs/flink-connector/flink-connector.md

@@ -34,16 +34,16 @@ This capability allows users to perform federation queries, accessing data from
 
 Set the flink configuration in flink-conf.yaml.
 ```yaml
-table.catalog-store.kind=gravitino
-table.catalog-store.gravitino.gravitino.metalake=test
-table.catalog-store.gravitino.gravitino.uri=http://localhost:8080
+table.catalog-store.kind: gravitino
+table.catalog-store.gravitino.gravitino.metalake: test
+table.catalog-store.gravitino.gravitino.uri: http://localhost:8090
 ```
 Or you can set the flink configuration in the `TableEnvironment`.
 ```java
 final Configuration configuration = new Configuration();
 configuration.setString("table.catalog-store.kind", "gravitino");
 configuration.setString("table.catalog-store.gravitino.gravitino.metalake", "test");
-configuration.setString("table.catalog-store.gravitino.gravitino.uri", "http://localhost:8080");
+configuration.setString("table.catalog-store.gravitino.gravitino.uri", "http://localhost:8090");
 EnvironmentSettings.Builder builder = EnvironmentSettings.newInstance().withConfiguration(configuration);
 TableEnvironment tableEnv = TableEnvironment.create(builder.inBatchMode().build());
 ```

flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java

@@ -656,11 +656,11 @@ static SchemaChange[] getSchemaChange(CatalogDatabase current, CatalogDatabase u
     return schemaChanges.toArray(new SchemaChange[0]);
   }
 
-  private Catalog catalog() {
+  protected Catalog catalog() {
     return GravitinoCatalogManager.get().getGravitinoCatalogInfo(getName());
   }
 
-  private String catalogName() {
+  protected String catalogName() {
     return getName();
   }
 }

flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalog.java

@@ -19,10 +19,17 @@
 
 package org.apache.gravitino.flink.connector.paimon;
 
+import java.util.Optional;
 import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.factories.Factory;
+import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.flink.connector.PartitionConverter;
 import org.apache.gravitino.flink.connector.PropertiesConverter;
 import org.apache.gravitino.flink.connector.catalog.BaseCatalog;
+import org.apache.paimon.flink.FlinkTableFactory;
 
 /**
  * The GravitinoPaimonCatalog class is an implementation of the BaseCatalog class that is used to
@@ -45,4 +52,21 @@ protected GravitinoPaimonCatalog(
   protected AbstractCatalog realCatalog() {
     return paimonCatalog;
   }
+
+  @Override
+  public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists)
+      throws TableNotExistException, CatalogException {
+    boolean dropped =
+        catalog()
+            .asTableCatalog()
+            .purgeTable(NameIdentifier.of(tablePath.getDatabaseName(), tablePath.getObjectName()));
+    if (!dropped && !ignoreIfNotExists) {
+      throw new TableNotExistException(catalogName(), tablePath);
+    }
+  }
+
+  @Override
+  public Optional<Factory> getFactory() {
+    return Optional.of(new FlinkTableFactory());
+  }
 }

flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/FlinkEnvIT.java

@@ -19,7 +19,6 @@
 package org.apache.gravitino.flink.connector.integration.test;
 
 import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableMap;
 import com.google.errorprone.annotations.FormatMethod;
 import com.google.errorprone.annotations.FormatString;
 import java.io.IOException;
@@ -159,17 +158,14 @@ protected TableResult sql(@FormatString String sql, Object... args) {
     return tableEnv.executeSql(String.format(sql, args));
   }
 
-  protected static void doWithSchema(
+  protected void doWithSchema(
       Catalog catalog, String schemaName, Consumer<Catalog> action, boolean dropSchema) {
     Preconditions.checkNotNull(catalog);
     Preconditions.checkNotNull(schemaName);
     try {
       tableEnv.useCatalog(catalog.name());
       if (!catalog.asSchemas().schemaExists(schemaName)) {
-        catalog
-            .asSchemas()
-            .createSchema(
-                schemaName, null, ImmutableMap.of("location", warehouse + "/" + schemaName));
+        catalog.asSchemas().createSchema(schemaName, null, null);
       }
       tableEnv.useDatabase(schemaName);
       action.accept(catalog);

flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/hive/FlinkHiveCatalogIT.java

@@ -29,6 +29,7 @@
 import java.util.Arrays;
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.api.DataTypes;
@@ -586,4 +587,28 @@ public void testGetHiveTable() {
   protected org.apache.gravitino.Catalog currentCatalog() {
     return hiveCatalog;
   }
+
+  protected void doWithSchema(
+      org.apache.gravitino.Catalog catalog,
+      String schemaName,
+      Consumer<org.apache.gravitino.Catalog> action,
+      boolean dropSchema) {
+    Preconditions.checkNotNull(catalog);
+    Preconditions.checkNotNull(schemaName);
+    try {
+      tableEnv.useCatalog(catalog.name());
+      if (!catalog.asSchemas().schemaExists(schemaName)) {
+        catalog
+            .asSchemas()
+            .createSchema(
+                schemaName, null, ImmutableMap.of("location", warehouse + "/" + schemaName));
+      }
+      tableEnv.useDatabase(schemaName);
+      action.accept(catalog);
+    } finally {
+      if (dropSchema) {
+        catalog.asSchemas().dropSchema(schemaName, true);
+      }
+    }
+  }
 }

flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/integration/test/paimon/FlinkPaimonCatalogIT.java

@@ -42,16 +42,6 @@ public class FlinkPaimonCatalogIT extends FlinkCommonIT {
 
   private static org.apache.gravitino.Catalog catalog;
 
-  @Override
-  protected boolean supportColumnOperation() {
-    return false;
-  }
-
-  @Override
-  protected boolean supportTableOperation() {
-    return false;
-  }
-
   @Override
   protected boolean supportSchemaOperationWithCommentAndOptions() {
     return false;