This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new f39f99713f9f fix(spark): create Hudi tables via the DSv2 Column API
and accept unchanged properties in HoodieCatalog (#20426)
f39f99713f9f is described below
commit f39f99713f9f8327d403947f4e38aad4259eb55a
Author: Sivabalan Narayanan <[email protected]>
AuthorDate: Mon Oct 26 22:39:30 2026 -0700
fix(spark): create Hudi tables via the DSv2 Column API and accept unchanged
properties in HoodieCatalog (#20426)
* fix(spark): create Hudi tables via the DSv2 Column API and accept
unchanged properties in HoodieCatalog
Spark 4.1's TableCatalog.createTable(Identifier, TableInfo), used by Spark
Declarative Pipelines, defaults to the Column[]-based createTable overload,
which DelegatingCatalogExtension forwards to the session catalog: the table
is
registered without hoodie.properties. Route that overload to HoodieCatalog's
StructType-based createTable for Hudi tables. The Column API exists from
Spark
3.4 on, so the override lives in a trait with a Spark 3.4+ and an empty
Spark
3.3 version, added from a source directory chosen by the Spark profile.
Declarative Pipelines also re-sends every table property as SetProperty on
each run, which HoodieCatalog.alterTable rejected. Accept SetProperty and
RemoveProperty: changes that keep the current value are no-ops, changes to
properties persisted in hoodie.properties are rejected, and other properties
are applied to the catalog table.
* fix(spark): validate and preserve catalog property changes
---------
Co-authored-by: danny0405 <[email protected]>
---
hudi-spark-datasource/hudi-spark-common/pom.xml | 20 ++
.../hudi/catalog/HoodieCatalogColumnsSupport.scala | 27 +++
.../hudi/catalog/HoodieCatalogColumnsSupport.scala | 49 ++++
.../apache/spark/sql/hudi/HoodieOptionConfig.scala | 60 ++++-
.../spark/sql/hudi/catalog/HoodieCatalog.scala | 67 +++++-
.../sql/hudi/catalog/TestHoodieCatalogDDL.scala | 246 ++++++++++++++++++++-
.../sql/hudi/common/TestHoodieOptionConfig.scala | 42 ++++
pom.xml | 3 +
8 files changed, 490 insertions(+), 24 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark-common/pom.xml
b/hudi-spark-datasource/hudi-spark-common/pom.xml
index 1dca8ee367a8..20f98fe3dab3 100644
--- a/hudi-spark-datasource/hudi-spark-common/pom.xml
+++ b/hudi-spark-datasource/hudi-spark-common/pom.xml
@@ -39,6 +39,26 @@
</resources>
<plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <version>${build-helper-maven-plugin.version}</version>
+ <executions>
+ <execution>
+ <id>add-spark-version-source</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <configuration>
+ <!-- Code that only compiles against some Spark versions, e.g.
the DSv2 Column API added in Spark 3.4 -->
+ <sources>
+ <source>${hudi.spark.common.version.source}</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-dependency-plugin</artifactId>
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala-spark3.3/org/apache/spark/sql/hudi/catalog/HoodieCatalogColumnsSupport.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala-spark3.3/org/apache/spark/sql/hudi/catalog/HoodieCatalogColumnsSupport.scala
new file mode 100644
index 000000000000..d0e43e111f8e
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala-spark3.3/org/apache/spark/sql/hudi/catalog/HoodieCatalogColumnsSupport.scala
@@ -0,0 +1,27 @@
+/*
+ * 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.spark.sql.hudi.catalog
+
+import org.apache.spark.sql.connector.catalog.DelegatingCatalogExtension
+
+/**
+ * Spark 3.3 has no `Column`-based `createTable` overload, so there is nothing
to route; see the Spark 3.4+ version.
+ */
+trait HoodieCatalogColumnsSupport extends DelegatingCatalogExtension
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala-spark3.4plus/org/apache/spark/sql/hudi/catalog/HoodieCatalogColumnsSupport.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala-spark3.4plus/org/apache/spark/sql/hudi/catalog/HoodieCatalogColumnsSupport.scala
new file mode 100644
index 000000000000..169637e40427
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala-spark3.4plus/org/apache/spark/sql/hudi/catalog/HoodieCatalogColumnsSupport.scala
@@ -0,0 +1,49 @@
+/*
+ * 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.spark.sql.hudi.catalog
+
+import org.apache.hudi.SparkAdapterSupport
+
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Column,
DelegatingCatalogExtension, Identifier, Table}
+import org.apache.spark.sql.connector.expressions.Transform
+
+import java.util
+
+/**
+ * Routes the [[Column]]-based `createTable` overload (Spark 3.4+) to the
`StructType`-based one that
+ * [[HoodieCatalog]] implements, for Hudi tables.
+ *
+ * [[DelegatingCatalogExtension]] forwards this overload straight to the
session catalog, which registers the
+ * table without initializing it as a Hudi table (no `hoodie.properties`).
Spark 4.1's
+ * `createTable(Identifier, TableInfo)`, used by Spark Declarative Pipelines,
ends up in this overload as well.
+ */
+trait HoodieCatalogColumnsSupport extends DelegatingCatalogExtension with
SparkAdapterSupport {
+
+ override def createTable(ident: Identifier,
+ columns: Array[Column],
+ partitions: Array[Transform],
+ properties: util.Map[String, String]): Table = {
+ if (sparkAdapter.isHoodieTable(properties)) {
+ createTable(ident, CatalogV2Util.v2ColumnsToStructType(columns),
partitions, properties)
+ } else {
+ super.createTable(ident, columns, partitions, properties)
+ }
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/HoodieOptionConfig.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/HoodieOptionConfig.scala
index ce95da64bb8e..d22bf261e507 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/HoodieOptionConfig.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/HoodieOptionConfig.scala
@@ -19,14 +19,18 @@ package org.apache.spark.sql.hudi
import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.common.avro.HoodieAvroUtils.getRootLevelFieldName
+import org.apache.hudi.common.config.ConfigProperty
import org.apache.hudi.common.model.HoodieTableType
import org.apache.hudi.common.table.HoodieTableConfig
-import org.apache.hudi.common.util.ValidationUtils
+import org.apache.hudi.common.util.{ConfigUtils, ValidationUtils}
import org.apache.hudi.config.{HoodieIndexConfig, HoodieWriteConfig}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.StructType
+import java.util.Locale
+
+import scala.collection.JavaConverters._
/**
* The HoodieOptionConfig defines some short name for the hoodie
@@ -133,6 +137,25 @@ object HoodieOptionConfig {
private lazy val tableConfigKeyToSqlOptionKey: Map[String, String] =
(sqlOptionKeyToTableConfigKey -- legacySqlOptionKeys).map(f => f._2 ->
f._1)
+ private lazy val tableConfigProperties: Map[String, ConfigProperty[_]] =
+ classOf[HoodieTableConfig].getFields
+ .filter(_.getType == classOf[ConfigProperty[_]])
+ .map(_.get(null).asInstanceOf[ConfigProperty[_]])
+ .map(config => config.key -> config).toMap
+
+ private lazy val tableConfigKeys: Map[String, String] = {
+ // Alternatives take precedence over deprecated declarations, e.g.
PRECOMBINE_FIELD -> ORDERING_FIELDS.
+ tableConfigProperties.keys.map(key => key -> key).toMap ++
tableConfigProperties.values.flatMap { config =>
+ config.getAlternatives.asScala.map(_ -> config.key)
+ }.toMap
+ }
+
+ private lazy val caseInsensitiveSqlOptionKeys: Map[String, String] =
+ sqlOptionKeyToWriteConfigKey.keys.map(key => key.toLowerCase(Locale.ROOT)
-> key).toMap
+
+ private[hudi] def canonicalizeSqlOptionKey(key: String): String =
+ caseInsensitiveSqlOptionKeys.getOrElse(key.toLowerCase(Locale.ROOT), key)
+
/**
* Mapping of the short sql value to the hoodie's config value
*/
@@ -182,6 +205,29 @@ object HoodieOptionConfig {
}
}
+ /**
+ * Map a SQL option, data source write config or table config, given by key
and value, to the table config
+ * it is persisted as in hoodie.properties. Returns None if it is not
persisted as a table config.
+ */
+ def mapToTableConfig(key: String, value: String): Option[(String, String)] =
{
+ val sqlKey = writeConfigKeyToSqlOptionKey.getOrElse(key,
canonicalizeSqlOptionKey(key))
+
sqlOptionKeyToTableConfigKey.get(sqlKey).orElse(tableConfigKeys.get(key)).map {
tableKey =>
+ val tableValue = if (tableKey == HoodieTableConfig.TYPE.key) {
+
sqlOptionValueToHoodieConfigValue.getOrElse(value.toLowerCase(Locale.ROOT),
value.toUpperCase(Locale.ROOT))
+ } else {
+ value
+ }
+ tableKey -> tableValue
+ }
+ }
+
+ /** Read a persisted value using its declared alternative keys and the same
conversion as incoming options. */
+ private[hudi] def getTableConfigValue(tableConfig: HoodieTableConfig, key:
String): Option[String] = {
+ tableConfigProperties.get(key)
+ .flatMap(config =>
Option(ConfigUtils.getStringWithAltKeys(tableConfig.getProps, config, null:
String)))
+ .flatMap(value => mapToTableConfig(key, value).map(_._2))
+ }
+
val defaultSqlOptions: Map[String, String] = {
HoodieOptionConfig.getClass.getDeclaredFields
.filter(f => f.getType == classOf[HoodieSQLOption[_]])
@@ -261,17 +307,7 @@ object HoodieOptionConfig {
}
def makeOptionsCaseInsensitive(sqlOptions: Map[String, String]): Map[String,
String] = {
- // Make Keys Case Insensitive
- val standardOptions = Seq(SQL_KEY_TABLE_PRIMARY_KEY,
SQL_KEY_ORDERING_FIELDS,
- SQL_KEY_TABLE_TYPE, SQL_PAYLOAD_CLASS, SQL_RECORD_MERGE_STRATEGY_ID,
SQL_RECORD_MERGE_MODE)
- .flatMap(key => (List(key.sqlKeyName) ++ key.alternatives).toStream)
-
- sqlOptions.map(option => {
- standardOptions.find(_.equalsIgnoreCase(option._1)) match {
- case Some(standardKey) => (standardKey, option._2)
- case None => (option._1, option._2)
- }
- })
+ sqlOptions.map { case (key, value) => canonicalizeSqlOptionKey(key) ->
value }
}
private def getSqlOptionWithAlternatives(sqlOption: HoodieSQLOption[String],
options: Map[String, String]): Option[String] = {
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
index 2810fcdbcb14..70a5853c4e73 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
@@ -39,10 +39,10 @@ import org.apache.spark.sql.catalyst.catalog._
import org.apache.spark.sql.catalyst.catalog.HoodieCatalogTable.needFilterProps
import org.apache.spark.sql.connector.catalog._
import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.IdentifierHelper
-import org.apache.spark.sql.connector.catalog.TableChange.{AddColumn,
ColumnChange, UpdateColumnComment, UpdateColumnType}
+import org.apache.spark.sql.connector.catalog.TableChange.{AddColumn,
ColumnChange, RemoveProperty, SetProperty, UpdateColumnComment,
UpdateColumnType}
import org.apache.spark.sql.connector.expressions.{FieldReference,
IdentityTransform, Transform}
import org.apache.spark.sql.execution.datasources.DataSource
-import org.apache.spark.sql.hudi.{HoodieSqlCommonUtils, ProvidesHoodieConfig}
+import org.apache.spark.sql.hudi.{HoodieOptionConfig, HoodieSqlCommonUtils,
ProvidesHoodieConfig}
import
org.apache.spark.sql.hudi.analysis.HoodieSparkBaseAnalysis.HoodieV1OrV2Table
import
org.apache.spark.sql.hudi.catalog.HoodieCatalog.{buildPartitionTransforms,
isTablePartitioned}
import org.apache.spark.sql.hudi.command._
@@ -56,6 +56,7 @@ import scala.collection.JavaConverters._
import scala.collection.mutable
class HoodieCatalog extends DelegatingCatalogExtension
+ with HoodieCatalogColumnsSupport
with StagingTableCatalog
with SparkAdapterSupport
with ProvidesHoodieConfig {
@@ -223,7 +224,12 @@ class HoodieCatalog extends DelegatingCatalogExtension
loadTable(ident) match {
case HoodieV1OrV2Table(table) => {
val tableIdent = TableIdentifier(ident.name(),
ident.namespace().lastOption)
- changes.groupBy(c => c.getClass).foreach {
+ val (propertyChanges, otherChanges) = changes.partition {
+ case _: SetProperty | _: RemoveProperty => true
+ case _ => false
+ }
+ val catalogPropertyChanges = if (propertyChanges.isEmpty) Seq.empty
else catalogPropertyChangesToApply(table, propertyChanges)
+ otherChanges.groupBy(c => c.getClass).foreach {
case (t, newColumns) if t == classOf[AddColumn] =>
AlterHoodieTableAddColumnsCommand(
tableIdent,
@@ -255,6 +261,9 @@ class HoodieCatalog extends DelegatingCatalogExtension
case (t, _) =>
throw new UnsupportedOperationException(s"not supported table
change: ${t.getName}")
}
+ if (catalogPropertyChanges.nonEmpty) {
+ super.alterTable(ident, catalogPropertyChanges: _*)
+ }
loadTable(ident)
}
@@ -262,6 +271,58 @@ class HoodieCatalog extends DelegatingCatalogExtension
}
}
+ /**
+ * Returns the property changes that still have to be applied to the catalog
table.
+ *
+ * Callers such as Spark Declarative Pipelines re-send every table property
on each run, so changes that keep the
+ * current value are dropped. Properties persisted in hoodie.properties
(record key, ordering fields, table type,
+ * ...) can't be changed through the catalog, since the catalog would no
longer match the table.
+ */
+ private def catalogPropertyChangesToApply(table: CatalogTable, changes:
Seq[TableChange]): Seq[TableChange] = {
+ lazy val tableConfig = HoodieCatalogTable(spark, table).tableConfig
+ val properties = table.properties
+ lazy val sqlProperties =
HoodieOptionConfig.makeOptionsCaseInsensitive(properties)
+ def catalogValue(key: String): Option[String] =
+ sqlProperties.get(HoodieOptionConfig.canonicalizeSqlOptionKey(key))
+
+ // Only filter immutable properties here. Spark applies mutable changes in
order, including repeated keys.
+ val catalogChanges = changes.filter {
+ case set: SetProperty =>
+ HoodieOptionConfig.mapToTableConfig(set.property, set.value) match {
+ case Some(_) if catalogValue(set.property).contains(set.value) =>
false
+ case Some((tableConfigKey, value)) =>
+ val persisted =
HoodieOptionConfig.getTableConfigValue(tableConfig, tableConfigKey)
+ if (!persisted.contains(value)) {
+ throw new HoodieAnalysisException(s"Can't set ${set.property} to
'${set.value}' on Hudi table " +
+ s"${table.identifier}: it is persisted in hoodie.properties as
$tableConfigKey " +
+ s"(current value: ${persisted.getOrElse("unset")}) and can't
be changed by altering the table.")
+ }
+ false
+ case None => true
+ }
+ case remove: RemoveProperty =>
+ val persisted = HoodieOptionConfig.mapToTableConfig(remove.property,
"").isDefined
+ if (persisted && catalogValue(remove.property).isDefined) {
+ throw new HoodieAnalysisException(s"Can't remove ${remove.property}
from Hudi table ${table.identifier}: " +
+ "it is persisted in hoodie.properties and can't be changed by
altering the table.")
+ }
+ !persisted
+ }
+
+ val updated = CatalogV2Util.applyPropertiesChanges(properties,
catalogChanges)
+ if (updated == properties) {
+ Seq.empty
+ } else {
+ val removed = (properties.keySet --
updated.keySet).toSeq.map(TableChange.removeProperty)
+ val set = updated.toSeq.collect {
+ // The session catalog reconstructs the comment from properties, even
though Hive stores it separately.
+ case (key, value) if key == TableCatalog.PROP_COMMENT ||
!properties.get(key).contains(value) =>
+ TableChange.setProperty(key, value)
+ }
+ set ++ removed
+ }
+ }
+
private def deduceTableLocationURIAndTableType(
ident: Identifier, properties: util.Map[String, String]): (URI,
CatalogTableType) = {
val locOpt = if (isPathIdentifier(ident)) {
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala
index f509a26078a9..68adad112b72 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala
@@ -17,19 +17,24 @@
package org.apache.spark.sql.hudi.catalog
+import org.apache.hudi.HoodieSparkUtils
+
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
-import org.apache.spark.sql.connector.catalog.{Identifier, TableChange}
+import org.apache.spark.sql.catalyst.catalog.HoodieCatalogTable
+import org.apache.spark.sql.connector.catalog.{Identifier, Table,
TableCatalog, TableChange}
import
org.apache.spark.sql.connector.catalog.TableCapability.{ACCEPT_ANY_SCHEMA,
BATCH_READ, OVERWRITE_BY_FILTER, TRUNCATE, V1_BATCH_WRITE}
import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.execution.datasources.v2.V2SessionCatalog
import org.apache.spark.sql.hudi.HoodieSqlCommonUtils
import org.apache.spark.sql.hudi.command.ShowHoodieCreateTableCommand
import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException
import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
-import org.apache.spark.sql.types.{IntegerType, LongType, StringType,
StructField, StructType}
+import org.apache.spark.sql.types.{DataType, IntegerType, LongType,
StringType, StructField, StructType}
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
import java.io.File
+import java.util.Locale
import scala.collection.JavaConverters._
@@ -89,13 +94,6 @@ class TestHoodieCatalogDDL extends HoodieSparkSqlTestBase {
assertEquals(IntegerType,
catalog.loadTable(ident).schema().fields.find(_.name ==
"age").get.dataType)
- // alterTable: any change that is neither AddColumn nor a ColumnChange
falls through to the
- // default arm of HoodieCatalog.alterTable and is reported as
unsupported.
- val unsupportedChange = intercept[UnsupportedOperationException] {
- catalog.alterTable(ident, TableChange.setProperty("some.key", "v"))
- }
- assertTrue(unsupportedChange.getMessage.contains("SetProperty"),
unsupportedChange.getMessage)
-
// renameTable moves the catalog entry.
val renamed = Identifier.of(Array("default"), s"${tableName}_renamed")
catalog.renameTable(ident, renamed)
@@ -110,6 +108,236 @@ class TestHoodieCatalogDDL extends HoodieSparkSqlTestBase
{
}
}
+ test("alterTable ignores re-sent table properties and rejects changes to
persisted table configs") {
+ withTempDir { tmp =>
+ val catalog = hoodieCatalog
+ val tableName = generateTableName
+ val ident = Identifier.of(Array("default"), tableName)
+ val tableIdent = TableIdentifier(tableName, Some("default"))
+ val schema = StructType(Seq(
+ StructField("id", IntegerType),
+ StructField("name", StringType),
+ StructField("ts", LongType)))
+ // The properties Spark Declarative Pipelines passes on create and
re-sends as SetProperty on every run.
+ val props = Map(
+ "provider" -> "hudi",
+ "comment" -> "a hudi table",
+ "hoodie.datasource.write.recordkey.field" -> "id",
+ "hoodie.datasource.write.precombine.field" -> "ts",
+ "hoodie.table.type" -> "COPY_ON_WRITE")
+ catalog.createTable(ident, schema, Array.empty[Transform],
+ (props + ("location" -> s"${tmp.getCanonicalPath}/$tableName")).asJava)
+ val propertiesBefore =
spark.sessionState.catalog.getTableMetadata(tableIdent).properties
+
+ // Re-sending unchanged values, including SQL aliases and differently
cased values, is a no-op.
+ catalog.alterTable(ident, props.map { case (k, v) =>
TableChange.setProperty(k, v) }.toSeq: _*)
+ catalog.alterTable(ident,
+ TableChange.setProperty("PRIMARYKEY", "id"),
+ TableChange.setProperty("TYPE", "cow"),
+ TableChange.setProperty("hoodie.table.type", "copy_on_write"),
+ TableChange.setProperty("PRECOMBINEFIELD", "ts"),
+ TableChange.setProperty("hoodie.table.precombine.field", "ts"),
+ TableChange.setProperty("hoodie.table.ordering.fields", "ts"))
+ assertEquals(propertiesBefore,
spark.sessionState.catalog.getTableMetadata(tableIdent).properties)
+
+ // Changing or removing a property persisted in hoodie.properties is
rejected.
+ val recordKeyChange = intercept[HoodieAnalysisException] {
+ catalog.alterTable(ident,
TableChange.setProperty("hoodie.datasource.write.recordkey.field", "name"))
+ }
+
assertTrue(recordKeyChange.getMessage.contains("hoodie.table.recordkey.fields"),
recordKeyChange.getMessage)
+ intercept[HoodieAnalysisException] {
+ catalog.alterTable(ident, TableChange.setProperty("type", "mor"))
+ }
+ intercept[HoodieAnalysisException] {
+ catalog.alterTable(ident,
TableChange.setProperty("hoodie.datasource.write.precombine.field", "name"))
+ }
+ intercept[HoodieAnalysisException] {
+ catalog.alterTable(ident, TableChange.removeProperty("primaryKey"))
+ }
+ Seq("PRIMARYKEY", "PrimaryKey").foreach { key =>
+ intercept[HoodieAnalysisException] {
+ catalog.alterTable(ident, TableChange.setProperty(key, "name"))
+ }
+ intercept[HoodieAnalysisException] {
+ catalog.alterTable(ident, TableChange.removeProperty(key))
+ }
+ }
+ // Field names and class names are case-sensitive, unlike table-type
enum values.
+ intercept[HoodieAnalysisException] {
+ catalog.alterTable(ident, TableChange.setProperty("primaryKey", "ID"))
+ }
+ intercept[HoodieAnalysisException] {
+ catalog.alterTable(ident, TableChange.setProperty("primaryKey", " id
"))
+ }
+ val tableConfig = HoodieCatalogTable(spark,
spark.sessionState.catalog.getTableMetadata(tableIdent)).tableConfig
+ intercept[HoodieAnalysisException] {
+ catalog.alterTable(ident,
TableChange.setProperty("hoodie.table.keygenerator.class",
+ tableConfig.getKeyGeneratorClassName.toUpperCase(Locale.ROOT)))
+ }
+ // A rejected change does not apply the other changes of the same call.
+ intercept[HoodieAnalysisException] {
+ catalog.alterTable(ident,
+ TableChange.setProperty("some.key", "v"),
TableChange.setProperty("primaryKey", "name"))
+ }
+ assertEquals(propertiesBefore,
spark.sessionState.catalog.getTableMetadata(tableIdent).properties)
+
+ // Other properties and the comment are applied to the catalog table.
+ catalog.alterTable(ident,
+ TableChange.setProperty("some.key", "v"),
TableChange.setProperty("comment", "new comment"))
+ val altered = spark.sessionState.catalog.getTableMetadata(tableIdent)
+ assertEquals((propertiesBefore - "comment") + ("some.key" -> "v"),
altered.properties - "comment")
+ assertEquals(Some("v"), altered.properties.get("some.key"))
+ assertEquals(Some("new comment"), altered.comment)
+ catalog.alterTable(ident, TableChange.removeProperty("some.key"))
+
assertFalse(spark.sessionState.catalog.getTableMetadata(tableIdent).properties.contains("some.key"))
+
+ // The table is still writable with its original configuration.
+ spark.sql(s"insert into $tableName values (1, 'a1', 1000)")
+ checkAnswer(s"select id, name, ts from $tableName")(Seq(1, "a1", 1000))
+ }
+ }
+
+ test("alterTable applies mutable property changes in order, including
table-service write configs") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val ident = Identifier.of(Array("default"), tableName)
+ val tableIdent = TableIdentifier(tableName, Some("default"))
+ val catalog = hoodieCatalog
+ catalog.createTable(ident, StructType(Seq(StructField("id",
IntegerType))), Array.empty[Transform],
+ Map("provider" -> "hudi", "primaryKey" -> "id", "location" ->
s"${tmp.getCanonicalPath}/$tableName").asJava)
+ def properties: Map[String, String] =
spark.sessionState.catalog.getTableMetadata(tableIdent).properties
+ val before = properties
+
+ catalog.alterTable(ident, TableChange.setProperty("some.key", "v"),
TableChange.removeProperty("some.key"))
+ assertEquals(before, properties)
+ catalog.alterTable(ident, TableChange.setProperty("some.key", "old"))
+ catalog.alterTable(ident, TableChange.setProperty("some.key", "new"),
TableChange.setProperty("some.key", "old"))
+ assertEquals(Some("old"), properties.get("some.key"))
+ catalog.alterTable(ident, TableChange.removeProperty("some.key"),
TableChange.setProperty("some.key", "old"))
+ assertEquals(Some("old"), properties.get("some.key"))
+ // Net no-ops on one key must also remain no-ops when another key is
changed in the same call.
+ catalog.alterTable(ident, TableChange.setProperty("some.key", "new"),
TableChange.setProperty("some.key", "old"),
+ TableChange.setProperty("another.key", "v"),
TableChange.removeProperty("absent.key"))
+ assertEquals(Some("old"), properties.get("some.key"))
+ assertEquals(Some("v"), properties.get("another.key"))
+
+ val writeConfigs = Seq("hoodie.table.services.enabled",
"hoodie.table.services.incremental.enabled",
+ "hoodie.table.service.manager.enabled")
+ catalog.alterTable(ident, writeConfigs.map(TableChange.setProperty(_,
"false")): _*)
+ writeConfigs.foreach(key => assertEquals(Some("false"),
properties.get(key)))
+ catalog.alterTable(ident, writeConfigs.map(TableChange.removeProperty):
_*)
+ writeConfigs.foreach(key => assertFalse(properties.contains(key)))
+ }
+ }
+
+ test("alterTable preserves unrelated properties changed after the initial
catalog read") {
+ withTempDir { tmp =>
+ val session = spark.sessionState.catalog
+ val tableName = generateTableName
+ val ident = Identifier.of(Array("default"), tableName)
+ val tableIdent = TableIdentifier(tableName, Some("default"))
+ hoodieCatalog.createTable(ident, StructType(Seq(StructField("id",
IntegerType))), Array.empty[Transform],
+ Map("provider" -> "hudi", "primaryKey" -> "id", "a" -> "old",
"location" -> s"${tmp.getCanonicalPath}/$tableName").asJava)
+ val delegate = new V2SessionCatalog(session)
+ def setA(value: String): Unit = {
+ val current = session.getTableMetadata(tableIdent)
+ session.alterTable(current.copy(properties = current.properties + ("a"
-> value)))
+ }
+ // The delegate preserves an unrelated update committed before its own
read.
+ setA("concurrent")
+ delegate.alterTable(ident, TableChange.setProperty("b", "control"))
+ assertEquals("concurrent",
session.getTableMetadata(tableIdent).properties("a"))
+ setA("old")
+
+ val interleaved = new HoodieCatalog {
+ private var inject = true
+ override def loadTable(id: Identifier): Table = {
+ val snapshot = super.loadTable(id)
+ if (inject) {
+ inject = false
+ setA("concurrent")
+ }
+ snapshot
+ }
+ }
+ interleaved.setDelegateCatalog(delegate)
+ interleaved.alterTable(ident, TableChange.setProperty("b", "requested"))
+ val properties = session.getTableMetadata(tableIdent).properties
+ assertEquals("concurrent", properties("a"))
+ assertEquals("requested", properties("b"))
+ }
+ }
+
+ test("alterTable preserves separately stored comments unless explicitly
removed") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val ident = Identifier.of(Array("default"), tableName)
+ val tableIdent = TableIdentifier(tableName, Some("default"))
+ val catalog = hoodieCatalog
+ catalog.createTable(ident, StructType(Seq(StructField("id",
IntegerType))), Array.empty[Transform],
+ Map("provider" -> "hudi", "primaryKey" -> "id", "comment" -> "keep me",
+ "location" -> s"${tmp.getCanonicalPath}/$tableName").asJava)
+ def metadata = spark.sessionState.catalog.getTableMetadata(tableIdent)
+ def storeCommentSeparately(): Unit = {
+ // HiveClientImpl excludes "comment" from properties and returns it
only in CatalogTable.comment.
+ spark.sessionState.catalog.alterTable(metadata.copy(properties =
metadata.properties - "comment"))
+ }
+
+ storeCommentSeparately()
+ catalog.alterTable(ident, TableChange.setProperty("some.key", "v"))
+ assertEquals(Some("keep me"), metadata.comment)
+ storeCommentSeparately()
+ catalog.alterTable(ident, TableChange.setProperty("comment", "keep me"),
TableChange.removeProperty("some.key"))
+ assertEquals(Some("keep me"), metadata.comment)
+ storeCommentSeparately()
+ catalog.alterTable(ident, TableChange.removeProperty("comment"),
TableChange.setProperty("comment", "keep me"),
+ TableChange.setProperty("some.key", "v2"))
+ assertEquals(Some("keep me"), metadata.comment)
+ storeCommentSeparately()
+ catalog.alterTable(ident, TableChange.removeProperty("comment"))
+ assertEquals(None, metadata.comment)
+ catalog.alterTable(ident, TableChange.setProperty("some.key", "v3"))
+ assertEquals(None, metadata.comment)
+ }
+ }
+
+ test("createTable with the Column-based overload initializes Hudi tables") {
+ // The Column-based overload exists from Spark 3.4 on; it is invoked
reflectively so that this suite also
+ // compiles against Spark 3.3. Spark 4.1's createTable(Identifier,
TableInfo) delegates to it.
+ if (HoodieSparkUtils.gteqSpark3_4) {
+ withTempDir { tmp =>
+ val catalog = hoodieCatalog
+ val columnClass =
Class.forName("org.apache.spark.sql.connector.catalog.Column")
+ val createColumn = columnClass.getMethod("create", classOf[String],
classOf[DataType])
+ def columns(fields: (String, DataType)*): AnyRef = {
+ val array = java.lang.reflect.Array.newInstance(columnClass,
fields.size)
+ fields.zipWithIndex.foreach { case ((name, dataType), i) =>
+ java.lang.reflect.Array.set(array, i, createColumn.invoke(null,
name, dataType))
+ }
+ array
+ }
+ val createTable = classOf[TableCatalog].getMethod("createTable",
classOf[Identifier],
+ java.lang.reflect.Array.newInstance(columnClass, 0).getClass,
classOf[Array[Transform]], classOf[java.util.Map[_, _]])
+ val cols = columns("id" -> IntegerType, "name" -> StringType, "ts" ->
LongType)
+
+ val hudiTable = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$hudiTable"
+ createTable.invoke(catalog, Identifier.of(Array("default"),
hudiTable), cols, Array.empty[Transform],
+ Map("provider" -> "hudi", "primaryKey" -> "id", "preCombineField" ->
"ts", "location" -> tablePath).asJava)
+ assertTrue(new File(s"$tablePath/.hoodie/hoodie.properties").exists())
+ spark.sql(s"insert into $hudiTable values (1, 'a1', 1000)")
+ checkAnswer(s"select id, name, ts from $hudiTable")(Seq(1, "a1", 1000))
+
+ // Non-Hudi tables still go to the session catalog.
+ val parquetTable = generateTableName
+ createTable.invoke(catalog, Identifier.of(Array("default"),
parquetTable), cols, Array.empty[Transform],
+ Map("provider" -> "parquet", "location" ->
s"${tmp.getCanonicalPath}/$parquetTable").asJava)
+ assertEquals(Some("parquet"),
+
spark.sessionState.catalog.getTableMetadata(TableIdentifier(parquetTable,
Some("default"))).provider)
+ }
+ }
+ }
+
test("SHOW CREATE TABLE regenerates Hudi DDL") {
val tableName = generateTableName
spark.sql(
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieOptionConfig.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieOptionConfig.scala
index 078ef5b35276..d9a2f7954925 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieOptionConfig.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieOptionConfig.scala
@@ -78,6 +78,48 @@ class TestHoodieOptionConfig extends
SparkClientFunctionalTestHarness {
assertEquals(expected,
HoodieOptionConfig.makeOptionsCaseInsensitive(options))
}
+ @Test
+ def testMapToTableConfigRecognizesAliasesAndDeclaredConfigs(): Unit = {
+ Seq("primaryKey", "PRIMARYKEY", "hoodie.datasource.write.recordkey.field",
"hoodie.table.recordkey.fields").foreach { key =>
+ assertEquals(Some(HoodieTableConfig.RECORDKEY_FIELDS.key -> "Id"),
HoodieOptionConfig.mapToTableConfig(key, "Id"))
+ }
+ Seq("orderingFields", "PRECOMBINEFIELD",
"hoodie.datasource.write.precombine.field",
"hoodie.table.precombine.field").foreach { key =>
+ assertEquals(Some(HoodieTableConfig.ORDERING_FIELDS.key -> "ts"),
HoodieOptionConfig.mapToTableConfig(key, "ts"))
+ }
+ Seq("TYPE", "hoodie.datasource.write.table.type",
"hoodie.table.type").foreach { key =>
+ assertEquals(Some(HoodieTableConfig.TYPE.key -> "MERGE_ON_READ"),
HoodieOptionConfig.mapToTableConfig(key, "MoR"))
+ assertEquals(Some(HoodieTableConfig.TYPE.key -> "COPY_ON_WRITE"),
HoodieOptionConfig.mapToTableConfig(key, "copy_on_write"))
+ }
+ assertEquals(Some(HoodieTableConfig.HIVE_STYLE_PARTITIONING_ENABLE.key ->
"true"),
+
HoodieOptionConfig.mapToTableConfig(HoodieTableConfig.HIVE_STYLE_PARTITIONING_ENABLE.key,
"true"))
+ assertEquals(Some(HoodieTableConfig.BASE_FILE_FORMAT.key -> "PARQUET"),
+ HoodieOptionConfig.mapToTableConfig("hoodie.table.ro.file.format",
"PARQUET"))
+ // Only table-type values are aliases: a field named "cow" must stay a
field name.
+ assertEquals(Some(HoodieTableConfig.RECORDKEY_FIELDS.key -> "cow"),
HoodieOptionConfig.mapToTableConfig("primaryKey", "cow"))
+ }
+
+ @Test
+ def testMapToTableConfigLeavesWriteAndCustomPropertiesMutable(): Unit = {
+ Seq("hoodie.table.services.enabled",
"hoodie.table.services.incremental.enabled",
+ "hoodie.table.service.manager.enabled", "hoodie.table.custom.setting",
"some.key").foreach { key =>
+ assertEquals(None, HoodieOptionConfig.mapToTableConfig(key, "false"),
key)
+ }
+ }
+
+ @Test
+ def testGetTableConfigValueUsesDeclaredAlternativesWithoutDefaults(): Unit =
{
+ val config = new HoodieTableConfig()
+ assertEquals(None, HoodieOptionConfig.getTableConfigValue(config,
HoodieTableConfig.TYPE.key))
+ config.setValue("hoodie.table.precombine.field", "ts")
+ config.setValue("hoodie.table.ro.file.format", "PARQUET")
+ assertEquals(Some("ts"), HoodieOptionConfig.getTableConfigValue(config,
HoodieTableConfig.ORDERING_FIELDS.key))
+ assertEquals(Some("PARQUET"),
HoodieOptionConfig.getTableConfigValue(config,
HoodieTableConfig.BASE_FILE_FORMAT.key))
+ config.setValue(HoodieTableConfig.ORDERING_FIELDS.key, "new_ts")
+ assertEquals(Some("new_ts"),
HoodieOptionConfig.getTableConfigValue(config,
HoodieTableConfig.ORDERING_FIELDS.key))
+ config.setValue(HoodieTableConfig.TYPE.key, "copy_on_write")
+ assertEquals(Some("COPY_ON_WRITE"),
HoodieOptionConfig.getTableConfigValue(config, HoodieTableConfig.TYPE.key))
+ }
+
@Test
def testWithDefaultSqlOptions(): Unit = {
val ops1 = Map("primaryKey" -> "id")
diff --git a/pom.xml b/pom.xml
index 91fba5ec295f..2080f34e42ec 100644
--- a/pom.xml
+++ b/pom.xml
@@ -273,6 +273,8 @@
<vortex.version>0.76.0</vortex.version>
<vortex.spark.artifact>vortex-spark_${scala.binary.version}</vortex.spark.artifact>
<vortex.skip.tests>false</vortex.skip.tests>
+ <!-- Spark-version specific sources of hudi-spark-common: Spark 3.4+ by
default, overridden by the spark3.3 profile -->
+
<hudi.spark.common.version.source>src/main/scala-spark3.4plus</hudi.spark.common.version.source>
<!-- The following properties are only used for Jacoco coverage report
aggregation -->
<copy.files>false</copy.files>
<copy.files.target.dir>${maven.multiModuleProjectDirectory}</copy.files.target.dir>
@@ -2731,6 +2733,7 @@
<lance.skip.tests>true</lance.skip.tests>
<!-- Vortex: Skip tests for Spark 3.3 due to lack of support -->
<vortex.skip.tests>true</vortex.skip.tests>
+
<hudi.spark.common.version.source>src/main/scala-spark3.3</hudi.spark.common.version.source>
<!-- NOTE: Some Hudi modules require standalone Parquet/Orc/etc
file-format dependency (hudi-hive-sync,
hudi-hadoop-mr, for ex). Since these Hudi modules might be
used from w/in the execution engine(s)
bringing these file-formats as dependencies as well, we
need to make sure that versions are
