[#4733] feat(paimon-spark-connector): support hive backend catalog in paimon spark connector

spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/paimon/SparkPaimonCatalogFilesystemBackendIT.java

@@ -48,17 +48,22 @@ protected void testCreateAndLoadSchema() {
     Map<String, String> databaseMeta = getDatabaseMetadata(testDatabaseName);
     // The database of the Paimon filesystem backend do not store any properties.
     Assertions.assertFalse(databaseMeta.containsKey("ID"));
+    Assertions.assertFalse(databaseMeta.containsKey("comment"));
   }
 
   @Test
   @Override
   protected void testAlterSchema() {
     String testDatabaseName = "t_alter";
     dropDatabaseIfExists(testDatabaseName);
-    sql("CREATE DATABASE " + testDatabaseName + " WITH DBPROPERTIES (ID=001);");
+    sql(
+        "CREATE DATABASE "
+            + testDatabaseName
+            + " COMMENT 'db comment' WITH DBPROPERTIES (ID=001);");
     Map<String, String> databaseMeta = getDatabaseMetadata(testDatabaseName);
     // The database of the Paimon filesystem backend do not store any properties.
-    Assertions.assertFalse(databaseMeta.containsKey("ID"));
+    Assertions.assertFalse(databaseMeta.get("Properties").contains("(ID,001)"));
+    Assertions.assertFalse(databaseMeta.containsKey("Comment"));
 
     // The Paimon filesystem backend do not support alter database operation.
     Assertions.assertThrows(

spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/paimon/SparkPaimonCatalogHiveBackendIT.java

@@ -0,0 +1,64 @@
+/*
+ * 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.spark.connector.integration.test.paimon;
+
+import com.google.common.collect.Maps;
+import java.util.Map;
+import org.apache.gravitino.spark.connector.paimon.PaimonPropertiesConstants;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+/** This class use Apache Paimon HiveCatalog for backend catalog. */
+@Tag("gravitino-docker-test")
+public abstract class SparkPaimonCatalogHiveBackendIT extends SparkPaimonCatalogIT {
+
+  @Override
+  protected Map<String, String> getCatalogConfigs() {
+    Map<String, String> catalogProperties = Maps.newHashMap();
+    catalogProperties.put(
+        PaimonPropertiesConstants.GRAVITINO_PAIMON_CATALOG_BACKEND,
+        PaimonPropertiesConstants.PAIMON_CATALOG_BACKEND_HIVE);
+    catalogProperties.put(PaimonPropertiesConstants.GRAVITINO_PAIMON_CATALOG_URI, hiveMetastoreUri);
+    catalogProperties.put(PaimonPropertiesConstants.GRAVITINO_PAIMON_CATALOG_WAREHOUSE, warehouse);
+    return catalogProperties;
+  }
+
+  @Test
+  @Override
+  protected void testAlterSchema() {
+    String testDatabaseName = "t_alter";
+    dropDatabaseIfExists(testDatabaseName);
+    sql(
+        "CREATE DATABASE "
+            + testDatabaseName
+            + " COMMENT 'db comment' WITH DBPROPERTIES (ID=001);");
+    Map<String, String> databaseMeta = getDatabaseMetadata(testDatabaseName);
+    Assertions.assertTrue(databaseMeta.get("Properties").contains("(ID,001)"));
+    Assertions.assertEquals("db comment", databaseMeta.get("Comment"));
+
+    // The Paimon filesystem backend do not support alter database operation.
+    Assertions.assertThrows(
+        UnsupportedOperationException.class,
+        () ->
+            sql(
+                String.format(
+                    "ALTER DATABASE %s SET DBPROPERTIES ('ID'='002')", testDatabaseName)));
+  }
+}

spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/paimon/SparkPaimonCatalogIT.java

@@ -104,6 +104,38 @@ void testPaimonPartitions() {
     checkDirExists(partitionPath);
   }
 
+  @Test
+  void testPaimonPartitionManagement() {
+    // replace, add and load partition operations are unsupported in Paimon now.
+    // Therefore, Paimon spark runtime only supports list and drop partition operations.
+    testPaimonListAndDropPartition();
+  }
+
+  private void testPaimonListAndDropPartition() {
+    String tableName = "test_paimon_drop_partition";
+    dropTableIfExists(tableName);
+    String createTableSQL = getCreatePaimonSimpleTableString(tableName);
+    createTableSQL = createTableSQL + " PARTITIONED BY (name);";
+    sql(createTableSQL);
+
+    String insertData =
+        String.format(
+            "INSERT into %s values(1,'a','beijing'), (2,'b','beijing'), (3,'c','beijing');",
+            tableName);
+    sql(insertData);
+    List<String> queryResult = getTableData(tableName);
+    Assertions.assertEquals(3, queryResult.size());
+
+    List<String> partitions = getQueryData(String.format("show partitions %s", tableName));
+    Assertions.assertEquals(3, partitions.size());
+    Assertions.assertEquals("name=a;name=b;name=c", String.join(";", partitions));
+
+    sql(String.format("ALTER TABLE %s DROP PARTITION (`name`='a')", tableName));
+    partitions = getQueryData(String.format("show partitions %s", tableName));
+    Assertions.assertEquals(2, partitions.size());
+    Assertions.assertEquals("name=b;name=c", String.join(";", partitions));
+  }
+
   private String getCreatePaimonSimpleTableString(String tableName) {
     return String.format(
         "CREATE TABLE %s (id INT COMMENT 'id comment', name STRING COMMENT '', address STRING COMMENT '') USING paimon",

spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/paimon/SparkPaimonCatalogHiveBackendIT33.java

@@ -0,0 +1,32 @@
+/*
+ * 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.spark.connector.integration.test.paimon;
+
+import org.apache.gravitino.spark.connector.paimon.GravitinoPaimonCatalogSpark33;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class SparkPaimonCatalogHiveBackendIT33 extends SparkPaimonCatalogHiveBackendIT {
+  @Test
+  void testCatalogClassName() {
+    String catalogClass =
+        getSparkSession().sparkContext().conf().get("spark.sql.catalog." + getCatalogName());
+    Assertions.assertEquals(GravitinoPaimonCatalogSpark33.class.getName(), catalogClass);
+  }
+}

spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/paimon/SparkPaimonCatalogHiveBackendIT34.java

@@ -0,0 +1,33 @@
+/*
+ * 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.spark.connector.integration.test.paimon;
+
+import org.apache.gravitino.spark.connector.paimon.GravitinoPaimonCatalogSpark34;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class SparkPaimonCatalogHiveBackendIT34 extends SparkPaimonCatalogHiveBackendIT {
+
+  @Test
+  void testCatalogClassName() {
+    String catalogClass =
+        getSparkSession().sparkContext().conf().get("spark.sql.catalog." + getCatalogName());
+    Assertions.assertEquals(GravitinoPaimonCatalogSpark34.class.getName(), catalogClass);
+  }
+}

spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/paimon/SparkPaimonCatalogHiveBackendIT35.java

@@ -0,0 +1,33 @@
+/*
+ * 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.spark.connector.integration.test.paimon;
+
+import org.apache.gravitino.spark.connector.paimon.GravitinoPaimonCatalogSpark35;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class SparkPaimonCatalogHiveBackendIT35 extends SparkPaimonCatalogHiveBackendIT {
+
+  @Test
+  void testCatalogClassName() {
+    String catalogClass =
+        getSparkSession().sparkContext().conf().get("spark.sql.catalog." + getCatalogName());
+    Assertions.assertEquals(GravitinoPaimonCatalogSpark35.class.getName(), catalogClass);
+  }
+}