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 d42148b54638 fix(spark): tolerate metadata column reconciliation in
HoodieCatalog (#20492)
d42148b54638 is described below
commit d42148b546388eaae2f82e1ee2160a3d03b8f05c
Author: Danny Chan <[email protected]>
AuthorDate: Wed Nov 4 19:43:50 2026 +0800
fix(spark): tolerate metadata column reconciliation in HoodieCatalog
(#20492)
* fix(spark): tolerate metadata column reconciliation in HoodieCatalog
* fix(spark): complete catalog schema reconciliation guards
* fix(spark): warn when ignoring catalog nullability changes
---
.../spark/sql/hudi/catalog/HoodieCatalog.scala | 41 ++++++++++++++-
.../sql/hudi/catalog/TestHoodieCatalogDDL.scala | 58 ++++++++++++++++++++++
2 files changed, 97 insertions(+), 2 deletions(-)
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 70a5853c4e73..6ae0f9a91ee2 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
@@ -22,6 +22,7 @@ package org.apache.spark.sql.hudi.catalog
import org.apache.hudi.{DataSourceWriteOptions, SparkAdapterSupport}
import org.apache.hudi.client.common.HoodieSparkEngineContext
import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.model.HoodieRecord
import org.apache.hudi.common.table.HoodieTableMetaClient
import org.apache.hudi.common.table.view.FileSystemViewManager
import org.apache.hudi.common.util.ConfigUtils
@@ -39,7 +40,7 @@ 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, RemoveProperty, SetProperty, UpdateColumnComment,
UpdateColumnType}
+import org.apache.spark.sql.connector.catalog.TableChange.{AddColumn,
ColumnChange, DeleteColumn, RemoveProperty, SetProperty, UpdateColumnComment,
UpdateColumnNullability, 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.{HoodieOptionConfig, HoodieSqlCommonUtils,
ProvidesHoodieConfig}
@@ -48,6 +49,7 @@ import
org.apache.spark.sql.hudi.catalog.HoodieCatalog.{buildPartitionTransforms
import org.apache.spark.sql.hudi.command._
import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException
import org.apache.spark.sql.types.{StructField, StructType}
+import org.slf4j.LoggerFactory
import java.net.URI
import java.util
@@ -229,7 +231,42 @@ class HoodieCatalog extends DelegatingCatalogExtension
case _ => false
}
val catalogPropertyChanges = if (propertyChanges.isEmpty) Seq.empty
else catalogPropertyChangesToApply(table, propertyChanges)
- otherChanges.groupBy(c => c.getClass).foreach {
+ // SDP reconciles its business schema against the catalog's schema on
each refresh.
+ // Hudi owns its metadata columns; deleting them must not change the
stored schema.
+ val schemaChanges = otherChanges.filterNot {
+ case delete: DeleteColumn =>
+ delete.fieldNames().length == 1 &&
HoodieRecord.HOODIE_META_COLUMNS_WITH_OPERATION.contains(delete.fieldNames()(0))
+ // Hudi exposes nullable fields and does not enforce NOT NULL
constraints. SDP
+ // re-sends nullability on refresh; retain the stored schema in
either direction.
+ case change: UpdateColumnNullability =>
+ val exists = table.schema.findNestedField(change.fieldNames(),
includeCollections = true,
+ spark.sessionState.conf.resolver).isDefined
+ if (exists) {
+ LoggerFactory.getLogger(classOf[HoodieCatalog]).warn(
+ s"Ignoring nullability change for
${ident.toString}.${change.fieldNames().mkString(".")}: " +
+ "Hudi preserves the stored schema and does not enforce NOT
NULL constraints.")
+ }
+ exists
+ case _ => false
+ }
+ // Reject unsupported changes before applying any supported change in
the same request.
+ schemaChanges.foreach {
+ case change: UpdateColumnType =>
+ val colName = UnresolvedAttribute(change.fieldNames()).name
+ val field = table.schema.findNestedField(change.fieldNames(),
includeCollections = true,
+ spark.sessionState.conf.resolver).map(_._2).getOrElse {
+ throw new HoodieAnalysisException(s"Couldn't find column
$colName in:\n${table.schema.treeString}")
+ }
+ if (field.dataType != change.newDataType()) {
+ throw new HoodieAnalysisException(
+ "ALTER TABLE CHANGE COLUMN is not supported for changing
column " +
+ s"'${field.name}' with type '${field.dataType}' to
'$colName' with type '${change.newDataType()}'")
+ }
+ case _: AddColumn | _: UpdateColumnComment =>
+ case change =>
+ throw new UnsupportedOperationException(s"not supported table
change: ${change.getClass.getName}")
+ }
+ schemaChanges.groupBy(c => c.getClass).foreach {
case (t, newColumns) if t == classOf[AddColumn] =>
AlterHoodieTableAddColumnsCommand(
tableIdent,
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 68adad112b72..211028bd6241 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
@@ -18,6 +18,7 @@
package org.apache.spark.sql.hudi.catalog
import org.apache.hudi.HoodieSparkUtils
+import org.apache.hudi.common.model.HoodieRecord
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
@@ -108,6 +109,63 @@ class TestHoodieCatalogDDL extends HoodieSparkSqlTestBase {
}
}
+ Seq(false, true).foreach { operationMetadata =>
+ test(s"alterTable preserves metadata columns during repeated schema
reconciliation, operation=$operationMetadata") {
+ withSQLConf("hoodie.schema.on.read.enable" -> "true") {
+ withTempDir { tmp =>
+ val catalog = hoodieCatalog
+ val tableName = generateTableName
+ val ident = Identifier.of(Array("default"), tableName)
+ catalog.createTable(ident, StructType(Seq(StructField("id",
IntegerType, nullable = false))),
+ Array.empty[Transform], Map("provider" -> "hudi", "primaryKey" ->
"id",
+ "hoodie.allow.operation.metadata.field" ->
operationMetadata.toString,
+ "location" -> s"${tmp.getCanonicalPath}/$tableName").asJava)
+ spark.sql("select 1 as id").write.format("hudi")
+ .option("hoodie.table.name", tableName)
+ .option("hoodie.datasource.write.recordkey.field", "id")
+ .option("hoodie.allow.operation.metadata.field",
operationMetadata.toString)
+ .option("hoodie.datasource.write.operation", "upsert")
+ .mode("append").save(s"${tmp.getCanonicalPath}/$tableName")
+ val schemaBefore = catalog.loadTable(ident).schema()
+ assertEquals(operationMetadata,
schemaBefore.fieldNames.contains(HoodieRecord.OPERATION_METADATA_FIELD))
+ val deletes =
schemaBefore.fieldNames.filter(_.startsWith("_hoodie_")).toSeq.map { name =>
+ TableChange.deleteColumn(Array(name), false)
+ }
+ (1 to 3).foreach { _ =>
+ catalog.alterTable(ident, (deletes ++
Seq(TableChange.setProperty("primaryKey", "id"),
+ TableChange.updateColumnNullability(Array("id"), false))): _*)
+ catalog.alterTable(ident,
TableChange.updateColumnNullability(Array("id"), true))
+ assertEquals(schemaBefore, catalog.loadTable(ident).schema())
+ }
+
+ val unsupported = Seq(
+ TableChange.deleteColumn(Array("id"), false),
+ TableChange.deleteColumn(Array("_hoodie_custom"), false),
+ TableChange.deleteColumn(Array("nested", "_hoodie_record_key"),
false),
+ TableChange.renameColumn(Array("id"), "renamed"),
+ TableChange.updateColumnNullability(Array("missing"), false),
+ TableChange.updateColumnPosition(Array("id"),
TableChange.ColumnPosition.first()))
+ unsupported.foreach { change =>
+ val error = intercept[UnsupportedOperationException] {
+ catalog.alterTable(ident, TableChange.addColumn(Array("added"),
StringType), change)
+ }
+ assertTrue(error.getMessage.contains(change.getClass.getName),
error.getMessage)
+ assertEquals(schemaBefore, catalog.loadTable(ident).schema())
+ }
+ Seq(true, false).foreach { addFirst =>
+ val add = TableChange.addColumn(Array("added"), StringType)
+ val widen = TableChange.updateColumnType(Array("id"), LongType)
+ intercept[HoodieAnalysisException] {
+ catalog.alterTable(ident, (if (addFirst) Seq(add, widen) else
Seq(widen, add)): _*)
+ }
+ assertEquals(schemaBefore, catalog.loadTable(ident).schema())
+ }
+ catalog.dropTable(ident)
+ }
+ }
+ }
+ }
+
test("alterTable ignores re-sent table properties and rejects changes to
persisted table configs") {
withTempDir { tmp =>
val catalog = hoodieCatalog
