This is an automated email from the ASF dual-hosted git repository.
voonhous 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 6fd80965527f fix(spark): make SparkCatalogMetaStoreClient sync-safe
and cover the Spark catalog DDL paths (#19162)
6fd80965527f is described below
commit 6fd80965527f4a5c8476ac188a52d8d293c22807
Author: Y Ethan Guo <[email protected]>
AuthorDate: Sat Aug 29 03:47:31 2026 -0700
fix(spark): make SparkCatalogMetaStoreClient sync-safe and cover the Spark
catalog DDL paths (#19162)
Production (SparkCatalogMetaStoreClient, the IMetaStoreClient used when
hoodie.datasource.hive_sync.use_spark_catalog=true):
- getSchema returned every partition column twice (CatalogTable.schema
already
contains them) and rendered column comments as "Some(x)"/"None";
HoodieHiveSyncClient.updateTableComments then threw "Duplicate key" for
any
partitioned table with sync_comment=true.
- fromCatalogTable dropped CatalogTable.comment, so a "comment" table
property never
round-tripped and updateTableProperties re-altered the table on every
sync.
- close() threw UnsupportedOperationException although
HoodieHiveSyncClient.close
calls it on every sync; it is now a no-op.
- getPartition(db, table, values) threw UnsupportedOperationException
although
HMSDDLExecutor.dropPartitionsToTable reaches it through
HivePartitionUtil.partitionExists, which only handles
NoSuchObjectException; it is
now implemented on ExternalCatalog.getPartitionOption.
- alter_table forwarded to HiveExternalCatalog.alterTable, which keeps the
stored
schema by design, so a column added by
HMSDDLExecutor.updateTableDefinition never
reached the catalog. It now merges the synced columns conservatively and
writes
the result through alterTableDataSchema: new columns are appended and
non-empty
comments applied, while an existing column keeps its type, nullability
and field
metadata (the catalog holds the logical Spark schema, including the
hudi_type
markers for VECTOR/BLOB and the original VariantType, whereas the sync
only speaks
Hive type strings). The merged schema is also handed to alterTable itself
so the
behaviour holds under InMemoryCatalog. toCatalogTable carries column
comments
across so the getTable -> alter_table round trips do not erase them.
- alter_table pinned the identifier to the incoming name, so the rename
step of
HoodieHiveSyncClient.createOrReplaceTable (force-recreate,
recreate-on-error,
base-path change) dropped the real table and left only <table>_temp; a
Table
carrying a different name now goes through ExternalCatalog.renameTable
first.
- HoodieCatalog.alterTable's unsupported-change message printed
"class java.lang.Class" instead of the change class.
Tests:
- TestHoodieCatalogDDL: HoodieCatalog create/load/alter/rename/drop through
the V2
TableCatalog API (external location survives drop, unsupported change
rejected),
SHOW CREATE TABLE through ShowHoodieCreateTableCommand (under
spark.sql.legacy.useV1Command, the only route Spark gives that command),
existing-location config conflict, HoodieInternalV2Table capabilities and
the
V2-to-V1 relation conversion for inserts under schema-on-read.
- TestSparkCatalogMetaStoreClient: pins the merge contract (comment lands,
property-
only alter keeps comment and schema, metadata and catalog type survive a
retype
through the alterTableDataSchema write, comment-less column list keeps
comments,
shorter list drops nothing, rename round trip), getPartition hit and
miss, close
and setMetaConf no-ops, the warehouse fallback, and the
UnsupportedOperationException
contract for the rest of IMetaStoreClient. Helpers register input/output
formats
and a serde like HMSDDLExecutor, which the pre-existing tests needed to
run at all.
- TestSparkCatalogSync: forced table recreation through HiveSyncTool
against the
in-memory catalog.
- TestHoodieSqlCommonUtils: partition-style detection incl. multi-column
tables,
config filtering, meta fields, partition-spec normalization, path
qualification.
- TestCreateTable: the illegal-CTAS cleanup test also covers an explicit
location.
CI: org.apache.spark.sql.hive is added to the scalatest wildcard lists in
bot.yml and
azure-pipelines-20230430.yml; TestSparkCatalogMetaStoreClient had never
executed in
CI and three of its tests had been failing on master unnoticed.
Not covered: HoodieStagedTable. SQL CTAS on the session catalog is
rewritten to the
V1 CreateHoodieTableAsSelectCommand, so the staged path is only reachable
when
HoodieCatalog is registered under a non-session catalog name.
---------
Co-authored-by: voon <[email protected]>
---
.github/workflows/bot.yml | 2 +-
azure-pipelines-20230430.yml | 1 +
.../sql/hive/SparkCatalogMetaStoreClient.scala | 91 ++++++-
.../spark/sql/hudi/catalog/HoodieCatalog.scala | 2 +-
.../sql/hive/TestSparkCatalogMetaStoreClient.scala | 291 ++++++++++++++++++++-
.../sql/hudi/catalog/TestHoodieCatalogDDL.scala | 209 +++++++++++++++
.../sql/hudi/common/TestHoodieSqlCommonUtils.scala | 129 +++++++++
.../spark/sql/hudi/ddl/TestCreateTable.scala | 19 ++
.../spark/sql/hudi/ddl/TestSparkCatalogSync.scala | 47 +++-
9 files changed, 777 insertions(+), 14 deletions(-)
diff --git a/.github/workflows/bot.yml b/.github/workflows/bot.yml
index 1025838cf8b6..dceae0761bfb 100644
--- a/.github/workflows/bot.yml
+++ b/.github/workflows/bot.yml
@@ -27,7 +27,7 @@ env:
JAVA_UT_FILTER1:
-Dtest=!TestCOWDataSource,!TestMORDataSource,!TestHoodieFileSystemViews
JAVA_UT_FILTER2:
-Dtest=TestCOWDataSource,TestMORDataSource,TestHoodieFileSystemViews
SCALA_TEST_DML_FILTER: -DwildcardSuites=org.apache.spark.sql.hudi.dml
- SCALA_TEST_OTHERS_FILTER:
-DwildcardSuites=org.apache.hudi,org.apache.spark.hudi,org.apache.spark.sql.avro,org.apache.spark.sql.execution,org.apache.spark.sql.hudi.analysis,org.apache.spark.sql.hudi.blob,org.apache.spark.sql.hudi.catalog,org.apache.spark.sql.hudi.command,org.apache.spark.sql.hudi.common,org.apache.spark.sql.hudi.ddl,org.apache.spark.sql.hudi.procedure,org.apache.spark.sql.hudi.feature
+ SCALA_TEST_OTHERS_FILTER:
-DwildcardSuites=org.apache.hudi,org.apache.spark.hudi,org.apache.spark.sql.avro,org.apache.spark.sql.execution,org.apache.spark.sql.hive,org.apache.spark.sql.hudi.analysis,org.apache.spark.sql.hudi.blob,org.apache.spark.sql.hudi.catalog,org.apache.spark.sql.hudi.command,org.apache.spark.sql.hudi.common,org.apache.spark.sql.hudi.ddl,org.apache.spark.sql.hudi.procedure,org.apache.spark.sql.hudi.feature
FLINK_IT_FILTER1: -Dit.test=ITTestHoodieDataSource
FLINK_IT_FILTER2: -Dit.test=!ITTestHoodieDataSource
diff --git a/azure-pipelines-20230430.yml b/azure-pipelines-20230430.yml
index cf9b84638edc..b863f65bb1c5 100644
--- a/azure-pipelines-20230430.yml
+++ b/azure-pipelines-20230430.yml
@@ -74,6 +74,7 @@ parameters:
- 'org.apache.spark.hudi'
- 'org.apache.spark.sql.avro'
- 'org.apache.spark.sql.execution'
+ - 'org.apache.spark.sql.hive'
- 'org.apache.spark.sql.hudi.analysis'
- 'org.apache.spark.sql.hudi.blob'
- 'org.apache.spark.sql.hudi.catalog'
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hive/SparkCatalogMetaStoreClient.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hive/SparkCatalogMetaStoreClient.scala
index 9fa225eeb876..254751b7e7ca 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hive/SparkCatalogMetaStoreClient.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hive/SparkCatalogMetaStoreClient.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.hive
import org.apache.hudi.hive.HiveSyncConfig
import org.apache.hadoop.hive.metastore.IMetaStoreClient
-import org.apache.hadoop.hive.metastore.api.{Database, EnvironmentContext,
FieldSchema, Partition, SerDeInfo, StorageDescriptor, Table}
+import org.apache.hadoop.hive.metastore.api.{Database, EnvironmentContext,
FieldSchema, MetaException, NoSuchObjectException, Partition, SerDeInfo,
StorageDescriptor, Table}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase,
CatalogStorageFormat, CatalogTable, CatalogTablePartition, CatalogTableType}
@@ -65,8 +65,48 @@ class SparkCatalogMetaStoreClient(syncConfig: HiveSyncConfig)
// scalastyle:off method.name
override def alter_table(dbName: String, tableName: String, table: Table):
Unit = {
- val updated = toCatalogTable(table).copy(identifier =
TableIdentifier(tableName, Some(dbName)))
- externalCatalog.alterTable(updated)
+ val current = externalCatalog.getTable(dbName, tableName)
+ // HoodieHiveSyncClient.createOrReplaceTable renames its temp table by
altering it under the
+ // final name, after it has already dropped the real table; Spark's
ExternalCatalog only does
+ // that through renameTable. It runs first so a failed alter still leaves
a usable table under
+ // the final name. Hive lower-cases table names, so a differently-cased
configured name is not
+ // a rename.
+ val targetName =
Option(table.getTableName).filter(_.nonEmpty).getOrElse(tableName)
+ if (!targetName.equalsIgnoreCase(tableName)) {
+ externalCatalog.renameTable(dbName, tableName, targetName)
+ }
+ // HiveExternalCatalog.alterTable deliberately keeps the existing schema
(Spark routes schema
+ // changes through alterTableDataSchema) while InMemoryCatalog replaces it
wholesale, so the
+ // merged schema is handed to both calls. The merge is deliberately
conservative: the catalog
+ // holds the logical Spark schema (nullability, field metadata such as the
VECTOR/BLOB type
+ // markers, the original VariantType) while the sync only speaks Hive type
strings, so an
+ // existing column keeps its type, nullability and metadata and is never
dropped; only new
+ // columns are appended and non-empty comments applied. Property-only
alters therefore leave
+ // the schema untouched.
+ val converted = toCatalogTable(table)
+ val merged = mergeDataSchema(current.dataSchema, converted.dataSchema)
+ val incoming = converted.copy(
+ identifier = TableIdentifier(targetName, Some(dbName)),
+ schema = StructType(merged ++ current.partitionSchema),
+ partitionColumnNames = current.partitionColumnNames)
+ externalCatalog.alterTable(incoming)
+ if (merged != current.dataSchema) {
+ externalCatalog.alterTableDataSchema(dbName, targetName, merged)
+ }
+ }
+
+ private def mergeDataSchema(current: StructType, incoming: StructType):
StructType = {
+ def key(name: String): String = name.toLowerCase(util.Locale.ROOT)
+ val incomingByName = incoming.fields.map(f => key(f.name) -> f).toMap
+ val kept = current.fields.map { existing =>
+ incomingByName.get(key(existing.name))
+ .flatMap(_.getComment().filter(_.nonEmpty))
+ .map(existing.withComment)
+ .getOrElse(existing)
+ }
+ val currentNames = current.fields.map(f => key(f.name)).toSet
+ val added = incoming.fields.filterNot(f =>
currentNames.contains(key(f.name)))
+ StructType(kept ++ added)
}
override def alter_table_with_environmentContext(dbName: String,
@@ -90,6 +130,25 @@ class SparkCatalogMetaStoreClient(syncConfig:
HiveSyncConfig)
listPartitions(dbName, tableName, max)
}
+ /**
+ * Looks up a single partition by its ordered values.
HivePartitionUtil.partitionExists relies on
+ * the Hive contract here: a missing partition surfaces as
NoSuchObjectException, not as null.
+ */
+ override def getPartition(dbName: String, tableName: String, values:
util.List[String]): Partition = {
+ val catalogTable = externalCatalog.getTable(dbName, tableName)
+ val partitionKeys = catalogTable.partitionColumnNames.toList
+ val partitionValues = Option(values).map(_.asScala.toList).getOrElse(Nil)
+ if (partitionValues.size != partitionKeys.size) {
+ val keys = partitionKeys.mkString(",")
+ throw new MetaException(
+ s"Expected ${partitionKeys.size} partition value(s) [$keys] for
$dbName.$tableName but got ${partitionValues.size}")
+ }
+ val spec = partitionKeys.zip(partitionValues).toMap
+ externalCatalog.getPartitionOption(dbName, tableName, spec)
+ .map(fromCatalogPartition(_, dbName, tableName, partitionKeys))
+ .getOrElse(throw new NoSuchObjectException(s"Partition $spec of
$dbName.$tableName does not exist"))
+ }
+
override def add_partitions(parts: util.List[Partition], ifNotExists:
Boolean, needResults: Boolean): util.List[Partition] = {
if (parts == null || parts.isEmpty) {
new util.ArrayList[Partition]()
@@ -128,8 +187,11 @@ class SparkCatalogMetaStoreClient(syncConfig:
HiveSyncConfig)
override def getSchema(dbName: String, tableName: String):
util.List[FieldSchema] = {
val table = externalCatalog.getTable(dbName, tableName)
- val cols = table.schema.fields.map { f =>
- new FieldSchema(f.name, f.dataType.catalogString,
Option(f.getComment()).map(_.toString).getOrElse(""))
+ // CatalogTable.schema already carries the partition columns, so they have
to be filtered out
+ // here or every partition column would be reported twice. Hive's
getSchema returns the data
+ // columns first and the partition columns last, which is the order
reproduced below.
+ val cols = table.schema.fields.filterNot(f =>
table.partitionColumnNames.contains(f.name)).map { f =>
+ new FieldSchema(f.name, f.dataType.catalogString,
f.getComment().getOrElse(""))
}
val partitionCols = table.partitionColumnNames.map { name =>
val dt = table.partitionSchema.fields.find(_.name ==
name).map(_.dataType.catalogString).getOrElse("string")
@@ -151,7 +213,9 @@ class SparkCatalogMetaStoreClient(syncConfig:
HiveSyncConfig)
override def setHiveAddedJars(arg0: String): Unit = unsupported[Unit]()
override def isLocalMetaStore(): Boolean = unsupported[Boolean]()
override def reconnect(): Unit = unsupported[Unit]()
- override def close(): Unit = unsupported[Unit]()
+ // close is a no-op: HoodieHiveSyncClient.close() calls it on every sync and
there is no
+ // connection to release, the Spark session outlives this client.
+ override def close(): Unit = {}
// setMetaConf is no-op: HoodieHiveSyncClient.setMetaConf forwards
// hive.metastore.callerContext.* values to the metastore for audit/tracing.
With Spark's
// external catalog there is no remote HMS to forward to, so accept the call
silently
@@ -176,7 +240,6 @@ class SparkCatalogMetaStoreClient(syncConfig:
HiveSyncConfig)
override def add_partition(arg0:
org.apache.hadoop.hive.metastore.api.Partition):
org.apache.hadoop.hive.metastore.api.Partition =
unsupported[org.apache.hadoop.hive.metastore.api.Partition]()
override def add_partitions(arg0:
java.util.List[org.apache.hadoop.hive.metastore.api.Partition]): Int =
unsupported[Int]()
override def add_partitions_pspec(arg0:
org.apache.hadoop.hive.metastore.partition.spec.PartitionSpecProxy): Int =
unsupported[Int]()
- override def getPartition(arg0: String, arg1: String, arg2:
java.util.List[String]): org.apache.hadoop.hive.metastore.api.Partition =
unsupported[org.apache.hadoop.hive.metastore.api.Partition]()
override def exchange_partition(arg0: java.util.Map[String, String], arg1:
String, arg2: String, arg3: String, arg4: String):
org.apache.hadoop.hive.metastore.api.Partition =
unsupported[org.apache.hadoop.hive.metastore.api.Partition]()
override def exchange_partitions(arg0: java.util.Map[String, String], arg1:
String, arg2: String, arg3: String, arg4: String):
java.util.List[org.apache.hadoop.hive.metastore.api.Partition] =
unsupported[java.util.List[org.apache.hadoop.hive.metastore.api.Partition]]()
override def getPartition(arg0: String, arg1: String, arg2: String):
org.apache.hadoop.hive.metastore.api.Partition =
unsupported[org.apache.hadoop.hive.metastore.api.Partition]()
@@ -303,8 +366,14 @@ class SparkCatalogMetaStoreClient(syncConfig:
HiveSyncConfig)
val cols =
Option(table.getSd).map(_.getCols).map(_.asScala.toList).getOrElse(Nil)
val partCols =
Option(table.getPartitionKeys).map(_.asScala.toList).getOrElse(Nil)
- val dataFields = cols.map(fs => StructField(fs.getName,
CatalystSqlParser.parseDataType(fs.getType), nullable = true, Metadata.empty))
- val partitionFields = partCols.map(fs => StructField(fs.getName,
CatalystSqlParser.parseDataType(fs.getType), nullable = true, Metadata.empty))
+ // Carry the Hive column comment across; fromCatalogTable emits it, and
since alter_table now
+ // writes the data schema back, dropping it here would erase comments on
every sync round trip.
+ def toField(fs: FieldSchema): StructField = {
+ val field = StructField(fs.getName,
CatalystSqlParser.parseDataType(fs.getType), nullable = true, Metadata.empty)
+
Option(fs.getComment).filter(_.nonEmpty).map(field.withComment).getOrElse(field)
+ }
+ val dataFields = cols.map(toField)
+ val partitionFields = partCols.map(toField)
// Strip "spark.sql.*" properties before handing off to Spark's external
catalog.
// HiveExternalCatalog.alterTable / createTable rejects such keys ("Cannot
persist ...
@@ -343,6 +412,10 @@ class SparkCatalogMetaStoreClient(syncConfig:
HiveSyncConfig)
t.setTableName(table.identifier.table)
t.setTableType(if (table.tableType == CatalogTableType.EXTERNAL)
"EXTERNAL_TABLE" else "MANAGED_TABLE")
t.setParameters(new util.HashMap[String, String](table.properties.asJava))
+ // Spark moves the "comment" table property into the dedicated
CatalogTable.comment field when it
+ // reads a table back (HiveClientImpl excludes it from
CatalogTable.properties), so it has to be
+ // put back here or a comment written through createTable/alter_table
would be lost on getTable.
+ table.comment.foreach(c => t.putToParameters("comment", c))
val nonPartitionFields = table.schema.fields.filterNot(f =>
table.partitionColumnNames.contains(f.name))
val cols = nonPartitionFields.map(f => new FieldSchema(f.name,
f.dataType.catalogString, f.getComment().orNull)).toList.asJava
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 f5ef66a04b92..2810fcdbcb14 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
@@ -253,7 +253,7 @@ class HoodieCatalog extends DelegatingCatalogExtension
AlterHoodieTableChangeColumnCommand(tableIdent, colName,
field.withComment(newComment)).run(spark)
}
case (t, _) =>
- throw new UnsupportedOperationException(s"not supported table
change: ${t.getClass}")
+ throw new UnsupportedOperationException(s"not supported table
change: ${t.getName}")
}
loadTable(ident)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hive/TestSparkCatalogMetaStoreClient.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hive/TestSparkCatalogMetaStoreClient.scala
index 93b2be295e3d..1e6dc3ce21d3 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hive/TestSparkCatalogMetaStoreClient.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hive/TestSparkCatalogMetaStoreClient.scala
@@ -18,11 +18,19 @@
package org.apache.spark.sql.hive
import org.apache.hudi.common.config.TypedProperties
+import org.apache.hudi.hadoop.HoodieParquetInputFormat
import org.apache.hudi.hive.HiveSyncConfig
import org.apache.hudi.testutils.HoodieClientTestUtils.getSparkConfForTest
-import org.apache.hadoop.hive.metastore.api.{Database, EnvironmentContext,
FieldSchema, Partition, SerDeInfo, StorageDescriptor, Table}
+import org.apache.hadoop.hive.common.ObjectPair
+import org.apache.hadoop.hive.conf.HiveConf
+import org.apache.hadoop.hive.metastore.{IMetaStoreClient,
PartitionDropOptions, TableType}
+import org.apache.hadoop.hive.metastore.api.{ColumnStatistics, CompactionType,
Database, DataOperationType, EnvironmentContext, FieldSchema, FireEventRequest,
ForeignKeysRequest, Function, GetPrincipalsInRoleRequest,
GetRoleGrantsForPrincipalRequest, HiveObjectRef, Index, LockRequest,
NoSuchObjectException, Partition, PartitionEventType, PartitionValuesRequest,
PrimaryKeysRequest, PrincipalType, PrivilegeBag, Role, SerDeInfo,
SetPartitionsStatsRequest, ShowLocksRequest, SQLForeignKey, SQ [...]
+import org.apache.hadoop.hive.metastore.partition.spec.PartitionSpecProxy
+import org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat
+import org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe
import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.types.{MetadataBuilder, StructType}
import org.apache.spark.util.Utils
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse,
assertNotNull, assertTrue}
import org.scalactic.source
@@ -125,6 +133,74 @@ class TestSparkCatalogMetaStoreClient extends FunSuite
with BeforeAndAfterAll {
client.alter_table_with_environmentContext(databaseName, tableName,
environmentAlteredTable, new EnvironmentContext())
assertEquals("env-context", client.getTable(databaseName,
tableName).getParameters.get("comment"))
+
+ // A column comment set through alter_table (the updateTableComments
path) must land in the
+ // catalog, and a property-only alter afterwards must keep both the
comment and the schema.
+ val commented = client.getTable(databaseName, tableName)
+ commented.getSd.getCols.asScala.find(_.getName ==
"name").get.setComment("the name column")
+ client.alter_table(databaseName, tableName, commented)
+ assertEquals("the name column", client.getSchema(databaseName,
tableName).asScala.find(_.getName == "name").get.getComment)
+
+ val propertyOnly = client.getTable(databaseName, tableName)
+ propertyOnly.putToParameters("comment", "v4")
+ client.alter_table(databaseName, tableName, propertyOnly)
+ assertEquals("v4", client.getTable(databaseName,
tableName).getParameters.get("comment"))
+ assertEquals(Seq("id", "name", "age", "dt"),
client.getSchema(databaseName, tableName).asScala.map(_.getName).toSeq)
+ assertEquals("the name column", client.getSchema(databaseName,
tableName).asScala.find(_.getName == "name").get.getComment)
+
+ // The sync only speaks Hive type strings, so a retyped existing column
must not overwrite the
+ // logical Spark type or the field metadata held by the catalog (the
VECTOR/BLOB type markers
+ // live there); only new columns and comments are written back.
+ val externalCatalog = spark.sessionState.catalog.externalCatalog
+ val markedFields = externalCatalog.getTable(databaseName,
tableName).dataSchema.fields.map {
+ case f if f.name == "name" =>
+ f.copy(metadata = new
MetadataBuilder().withMetadata(f.metadata).putString("hudi_type",
"VECTOR").build())
+ case f => f
+ }
+ externalCatalog.alterTableDataSchema(databaseName, tableName,
StructType(markedFields))
+ // Adding a column at the same time forces the alterTableDataSchema
write, which is the one
+ // call that could erase the marker if the merge did not keep the
existing field.
+ val retyped = client.getTable(databaseName, tableName)
+ retyped.getSd.getCols.asScala.find(_.getName ==
"name").get.setType("binary")
+ val retypedCols = new util.ArrayList[FieldSchema](retyped.getSd.getCols)
+ retypedCols.add(fieldSchema("extra", "int"))
+ retyped.getSd.setCols(retypedCols)
+ client.alter_table(databaseName, tableName, retyped)
+ assertEquals(Seq("id", "name", "age", "extra", "dt"),
client.getSchema(databaseName, tableName).asScala.map(_.getName).toSeq)
+ assertEquals("string", client.getSchema(databaseName,
tableName).asScala.find(_.getName == "name").get.getType)
+ val nameField = externalCatalog.getTable(databaseName,
tableName).dataSchema("name")
+ assertEquals("VECTOR", nameField.metadata.getString("hudi_type"))
+ assertEquals(Some("the name column"), nameField.getComment())
+
+ // updateTableDefinition sends every column with an empty comment
(HiveSchemaUtil), which must
+ // not erase the comments already in the catalog.
+ client.alter_table(databaseName, tableName, newTable(
+ databaseName,
+ tableName,
+ new File(tmp, s"${tableName}_v3").toURI.toString,
+ Seq("id" -> "int", "name" -> "string", "age" -> "int"),
+ Seq("dt" -> "string")))
+ assertEquals("the name column", client.getSchema(databaseName,
tableName).asScala.find(_.getName == "name").get.getComment)
+
+ // A shorter column list must not drop columns from the catalog either.
+ client.alter_table(databaseName, tableName, newTable(
+ databaseName,
+ tableName,
+ new File(tmp, s"${tableName}_v3").toURI.toString,
+ Seq("id" -> "int"),
+ Seq("dt" -> "string")))
+ assertEquals(Seq("id", "name", "age", "extra", "dt"),
client.getSchema(databaseName, tableName).asScala.map(_.getName).toSeq)
+
+ // HoodieHiveSyncClient.createOrReplaceTable renames its temp table by
altering it under the
+ // final name, which needs an explicit ExternalCatalog.renameTable.
+ val renamedName = s"${tableName}_renamed"
+ val renamed = client.getTable(databaseName, tableName)
+ renamed.setTableName(renamedName)
+ client.alter_table(databaseName, tableName, renamed)
+ assertFalse(client.tableExists(databaseName, tableName))
+ assertTrue(client.tableExists(databaseName, renamedName))
+ assertEquals(Seq("id", "name", "age", "extra", "dt"),
client.getSchema(databaseName, renamedName).asScala.map(_.getName).toSeq)
+ assertEquals("VECTOR", externalCatalog.getTable(databaseName,
renamedName).dataSchema("name").metadata.getString("hudi_type"))
}
}
@@ -148,6 +224,11 @@ class TestSparkCatalogMetaStoreClient extends FunSuite
with BeforeAndAfterAll {
val added = client.add_partitions(util.Arrays.asList(partitionOne,
partitionTwo), false, true)
assertEquals(2, added.size())
+ // HivePartitionUtil.partitionExists calls getPartition and treats
NoSuchObjectException as
+ // "absent", so a missing partition must raise that instead of returning
null.
+ assertEquals("2024-01-01", client.getPartition(databaseName, tableName,
util.Collections.singletonList("2024-01-01")).getValues.get(0))
+ assertThrows[NoSuchObjectException](client.getPartition(databaseName,
tableName, util.Collections.singletonList("2099-01-01")))
+
val listedPartitions = client.listPartitions(databaseName, tableName,
(-1).toShort).asScala.toSeq
assertEquals(Set("2024-01-01", "2024-01-02"),
listedPartitions.map(_.getValues.get(0)).toSet)
assertNotNull(listedPartitions.find(_.getValues.get(0) ==
"2024-01-02").orNull)
@@ -203,6 +284,206 @@ class TestSparkCatalogMetaStoreClient extends FunSuite
with BeforeAndAfterAll {
}
}
+ test("supported client edge cases: empty partitions, no-op setMetaConf,
default database location") {
+ val client = newClient()
+
+ // add_partitions with an empty list is a no-op that returns an empty
result.
+ assertTrue(client.add_partitions(new util.ArrayList[Partition](), false,
true).isEmpty)
+
+ // setMetaConf is intentionally a silent no-op; HoodieHiveSyncClient
forwards caller-context
+ // values through it on every sync and there is no remote metastore to
receive them.
+ client.setMetaConf("hive.metastore.callerContext", "hudi")
+
+ // close must not throw either: HoodieHiveSyncClient.close() calls it on
every sync.
+ client.close()
+
+ // createDatabase without an explicit location falls back to the warehouse
path.
+ val databaseName = generateName("db")
+ client.createDatabase(new Database(databaseName, "no-location db", null,
new util.HashMap[String, String]()))
+ val locationUri = client.getDatabase(databaseName).getLocationUri
+ assertTrue(locationUri.contains(warehouseDir.getCanonicalPath),
locationUri)
+ assertFalse(client.tableExists(databaseName, "missing_table"))
+ }
+
+ test("unsupported IMetaStoreClient operations throw
UnsupportedOperationException") {
+ // SparkCatalogMetaStoreClient only implements the subset of
IMetaStoreClient exercised by
+ // HoodieHiveSyncClient/HMSDDLExecutor, plus close and setMetaConf which
those callers invoke
+ // unconditionally and which are deliberate no-ops. Every method outside
that subset must fail
+ // fast rather than return a misleading default. This locks in that
contract across the
+ // delegated surface.
+ val client = newClient()
+
+ // Connection / config lifecycle.
+ assertUnsupported(client.isCompatibleWith(null: HiveConf))
+ assertUnsupported(client.isSameConfObj(null: HiveConf))
+ assertUnsupported(client.setHiveAddedJars(null: String))
+ assertUnsupported(client.isLocalMetaStore())
+ assertUnsupported(client.reconnect())
+ assertUnsupported(client.getMetaConf(null: String))
+ assertUnsupported(client.flushCache())
+
+ // Databases.
+ assertUnsupported(client.getDatabases(null: String))
+ assertUnsupported(client.getAllDatabases())
+ assertUnsupported(client.dropDatabase(null: String))
+ assertUnsupported(client.dropDatabase(null: String, false, false))
+ assertUnsupported(client.dropDatabase(null: String, false, false, false))
+ assertUnsupported(client.alterDatabase(null: String, null: Database))
+
+ // Tables.
+ assertUnsupported(client.getTables(null: String, null: String))
+ assertUnsupported(client.getTables(null: String, null: String, null:
TableType))
+ assertUnsupported(client.getTableMeta(null: String, null: String, null:
util.List[String]))
+ assertUnsupported(client.getAllTables(null: String))
+ assertUnsupported(client.listTableNamesByFilter(null: String, null:
String, 0.toShort))
+ assertUnsupported(client.dropTable(null: String, null: String, false,
false))
+ assertUnsupported(client.dropTable(null: String, null: String, false,
false, false))
+ assertUnsupported(client.dropTable(null: String, false))
+ assertUnsupported(client.tableExists(null: String))
+ assertUnsupported(client.getTable(null: String))
+ assertUnsupported(client.getTableObjectsByName(null: String, null:
util.List[String]))
+ assertUnsupported(client.getFields(null: String, null: String))
+ assertUnsupported(client.insertTable(null: Table, false))
+
+ // Partitions.
+ assertUnsupported(client.appendPartition(null: String, null: String, null:
util.List[String]))
+ assertUnsupported(client.appendPartition(null: String, null: String, null:
String))
+ assertUnsupported(client.add_partition(null: Partition))
+ assertUnsupported(client.add_partitions(null: util.List[Partition]))
+ assertUnsupported(client.add_partitions_pspec(null: PartitionSpecProxy))
+ assertUnsupported(client.getPartition(null: String, null: String, null:
String))
+ assertUnsupported(client.getPartitionWithAuthInfo(null: String, null:
String, null: util.List[String], null: String, null: util.List[String]))
+ assertUnsupported(client.exchange_partition(null: util.Map[String,
String], null: String, null: String, null: String, null: String))
+ assertUnsupported(client.exchange_partitions(null: util.Map[String,
String], null: String, null: String, null: String, null: String))
+ assertUnsupported(client.listPartitionSpecs(null: String, null: String, 0))
+ assertUnsupported(client.listPartitions(null: String, null: String, null:
util.List[String], 0.toShort))
+ assertUnsupported(client.listPartitionNames(null: String, null: String,
0.toShort))
+ assertUnsupported(client.listPartitionNames(null: String, null: String,
null: util.List[String], 0.toShort))
+ assertUnsupported(client.listPartitionValues(null: PartitionValuesRequest))
+ assertUnsupported(client.getNumPartitionsByFilter(null: String, null:
String, null: String))
+ assertUnsupported(client.listPartitionSpecsByFilter(null: String, null:
String, null: String, 0))
+ assertUnsupported(client.listPartitionsByExpr(null: String, null: String,
null: Array[Byte], null: String, 0.toShort, null: util.List[Partition]))
+ assertUnsupported(client.listPartitionsWithAuthInfo(null: String, null:
String, 0.toShort, null: String, null: util.List[String]))
+ assertUnsupported(client.listPartitionsWithAuthInfo(null: String, null:
String, null: util.List[String], 0.toShort, null: String, null:
util.List[String]))
+ assertUnsupported(client.getPartitionsByNames(null: String, null: String,
null: util.List[String]))
+ assertUnsupported(client.markPartitionForEvent(null: String, null: String,
null: util.Map[String, String], null: PartitionEventType))
+ assertUnsupported(client.isPartitionMarkedForEvent(null: String, null:
String, null: util.Map[String, String], null: PartitionEventType))
+ assertUnsupported(client.validatePartitionNameCharacters(null:
util.List[String]))
+ assertUnsupported(client.alter_table(null: String, null: String, null:
Table, false))
+ assertUnsupported(client.dropPartition(null: String, null: String, null:
util.List[String], false))
+ assertUnsupported(client.dropPartition(null: String, null: String, null:
util.List[String], null: PartitionDropOptions))
+ assertUnsupported(client.dropPartitions(null: String, null: String, null:
util.List[ObjectPair[java.lang.Integer, Array[Byte]]], false, false))
+ assertUnsupported(client.dropPartitions(null: String, null: String, null:
util.List[ObjectPair[java.lang.Integer, Array[Byte]]], false, false, false))
+ assertUnsupported(client.dropPartitions(null: String, null: String, null:
util.List[ObjectPair[java.lang.Integer, Array[Byte]]], null:
PartitionDropOptions))
+ assertUnsupported(client.alter_partition(null: String, null: String, null:
Partition))
+ assertUnsupported(client.alter_partition(null: String, null: String, null:
Partition, null: EnvironmentContext))
+ assertUnsupported(client.alter_partitions(null: String, null: String,
null: util.List[Partition]))
+ assertUnsupported(client.renamePartition(null: String, null: String, null:
util.List[String], null: Partition))
+ assertUnsupported(client.partitionNameToVals(null: String))
+ assertUnsupported(client.partitionNameToSpec(null: String))
+ assertUnsupported(client.getConfigValue(null: String, null: String))
+
+ // Indexes.
+ assertUnsupported(client.createIndex(null: Index, null: Table))
+ assertUnsupported(client.alter_index(null: String, null: String, null:
String, null: Index))
+ assertUnsupported(client.getIndex(null: String, null: String, null:
String))
+ assertUnsupported(client.listIndexes(null: String, null: String,
0.toShort))
+ assertUnsupported(client.listIndexNames(null: String, null: String,
0.toShort))
+ assertUnsupported(client.dropIndex(null: String, null: String, null:
String, false))
+
+ // Column statistics.
+ assertUnsupported(client.updateTableColumnStatistics(null:
ColumnStatistics))
+ assertUnsupported(client.updatePartitionColumnStatistics(null:
ColumnStatistics))
+ assertUnsupported(client.getTableColumnStatistics(null: String, null:
String, null: util.List[String]))
+ assertUnsupported(client.getPartitionColumnStatistics(null: String, null:
String, null: util.List[String], null: util.List[String]))
+ assertUnsupported(client.deletePartitionColumnStatistics(null: String,
null: String, null: String, null: String))
+ assertUnsupported(client.deleteTableColumnStatistics(null: String, null:
String, null: String))
+ assertUnsupported(client.getAggrColStatsFor(null: String, null: String,
null: util.List[String], null: util.List[String]))
+ assertUnsupported(client.setPartitionColumnStatistics(null:
SetPartitionsStatsRequest))
+
+ // Roles and privileges.
+ assertUnsupported(client.create_role(null: Role))
+ assertUnsupported(client.drop_role(null: String))
+ assertUnsupported(client.listRoleNames())
+ assertUnsupported(client.grant_role(null: String, null: String, null:
PrincipalType, null: String, null: PrincipalType, false))
+ assertUnsupported(client.revoke_role(null: String, null: String, null:
PrincipalType, false))
+ assertUnsupported(client.list_roles(null: String, null: PrincipalType))
+ assertUnsupported(client.get_privilege_set(null: HiveObjectRef, null:
String, null: util.List[String]))
+ assertUnsupported(client.list_privileges(null: String, null:
PrincipalType, null: HiveObjectRef))
+ assertUnsupported(client.grant_privileges(null: PrivilegeBag))
+ assertUnsupported(client.revoke_privileges(null: PrivilegeBag, false))
+ assertUnsupported(client.get_principals_in_role(null:
GetPrincipalsInRoleRequest))
+ assertUnsupported(client.get_role_grants_for_principal(null:
GetRoleGrantsForPrincipalRequest))
+
+ // Delegation tokens and master keys.
+ assertUnsupported(client.getDelegationToken(null: String, null: String))
+ assertUnsupported(client.renewDelegationToken(null: String))
+ assertUnsupported(client.cancelDelegationToken(null: String))
+ assertUnsupported(client.getTokenStrForm())
+ assertUnsupported(client.addToken(null: String, null: String))
+ assertUnsupported(client.removeToken(null: String))
+ assertUnsupported(client.getToken(null: String))
+ assertUnsupported(client.getAllTokenIdentifiers())
+ assertUnsupported(client.addMasterKey(null: String))
+ assertUnsupported(client.updateMasterKey(null: java.lang.Integer, null:
String))
+ assertUnsupported(client.removeMasterKey(null: java.lang.Integer))
+ assertUnsupported(client.getMasterKeys())
+
+ // Functions.
+ assertUnsupported(client.createFunction(null: Function))
+ assertUnsupported(client.alterFunction(null: String, null: String, null:
Function))
+ assertUnsupported(client.dropFunction(null: String, null: String))
+ assertUnsupported(client.getFunction(null: String, null: String))
+ assertUnsupported(client.getFunctions(null: String, null: String))
+ assertUnsupported(client.getAllFunctions())
+
+ // Transactions and locks.
+ assertUnsupported(client.getValidTxns())
+ assertUnsupported(client.getValidTxns(0L))
+ assertUnsupported(client.openTxn(null: String))
+ assertUnsupported(client.openTxns(null: String, 0))
+ assertUnsupported(client.rollbackTxn(0L))
+ assertUnsupported(client.commitTxn(0L))
+ assertUnsupported(client.abortTxns(null: util.List[java.lang.Long]))
+ assertUnsupported(client.showTxns())
+ assertUnsupported(client.lock(null: LockRequest))
+ assertUnsupported(client.checkLock(0L))
+ assertUnsupported(client.unlock(0L))
+ assertUnsupported(client.showLocks())
+ assertUnsupported(client.showLocks(null: ShowLocksRequest))
+ assertUnsupported(client.heartbeat(0L, 0L))
+ assertUnsupported(client.heartbeatTxnRange(0L, 0L))
+ assertUnsupported(client.compact(null: String, null: String, null: String,
null: CompactionType))
+ assertUnsupported(client.compact(null: String, null: String, null: String,
null: CompactionType, null: util.Map[String, String]))
+ assertUnsupported(client.compact2(null: String, null: String, null:
String, null: CompactionType, null: util.Map[String, String]))
+ assertUnsupported(client.showCompactions())
+ assertUnsupported(client.addDynamicPartitions(0L, null: String, null:
String, null: util.List[String]))
+ assertUnsupported(client.addDynamicPartitions(0L, null: String, null:
String, null: util.List[String], null: DataOperationType))
+
+ // Notifications and file metadata.
+ assertUnsupported(client.getNextNotification(0L, 0, null:
IMetaStoreClient.NotificationFilter))
+ assertUnsupported(client.getCurrentNotificationEventId())
+ assertUnsupported(client.fireListenerEvent(null: FireEventRequest))
+ assertUnsupported(client.getFileMetadata(null: util.List[java.lang.Long]))
+ assertUnsupported(client.getFileMetadataBySarg(null:
util.List[java.lang.Long], null: java.nio.ByteBuffer, false))
+ assertUnsupported(client.clearFileMetadata(null:
util.List[java.lang.Long]))
+ assertUnsupported(client.putFileMetadata(null: util.List[java.lang.Long],
null: util.List[java.nio.ByteBuffer]))
+ assertUnsupported(client.cacheFileMetadata(null: String, null: String,
null: String, false))
+
+ // Constraints.
+ assertUnsupported(client.getPrimaryKeys(null: PrimaryKeysRequest))
+ assertUnsupported(client.getForeignKeys(null: ForeignKeysRequest))
+ assertUnsupported(client.createTableWithConstraints(null: Table, null:
util.List[SQLPrimaryKey], null: util.List[SQLForeignKey]))
+ assertUnsupported(client.dropConstraint(null: String, null: String, null:
String))
+ assertUnsupported(client.addPrimaryKey(null: util.List[SQLPrimaryKey]))
+ assertUnsupported(client.addForeignKey(null: util.List[SQLForeignKey]))
+ }
+
+ private def assertUnsupported(fn: => Any): Unit = {
+ assertThrows[UnsupportedOperationException](fn)
+ ()
+ }
+
private def newClient(): SparkCatalogMetaStoreClient = {
SparkSession.setActiveSession(spark)
SparkSession.setDefaultSession(spark)
@@ -223,10 +504,15 @@ class TestSparkCatalogMetaStoreClient extends FunSuite
with BeforeAndAfterAll {
table.setPartitionKeys(partitionColumns.map { case (name, dataType) =>
fieldSchema(name, dataType) }.asJava)
val serdeInfo = new SerDeInfo()
+ serdeInfo.setSerializationLib(classOf[ParquetHiveSerDe].getName)
serdeInfo.setParameters(new util.HashMap[String, String]())
+ // Mirror the storage format HMSDDLExecutor registers; Spark's
HiveClientImpl loads the
+ // input and output format classes by name when it converts the table, so
they must be set.
val storageDescriptor = new StorageDescriptor()
storageDescriptor.setCols(columns.map { case (name, dataType) =>
fieldSchema(name, dataType) }.asJava)
+ storageDescriptor.setInputFormat(classOf[HoodieParquetInputFormat].getName)
+
storageDescriptor.setOutputFormat(classOf[MapredParquetOutputFormat].getName)
storageDescriptor.setLocation(location)
storageDescriptor.setSerdeInfo(serdeInfo)
table.setSd(storageDescriptor)
@@ -244,9 +530,12 @@ class TestSparkCatalogMetaStoreClient extends FunSuite
with BeforeAndAfterAll {
partition.setParameters(new util.HashMap[String, String]())
val serdeInfo = new SerDeInfo()
+ serdeInfo.setSerializationLib(classOf[ParquetHiveSerDe].getName)
serdeInfo.setParameters(new util.HashMap[String, String]())
val storageDescriptor = new StorageDescriptor()
+ storageDescriptor.setInputFormat(classOf[HoodieParquetInputFormat].getName)
+
storageDescriptor.setOutputFormat(classOf[MapredParquetOutputFormat].getName)
storageDescriptor.setLocation(location)
storageDescriptor.setSerdeInfo(serdeInfo)
partition.setSd(storageDescriptor)
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
new file mode 100644
index 000000000000..f509a26078a9
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala
@@ -0,0 +1,209 @@
+/*
+ * 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.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
+import org.apache.spark.sql.connector.catalog.{Identifier, 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.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.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
+
+import java.io.File
+
+import scala.collection.JavaConverters._
+
+/**
+ * DDL-level coverage for [[HoodieCatalog]], [[HoodieInternalV2Table]] and the
+ * create/show-create table commands, exercised end-to-end through the V2
session
+ * catalog wired up by [[HoodieSparkSqlTestBase]] (spark_catalog =
HoodieCatalog).
+ */
+class TestHoodieCatalogDDL extends HoodieSparkSqlTestBase {
+
+ private def hoodieCatalog: HoodieCatalog =
+
spark.sessionState.catalogManager.v2SessionCatalog.asInstanceOf[HoodieCatalog]
+
+ test("HoodieCatalog create, load, alter, rename and drop via the V2 catalog
API") {
+ withTempDir { tmp =>
+ val catalog = hoodieCatalog
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ val ident = Identifier.of(Array("default"), tableName)
+ val schema = StructType(Seq(
+ StructField("id", IntegerType),
+ StructField("name", StringType),
+ StructField("ts", LongType)))
+ val props = Map(
+ "provider" -> "hudi",
+ "primaryKey" -> "id",
+ "preCombineField" -> "ts",
+ "location" -> tablePath).asJava
+
+ // createTable routes to createHoodieTable(CREATE) and initializes the
table on disk.
+ catalog.createTable(ident, schema, Array.empty[Transform], props)
+ assertTrue(catalog.tableExists(ident))
+ assertTrue(new File(s"$tablePath/.hoodie/hoodie.properties").exists())
+
+ // loadTable returns a Hudi-backed table exposing the user schema.
+ val loaded = catalog.loadTable(ident)
+ assertEquals(Seq("id", "name", "ts"),
+
HoodieSqlCommonUtils.removeMetaFields(loaded.schema()).fieldNames.toSeq)
+
+ // alterTable: add a column.
+ catalog.alterTable(ident, TableChange.addColumn(Array("age"),
IntegerType, true))
+ assertTrue(catalog.loadTable(ident).schema().fieldNames.contains("age"))
+
+ // alterTable: update a column comment.
+ catalog.alterTable(ident, TableChange.updateColumnComment(Array("name"),
"the name column"))
+ val commented = catalog.loadTable(ident).schema().fields.find(_.name ==
"name").get
+ assertEquals("the name column", commented.getComment().getOrElse(""))
+
+ // alterTable: changing a column type is rejected by the V2 alter path,
which routes to
+ // AlterHoodieTableChangeColumnCommand and does not support column type
changes.
+ val typeChange = intercept[HoodieAnalysisException] {
+ catalog.alterTable(ident, TableChange.updateColumnType(Array("age"),
LongType))
+ }
+ assertTrue(typeChange.getMessage.contains(
+ "ALTER TABLE CHANGE COLUMN is not supported for changing column
'age'"),
+ typeChange.getMessage)
+ 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)
+ assertFalse(catalog.tableExists(ident))
+ assertTrue(catalog.tableExists(renamed))
+
+ // dropTable removes the Hudi table from the catalog.
+ assertTrue(catalog.dropTable(renamed))
+ assertFalse(catalog.tableExists(renamed))
+ // HoodieCatalog.dropTable passes purge = false, so the external
location survives the drop.
+ assertTrue(existsPath(tablePath))
+ }
+ }
+
+ test("SHOW CREATE TABLE regenerates Hudi DDL") {
+ val tableName = generateTableName
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | ts long,
+ | dt string
+ |) using hudi
+ |partitioned by (dt)
+ |comment 'a hudi table'
+ |tblproperties (primaryKey = 'id', preCombineField = 'ts')
+ |""".stripMargin)
+
+ // Spark's ResolveSessionCatalog only emits the V1 ShowCreateTableCommand
under
+ // `spark.sql.legacy.useV1Command`; HoodieAnalysis then rewrites it to
ShowHoodieCreateTableCommand.
+ withSQLConf("spark.sql.legacy.useV1Command" -> "true") {
+ val ddl = spark.sql(s"show create table $tableName").head().getString(0)
+ assertTrue(ddl.contains("CREATE TABLE IF NOT EXISTS"), ddl)
+ assertTrue(ddl.contains(s"`default`.`$tableName`"), ddl)
+ assertTrue(ddl.contains("USING hudi"), ddl)
+ assertTrue(ddl.contains("PARTITIONED BY (dt)"), ddl)
+ assertTrue(ddl.contains("COMMENT 'a hudi table'"), ddl)
+ assertTrue(ddl.contains("TBLPROPERTIES"), ddl)
+ assertTrue(ddl.contains("primaryKey='id'"), ddl)
+ }
+
+ intercept[NoSuchTableException] {
+
ShowHoodieCreateTableCommand(TableIdentifier("does_not_exist_tbl")).run(spark)
+ }
+ }
+
+ test("CREATE over an existing location rejects conflicting table
properties") {
+ withTempDir { tmp =>
+ val basePath = s"${tmp.getCanonicalPath}/shared"
+ val first = generateTableName
+ spark.sql(
+ s"""
+ |create table $first (id int, name string, ts long) using hudi
+ |tblproperties (primaryKey = 'id', preCombineField = 'ts')
+ |location '$basePath'
+ |""".stripMargin)
+
+ // A conflicting primaryKey against the on-disk table config is rejected
with a config-conflict
+ // error surfaced from the write-path validation (HoodieWriterUtils).
+ val conflicting = generateTableName
+ checkExceptionContain(
+ s"""
+ |create table $conflicting (id int, name string, ts long) using hudi
+ |tblproperties (primaryKey = 'name', preCombineField = 'ts')
+ |location '$basePath'
+ |""".stripMargin)("hoodie.table.recordkey.fields")
+ }
+ }
+
+ test("HoodieInternalV2Table exposes v2 capabilities and handles reads and
writes") {
+ withTempDir { tmp =>
+ withSQLConf("hoodie.schema.on.read.enable" -> "true") {
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ spark.sql(
+ s"""
+ |create table $tableName (id int, name string, ts long) using hudi
+ |tblproperties (primaryKey = 'id', preCombineField = 'ts')
+ |location '$tablePath'
+ |""".stripMargin)
+
+ // With schema evolution enabled, loadTable returns the V2 table
directly.
+ val ident = Identifier.of(Array("default"), tableName)
+ val loaded = hoodieCatalog.loadTable(ident)
+ assertTrue(loaded.isInstanceOf[HoodieInternalV2Table])
+ val v2 = loaded.asInstanceOf[HoodieInternalV2Table]
+ assertEquals(
+ Set(BATCH_READ, V1_BATCH_WRITE, OVERWRITE_BY_FILTER, TRUNCATE,
ACCEPT_ANY_SCHEMA).asJava,
+ v2.capabilities())
+ assertTrue(v2.schema().fieldNames.contains("id"))
+ assertTrue(v2.partitioning().isEmpty)
+ assertFalse(v2.properties().isEmpty)
+ // v2.name() is catalog-qualified on Spark 3.4+
(spark_catalog.default.<t>) but only
+ // db-qualified on Spark 3.3 (TableIdentifier has no catalog field
there), so match either.
+ assertTrue(
+ v2.name() == s"spark_catalog.default.$tableName" || v2.name() ==
s"default.$tableName",
+ v2.name())
+
+ // HoodieSpark35Analysis (and its per-version HoodieSpark3xAnalysis
siblings) rewrites the
+ // V2 relation behind an InsertIntoStatement into the V1
LogicalRelation before any write
+ // builder is created, so these inserts cover the V2-to-V1 relation
conversion rather than
+ // HoodieV1WriteBuilder.
+ spark.sql(s"insert into $tableName values (1, 'a1', 1000), (2, 'a2',
2000)")
+ checkAnswer(s"select id, name from $tableName")(Seq(1, "a1"), Seq(2,
"a2"))
+ spark.sql(s"insert overwrite table $tableName values (3, 'a3', 3000)")
+ checkAnswer(s"select id, name from $tableName")(Seq(3, "a3"))
+ }
+ }
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieSqlCommonUtils.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieSqlCommonUtils.scala
new file mode 100644
index 000000000000..6289cc78faaf
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieSqlCommonUtils.scala
@@ -0,0 +1,129 @@
+/*
+ * 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.common
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.analysis.caseInsensitiveResolution
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
CatalogTable, CatalogTableType}
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.hudi.HoodieSqlCommonUtils
+import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException
+import org.apache.spark.sql.types.{IntegerType, StringType, StructField,
StructType}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
+import org.scalatest.funsuite.AnyFunSuite
+
+import java.net.URI
+
+/**
+ * Unit coverage for the pure helper methods in [[HoodieSqlCommonUtils]] that
are otherwise
+ * only reached through heavier read/write code paths.
+ */
+class TestHoodieSqlCommonUtils extends AnyFunSuite {
+
+ private def partitionedTable(partitionCols: Seq[String]): CatalogTable = {
+ val fields = StructField("id", IntegerType) +:
partitionCols.map(StructField(_, StringType))
+ CatalogTable(
+ identifier = TableIdentifier("t", Some("default")),
+ tableType = CatalogTableType.MANAGED,
+ storage = CatalogStorageFormat.empty,
+ schema = StructType(fields),
+ provider = Some("hudi"),
+ partitionColumnNames = partitionCols)
+ }
+
+ private val nonPartitionedTable: CatalogTable = CatalogTable(
+ identifier = TableIdentifier("t0", Some("default")),
+ tableType = CatalogTableType.MANAGED,
+ storage = CatalogStorageFormat.empty,
+ schema = StructType(Seq(StructField("id", IntegerType))),
+ provider = Some("hudi"))
+
+ test("partition style detectors classify partition paths") {
+ val t = partitionedTable(Seq("dt"))
+
+
assertTrue(HoodieSqlCommonUtils.isHiveStyledPartitioning(Seq("dt=2021-04-01"),
t))
+
assertFalse(HoodieSqlCommonUtils.isHiveStyledPartitioning(Seq("2021-04-01"), t))
+ assertTrue(HoodieSqlCommonUtils.isHiveStyledPartitioning(Seq("anything"),
nonPartitionedTable))
+
+ assertTrue(HoodieSqlCommonUtils.isUrlEncodeEnabled(Seq("dt=2021-04-01"),
t))
+ assertFalse(HoodieSqlCommonUtils.isUrlEncodeEnabled(Seq("a/b"), t))
+ assertFalse(HoodieSqlCommonUtils.isUrlEncodeEnabled(Seq("x"),
nonPartitionedTable))
+
+ // Both detectors compare the slash-separated fragment count against the
partition columns,
+ // so a multi-column table only matches when every fragment is present.
+ val t2 = partitionedTable(Seq("dt", "hh"))
+
assertTrue(HoodieSqlCommonUtils.isHiveStyledPartitioning(Seq("dt=2021-04-01/hh=12"),
t2))
+
assertFalse(HoodieSqlCommonUtils.isHiveStyledPartitioning(Seq("dt=2021-04-01/12"),
t2))
+
assertTrue(HoodieSqlCommonUtils.isUrlEncodeEnabled(Seq("2021%2F04%2F01/12"),
t2))
+ assertFalse(HoodieSqlCommonUtils.isUrlEncodeEnabled(Seq("2021/04/01/12"),
t2))
+ }
+
+ test("config helpers and meta field utilities") {
+ val opts = Map("hoodie.a" -> "1", "spark.hoodie.b" -> "2", "other" -> "3")
+ assertEquals(Map("hoodie.a" -> "1"),
HoodieSqlCommonUtils.filterHoodieConfigs(opts))
+ assertEquals(Map("hoodie.b" -> "2"),
HoodieSqlCommonUtils.extractSparkPrefixedHoodieConfigs(opts))
+
+ val base = StructType(Seq(StructField("id", IntegerType),
StructField("name", StringType)))
+ val withMeta = HoodieSqlCommonUtils.addMetaFields(base)
+ assertTrue(withMeta.fieldNames.contains("_hoodie_commit_time"))
+ assertEquals(base.fields.length + 5, withMeta.fields.length)
+ assertEquals(base, HoodieSqlCommonUtils.removeMetaFields(withMeta))
+ assertTrue(HoodieSqlCommonUtils.isMetaField("_hoodie_commit_time"))
+ assertFalse(HoodieSqlCommonUtils.isMetaField("id"))
+
+ val attrs = Seq(
+ AttributeReference("_hoodie_commit_time", StringType)(),
+ AttributeReference("id", IntegerType)())
+ assertEquals(Seq("id"),
HoodieSqlCommonUtils.removeMetaFields(attrs).map(_.name))
+ }
+
+ test("normalizePartitionSpec normalizes keys and rejects invalid specs") {
+ val resolver = caseInsensitiveResolution
+ assertEquals(Map("dt" -> "2021"),
+ HoodieSqlCommonUtils.normalizePartitionSpec(Map("DT" -> "2021"),
Seq("dt"), "t", resolver))
+
+ // Unknown partition column.
+ intercept[HoodieAnalysisException] {
+ HoodieSqlCommonUtils.normalizePartitionSpec(Map("bad" -> "x"),
Seq("dt"), "t", resolver)
+ }
+ // Not all partition columns specified.
+ intercept[HoodieAnalysisException] {
+ HoodieSqlCommonUtils.normalizePartitionSpec(Map.empty[String, String],
Seq("dt"), "t", resolver)
+ }
+ // Duplicate partition columns.
+ intercept[HoodieAnalysisException] {
+ HoodieSqlCommonUtils.normalizePartitionSpec(
+ Map("dt" -> "a", "DT" -> "b"), Seq("dt", "DT"), "t", resolver)
+ }
+ }
+
+ test("path qualification and column helpers") {
+ val qualified = HoodieSqlCommonUtils.makePathQualified(new
URI("/tmp/hudi_test_path"), new Configuration())
+ assertTrue(qualified.startsWith("file:"))
+
+ val resolver = caseInsensitiveResolution
+ val schema = StructType(Seq(StructField("id", IntegerType),
StructField("name", StringType)))
+ assertEquals("id", HoodieSqlCommonUtils.findColumnByName(schema, "ID",
resolver).get.name)
+ assertTrue(HoodieSqlCommonUtils.findColumnByName(schema, "missing",
resolver).isEmpty)
+ assertTrue(HoodieSqlCommonUtils.columnEqual(
+ StructField("a", IntegerType), StructField("A", IntegerType), resolver))
+ assertFalse(HoodieSqlCommonUtils.columnEqual(
+ StructField("a", IntegerType), StructField("a", StringType), resolver))
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestCreateTable.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestCreateTable.scala
index 7c87f5c034a6..240982465f35 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestCreateTable.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestCreateTable.scala
@@ -1443,6 +1443,25 @@ class TestCreateTable extends HoodieSparkSqlTestBase
with ExtendedParserTestHelp
val dbPath =
spark.sessionState.catalog.getDatabaseMetadata("default").locationUri.getPath
val tablePath = s"${dbPath}/${tableName}"
assertResult(false)(existsPath(tablePath))
+
+ // The same illegal CTAS against an explicit location must clean the
external path up too.
+ withTempDir { tmp =>
+ val externalTableName = generateTableName
+ val externalTablePath = s"${tmp.getCanonicalPath}/$externalTableName"
+ checkExceptionContain(
+ s"""
+ | create table $externalTableName using hudi
+ | tblproperties(
+ | primaryKey = 'id',
+ | type = 'cow',
+ | hoodie.compact.inline='true'
+ | )
+ | location '$externalTablePath'
+ | AS
+ | select 1 as id, 'a1' as name, 10 as price, 1000 as ts
+ |""".stripMargin)("Compaction is not supported on a CopyOnWrite
table")
+ assertResult(false)(existsPath(externalTablePath))
+ }
}
test("Test Create Non-Hudi Table(Parquet Table)") {
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestSparkCatalogSync.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestSparkCatalogSync.scala
index 8f644d263a39..970adc4bde44 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestSparkCatalogSync.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestSparkCatalogSync.scala
@@ -22,12 +22,13 @@ import org.apache.hudi.common.config.TypedProperties
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.hive.{HiveStylePartitionValueExtractor,
HiveSyncConfigHolder, HiveSyncTool}
import org.apache.hudi.hive.ddl.HiveSyncMode
-import org.apache.hudi.sync.common.HoodieSyncConfig.{META_SYNC_BASE_PATH,
META_SYNC_DATABASE_NAME, META_SYNC_PARTITION_EXTRACTOR_CLASS,
META_SYNC_PARTITION_FIELDS, META_SYNC_TABLE_NAME}
+import org.apache.hudi.sync.common.HoodieSyncConfig.{META_SYNC_BASE_PATH,
META_SYNC_DATABASE_NAME, META_SYNC_FORCE_RECREATE_TABLE,
META_SYNC_PARTITION_EXTRACTOR_CLASS, META_SYNC_PARTITION_FIELDS,
META_SYNC_TABLE_NAME}
import org.apache.hadoop.hive.conf.HiveConf
import org.apache.spark.sql.{DataFrame, SaveMode}
+import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
-import org.junit.jupiter.api.Assertions.{assertFalse, assertTrue}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
class TestSparkCatalogSync extends HoodieSparkSqlTestBase {
@@ -113,6 +114,48 @@ class TestSparkCatalogSync extends HoodieSparkSqlTestBase {
}
}
+ test("Test Spark catalog sync with forced table recreation") {
+ withTempDir { tmp =>
+ import spark.implicits._
+
+ val tableName = generateTableName
+ val databaseName = "testdb"
+ val basePath = s"${tmp.getCanonicalPath}/$tableName"
+ val syncProps = buildSyncProps(databaseName, tableName, basePath)
+
+ try {
+ spark.sql(s"create database if not exists $databaseName")
+ writeToHudi(
+ Seq((1, "a1", 1000L, "2024-01-01")).toDF("id", "name", "ts", "dt"),
+ tableName,
+ basePath,
+ SaveMode.Overwrite)
+ syncOnce(syncProps)
+ assertTrue(spark.catalog.tableExists(databaseName, tableName), "Table
should exist after the first sync")
+
+ // A property set between the syncs must not survive, which proves the
table was really
+ // recreated rather than left as it was.
+ val identifier = TableIdentifier(tableName, Some(databaseName))
+ val before = spark.sessionState.catalog.getTableMetadata(identifier)
+ spark.sessionState.catalog.alterTable(before.copy(properties =
before.properties + ("drift_marker" -> "true")))
+
assertTrue(spark.sessionState.catalog.getTableMetadata(identifier).properties.contains("drift_marker"))
+
+ // Recreation syncs into a temp table, drops the real one and renames
the temp table over
+ // it through alter_table, which needs SparkCatalogMetaStoreClient to
honor the new name.
+ syncProps.setProperty(META_SYNC_FORCE_RECREATE_TABLE.key, "true")
+ syncOnce(syncProps)
+ val remaining = spark.catalog.listTables(databaseName).collect()
+
.map(_.name.toLowerCase).filter(_.startsWith(tableName.toLowerCase)).toSeq
+ assertEquals(Seq(tableName.toLowerCase), remaining, "Only the
recreated table should remain after a forced recreation")
+
assertFalse(spark.sessionState.catalog.getTableMetadata(identifier).properties.contains("drift_marker"),
+ "The recreated table must not carry the pre-recreation property")
+ assertEquals(1L, spark.table(s"$databaseName.$tableName").count())
+ } finally {
+ spark.sql(s"drop table if exists $databaseName.$tableName")
+ }
+ }
+ }
+
private def syncOnce(syncProps: TypedProperties): Unit = {
val syncTool = new HiveSyncTool(syncProps, new HiveConf())
try {
