[WIP][spark] Add sparksql lake table catalog DDL suite for paimon

fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkCatalog.scala

@@ -74,13 +74,34 @@ class SparkCatalog extends TableCatalog with SupportsFlussNamespaces with WithFl
         } else if (e.getCause.isInstanceOf[TableAlreadyExistException]) {
           throw new TableAlreadyExistsException(ident)
         } else {
-          throw new RuntimeException(e)
+          throw e
         }
     }
   }
 
   override def alterTable(ident: Identifier, changes: TableChange*): Table = {
-    throw new UnsupportedOperationException("Altering table is not supported")
+    if (
+      !changes.forall(
+        e => e.isInstanceOf[TableChange.SetProperty] || e.isInstanceOf[TableChange.RemoveProperty])
+    ) {
+      throw new IllegalArgumentException(
+        "Altering table only supports set or remove properties for now")
+    }
+    try {
+      admin
+        .alterTable(toTablePath(ident), SparkConversions.toFlussTableChanges(changes).asJava, false)
+        .get()
+      loadTable(ident)
+    } catch {
+      case e: ExecutionException =>
+        if (e.getCause.isInstanceOf[TableNotExistException]) {
+          throw new NoSuchTableException(ident)
+        } else {
+          throw e
+        }
+      case e: UnsupportedOperationException =>
+        throw new IllegalArgumentException(e)
+    }
   }
 
   override def dropTable(ident: Identifier): Boolean = {

fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkConversions.scala

@@ -25,6 +25,7 @@ import org.apache.fluss.types.RowType
 
 import org.apache.spark.sql.FlussIdentityTransform
 import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
+import org.apache.spark.sql.connector.catalog.TableChange
 import org.apache.spark.sql.connector.expressions.Transform
 import org.apache.spark.sql.types.StructType
 
@@ -54,7 +55,7 @@ object SparkConversions {
     tableDescriptorBuilder.partitionedBy(partitionKey: _*)
 
     val primaryKeys = if (caseInsensitiveProps.contains(PRIMARY_KEY.key)) {
-      val pks = caseInsensitiveProps.get(PRIMARY_KEY.key).get.split(",")
+      val pks = caseInsensitiveProps.get(PRIMARY_KEY.key).get.split(",").map(_.trim)
       schemaBuilder.primaryKey(pks: _*)
       pks
     } else {
@@ -64,7 +65,7 @@ object SparkConversions {
     if (caseInsensitiveProps.contains(BUCKET_NUMBER.key)) {
       val bucketNum = caseInsensitiveProps.get(BUCKET_NUMBER.key).get.toInt
       val bucketKeys = if (caseInsensitiveProps.contains(BUCKET_KEY.key)) {
-        caseInsensitiveProps.get(BUCKET_KEY.key).get.split(",")
+        caseInsensitiveProps.get(BUCKET_KEY.key).get.split(",").map(_.trim)
       } else {
         primaryKeys.filterNot(partitionKey.contains)
       }
@@ -76,7 +77,7 @@ object SparkConversions {
     }
 
     val (tableProps, customProps) =
-      caseInsensitiveProps.filterNot(SPARK_TABLE_OPTIONS.contains).partition {
+      caseInsensitiveProps.filterNot(e => SPARK_TABLE_OPTIONS.contains(e._1)).partition {
         case (key, _) => key.startsWith(FlussConfigUtils.TABLE_PREFIX)
       }
 
@@ -97,4 +98,15 @@ object SparkConversions {
     }
     partitionKeys.toArray
   }
+
+  def toFlussTableChanges(changes: Seq[TableChange]): Seq[org.apache.fluss.metadata.TableChange] = {
+    changes.map {
+      case p: TableChange.SetProperty =>
+        org.apache.fluss.metadata.TableChange.set(p.property(), p.value())
+      case p: TableChange.RemoveProperty =>
+        org.apache.fluss.metadata.TableChange.reset(p.property())
+      // TODO Add full support for table changes
+      case _ => throw new UnsupportedOperationException("Unsupported table change")
+    }
+  }
 }

fluss-spark/fluss-spark-ut/pom.xml

@@ -81,6 +81,20 @@
             <artifactId>fluss-test-utils</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.fluss</groupId>
+            <artifactId>fluss-lake-paimon</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.paimon</groupId>
+            <artifactId>paimon-bundle</artifactId>
+            <version>${paimon.version}</version>
+            <scope>test</scope>
+        </dependency>
+
         <!-- for curator TestingServer  -->
         <dependency>
             <groupId>org.apache.curator</groupId>

fluss-spark/fluss-spark-ut/src/test/resources/META-INF/services/org.apache.fluss.lake.lakestorage.LakeStoragePlugin

@@ -0,0 +1,19 @@
+#
+# 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.
+#
+
+org.apache.fluss.lake.paimon.PaimonLakeStoragePlugin

fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/FlussSparkTestBase.scala

@@ -26,43 +26,58 @@ import org.apache.fluss.metadata.{TableDescriptor, TablePath}
 import org.apache.fluss.row.InternalRow
 import org.apache.fluss.server.testutils.FlussClusterExtension
 
-import org.apache.spark.SparkConf
 import org.apache.spark.sql.QueryTest
 import org.apache.spark.sql.test.SharedSparkSession
-import org.junit.jupiter.api.extension.RegisterExtension
 
 import java.time.Duration
 
 import scala.collection.JavaConverters._
 
 class FlussSparkTestBase extends QueryTest with SharedSparkSession {
 
-  import FlussSparkTestBase._
-
   protected val DEFAULT_CATALOG = "fluss_catalog"
   protected val DEFAULT_DATABASE = "fluss"
 
   protected var conn: Connection = _
   protected var admin: Admin = _
 
-  override protected def sparkConf: SparkConf = {
-    super.sparkConf
-      .set(s"spark.sql.catalog.$DEFAULT_CATALOG", classOf[SparkCatalog].getName)
-      .set(s"spark.sql.catalog.$DEFAULT_CATALOG.bootstrap.servers", bootstrapServers)
-      .set("spark.sql.defaultCatalog", DEFAULT_CATALOG)
-      // Enable read optimized by default temporarily.
-      // TODO: remove this when https://github.com/apache/fluss/issues/2427 is done.
-      .set("spark.sql.fluss.readOptimized", "true")
-  }
+  val flussServer: FlussClusterExtension =
+    FlussClusterExtension.builder
+      .setClusterConf(flussConf)
+      .setNumOfTabletServers(3)
+      .build
 
   override protected def beforeAll(): Unit = {
     super.beforeAll()
-    conn = ConnectionFactory.createConnection(clientConf)
+    flussServer.start()
+    conn = ConnectionFactory.createConnection(flussServer.getClientConfig)
     admin = conn.getAdmin
 
+    spark.conf.set(s"spark.sql.catalog.$DEFAULT_CATALOG", classOf[SparkCatalog].getName)
+    spark.conf.set(
+      s"spark.sql.catalog.$DEFAULT_CATALOG.bootstrap.servers",
+      flussServer.getBootstrapServers)
+    spark.conf.set("spark.sql.defaultCatalog", DEFAULT_CATALOG)
+    // Enable read optimized by default temporarily.
+    // TODO: remove this when https://github.com/apache/fluss/issues/2427 is done.
+    spark.conf.set("spark.sql.fluss.readOptimized", "true")
+
     sql(s"USE $DEFAULT_DATABASE")
   }
 
+  override protected def afterAll(): Unit = {
+    super.afterAll()
+    if (admin != null) {
+      admin.close()
+      admin = null
+    }
+    if (conn != null) {
+      conn.close()
+      conn = null
+    }
+    flussServer.close()
+  }
+
   def createTablePath(tableName: String): TablePath = {
     TablePath.of(DEFAULT_DATABASE, tableName)
   }
@@ -98,21 +113,9 @@ class FlussSparkTestBase extends QueryTest with SharedSparkSession {
       .map(record => (record.getChangeType.shortString(), record.getRow))
       .toArray
   }
-}
-
-@RegisterExtension
-object FlussSparkTestBase {
-  val FLUSS_CLUSTER_EXTENSION: FlussClusterExtension =
-    FlussClusterExtension.builder
-      .setClusterConf(
-        new Configuration()
-          .set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1))
-      )
-      .setNumOfTabletServers(3)
-      .build
 
-  FLUSS_CLUSTER_EXTENSION.start()
-
-  val clientConf: Configuration = FLUSS_CLUSTER_EXTENSION.getClientConfig
-  val bootstrapServers: String = FLUSS_CLUSTER_EXTENSION.getBootstrapServers
+  protected def flussConf: Configuration = {
+    val conf = new Configuration
+    conf.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1))
+  }
 }

fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkCatalogTest.scala

@@ -17,6 +17,7 @@
 
 package org.apache.fluss.spark
 
+import org.apache.fluss.config.ConfigOptions
 import org.apache.fluss.metadata._
 import org.apache.fluss.types.{DataTypes, RowType}
 
@@ -192,6 +193,33 @@ class SparkCatalogTest extends FlussSparkTestBase {
     checkAnswer(sql("SHOW DATABASES"), Row(DEFAULT_DATABASE) :: Nil)
   }
 
+  test("Catalog: set/remove table properties") {
+    withTable("t") {
+      sql(
+        s"CREATE TABLE $DEFAULT_DATABASE.t (id int, name string) TBLPROPERTIES('key1' = 'value1', '${SparkConnectorOptions.BUCKET_NUMBER.key()}' = 3)")
+      var flussTable = admin.getTableInfo(createTablePath("t")).get()
+      assertResult(
+        Map(ConfigOptions.TABLE_REPLICATION_FACTOR.key() -> "1"),
+        "check table properties")(flussTable.getProperties.toMap.asScala)
+      assert(
+        flussTable.getCustomProperties.toMap.asScala.getOrElse("key1", "non-exists") == "value1")
+
+      sql(s"ALTER TABLE t SET TBLPROPERTIES('key1' = 'value2')")
+      flussTable = admin.getTableInfo(createTablePath("t")).get()
+      assertResult(
+        Map(ConfigOptions.TABLE_REPLICATION_FACTOR.key() -> "1"),
+        "check table properties")(flussTable.getProperties.toMap.asScala)
+      assert(
+        flussTable.getCustomProperties.toMap.asScala.getOrElse("key1", "non-exists") == "value2")
+
+      sql(s"ALTER TABLE t UNSET TBLPROPERTIES('key1')")
+      flussTable = admin.getTableInfo(createTablePath("t")).get()
+      assert(!flussTable.getCustomProperties.toMap.asScala.contains("key1"))
+
+      sql(s"ALTER TABLE t UNSET TBLPROPERTIES('key1')")
+    }
+  }
+
   test("Partition: show partitions") {
     withTable("t") {
       sql(s"CREATE TABLE t (id int, name string, pt1 string, pt2 int) PARTITIONED BY (pt1, pt2)")