Skip to content

[spark] Add sparksql alter table partial support for set/remove table properties #2437

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wuchong merged 4 commits into from
Jan 26, 2026
Merged

[spark] Add sparksql alter table partial support for set/remove table properties #2437

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1388fa1
support set/remove table properties
Yohahaha Jan 22, 2026
6901542
fix comments
Yohahaha Jan 26, 2026
20078df
add test to alter "table.datalake.enabled" config
wuchong Jan 26, 2026
2415ecc
fix spotless
wuchong Jan 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkCatalog.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,19 @@ class SparkCatalog extends TableCatalog with SupportsFlussNamespaces with WithFl
}

override def alterTable(ident: Identifier, changes: TableChange*): Table = {
throw new UnsupportedOperationException("Altering table is not supported")
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
}
}
}

override def dropTable(ident: Identifier): Boolean = {
Expand Down
18 changes: 15 additions & 3 deletions fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkConversions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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 {
Expand All @@ -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)
}
Expand All @@ -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)
}

Expand All @@ -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")
}
}
}
3 changes: 2 additions & 1 deletion fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/FlussSparkTestBase.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ import org.apache.fluss.client.admin.Admin
import org.apache.fluss.client.table.Table
import org.apache.fluss.client.table.scanner.log.LogScanner
import org.apache.fluss.config.{ConfigOptions, Configuration}
import org.apache.fluss.metadata.{TableDescriptor, TablePath}
import org.apache.fluss.metadata.{DataLakeFormat, TableDescriptor, TablePath}
import org.apache.fluss.row.InternalRow
import org.apache.fluss.server.testutils.FlussClusterExtension

Expand Down Expand Up @@ -107,6 +107,7 @@ object FlussSparkTestBase {
.setClusterConf(
new Configuration()
.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1))
.set(ConfigOptions.DATALAKE_FORMAT, DataLakeFormat.PAIMON)
)
.setNumOfTabletServers(3)
.build
Expand Down
59 changes: 59 additions & 0 deletions fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkCatalogTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,18 @@

package org.apache.fluss.spark

import org.apache.fluss.config.ConfigOptions
import org.apache.fluss.exception.InvalidAlterTableException
import org.apache.fluss.metadata._
import org.apache.fluss.types.{DataTypes, RowType}

import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.catalyst.analysis.PartitionsAlreadyExistException
import org.apache.spark.sql.connector.catalog.Identifier
import org.assertj.core.api.Assertions.{assertThat, assertThatList}
import org.scalatest.matchers.should.Matchers.{a, convertToAnyShouldWrapper}

import java.util.concurrent.ExecutionException

import scala.collection.JavaConverters._

Expand Down Expand Up @@ -192,6 +197,60 @@ class SparkCatalogTest extends FlussSparkTestBase {
checkAnswer(sql("SHOW DATABASES"), Row(DEFAULT_DATABASE) :: Nil)
}

test("Catalog: set/remove table properties") {
Copy link
Contributor

@YannByron YannByron Jan 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe it'll be nice to have a separate SparkTableChangeTest, which includes all Alter Table tests.

Copy link
Contributor Author

@Yohahaha Yohahaha Jan 26, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

#2314 (comment)

above comments suggests add all DDL tests into one suite.

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(flussTable.getNumBuckets, "check bucket num")(3)
assertResult(
Map(
ConfigOptions.TABLE_REPLICATION_FACTOR.key() -> "1",
ConfigOptions.TABLE_DATALAKE_FORMAT.key() -> "paimon"),
"check table properties")(flussTable.getProperties.toMap.asScala)
assert(
flussTable.getCustomProperties.toMap.asScala.getOrElse("key1", "non-exists") == "value1")

sql("ALTER TABLE t SET TBLPROPERTIES('key1' = 'value2', 'key2' = 'value2')")
flussTable = admin.getTableInfo(createTablePath("t")).get()
assertResult(
Map(
ConfigOptions.TABLE_REPLICATION_FACTOR.key() -> "1",
ConfigOptions.TABLE_DATALAKE_FORMAT.key() -> "paimon"),
"check table properties")(flussTable.getProperties.toMap.asScala)
assert(
flussTable.getCustomProperties.toMap.asScala.getOrElse("key1", "non-exists") == "value2")
assert(
flussTable.getCustomProperties.toMap.asScala.getOrElse("key2", "non-exists") == "value2")

sql("ALTER TABLE t UNSET TBLPROPERTIES('key1', 'key2')")
flussTable = admin.getTableInfo(createTablePath("t")).get()
assert(!flussTable.getCustomProperties.toMap.asScala.contains("key1"))
assert(!flussTable.getCustomProperties.toMap.asScala.contains("key2"))

// no error if unset not-exists key
sql("ALTER TABLE t UNSET TBLPROPERTIES('key1')")

sql(
s"ALTER TABLE t SET TBLPROPERTIES('${ConfigOptions.TABLE_DATALAKE_ENABLED.key()}' = 'true')")
flussTable = admin.getTableInfo(createTablePath("t")).get()
assertResult(
Map(
ConfigOptions.TABLE_REPLICATION_FACTOR.key() -> "1",
ConfigOptions.TABLE_DATALAKE_FORMAT.key() -> "paimon",
ConfigOptions.TABLE_DATALAKE_ENABLED.key() -> "true"
),
"check table properties"
)(flussTable.getProperties.toMap.asScala)

// Most table properties with prefix of 'table.' are not allowed to be modified.
intercept[ExecutionException] {
sql(
s"ALTER TABLE t SET TBLPROPERTIES('${ConfigOptions.TABLE_REPLICATION_FACTOR.key()}' = '2')")
}.getCause.shouldBe(a[InvalidAlterTableException])
}
}

test("Partition: show partitions") {
withTable("t") {
sql(s"CREATE TABLE t (id int, name string, pt1 string, pt2 int) PARTITIONED BY (pt1, pt2)")
Expand Down