nsivabalan commented on code in PR #20423:
URL: https://github.com/apache/hudi/pull/20423#discussion_r4395384091
##########
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 331a3eac, thanks.
##########
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:
**A `path` option that differs from the existing table's location is
silently ignored.** `saveAsTable` puts the path in `table.storage.locationUri`,
and nothing compares it with `existingTable.storage.locationUri`, so the rows
go to the existing table's location.
Reproduced on this branch:
`df.write.format("hudi").mode("append").option("path", other).saveAsTable(t)`
succeeds and writes to `t`'s location, and `other` is never created. For a
parquet table, Spark fails with "The location of the existing table ... doesn't
match the specified location ...". Since this method already mirrors Spark's
partitioning check, could it do the same for the location (when
`table.storage.locationUri` is defined and differs, after qualifying both)?
##########
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:
Non-blocking: whether the table exists is decided twice, here at analysis
and again in `CreateHoodieTableAsSelectCommand.run`. If another writer creates
the table between the two, `run` takes the new Append branch with a query that
was aligned for CTAS (`alignCtasQueryByPartitionOrder`), not aligned to the
table by name. The window is small for the DataFrame API. Passing the
analysis-time decision into the command (or failing in `run` when the two
disagree) would close it.
##########
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:
**A misnamed DataFrame column is written into a different column.**
`alignQueryOutput` comes from `INSERT INTO`: when by-name resolution fails,
`coerceQueryOutputColumns` retries by position. That is correct for SQL `INSERT
INTO`, but `saveAsTable` append is by name. Spark rejects an unresolved column
name for its own tables (`PreprocessTableCreation`), and Hudi preempts that
rule here.
Reproduced on this branch (Spark 3.5) with a table `(id INT, first STRING,
last STRING)`:
```scala
spark.sql("SELECT 1 AS id, 'Doe' AS last, 'John' AS first_name")
.write.format("hudi").mode("append").saveAsTable(t)
```
- **Hudi:** the append succeeds and stores `first = 'Doe'`, `last = 'John'`.
Both values land in the wrong column, with no error.
- **Same table `USING parquet`:** fails with `Cannot resolve 'first' given
input columns: [id, last, first_name]`.
Before this PR the call failed outright, so this is a new way to write wrong
data. Could the append path resolve by name only (`byName = true` with no
positional fallback, e.g. a flag on `alignQueryOutput`), so it fails like Spark
does? A test with a renamed column would cover it.
##########
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
Review Comment:
Addressed in 331a3eac, thanks.
--
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]
