[spark] Support ADD COLUMN in Spark catalog

Author: Xtpacz

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

@@ -105,8 +105,28 @@ object SparkConversions {
         org.apache.fluss.metadata.TableChange.set(p.property(), p.value())
       case p: TableChange.RemoveProperty =>
         org.apache.fluss.metadata.TableChange.reset(p.property())
+      case p: TableChange.AddColumn =>
+        if (p.fieldNames().length != 1) {
+          throw new UnsupportedOperationException(
+            s"Adding nested columns is not supported: ${p.fieldNames().mkString(".")}")
+        }
+        org.apache.fluss.metadata.TableChange.addColumn(
+          p.fieldNames().head,
+          SparkToFlussTypeVisitor.visit(p.dataType()).copy(p.isNullable()),
+          p.comment(),
+          toFlussColumnPosition(p.position()))
       // TODO Add full support for table changes
       case _ => throw new UnsupportedOperationException("Unsupported table change")
     }
   }
+
+  private def toFlussColumnPosition(position: TableChange.ColumnPosition)
+      : org.apache.fluss.metadata.TableChange.ColumnPosition = {
+    position match {
+      case _: TableChange.First => org.apache.fluss.metadata.TableChange.ColumnPosition.first()
+      case p: TableChange.After =>
+        org.apache.fluss.metadata.TableChange.ColumnPosition.after(p.column())
+      case _ => org.apache.fluss.metadata.TableChange.ColumnPosition.last()
+    }
+  }
 }

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

@@ -36,6 +36,23 @@ class SparkCatalogTest extends FlussSparkTestBase {
 
   protected def lakeFormat: Option[DataLakeFormat] = None
 
+  test("Catalog: add columns") {
+    withTable("t") {
+      sql("CREATE TABLE t (id int, name string)")
+
+      sql("ALTER TABLE t ADD COLUMN age bigint")
+      checkAnswer(
+        sql("DESC t"),
+        Row("id", "int", null) ::
+          Row("name", "string", null) ::
+          Row("age", "bigint", null) :: Nil)
+
+      val table = admin.getTableInfo(createTablePath("t")).get()
+      assertThat(table.getRowType.getFieldCount).isEqualTo(3)
+      assertThat(table.getRowType.getFieldNames).containsExactly("id", "name", "age")
+    }
+  }
+
   test("Catalog: namespaces") {
     // Always a default database 'fluss'.
     checkAnswer(sql("SHOW DATABASES"), Row(DEFAULT_DATABASE) :: Nil)