voonhous commented on code in PR #19163:
URL: https://github.com/apache/hudi/pull/19163#discussion_r3871122443
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestInsertTable2.scala:
##########
@@ -693,6 +695,164 @@ class TestInsertTable2 extends HoodieSparkSqlTestBase {
}
}
+ test("Test bulk insert with insert overwrite partition in dynamic mode") {
+ // The static-mode test above resolves the partitions to replace from the
PARTITION clause;
+ // dynamic mode resolves them from the incoming rows, which on the
row-writer path is
+ // DatasetBulkInsertOverwriteCommitActionExecutor reading
_hoodie_partition_path off the
+ // prepared dataset.
+ withSQLConf(SPARK_SQL_INSERT_INTO_OPERATION.key ->
WriteOperationType.BULK_INSERT.value(),
+ "hoodie.datasource.overwrite.mode" -> "dynamic") {
+ withTempDir { tmp =>
+ withTable(generateTableName) { tableName =>
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | dt string
+ |) using hudi
+ | tblproperties (
+ | type = 'cow',
+ | primaryKey = 'id'
+ | )
+ | partitioned by (dt)
+ | location '$tablePath'
+ """.stripMargin)
+ spark.sql(s"insert into $tableName values(1, 'a1', 10,
'2021-07-18'), (2, 'a2', 20, '2021-07-19')")
+
+ // Only the partition present in the incoming rows is replaced;
2021-07-19 survives.
+ spark.sql(s"insert overwrite table $tableName partition (dt)
values(3, 'b1', 11, '2021-07-18')")
+ checkAnswer(s"select id, name, price, dt from $tableName order by
id")(
+ Seq(2, "a2", 20.0, "2021-07-19"),
+ Seq(3, "b1", 11.0, "2021-07-18")
+ )
+ assertResult(WriteOperationType.INSERT_OVERWRITE) {
+ getLastCommitMetadata(spark, tablePath).getOperationType
+ }
+ assertResult(Set("dt=2021-07-18"))(getReplacedPartitions(tablePath))
+ }
+ }
+ }
+ }
+
+ test("Test bulk insert with insert overwrite against pending clustering") {
+ // The only coverage of rejectIfOverlappingPendingClustering on the
row-writer path; the RDD
+ // path is covered by TestInsertOverwriteWithClustering. Clustering is
pending on one of two
+ // partitions: overwriting the other must succeed and leave the plan
pending, overwriting the
+ // clustered one must be rejected before anything is written. Static mode
resolves the target
+ // partition from the PARTITION clause and dynamic mode from the rows, so
both arms of
+ // DatasetBulkInsertOverwriteCommitActionExecutor.resolveTargetPartitions
are driven.
+ Seq("static", "dynamic").foreach { overwriteMode =>
+ withSQLConf(SPARK_SQL_INSERT_INTO_OPERATION.key ->
WriteOperationType.BULK_INSERT.value(),
+ "hoodie.datasource.overwrite.mode" -> overwriteMode) {
+ withTempDir { tmp =>
+ withTable(generateTableName) { tableName =>
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | dt string
+ |) using hudi
+ | tblproperties (
+ | type = 'cow',
+ | primaryKey = 'id'
+ | )
+ | partitioned by (dt)
+ | location '$tablePath'
+ """.stripMargin)
+ // Two commits into 2021-07-18 give the size-based planner two
file groups to cluster.
+ spark.sql(s"insert into $tableName values(1, 'a1', 10,
'2021-07-18')")
+ spark.sql(s"insert into $tableName values(2, 'a2', 20,
'2021-07-18')")
+ spark.sql(s"insert into $tableName values(3, 'a3', 30,
'2021-07-19')")
+ spark.sql(s"call run_clustering(table => '$tableName', op =>
'schedule', selected_partitions => 'dt=2021-07-18')")
+ val metaClient = createMetaClient(spark, tablePath)
+
assertResult(1)(metaClient.getActiveTimeline.filterPendingClusteringTimeline().countInstants())
+
+ def overwriteSql(dt: String, id: Int, name: String, price: Int):
String = overwriteMode match {
+ case "static" => s"insert overwrite table $tableName partition
(dt = '$dt') values($id, '$name', $price)"
+ case "dynamic" => s"insert overwrite table $tableName partition
(dt) values($id, '$name', $price, '$dt')"
+ }
+
+ // Non-overlapping partition: the overwrite goes through and the
plan stays pending.
+ spark.sql(overwriteSql("2021-07-19", 4, "b1", 40))
Review Comment:
Static and dynamic assert identical outcomes here, so a static resolution
that silently fell through to the row-derived branch would still pass. A
statement with no partition value discriminates: under static, `partition (dt)`
resolves to every existing partition (a two-element
`STATIC_OVERWRITE_PARTITION_PATHS`, the exact HUDI-7183 / #10254 shape, whose
regression test only runs the RDD executor) and must be rejected because it
overlaps `dt=2021-07-18`; under dynamic it is just the row's partition. Could
we add `insert overwrite table t partition (dt) values(5, 'c1', 50,
'2021-07-19')` to the static arm and assert the rejection?
--
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]
