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 ba4615b864f0 fix(spark): support saveAsTable append to existing Hudi
tables (#20423)
ba4615b864f0 is described below
commit ba4615b864f07c71016b043af84fe37dba8b0b69
Author: Danny Chan <[email protected]>
AuthorDate: Mon Nov 2 11:31:35 2026 +0800
fix(spark): support saveAsTable append to existing Hudi tables (#20423)
---
.../spark/sql/hudi/analysis/HoodieAnalysis.scala | 45 +++-
.../spark/sql/hudi/ddl/TestCreateTable.scala | 248 ++++++++++++++++++++-
.../command/CreateHoodieTableAsSelectCommand.scala | 24 +-
.../command/InsertIntoHoodieTableCommand.scala | 11 +-
.../command/CreateHoodieTableAsSelectCommand.scala | 24 +-
.../command/InsertIntoHoodieTableCommand.scala | 11 +-
6 files changed, 344 insertions(+), 19 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala
index f9d99cbc0602..ad240a8e7876 100644
---
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala
+++
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala
@@ -18,10 +18,11 @@
package org.apache.spark.sql.hudi.analysis
import org.apache.hudi.{HoodieSchemaUtils, HoodieSparkUtils,
SparkAdapterSupport}
+import
org.apache.hudi.HoodieWriterUtils.{mappingSparkDatasourceConfigsToTableConfigs,
validateTableConfig}
import org.apache.hudi.common.util.ReflectionUtils.loadClass
import org.apache.hudi.common.util.ValidationUtils
-import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
CatalogTable, HoodieCatalogTable}
@@ -31,7 +32,8 @@ import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.command._
import org.apache.spark.sql.execution.datasources.{CreateTable,
LogicalRelation}
-import org.apache.spark.sql.hudi.HoodieSqlCommonUtils.{isMetaField,
removeMetaFields}
+import
org.apache.spark.sql.hudi.HoodieOptionConfig.mapSqlOptionsToDataSourceWriteConfigs
+import org.apache.spark.sql.hudi.HoodieSqlCommonUtils.{isMetaField,
makePathQualified, removeMetaFields}
import org.apache.spark.sql.hudi.analysis.HoodieAnalysis.{sparkAdapter,
MatchCreateIndex, MatchCreateTableLike, MatchDropIndex,
MatchInsertIntoStatement, MatchMergeIntoTable, MatchRefreshIndex,
MatchShowIndexes, ResolvesToHudiTable}
import org.apache.spark.sql.hudi.blob.ReadBlobRule
import org.apache.spark.sql.hudi.command._
@@ -40,6 +42,7 @@ import
org.apache.spark.sql.hudi.command.InsertIntoHoodieTableCommand.alignQuery
import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException
import org.apache.spark.sql.hudi.command.procedures.{HoodieProcedures,
Procedure, ProcedureArgs}
+import java.net.URI
import java.util
import scala.collection.mutable.ListBuffer
@@ -410,9 +413,13 @@ case class ResolveImplementationsEarly(spark:
SparkSession) extends Rule[Logical
// Convert to CreateHoodieTableAsSelectCommand
case ct @ CreateTable(table, mode, Some(query))
if sparkAdapter.isHoodieTable(table) && ct.query.forall(_.resolved) =>
- val alignedQuery = alignCtasQueryByPartitionOrder(
- stripMetaFieldAttributes(query), table.partitionColumnNames)
- CreateHoodieTableAsSelectCommand(table, mode, alignedQuery)
+ val appendToExistingTable = mode == SaveMode.Append &&
spark.sessionState.catalog.tableExists(table.identifier)
+ val alignedQuery = if (appendToExistingTable) {
+ alignAppendQuery(table, query)
+ } else {
+ alignCtasQueryByPartitionOrder(stripMetaFieldAttributes(query),
table.partitionColumnNames)
+ }
+ CreateHoodieTableAsSelectCommand(table, mode, alignedQuery,
appendToExistingTable)
case ct: CreateTable =>
try {
@@ -434,6 +441,34 @@ case class ResolveImplementationsEarly(spark:
SparkSession) extends Rule[Logical
}
}
+ private def alignAppendQuery(table: CatalogTable, query: LogicalPlan):
LogicalPlan = {
+ val existingTable =
spark.sessionState.catalog.getTableMetadata(table.identifier)
+ if (!sparkAdapter.isHoodieTable(existingTable)) {
+ throw new HoodieAnalysisException(s"Cannot append using hudi to table
${table.identifier.unquotedString}: " +
+ s"the existing table uses ${existingTable.provider.getOrElse("an
unspecified provider")}")
+ }
+ table.storage.locationUri.foreach { location =>
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ val specifiedLocation = new URI(makePathQualified(location,
hadoopConf)).normalize()
+ val existingLocation = new URI(makePathQualified(existingTable.location,
hadoopConf)).normalize()
+ if (specifiedLocation != existingLocation) {
+ throw new HoodieAnalysisException(s"The location of the existing table
${existingTable.identifier.unquotedString} " +
+ s"($existingLocation) doesn't match the specified location
($specifiedLocation)")
+ }
+ }
+ val conf = spark.sessionState.conf
+ val partitionColumns = table.partitionColumnNames
+ if (partitionColumns.nonEmpty && (partitionColumns.size !=
existingTable.partitionColumnNames.size ||
+ !partitionColumns.zip(existingTable.partitionColumnNames).forall { case
(specified, existing) => conf.resolver(specified, existing) })) {
+ throw new HoodieAnalysisException(s"Specified partitioning
${partitionColumns.mkString("[", ", ", "]")} " +
+ s"does not match that of the existing table
${existingTable.partitionColumnNames.mkString("[", ", ", "]")}")
+ }
+ val catalogTable = new HoodieCatalogTable(spark, existingTable)
+ val options =
mapSqlOptionsToDataSourceWriteConfigs(table.storage.properties --
HoodieCatalogTable.needFilterProps - "path")
+ validateTableConfig(spark, options ++
mappingSparkDatasourceConfigsToTableConfigs(options), catalogTable.tableConfig)
+ alignQueryOutput(query, catalogTable, Map.empty, conf,
allowPositionalFallback = false)
+ }
+
private def alignCtasQueryByPartitionOrder(query: LogicalPlan,
partitionColumns: Seq[String]): LogicalPlan = {
if (partitionColumns.isEmpty) {
query
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 3e3f3cf774ec..67c89050c1f3 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
@@ -35,9 +35,10 @@ import
org.apache.hudi.storage.hadoop.HadoopStorageConfiguration
import org.apache.hudi.testutils.Assertions
import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
-import org.apache.spark.sql.{SaveMode, SparkSession}
+import org.apache.spark.sql.{AnalysisException, SaveMode, SparkSession}
import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.catalog.{CatalogTableType,
HoodieCatalogTable}
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
CatalogTable, CatalogTableType, HoodieCatalogTable}
+import org.apache.spark.sql.execution.datasources.CreateTable
import org.apache.spark.sql.functions.{col, concat, expr, lit}
import org.apache.spark.sql.hudi.HoodieSqlCommonUtils
import org.apache.spark.sql.hudi.command.CreateHoodieTableCommand
@@ -1654,6 +1655,249 @@ class TestCreateTable extends HoodieSparkSqlTestBase
with ExtendedParserTestHelp
spark.sql(s"drop table $tableName")
}
+ for (tableType <- Seq("cow", "mor"); external <- Seq(false, true);
partitioned <- Seq(false, true)) {
+ test(s"saveAsTable append preserves existing $tableType table
(external=$external, partitioned=$partitioned)") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val location = if (external) s"LOCATION
'${tmp.getCanonicalPath}/$tableName'" else ""
+ val partition = if (partitioned) "PARTITIONED BY (dt)" else ""
+ spark.sql(
+ s"""CREATE TABLE $tableName (id INT, name STRING, ts BIGINT, dt
STRING)
+ |USING hudi $partition $location
+ |TBLPROPERTIES (type = '$tableType', primaryKey = 'id',
orderingFields = 'ts')
+ |""".stripMargin)
+ try {
+ spark.sql(s"INSERT INTO $tableName VALUES (1, 'first', 1, 'p1')")
+ val identifier = TableIdentifier(tableName)
+ val before = spark.sessionState.catalog.getTableMetadata(identifier)
+ val tablePath = HoodieCatalogTable(spark, before).tableLocation
+ val configBefore = createMetaClient(spark,
tablePath).getTableConfig.getProps.asScala.toMap
+
+ // Omit table options and partitionBy: these must come from the
existing table.
+ spark.sql("SELECT 2 AS id, 'second' AS name, CAST(2 AS BIGINT) AS
ts, 'p2' AS dt")
+ .write.format("hudi").mode(SaveMode.Append).saveAsTable(tableName)
+ checkAnswer(s"SELECT id, name, ts, dt FROM $tableName")(
+ Seq(1, "first", 1L, "p1"), Seq(2, "second", 2L, "p2"))
+ // saveAsTable appends by column name even when the input order
differs.
+ spark.sql("SELECT 'p3' AS dt, CAST(3 AS BIGINT) AS ts, 'third' AS
name, 3 AS id")
+ .write.format("hudi").mode(SaveMode.Append).saveAsTable(tableName)
+ checkAnswer(s"SELECT id, name, ts, dt FROM $tableName")(
+ Seq(1, "first", 1L, "p1"), Seq(2, "second", 2L, "p2"), Seq(3,
"third", 3L, "p3"))
+ spark.sql("SELECT 2 AS id, 'updated' AS name, CAST(4 AS BIGINT) AS
ts, 'p2' AS dt")
+ .write.format("hudi").mode(SaveMode.Append)
+ .option("hoodie.spark.sql.insert.into.operation",
"upsert").saveAsTable(tableName)
+ // The snapshot view must contain the updated row exactly once,
including for MOR tables.
+ checkAnswer(s"SELECT id, name, ts, dt FROM $tableName")(
+ Seq(1, "first", 1L, "p1"), Seq(2, "updated", 4L, "p2"), Seq(3,
"third", 3L, "p3"))
+ val after = spark.sessionState.catalog.getTableMetadata(identifier)
+ assertEquals(before.storage.locationUri, after.storage.locationUri)
+ assertEquals(before.partitionColumnNames, after.partitionColumnNames)
+ assertEquals(before.schema, after.schema)
+ assertEquals(configBefore, createMetaClient(spark,
tablePath).getTableConfig.getProps.asScala.toMap)
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $tableName")
+ }
+ }
+ }
+ }
+
+ for (tableType <- Seq("cow", "mor")) {
+ test(s"saveAsTable append aligns types and rejects extra columns for
$tableType") {
+ val tableName = generateTableName
+ spark.sql(s"CREATE TABLE $tableName (id INT, name STRING) USING hudi " +
+ s"TBLPROPERTIES (type = '$tableType', primaryKey = 'id')")
+ try {
+ spark.sql("SELECT 'first' AS name, CAST(1 AS BIGINT) AS id")
+ .write.format("hudi").mode(SaveMode.Append).saveAsTable(tableName)
+ checkAnswer(s"SELECT id, name FROM $tableName")(Seq(1, "first"))
+ assertEquals(IntegerType, spark.table(tableName).schema("id").dataType)
+ val path = HoodieCatalogTable(spark,
spark.sessionState.catalog.getTableMetadata(TableIdentifier(tableName))).tableLocation
+ assertEquals(IntegerType,
spark.read.format("hudi").load(path).schema("id").dataType)
+ val metaClient = createMetaClient(spark, path)
+ val commitsBefore =
metaClient.getActiveTimeline.getWriteTimeline.countInstants()
+ val error = intercept[AnalysisException] {
+ spark.sql("SELECT 2 AS id, 'second' AS name, 'unexpected' AS extra")
+ .write.format("hudi").mode(SaveMode.Append).saveAsTable(tableName)
+ }
+ assertTrue(error.getMessage.toLowerCase.contains("too many data
columns"))
+ assertEquals(commitsBefore,
metaClient.reloadActiveTimeline().getWriteTimeline.countInstants())
+ // A rejected append must leave the persisted and catalog schemas
usable for later inserts.
+ spark.sql(s"INSERT INTO $tableName VALUES (2, 'second')")
+ checkAnswer(s"SELECT id, name FROM $tableName")(Seq(1, "first"),
Seq(2, "second"))
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $tableName")
+ }
+ }
+ }
+
+ for (partitionColumns <- Seq(Seq.empty[String], Seq("dt"), Seq("name",
"dt"))) {
+ test(s"saveAsTable append validates specified partitioning
${partitionColumns.mkString(",")}") {
+ val tableName = generateTableName
+ val partition = if (partitionColumns.isEmpty) "" else s"PARTITIONED BY
(${partitionColumns.mkString(",")})"
+ spark.sql(s"CREATE TABLE $tableName (id INT, name STRING, dt STRING)
USING hudi $partition " +
+ "TBLPROPERTIES (primaryKey = 'id')")
+ try {
+ val df = spark.sql("SELECT 1 AS id, 'first' AS name, 'p1' AS dt")
+ val error = intercept[AnalysisException] {
+
df.write.format("hudi").mode(SaveMode.Append).partitionBy("name").saveAsTable(tableName)
+ }
+ assertTrue(error.getMessage.contains("does not match that of the
existing table"))
+ if (partitionColumns.size > 1) {
+ intercept[AnalysisException] {
+
df.write.format("hudi").mode(SaveMode.Append).partitionBy(partitionColumns.reverse:
_*).saveAsTable(tableName)
+ }
+ }
+ withSQLConf("spark.sql.caseSensitive" -> "false") {
+ df.write.format("hudi").mode(SaveMode.Append)
+
.partitionBy(partitionColumns.map(_.toUpperCase(java.util.Locale.ROOT)):
_*).saveAsTable(tableName)
+ }
+ checkAnswer(s"SELECT id, name, dt FROM $tableName")(Seq(1, "first",
"p1"))
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $tableName")
+ }
+ }
+ }
+
+ test("saveAsTable append rejects a non-Hudi table") {
+ val tableName = generateTableName
+ spark.sql(s"CREATE TABLE $tableName (id INT) USING parquet")
+ try {
+ val error = intercept[AnalysisException] {
+ spark.sql("SELECT 1 AS
id").write.format("hudi").mode(SaveMode.Append).saveAsTable(tableName)
+ }
+ assertTrue(error.getMessage.contains("existing table uses parquet"))
+ spark.sql(s"INSERT INTO $tableName VALUES (2)")
+ checkAnswer(s"SELECT * FROM $tableName")(Seq(2))
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $tableName")
+ }
+ }
+
+ test("saveAsTable append rejects conflicting table options") {
+ val tableName = generateTableName
+ spark.sql(s"CREATE TABLE $tableName (id INT, name STRING, ts BIGINT, dt
STRING) USING hudi " +
+ "PARTITIONED BY (dt) TBLPROPERTIES (primaryKey = 'id', orderingFields =
'ts')")
+ try {
+ val conflicts = Seq(
+ "hoodie.datasource.write.recordkey.field" -> "name",
+ "hoodie.datasource.write.precombine.field" -> "id",
+ "hoodie.datasource.write.table.type" -> "MERGE_ON_READ",
+ "hoodie.datasource.write.keygenerator.class" ->
"org.apache.hudi.keygen.ComplexKeyGenerator",
+ "primaryKey" -> "name",
+ "type" -> "mor")
+ val df = spark.sql("SELECT 1 AS id, 'first' AS name, CAST(1 AS BIGINT)
AS ts, 'p1' AS dt")
+ conflicts.foreach { case (key, value) =>
+ withClue(s"Conflicting option $key=$value: ") {
+ val error = intercept[org.apache.hudi.exception.HoodieException] {
+ df.write.format("hudi").mode(SaveMode.Append).option(key,
value).saveAsTable(tableName)
+ }
+ assertTrue(error.getMessage.contains("Config conflict"))
+ }
+ }
+ df.write.format("hudi").mode(SaveMode.Append)
+ .option("primaryKey", "id").option("type",
"cow").saveAsTable(tableName)
+ checkAnswer(s"SELECT id, name, ts, dt FROM $tableName")(Seq(1, "first",
1L, "p1"))
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $tableName")
+ }
+ }
+
+ for (caseSensitive <- Seq(false, true)) {
+ test(s"saveAsTable append matches columns only by name
(caseSensitive=$caseSensitive)") {
+ withSQLConf("spark.sql.caseSensitive" -> caseSensitive.toString) {
+ val tableName = generateTableName
+ spark.sql(s"CREATE TABLE $tableName (id INT, first STRING, last
STRING) USING hudi " +
+ "TBLPROPERTIES (primaryKey = 'id')")
+ try {
+ val error = intercept[AnalysisException] {
+ spark.sql("SELECT 1 AS id, 'Doe' AS last, 'John' AS first_name")
+
.write.format("hudi").mode(SaveMode.Append).saveAsTable(tableName)
+ }
+ assertTrue(error.getMessage.contains("first"))
+ assertEquals(0L, spark.table(tableName).count())
+ val upperCase = spark.sql("SELECT 'Doe' AS LAST, 1 AS ID, 'John' AS
FIRST")
+ if (caseSensitive) {
+ intercept[AnalysisException] {
+
upperCase.write.format("hudi").mode(SaveMode.Append).saveAsTable(tableName)
+ }
+ spark.sql("SELECT 'Doe' AS last, 1 AS id, 'John' AS first")
+
.write.format("hudi").mode(SaveMode.Append).saveAsTable(tableName)
+ } else {
+
upperCase.write.format("hudi").mode(SaveMode.Append).saveAsTable(tableName)
+ }
+ // SQL INSERT INTO retains its positional fallback for differently
named input columns.
+ spark.sql(s"INSERT INTO $tableName SELECT 2 AS other_id, 'Jane' AS
other_first, 'Roe' AS other_last")
+ checkAnswer(s"SELECT id, first, last FROM $tableName")(
+ Seq(1, "John", "Doe"), Seq(2, "Jane", "Roe"))
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $tableName")
+ }
+ }
+ }
+ }
+
+ for (external <- Seq(false, true)) {
+ test(s"saveAsTable append validates the qualified location
(external=$external)") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val locationClause = if (external) s"LOCATION
'${tmp.getCanonicalPath}/$tableName'" else ""
+ spark.sql(s"CREATE TABLE $tableName (id INT) USING hudi
$locationClause TBLPROPERTIES (primaryKey = 'id')")
+ try {
+ spark.sql(s"INSERT INTO $tableName VALUES (1)")
+ val location =
spark.sessionState.catalog.getTableMetadata(TableIdentifier(tableName)).location
+ val other = new java.io.File(tmp, "other")
+ val error = intercept[AnalysisException] {
+ spark.sql("SELECT 2 AS
id").write.format("hudi").mode(SaveMode.Append)
+ .option("path", other.getCanonicalPath).saveAsTable(tableName)
+ }
+ assertTrue(error.getMessage.contains("doesn't match the specified
location"))
+ assertFalse(other.exists())
+ checkAnswer(s"SELECT id FROM $tableName")(Seq(1))
+ // Qualified URIs and unqualified paths referring to the same
location are both valid.
+ spark.sql("SELECT 2 AS
id").write.format("hudi").mode(SaveMode.Append)
+ .option("path", location.getPath).saveAsTable(tableName)
+ spark.sql("SELECT 3 AS
id").write.format("hudi").mode(SaveMode.Append)
+ .option("path", location.toString + "/").saveAsTable(tableName)
+ checkAnswer(s"SELECT id FROM $tableName")(Seq(1), Seq(2), Seq(3))
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $tableName")
+ }
+ }
+ }
+ }
+
+ for (existsAtAnalysis <- Seq(false, true)) {
+ test(s"saveAsTable append rejects table existence changes after analysis
(existsAtAnalysis=$existsAtAnalysis)") {
+ val tableName = generateTableName
+ val createTable = s"CREATE TABLE $tableName (id INT) USING hudi
TBLPROPERTIES (primaryKey = 'id')"
+ if (existsAtAnalysis) {
+ spark.sql(createTable)
+ }
+ try {
+ val table = CatalogTable(TableIdentifier(tableName),
CatalogTableType.MANAGED,
+ CatalogStorageFormat.empty, new StructType(), provider =
Some("hudi"))
+ val query = spark.sql("SELECT 1 AS id").queryExecution.analyzed
+ val analyzed = spark.sessionState.executePlan(CreateTable(table,
SaveMode.Append, Some(query))).analyzed
+ if (existsAtAnalysis) {
+ spark.sql(s"DROP TABLE $tableName")
+ } else {
+ spark.sql(createTable)
+ }
+ val error = intercept[AnalysisException] {
+
spark.sessionState.executePlan(analyzed).executedPlan.executeCollect()
+ }
+ assertTrue(error.getMessage.contains("existence changed after
analysis"))
+ if (existsAtAnalysis) {
+ assertFalse(spark.catalog.tableExists(tableName))
+ } else {
+ assertEquals(0L, spark.table(tableName).count())
+ }
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $tableName")
+ }
+ }
+ }
+
test("Test CTAS not de-duplicating (by default)") {
withRecordType() {
withTempDir { tmp =>
diff --git
a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/CreateHoodieTableAsSelectCommand.scala
b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/CreateHoodieTableAsSelectCommand.scala
index ffb45bd80746..d1f70b6740a7 100644
---
a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/CreateHoodieTableAsSelectCommand.scala
+++
b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/CreateHoodieTableAsSelectCommand.scala
@@ -34,6 +34,7 @@ import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.command.DataWritingCommand
import org.apache.spark.sql.execution.metric.SQLMetric
+import
org.apache.spark.sql.hudi.HoodieOptionConfig.mapSqlOptionsToDataSourceWriteConfigs
import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException
import scala.collection.JavaConverters._
@@ -44,7 +45,8 @@ import scala.collection.JavaConverters._
case class CreateHoodieTableAsSelectCommand(
table: CatalogTable,
mode: SaveMode,
- query: LogicalPlan) extends DataWritingCommand {
+ query: LogicalPlan,
+ appendToExistingTable: Boolean = false) extends DataWritingCommand {
override def innerChildren: Seq[QueryPlan[_]] = Seq(query)
override lazy val metrics: Map[String, SQLMetric] =
HoodieCommandMetrics.metrics
@@ -63,7 +65,12 @@ case class CreateHoodieTableAsSelectCommand(
val qualifiedTableIdentifier = table.identifier.copy(database = Some(db))
val tableName = qualifiedTableIdentifier.unquotedString
- if (sessionState.catalog.tableExists(qualifiedTableIdentifier)) {
+ val tableExists =
sessionState.catalog.tableExists(qualifiedTableIdentifier)
+ if (mode == SaveMode.Append && tableExists != appendToExistingTable) {
+ throw new HoodieAnalysisException(s"Table $tableName existence changed
after analysis. Retry the append.")
+ }
+
+ if (tableExists) {
checkState(mode != SaveMode.Overwrite,
s"Expect the table $tableName has been dropped when the save mode is
Overwrite")
@@ -77,6 +84,19 @@ case class CreateHoodieTableAsSelectCommand(
return Seq.empty
// scalastyle:on
}
+
+ if (mode == SaveMode.Append) {
+ val existingTable =
sessionState.catalog.getTableMetadata(qualifiedTableIdentifier)
+ val partitionSpec = existingTable.partitionColumnNames.map((_,
None)).toMap
+ InsertIntoHoodieTableCommand.run(sparkSession, existingTable, plan,
partitionSpec,
+ overwrite = false,
+ extraOptions =
mapSqlOptionsToDataSourceWriteConfigs(table.storage.properties --
needFilterProps - "path"),
+ metrics = metrics)
+ DataWritingCommand.propogateMetrics(sparkSession.sparkContext, this,
metrics)
+ // scalastyle:off
+ return Seq.empty
+ // scalastyle:on
+ }
}
// Remove some properties should not be used
diff --git
a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala
b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala
index 3fc336167e89..155391717af8 100644
---
a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala
+++
b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala
@@ -149,11 +149,13 @@ object InsertIntoHoodieTableCommand extends Logging with
ProvidesHoodieConfig wi
* @param catalogTable catalog table
* @param partitionsSpec partition spec specifying static/dynamic partition
values
* @param conf Spark's [[SQLConf]]
+ * @param allowPositionalFallback whether to retry positional matching when
column names do not match
*/
def alignQueryOutput(query: LogicalPlan,
catalogTable: HoodieCatalogTable,
partitionsSpec: Map[String, Option[String]],
- conf: SQLConf): LogicalPlan = {
+ conf: SQLConf,
+ allowPositionalFallback: Boolean = true):
LogicalPlan = {
val targetPartitionSchema = catalogTable.partitionSchema
val staticPartitionValues = filterStaticPartitionValues(partitionsSpec)
@@ -168,7 +170,7 @@ object InsertIntoHoodieTableCommand extends Logging with
ProvidesHoodieConfig wi
// since such columns wouldn't be otherwise specified w/in the query
itself and therefore couldn't be matched
// positionally for example
val expectedQueryColumns =
catalogTable.tableSchemaWithoutMetaFields.filterNot(f =>
staticPartitionValues.contains(f.name))
- val coercedQueryOutput =
coerceQueryOutputColumns(StructType(expectedQueryColumns), cleanedQuery,
catalogTable, conf)
+ val coercedQueryOutput =
coerceQueryOutputColumns(StructType(expectedQueryColumns), cleanedQuery,
catalogTable, conf, allowPositionalFallback)
// After potential reshaping validate that the output of the query
conforms to the table's schema
validate(removeMetaFields(coercedQueryOutput.schema), partitionsSpec,
catalogTable)
@@ -180,7 +182,8 @@ object InsertIntoHoodieTableCommand extends Logging with
ProvidesHoodieConfig wi
private def coerceQueryOutputColumns(expectedSchema: StructType,
query: LogicalPlan,
catalogTable: HoodieCatalogTable,
- conf: SQLConf): LogicalPlan = {
+ conf: SQLConf,
+ allowPositionalFallback: Boolean):
LogicalPlan = {
val planUtils = sparkAdapter.getCatalystPlanUtils
try {
planUtils.resolveOutputColumns(
@@ -188,7 +191,7 @@ object InsertIntoHoodieTableCommand extends Logging with
ProvidesHoodieConfig wi
} catch {
// NOTE: In case matching by name didn't match the query output, we will
attempt positional matching
// SPARK-42309 Error message changed in Spark 3.5.0 so we need to match
two strings here
- case ae: AnalysisException if
(ae.getMessage().startsWith("[INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA]
Cannot write incompatible data for the table")
+ case ae: AnalysisException if allowPositionalFallback &&
(ae.getMessage().startsWith("[INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA]
Cannot write incompatible data for the table")
|| ae.getMessage().startsWith("Cannot write incompatible data to
table")) =>
planUtils.resolveOutputColumns(
catalogTable.catalogTableName,
sparkAdapter.getSchemaUtils.toAttributes(expectedSchema), query, byName =
false, conf)
diff --git
a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/command/CreateHoodieTableAsSelectCommand.scala
b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/command/CreateHoodieTableAsSelectCommand.scala
index 932cd900580c..459225db2656 100644
---
a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/command/CreateHoodieTableAsSelectCommand.scala
+++
b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/command/CreateHoodieTableAsSelectCommand.scala
@@ -35,6 +35,7 @@ import org.apache.spark.sql.classic.SparkSession
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.command.DataWritingCommand
import org.apache.spark.sql.execution.metric.SQLMetric
+import
org.apache.spark.sql.hudi.HoodieOptionConfig.mapSqlOptionsToDataSourceWriteConfigs
import org.apache.spark.sql.hudi.command.exception.HoodieAnalysisException
import scala.collection.JavaConverters._
@@ -45,7 +46,8 @@ import scala.collection.JavaConverters._
case class CreateHoodieTableAsSelectCommand(
table: CatalogTable,
mode: SaveMode,
- query: LogicalPlan) extends DataWritingCommand {
+ query: LogicalPlan,
+ appendToExistingTable: Boolean = false) extends DataWritingCommand {
override def innerChildren: Seq[QueryPlan[_]] = Seq(query)
override lazy val metrics: Map[String, SQLMetric] =
HoodieCommandMetrics.metrics
@@ -64,7 +66,12 @@ case class CreateHoodieTableAsSelectCommand(
val qualifiedTableIdentifier = table.identifier.copy(database = Some(db))
val tableName = qualifiedTableIdentifier.unquotedString
- if (sessionState.catalog.tableExists(qualifiedTableIdentifier)) {
+ val tableExists =
sessionState.catalog.tableExists(qualifiedTableIdentifier)
+ if (mode == SaveMode.Append && tableExists != appendToExistingTable) {
+ throw new HoodieAnalysisException(s"Table $tableName existence changed
after analysis. Retry the append.")
+ }
+
+ if (tableExists) {
checkState(mode != SaveMode.Overwrite,
s"Expect the table $tableName has been dropped when the save mode is
Overwrite")
@@ -78,6 +85,19 @@ case class CreateHoodieTableAsSelectCommand(
return Seq.empty
// scalastyle:on
}
+
+ if (mode == SaveMode.Append) {
+ val existingTable =
sessionState.catalog.getTableMetadata(qualifiedTableIdentifier)
+ val partitionSpec = existingTable.partitionColumnNames.map((_,
None)).toMap
+ InsertIntoHoodieTableCommand.run(sparkSession, existingTable, plan,
partitionSpec,
+ overwrite = false,
+ extraOptions =
mapSqlOptionsToDataSourceWriteConfigs(table.storage.properties --
needFilterProps - "path"),
+ metrics = metrics)
+ DataWritingCommand.propogateMetrics(sparkSession.sparkContext, this,
metrics)
+ // scalastyle:off
+ return Seq.empty
+ // scalastyle:on
+ }
}
// Remove some properties should not be used
diff --git
a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala
b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala
index b1a2a3a2c6e7..5ed8ec54980f 100644
---
a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala
+++
b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala
@@ -149,11 +149,13 @@ object InsertIntoHoodieTableCommand extends Logging with
ProvidesHoodieConfig wi
* @param catalogTable catalog table
* @param partitionsSpec partition spec specifying static/dynamic partition
values
* @param conf Spark's [[SQLConf]]
+ * @param allowPositionalFallback whether to retry positional matching when
column names do not match
*/
def alignQueryOutput(query: LogicalPlan,
catalogTable: HoodieCatalogTable,
partitionsSpec: Map[String, Option[String]],
- conf: SQLConf): LogicalPlan = {
+ conf: SQLConf,
+ allowPositionalFallback: Boolean = true):
LogicalPlan = {
val targetPartitionSchema = catalogTable.partitionSchema
val staticPartitionValues = filterStaticPartitionValues(partitionsSpec)
@@ -168,7 +170,7 @@ object InsertIntoHoodieTableCommand extends Logging with
ProvidesHoodieConfig wi
// since such columns wouldn't be otherwise specified w/in the query
itself and therefore couldn't be matched
// positionally for example
val expectedQueryColumns =
catalogTable.tableSchemaWithoutMetaFields.filterNot(f =>
staticPartitionValues.contains(f.name))
- val coercedQueryOutput =
coerceQueryOutputColumns(StructType(expectedQueryColumns), cleanedQuery,
catalogTable, conf)
+ val coercedQueryOutput =
coerceQueryOutputColumns(StructType(expectedQueryColumns), cleanedQuery,
catalogTable, conf, allowPositionalFallback)
// After potential reshaping validate that the output of the query
conforms to the table's schema
validate(removeMetaFields(coercedQueryOutput.schema), partitionsSpec,
catalogTable)
@@ -180,7 +182,8 @@ object InsertIntoHoodieTableCommand extends Logging with
ProvidesHoodieConfig wi
private def coerceQueryOutputColumns(expectedSchema: StructType,
query: LogicalPlan,
catalogTable: HoodieCatalogTable,
- conf: SQLConf): LogicalPlan = {
+ conf: SQLConf,
+ allowPositionalFallback: Boolean):
LogicalPlan = {
val planUtils = sparkAdapter.getCatalystPlanUtils
try {
planUtils.resolveOutputColumns(
@@ -188,7 +191,7 @@ object InsertIntoHoodieTableCommand extends Logging with
ProvidesHoodieConfig wi
} catch {
// NOTE: In case matching by name didn't match the query output, we will
attempt positional matching
// SPARK-42309 Error message changed in Spark 3.5.0 so we need to match
two strings here
- case ae: AnalysisException if
(ae.getMessage().startsWith("[INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA]
Cannot write incompatible data for the table")
+ case ae: AnalysisException if allowPositionalFallback &&
(ae.getMessage().startsWith("[INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA]
Cannot write incompatible data for the table")
|| ae.getMessage().startsWith("Cannot write incompatible data to
table")) =>
planUtils.resolveOutputColumns(
catalogTable.catalogTableName,
sparkAdapter.getSchemaUtils.toAttributes(expectedSchema), query, byName =
false, conf)
