danny0405 commented on code in PR #20423:
URL: https://github.com/apache/hudi/pull/20423#discussion_r4396489259
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala:
##########
@@ -434,6 +439,25 @@ 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")}")
+ }
+ 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)
Review Comment:
Fixed in f779a1e76122. The append path now calls `alignQueryOutput` with
`allowPositionalFallback = false` in both Spark implementations; SQL `INSERT
INTO` retains the default positional fallback. Added your `first_name` example
under both case-sensitivity settings, plus checks for reordered columns,
case-only differences, and SQL positional inserts. All 76 `TestCreateTable`
tests passed on Spark 4.1, and all 21 append regressions passed on Spark 3.5.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala:
##########
@@ -434,6 +439,25 @@ case class ResolveImplementationsEarly(spark:
SparkSession) extends Rule[Logical
}
}
+ private def alignAppendQuery(table: CatalogTable, query: LogicalPlan):
LogicalPlan = {
Review Comment:
Fixed in f779a1e76122. When `table.storage.locationUri` is supplied,
analysis now qualifies and normalizes both locations and rejects a mismatch
before writing. Added managed/external table tests that verify rejection leaves
the existing rows unchanged and does not create the other path. The tests also
accept equivalent unqualified paths and qualified URIs with a trailing slash.
Passed on Spark 3.5 and 4.1.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala:
##########
@@ -410,8 +412,11 @@ 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)
+ val alignedQuery = if (mode == SaveMode.Append &&
spark.sessionState.catalog.tableExists(table.identifier)) {
Review Comment:
Addressed in f779a1e76122 as well. The command now carries the analysis-time
`appendToExistingTable` decision and checks it against table existence at
execution. If the table appeared or disappeared, it fails with a retry message
before initialization or insertion. Added deterministic tests that analyze the
command, then create or drop the table before executing it; both pass on Spark
3.5 and 4.1.
##########
hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/CreateHoodieTableAsSelectCommand.scala:
##########
@@ -77,6 +77,17 @@ 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,
Review Comment:
Addressed in 331a3eacd56f: `ResolveImplementationsEarly` now aligns
existing-table append queries against the catalog schema. COW/MOR tests verify
BIGINT-to-INT coercion, rejection of extra columns without a new commit, and a
successful subsequent SQL insert. The follow-up f779a1e76122 also disables
positional fallback specifically for `saveAsTable` append. These regressions
pass on Spark 3.5 and 4.1.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
