voonhous commented on code in PR #19791:
URL: https://github.com/apache/hudi/pull/19791#discussion_r3886679111
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -617,8 +618,29 @@ class HoodieSparkSqlWriterInternal {
}
/**
- * Resolve wildcards in partitions
+ * Reject insert overwrite combined with non-blocking concurrency control.
*
+ * Insert overwrite reuses the deterministic bucket file id under
non-blocking concurrency
+ * control, but the replace commit records that same file id as replaced.
The file system view
+ * hides a replaced file group by file id (ignoring the replace instant), so
the freshly
+ * overwritten data would become invisible. Reject the combination to avoid
data loss.
+ */
+ private def validateNonBlockingConcurrencyControl(hoodieConfig:
HoodieConfig, operation: WriteOperationType): Unit = {
+ val isNonBlockingConcurrencyControl =
WriteConcurrencyMode.isNonBlockingConcurrencyControl(
+
hoodieConfig.getStringOrDefault(HoodieWriteConfig.WRITE_CONCURRENCY_MODE))
+ val rowWriterOverwriteType =
Option(hoodieConfig.getString(HoodieInternalConfig.BULKINSERT_OVERWRITE_OPERATION_TYPE))
+ .map(WriteOperationType.fromValue)
+ .orNull
+ val isInsertOverwrite = operation == WriteOperationType.INSERT_OVERWRITE ||
Review Comment:
Resolving: `BUCKET_RESCALE` is still exempt from this early guard, but
`DatasetBucketRescaleCommitActionExecutor.preExecute` calls `super`, which
reaches `preWrite` with the inherited `INSERT_OVERWRITE`, so the backstop now
rejects rescale under NB-CC (after `startCommit`, before the hashing config is
saved). Good enough for this PR.
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java:
##########
@@ -1468,6 +1469,48 @@ void testInsertOverwrite(String indexType,
HoodieTableType tableType) {
assertRowsEquals(result5, expected);
}
+ @Test
+ void testInsertOverwriteWithNonBlockingConcurrencyControlThrows() {
+ TableEnvironment tableEnv = batchTableEnv;
+ // MOR + simple bucket index + non-blocking concurrency control.
+ String hoodieTableDDL = sql("t1")
+ .option(FlinkOptions.PATH, tempFile.getAbsolutePath())
+ .options(getDefaultKeys())
+ .option(FlinkOptions.TABLE_TYPE, FlinkOptions.TABLE_TYPE_MERGE_ON_READ)
+ .option(FlinkOptions.INDEX_TYPE, HoodieIndex.IndexType.BUCKET.name())
+ .option(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS, 1)
+ .option(HoodieWriteConfig.WRITE_CONCURRENCY_MODE.key(),
+ WriteConcurrencyMode.NON_BLOCKING_CONCURRENCY_CONTROL.name())
+ .end();
+ tableEnv.executeSql(hoodieTableDDL);
+
+ // Whole-table overwrite resolves to INSERT_OVERWRITE_TABLE
+ final String overwriteTable = "insert overwrite t1 values\n"
+ + "('id1','Danny',24,TIMESTAMP '1970-01-01 00:00:01', 'par1')\n";
+ // static-partition overwrite resolves to INSERT_OVERWRITE
+ final String overwriteStaticPartition = "insert overwrite t1
partition(`partition`='par1') values\n"
+ + "('id1','Danny',24,TIMESTAMP '1970-01-01 00:00:01')\n";
+ // dynamic-partition overwrite resolves to INSERT_OVERWRITE
+ final String overwriteDynamicPartition = "insert overwrite t1
partition(`partition`='par1') values\n"
+ + "('id1','Danny',24,TIMESTAMP '1970-01-01 00:00:01')\n";
Review Comment:
Verified at 7c36e9ba: dynamic leg uses the `OPTIONS` hint. Resolving.
--
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]
