[test][spark] Add alter with incompatible col type test case (#6689)

Author: Zouxxyy

paimon-hive/paimon-hive-common/src/test/java/org/apache/paimon/hive/TestHiveMetastore.java

@@ -115,20 +115,24 @@ private static void setup() {
     private TServer server;
     private HiveMetaStore.HMSHandler baseHandler;
 
+    public void start(Configuration configuration, int port) {
+        start(configuration, DEFAULT_POOL_SIZE, port);
+    }
+
     /**
      * Starts a TestHiveMetastore with the default connection pool size (5) and the default
      * HiveConf.
      */
     public void start(int port) {
-        start(new HiveConf(new Configuration(), TestHiveMetastore.class), DEFAULT_POOL_SIZE, port);
+        start(new Configuration(), DEFAULT_POOL_SIZE, port);
     }
 
     /**
      * Starts a TestHiveMetastore with the default connection pool size (5) and the default
      * HiveConf.
      */
     public void start() {
-        start(new HiveConf(new Configuration(), TestHiveMetastore.class), DEFAULT_POOL_SIZE, 9083);
+        start(new Configuration(), DEFAULT_POOL_SIZE, 9083);
     }
 
     /**
@@ -137,13 +141,11 @@ public void start() {
      * @param conf The hive configuration to use
      * @param poolSize The number of threads in the executor pool
      */
-    public void start(HiveConf conf, int poolSize, int portNum) {
+    public void start(Configuration conf, int poolSize, int portNum) {
         try {
             TServerSocket socket = new TServerSocket(portNum);
             int port = socket.getServerSocket().getLocalPort();
-            initConf(conf, port);
-
-            this.hiveConf = conf;
+            this.hiveConf = initConf(conf, port);
             this.server = newThriftServer(socket, poolSize, hiveConf);
             this.executorService = Executors.newSingleThreadExecutor();
             this.executorService.submit(() -> server.serve());
@@ -211,16 +213,17 @@ private TServer newThriftServer(TServerSocket socket, int poolSize, HiveConf con
         return new TThreadPoolServer(args);
     }
 
-    private void initConf(HiveConf conf, int port) {
-        conf.set(HiveConf.ConfVars.METASTOREURIS.varname, "thrift://localhost:" + port);
-        conf.set(HiveConf.ConfVars.METASTOREWAREHOUSE.varname, warehouseDir());
-        conf.set(HiveConf.ConfVars.METASTORE_TRY_DIRECT_SQL.varname, "false");
-        conf.set(
+    private HiveConf initConf(Configuration conf, int port) {
+        conf.setIfUnset(HiveConf.ConfVars.METASTOREURIS.varname, "thrift://localhost:" + port);
+        conf.setIfUnset(HiveConf.ConfVars.METASTOREWAREHOUSE.varname, warehouseDir());
+        conf.setIfUnset(HiveConf.ConfVars.METASTORE_TRY_DIRECT_SQL.varname, "false");
+        conf.setIfUnset(
                 HiveConf.ConfVars.METASTORE_DISALLOW_INCOMPATIBLE_COL_TYPE_CHANGES.varname,
                 "false");
-        conf.set(
+        conf.setIfUnset(
                 HiveConf.ConfVars.HIVE_IN_TEST.varname,
                 HiveConf.ConfVars.HIVE_IN_TEST.getDefaultValue());
+        return new HiveConf(conf, TestHiveMetastore.class);
     }
 
     private static void setupMetastoreDB(String dbURL) throws SQLException, IOException {

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonHiveTestBase.scala

@@ -20,6 +20,7 @@ package org.apache.paimon.spark
 
 import org.apache.paimon.catalog.{Catalog, DelegateCatalog}
 import org.apache.paimon.hive.{HiveCatalog, TestHiveMetastore}
+import org.apache.paimon.spark.PaimonHiveTestBase._
 import org.apache.paimon.table.FileStoreTable
 
 import org.apache.hadoop.conf.Configuration
@@ -41,7 +42,7 @@ class PaimonHiveTestBase extends PaimonSparkTestBase {
 
   protected val hiveDbName: String = "test_hive"
 
-  val hiveUri: String = PaimonHiveTestBase.hiveUri
+  protected def configuration: Configuration = new Configuration
 
   /**
    * Add spark_catalog ([[SparkGenericCatalog]] in hive) and paimon_hive ([[SparkCatalog]] in hive)
@@ -61,7 +62,7 @@ class PaimonHiveTestBase extends PaimonSparkTestBase {
   }
 
   override protected def beforeAll(): Unit = {
-    testHiveMetastore.start(PaimonHiveTestBase.hivePort)
+    testHiveMetastore.start(configuration, hivePort)
     super.beforeAll()
     spark.sql(s"USE $sparkCatalogName")
     spark.sql(s"CREATE DATABASE IF NOT EXISTS $hiveDbName")

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLWithIncompatibleColType.scala

@@ -0,0 +1,60 @@
+/*
+ * 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.paimon.spark.sql
+
+import org.apache.paimon.spark.PaimonHiveTestBase
+
+import org.apache.hadoop.conf.Configuration
+
+class DDLWithDisallowIncompatibleColType extends DDLWithIncompatibleColType {
+  val disallowIncompatible: Boolean = true
+}
+
+class DDLWithAllowIncompatibleColType extends DDLWithIncompatibleColType {
+  val disallowIncompatible: Boolean = false
+}
+
+abstract class DDLWithIncompatibleColType extends PaimonHiveTestBase {
+
+  def disallowIncompatible: Boolean
+
+  override def configuration: Configuration = {
+    val conf_ = super.configuration
+    conf_.set(
+      "hive.metastore.disallow.incompatible.col.type.changes",
+      disallowIncompatible.toString)
+    conf_
+  }
+
+  test("Paimon DDL with hive catalog: alter with incompatible col type") {
+    withTable("t") {
+      spark.sql("CREATE TABLE t (a INT, b INT, c STRUCT<f1: INT>) USING paimon")
+      if (disallowIncompatible) {
+        val e = intercept[Exception] {
+          spark.sql("ALTER TABLE t DROP COLUMN b")
+        }
+        assert(
+          e.getMessage.contains(
+            "The following columns have types incompatible with the existing columns"))
+      } else {
+        spark.sql("ALTER TABLE t DROP COLUMN b")
+      }
+    }
+  }
+}

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala

@@ -22,6 +22,7 @@ import org.apache.paimon.catalog.{DelegateCatalog, Identifier}
 import org.apache.paimon.fs.Path
 import org.apache.paimon.hive.HiveCatalog
 import org.apache.paimon.spark.PaimonHiveTestBase
+import org.apache.paimon.spark.PaimonHiveTestBase.hiveUri
 import org.apache.paimon.table.FormatTable
 import org.apache.paimon.utils.CompressUtils