diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkConversions.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkConversions.scala index eeb726678d4..1e969d458dc 100644 --- a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkConversions.scala +++ b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkConversions.scala @@ -105,8 +105,32 @@ 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(".")}") + } + if (p.defaultValue() != null) { + throw new UnsupportedOperationException( + s"Adding column with default value is not supported: ${p.fieldNames().head}") + } + 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() + } + } } diff --git a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkCatalogTest.scala b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkCatalogTest.scala index e30b03baacb..37f76f6ca5b 100644 --- a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkCatalogTest.scala +++ b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkCatalogTest.scala @@ -22,6 +22,7 @@ import org.apache.fluss.exception.InvalidAlterTableException import org.apache.fluss.metadata._ import org.apache.fluss.types.{DataTypes, RowType} +import org.apache.spark.SparkException import org.apache.spark.sql.{AnalysisException, Row} import org.apache.spark.sql.catalyst.analysis.PartitionsAlreadyExistException import org.apache.spark.sql.connector.catalog.Identifier @@ -36,6 +37,87 @@ class SparkCatalogTest extends FlussSparkTestBase { protected def lakeFormat: Option[DataLakeFormat] = None + test("Catalog: add columns") { + withTable("t") { + sql("CREATE TABLE t (id int, name string)") + + // add column without comment, default is nullable and appended at the end + sql("ALTER TABLE t ADD COLUMN age bigint") + // add column with comment + sql("ALTER TABLE t ADD COLUMN addr string COMMENT 'address of the user'") + checkAnswer( + sql("DESC t"), + Row("id", "int", null) :: + Row("name", "string", null) :: + Row("age", "bigint", null) :: + Row("addr", "string", null) :: Nil) + + val table = admin.getTableInfo(createTablePath("t")).get() + assertThat(table.getRowType.getFieldCount).isEqualTo(4) + assertThatList(table.getRowType.getFieldNames).containsExactly("id", "name", "age", "addr") + // the comment of the added column should be persisted + assertThat(table.getSchema.getColumn("addr").getComment.get).isEqualTo("address of the user") + } + } + + test("Catalog: add columns with unsupported position") { + withTable("t") { + sql("CREATE TABLE t (id int, name string)") + + // only the last position is supported: FIRST/AFTER fail at the Fluss RPC serialization + // layer (ColumnPositionType only knows LAST), surfaced by Spark as a SparkException + val firstException = intercept[SparkException] { + sql("ALTER TABLE t ADD COLUMN age bigint FIRST") + } + assertThat(firstException).hasMessageContaining("Unsupported ColumnPositionType: FIRST") + val afterException = intercept[SparkException] { + sql("ALTER TABLE t ADD COLUMN age bigint AFTER id") + } + assertThat(afterException).hasMessageContaining("Unsupported ColumnPositionType: AFTER") + + val table = admin.getTableInfo(createTablePath("t")).get() + assertThatList(table.getRowType.getFieldNames).containsExactly("id", "name") + } + } + + test("Catalog: add non-nullable column is not supported") { + withTable("t") { + sql("CREATE TABLE t (id int, name string)") + + // fluss only supports adding nullable columns currently, the server rejects the change + // and the IllegalArgumentException is wrapped as UnknownServerException over the RPC + val notNullException = intercept[ExecutionException] { + sql("ALTER TABLE t ADD COLUMN age bigint NOT NULL") + } + assertThat(notNullException).hasMessageContaining("Column age must be nullable.") + + val table = admin.getTableInfo(createTablePath("t")).get() + assertThatList(table.getRowType.getFieldNames).containsExactly("id", "name") + } + } + + test("Catalog: column default value is not supported") { + // Creating a table with a column default value is rejected by Spark, because Fluss does not + // implement SupportsColumnDefaultValue. + val createException = intercept[AnalysisException] { + sql("CREATE TABLE t (id int, name string DEFAULT 'abc')") + } + assertThat(createException) + .hasMessageContaining( + "Table `fluss_catalog`.`fluss`.`t` does not support column default value") + + withTable("t") { + sql("CREATE TABLE t (id int, name string)") + + // ALTER TABLE ADD COLUMN with a DEFAULT clause behaves differently: Spark silently drops + // the default value (the catalog does not support it) and still adds a nullable column at + // the end. + sql("ALTER TABLE t ADD COLUMN age bigint DEFAULT 18") + val table = admin.getTableInfo(createTablePath("t")).get() + assertThatList(table.getRowType.getFieldNames).containsExactly("id", "name", "age") + } + } + test("Catalog: namespaces") { // Always a default database 'fluss'. checkAnswer(sql("SHOW DATABASES"), Row(DEFAULT_DATABASE) :: Nil)