nsivabalan commented on code in PR #20423:
URL: https://github.com/apache/hudi/pull/20423#discussion_r4364729551
##########
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:
**Options passed with `.option(...)` are dropped on append.** V1
`saveAsTable` puts the writer options in `table.storage.properties`, but this
branch builds the write from `existingTable` (the catalog entry) and passes no
`extraOptions`, so `ProvidesHoodieConfig.combineOptions` never sees them. For
example,
`df.write.format("hudi").option("hoodie.spark.sql.insert.into.operation",
"upsert").mode("append").saveAsTable(t)` runs the default `insert`, so a row
whose key already exists lands as a duplicate, with no error. Per-write tuning
(parallelism, `bulk_insert`) is lost the same way, and SDP materialized views
(#20405) rely on it. The CTAS path below keeps these options because it builds
`updatedTable` from `table`.
Could we pass them through, e.g. `extraOptions = table.storage.properties --
needFilterProps - "path"`? It may also be worth rejecting options that conflict
with the persisted table identity (record key, ordering fields, table type, key
generator) with a clear error instead of letting them be silently overridden.
##########
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:
**A `partitionBy(...)` that differs from the existing table is silently
ignored.** `partitionSpec` comes from `existingTable`, so
`df.write.format("hudi").partitionBy("name").mode("append").saveAsTable(t)` on
a table partitioned by `dt` succeeds and writes by `dt`. For non-Hudi tables,
Spark's `PreprocessTableCreation` rejects this ("Specified partitioning does
not match that of the existing table"), but it doesn't run here because
`CreateTable` is converted to this command first. Could we fail when
`table.partitionColumnNames` is non-empty and differs from
`existingTable.partitionColumnNames`?
--
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]
