voonhous commented on code in PR #19163:
URL: https://github.com/apache/hudi/pull/19163#discussion_r3871265568
##########
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:
Done in 72a412e96a62. The static arm now also runs `partition (dt)` without
a value and asserts the rejection; the two-element split goes through
`resolveTargetPartitions` on the Dataset executor.
--
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]
