This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 85033e339f66 test(spark): add write and commit path coverage (#19163)
85033e339f66 is described below
commit 85033e339f667c0d0bd8ed16de465bf27c517e3a
Author: Y Ethan Guo <[email protected]>
AuthorDate: Thu Aug 27 06:56:02 2026 -0700
test(spark): add write and commit path coverage (#19163)
The row-writer bulk_insert overwrite path in hudi-spark-common had no
SQL-level coverage of rejectIfOverlappingPendingClustering (#18829 only
drives the RDD path) or of the static, dynamic and unpartitioned arms of
DatasetBulkInsertOverwriteCommitActionExecutor.resolveTargetPartitions.
Tests, both in TestInsertTable2 and pinned to the row writer via
hoodie.datasource.write.row.writer.enable=true:
- "Test bulk insert with insert overwrite against pending clustering":
two partitions, clustering scheduled on one via run_clustering,
parameterized over static and dynamic overwrite mode. Overwriting the
non-clustered partition succeeds (INSERT_OVERWRITE, exactly one file
id replaced, plan still pending); overwriting the clustered one is
rejected by SparkRejectUpdateStrategy before any data file is written.
The static arm also rejects "partition (dt)" without a value, which
resolves to every partition.
- "Test bulk insert with insert overwrite on unpartitioned table against
pending clustering": a forced insert_overwrite (rejected), the deduced
INSERT_OVERWRITE_TABLE through
SparkInsertOverwriteTableCommitActionExecutor (rejected), and
SparkAllowUpdateStrategy (goes through, plan left pending).
Fix: HoodieSparkSqlWriter.bulkInsertAsRow now closes the write client
when executor.execute throws, mirroring the non-row-writer path. Every
rejection above used to leak the client and its embedded timeline
server.
Test base: HoodieSparkSqlTestBase clears the INMEMORY index's JVM-static
record-location map after every test, before the catalog cleanup so it
holds even if a drop throws, and gains getLastReplaceCommitMetadata next
to getLastCommitMetadata.
Not covered: the row-writer INSERT_OVERWRITE_TABLE variant deletes and
re-initializes the table before the check runs (#19771). Found on the
way, not fixed here: ALTER TABLE RENAME COLUMN / DROP COLUMN fail under
default configs on a schema-on-read table since #13595 (#19766), and the
pending-clustering check is bypassed when meta fields are not populated
(#19770).
---------
Co-authored-by: voon <[email protected]>
---
.../org/apache/hudi/HoodieSparkSqlWriter.scala | 9 +-
.../sql/hudi/common/HoodieSparkSqlTestBase.scala | 14 +-
.../sql/hudi/dml/insert/TestInsertTable2.scala | 162 ++++++++++++++++++++-
3 files changed, 181 insertions(+), 4 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
index 226d8dd1e568..042af12a2954 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
@@ -867,7 +867,14 @@ class HoodieSparkSqlWriterInternal {
throw new HoodieException(s"$mode with bulk_insert in row writer path
is not supported yet");
}
- val writeResult = executor.execute(df, tableConfig.isTablePartitioned)
+ val writeResult = try {
+ executor.execute(df, tableConfig.isTablePartitioned)
+ } catch {
+ case e: HoodieException =>
+ // close the write client in all cases
+ closeWriteClient(writeClient, tableConfig, parameters,
jsc.hadoopConfiguration())
+ throw e
+ }
try {
val (writeSuccessful, compactionInstant, clusteringInstant) =
commitAndPerformPostOperations(
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/HoodieSparkSqlTestBase.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/HoodieSparkSqlTestBase.scala
index 137620a4cdbe..64b9b1275e46 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/HoodieSparkSqlTestBase.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/HoodieSparkSqlTestBase.scala
@@ -22,7 +22,7 @@ import org.apache.hudi.HoodieFileIndex.DataSkippingFailureMode
import org.apache.hudi.common.avro.AvroSchemaCache
import org.apache.hudi.common.config.{HoodieCommonConfig,
HoodieMetadataConfig, HoodieStorageConfig}
import org.apache.hudi.common.engine.HoodieLocalEngineContext
-import org.apache.hudi.common.model.{FileSlice, HoodieAvroRecordMerger,
HoodieLogFile, HoodieRecord}
+import org.apache.hudi.common.model.{FileSlice, HoodieAvroRecordMerger,
HoodieLogFile, HoodieRecord, HoodieReplaceCommitMetadata}
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient,
TableSchemaResolver}
import org.apache.hudi.common.table.log.HoodieLogFormat
@@ -113,6 +113,12 @@ class HoodieSparkSqlTestBase extends FunSuite with
BeforeAndAfterAll {
try {
testFun
} finally {
+ // The INMEMORY index keeps a JVM-static record-location map; reset it
after every test so
+ // stale keys from an earlier test cannot misroute writes in a later
one. withRecordType
+ // clears it between record-type iterations, but only on success and
only for tests that use
+ // it, so a throwing or non-withRecordType INMEMORY test would
otherwise leak state here.
+ // Runs before the catalog cleanup so it holds even if a drop throws.
+ HoodieInMemoryHashIndex.clear()
val catalog = spark.sessionState.catalog
catalog.listDatabases().foreach { db =>
catalog.listTables(db).foreach { table =>
@@ -403,6 +409,12 @@ object HoodieSparkSqlTestBase {
metaClient.getActiveTimeline.getLastCommitMetadataWithValidData.get.getRight
}
+ def getLastReplaceCommitMetadata(spark: SparkSession, tablePath: String):
HoodieReplaceCommitMetadata = {
+ val metaClient = createMetaClient(spark, tablePath)
+ val lastInstant =
metaClient.getActiveTimeline.getLastCommitMetadataWithValidData.get.getLeft
+ metaClient.getActiveTimeline.readReplaceCommitMetadata(lastInstant)
+ }
+
def getLastCleanMetadata(spark: SparkSession, tablePath: String) = {
val metaClient = createMetaClient(spark, tablePath)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestInsertTable2.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestInsertTable2.scala
index cd2ea668b849..126b680a67a9 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestInsertTable2.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestInsertTable2.scala
@@ -19,24 +19,28 @@
package org.apache.spark.sql.hudi.dml.insert
-import org.apache.hudi.DataSourceWriteOptions.{COW_TABLE_TYPE_OPT_VAL,
MOR_TABLE_TYPE_OPT_VAL, PARTITIONPATH_FIELD, RECORDKEY_FIELD,
SPARK_SQL_INSERT_INTO_OPERATION, TABLE_TYPE}
+import org.apache.hudi.DataSourceWriteOptions.{COW_TABLE_TYPE_OPT_VAL,
ENABLE_ROW_WRITER, MOR_TABLE_TYPE_OPT_VAL, PARTITIONPATH_FIELD,
RECORDKEY_FIELD, SPARK_SQL_INSERT_INTO_OPERATION, TABLE_TYPE}
import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.HoodieSparkUtils
import org.apache.hudi.client.WriteClientTestUtils
+import
org.apache.hudi.client.clustering.update.strategy.SparkAllowUpdateStrategy
import org.apache.hudi.common.model.WriteOperationType
import org.apache.hudi.common.table.HoodieTableConfig
import org.apache.hudi.common.table.TableSchemaResolver
import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
import org.apache.hudi.common.testutils.HoodieTestUtils
+import org.apache.hudi.config.HoodieClusteringConfig
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
-import
org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase.getLastCommitMetadata
+import
org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase.{getLastCommitMetadata,
getLastReplaceCommitMetadata}
import java.io.File
+import scala.collection.JavaConverters._
+
class TestInsertTable2 extends HoodieSparkSqlTestBase {
test("Test Different Type of Partition Column") {
@@ -693,6 +697,160 @@ class TestInsertTable2 extends HoodieSparkSqlTestBase {
}
}
+ 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, with meta
+ // fields populated (the unpopulated case bypasses the check today, see
#19770).
+ Seq("static", "dynamic").foreach { overwriteMode =>
+ withSQLConf(SPARK_SQL_INSERT_INTO_OPERATION.key ->
WriteOperationType.BULK_INSERT.value(),
+ ENABLE_ROW_WRITER.key -> "true",
+ "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))
+ checkAnswer(s"select id, name, price, dt from $tableName order by
id")(
+ Seq(1, "a1", 10.0, "2021-07-18"),
+ Seq(2, "a2", 20.0, "2021-07-18"),
+ Seq(4, "b1", 40.0, "2021-07-19")
+ )
+ assertResult(WriteOperationType.INSERT_OVERWRITE) {
+ getLastCommitMetadata(spark, tablePath).getOperationType
+ }
+ val replaced = getLastReplaceCommitMetadata(spark,
tablePath).getPartitionToReplaceFileIds
+ assertResult(Set("dt=2021-07-19"))(replaced.keySet().asScala.toSet)
+ assertResult(1)(replaced.get("dt=2021-07-19").size())
+
assertResult(1)(metaClient.reloadActiveTimeline().filterPendingClusteringTimeline().countInstants())
+
+ // Overlapping partition: rejected by the default
SparkRejectUpdateStrategy before any
+ // write materializes, so no data files are written (the requested
instant is left behind).
+ checkExceptionContain(overwriteSql("2021-07-18", 1, "a1_new", 11))(
+ "Not allowed to update the clustering file group")
+ checkAnswer(s"select id, name, price, dt from $tableName order by
id")(
+ Seq(1, "a1", 10.0, "2021-07-18"),
+ Seq(2, "a2", 20.0, "2021-07-18"),
+ Seq(4, "b1", 40.0, "2021-07-19")
+ )
+
+ if (overwriteMode == "static") {
+ // Without a partition value, static mode expands `partition
(dt)` to every existing
+ // partition, a two-element STATIC_OVERWRITE_PARTITION_PATHS
(the HUDI-7183 shape),
+ // which overlaps the plan and is rejected. Dynamic mode
resolves only the row's
+ // partition, which is the successful statement above.
+ checkExceptionContain(s"insert overwrite table $tableName
partition (dt) values(5, 'c1', 50, '2021-07-19')")(
+ "Not allowed to update the clustering file group")
+ checkAnswer(s"select id, name, price, dt from $tableName order
by id")(
+ Seq(1, "a1", 10.0, "2021-07-18"),
+ Seq(2, "a2", 20.0, "2021-07-18"),
+ Seq(4, "b1", 40.0, "2021-07-19")
+ )
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("Test bulk insert with insert overwrite on unpartitioned table against
pending clustering") {
+ // Forcing INSERT_OVERWRITE drives the unpartitioned arm of
+ // DatasetBulkInsertOverwriteCommitActionExecutor.resolveTargetPartitions.
The deduced
+ // INSERT_OVERWRITE_TABLE is covered off the row writer only: on that path
SaveMode.Overwrite
+ // recreates the table before the executor runs, taking the pending plan
with it (#15984).
+ withSQLConf(SPARK_SQL_INSERT_INTO_OPERATION.key ->
WriteOperationType.BULK_INSERT.value(),
+ ENABLE_ROW_WRITER.key -> "true") {
+ withTempDir { tmp =>
+ withTable(generateTableName) { tableName =>
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double
+ |) using hudi
+ | tblproperties (
+ | type = 'cow',
+ | primaryKey = 'id'
+ | )
+ | location '$tablePath'
+ """.stripMargin)
+ spark.sql(s"insert into $tableName values(1, 'a1', 10)")
+ spark.sql(s"insert into $tableName values(2, 'a2', 20)")
+ spark.sql(s"call run_clustering(table => '$tableName', op =>
'schedule')")
+ assertResult(1)(createMetaClient(spark,
tablePath).getActiveTimeline.filterPendingClusteringTimeline().countInstants())
+
+ // Scoped to the overwrite statement only, so the seeding inserts
above stay inserts rather
+ // than overwrites and each seeds its own file group.
+ withSQLConf(DataSourceWriteOptions.OPERATION.key ->
WriteOperationType.INSERT_OVERWRITE.value()) {
+ checkExceptionContain(s"insert overwrite table $tableName
values(3, 'b1', 30)")(
+ "Not allowed to update the clustering file group")
+ }
+ checkAnswer(s"select id, name, price from $tableName order by id")(
+ Seq(1, "a1", 10.0),
+ Seq(2, "a2", 20.0)
+ )
+
+ // The deduced INSERT_OVERWRITE_TABLE is covered off the row writer,
since the row-writer
+ // variant recreates the table before the check (#15984): the
RDD-side
+ // SparkInsertOverwriteTableCommitActionExecutor override must
reject as well.
+ withSQLConf(SPARK_SQL_INSERT_INTO_OPERATION.key ->
WriteOperationType.INSERT.value()) {
+ checkExceptionContain(s"insert overwrite table $tableName
values(3, 'b1', 30)")(
+ "Not allowed to update the clustering file group")
+ }
+ checkAnswer(s"select id, name, price from $tableName order by id")(
+ Seq(1, "a1", 10.0),
+ Seq(2, "a2", 20.0)
+ )
+
+ // SparkAllowUpdateStrategy is the other branch of
rejectIfOverlappingPendingClustering:
+ // the overlap is deferred to conflict resolution, the overwrite
goes through and the plan
+ // is left pending.
+ withSQLConf(DataSourceWriteOptions.OPERATION.key ->
WriteOperationType.INSERT_OVERWRITE.value(),
+ HoodieClusteringConfig.UPDATES_STRATEGY.key ->
classOf[SparkAllowUpdateStrategy[_]].getName) {
+ spark.sql(s"insert overwrite table $tableName values(3, 'b1', 30)")
+ }
+ checkAnswer(s"select id, name, price from $tableName order by id")(
+ Seq(3, "b1", 30.0)
+ )
+ assertResult(1)(createMetaClient(spark,
tablePath).getActiveTimeline.filterPendingClusteringTimeline().countInstants())
+ }
+ }
+ }
+ }
+
test("Test combine before insert") {
Seq("cow", "mor").foreach { tableType =>
withSQLConf("hoodie.sql.bulk.insert.enable" -> "false",
"hoodie.merge.allow.duplicate.on.inserts" -> "false",
