voonhous commented on code in PR #19791:
URL: https://github.com/apache/hudi/pull/19791#discussion_r3886286226
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java:
##########
@@ -79,6 +79,10 @@ public HoodieTableSink(Configuration conf, ResolvedSchema
schema, boolean overwr
public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
return (DataStreamSinkProviderAdapter) dataStream -> {
+ // validate the finalized write operation (after #applyOverwrite /
#applyStaticPartition)
+ // before any table initialization takes place.
+ OptionsResolver.checkNonBlockingConcurrencyControl(conf);
Review Comment:
This misses `HoodieFlinkStreamer --op insert_overwrite`: it calls
`Pipelines.bootstrap` + `Pipelines.hoodieStreamWrite` directly
(`HoodieFlinkStreamer.java:107-108`) and never passes through
`HoodieTableSink`, and Flink has no partitioner-level backstop like Spark's, so
that path still loses data. `PipelinesV2.sink` also has no production caller,
so the V2 copy is unreachable today. Could we move the single call to the top
of `Pipelines.hoodieStreamWrite`, which all three callers funnel through
(`HoodieTableSink:130`, `PipelinesV2:135`, `HoodieFlinkStreamer:108`), beside
the consistent-hashing overwrite rejection at `Pipelines.java:578-581`?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala:
##########
@@ -232,23 +232,23 @@ trait ProvidesHoodieConfig extends Logging {
}
)
- val overwriteTableOpts = if
(operation.equals(BULK_INSERT_OPERATION_OPT_VAL)) {
+ val overwriteTableOpts = if
(operation.equalsIgnoreCase(BULK_INSERT_OPERATION_OPT_VAL)) {
Review Comment:
`deduceOverwriteConfig` (`:345-346`) still compares case-sensitively, with a
worse outcome: `hoodie.datasource.write.operation=INSERT_OVERWRITE` (uppercase)
plus `insert overwrite t values (...)` on a partitioned table makes
`isOverwriteOperation` false, so `isStaticOverwrite` and `isOverWriteTable`
turn true, the mode becomes `SaveMode.Overwrite`, and `handleSaveModes` deletes
the table path (`HoodieSparkSqlWriter.scala:925-930`); lowercase gives
`SaveMode.Append` and a dynamic partition overwrite. Since
`WriteOperationType.fromValue` accepts any case, could we lower-case
`operation` once (`Locale.ROOT`) in both methods instead of converting
comparisons one at a time?
##########
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 = {
Review Comment:
Every overwrite entry point on every engine reaches
`BaseHoodieWriteClient.preWrite` with the overwrite op:
`SparkRDDWriteClient:258,274`, `HoodieFlinkWriteClient:294,307`, and the row
writer via `BaseDatasetBulkInsertCommitActionExecutor:86`
(`getWriteOperationType()` is `INSERT_OVERWRITE`). Hudi Streamer
(`StreamSync.java:1118-1124`) bypasses this writer and today only fails via the
partitioner check, after the requested and inflight replacecommit and the
workload-profile shuffle. Could we add
`config.isNonBlockingConcurrencyControl() &&
WriteOperationType.isOverwrite(writeOperationType)` to `preWrite` as the
engine-agnostic backstop, keeping this guard and the Flink one for the early,
clean-timeline error?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestInsertTable5.scala:
##########
@@ -358,4 +358,80 @@ class TestInsertTable5 extends HoodieSparkSqlTestBase {
}
}
}
+
+ test("Test Insert Overwrite With Non Blocking Concurrency Control Is
Rejected") {
Review Comment:
This test passes on master without the `HoodieSparkSqlWriter` change:
`insert into` with the forced op takes the RDD path, where
`BaseSparkBucketIndexBucketInfoGetter.java:40-42` already throws the
byte-identical message, and `checkExceptionContain` matches it via root cause
(`HoodieSparkSqlTestBase.scala:241`). The behavioural delta of the new guard is
that nothing lands on the timeline; the old check fires after the requested and
inflight replacecommit. Could we use plain `insert overwrite $tableName` here
(the DML users actually run, currently untested) and assert
`HoodieClientTestUtils.createMetaClient(spark,
tablePath).getActiveTimeline.empty()` after the failure, so the test is red
against the old late check?
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/ITTestDataStreamV2Write.java:
##########
@@ -100,6 +107,34 @@ public void testAppendWrite() throws Exception {
writeAndCheckExpected(conf, "append_write", 1);
}
+ @ParameterizedTest
Review Comment:
nit: this spends the class-level `FlinkMiniCluster` on a synchronous config
assertion against `composePipeline`, whose only caller is `HoodieSink`, which
nothing in production constructs, and it re-asserts what
`TestOptionsResolver#testCheckNonBlockingConcurrencyControl` already pins. If
the check moves into `Pipelines.hoodieStreamWrite` this test goes away with it;
otherwise could it move to `TestPipelinesV2` (reusing its `input` stream
instead of the `(RowData) null` map, and outside its
`mockStatic(OptionsResolver)` block, which would stub the guard away)?
##########
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";
+
+ for (String overwriteSql : new String[] {overwriteTable,
overwriteStaticPartition, overwriteDynamicPartition}) {
+ Throwable thrown = assertThrows(Throwable.class, () ->
tableEnv.executeSql(overwriteSql));
+ assertTrue(exceptionChainContains(thrown,
+ "Insert overwrite is not supported with non-blocking concurrency
control"),
+ "Unexpected exception: " + thrown);
+ }
+ }
+
+ private static boolean exceptionChainContains(Throwable thrown, String
message) {
Review Comment:
nit, feel free to ignore: Flink ships this walker as
`org.apache.flink.util.ExceptionUtils.findThrowableWithMessage(thrown, msg)`,
and the test tree already has three private copies
(`ITTestVectorDataSource.java:787`,
`ITTestVariantCrossEngineCompatibility.java:173`,
`ITTestDataStreamWrite.java:600`). Could the call site use
`ExceptionUtils.findThrowableWithMessage(thrown, "...").isPresent()` and this
helper be dropped?
##########
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:
nit: this string is byte-identical to `overwriteStaticPartition`, so the
loop runs the static statement twice and the dynamic arm
(`HoodieTableSink.java:181-182`, selected by `write.partition.overwrite.mode`)
is never reached. Could we use the hint form already used at `:1437`?
```suggestion
final String overwriteDynamicPartition = "insert overwrite t1 /*+
OPTIONS('write.partition.overwrite.mode'='dynamic') */ values\n"
+ "('id1','Danny',24,TIMESTAMP '1970-01-01 00:00:01', 'par1')\n";
```
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestInsertTable5.scala:
##########
@@ -358,4 +358,80 @@ class TestInsertTable5 extends HoodieSparkSqlTestBase {
}
}
}
+
+ test("Test Insert Overwrite With Non Blocking Concurrency Control Is
Rejected") {
+ Seq("INSERT_OVERWRITE", "INSERT_OVERWRITE_TABLE").foreach { operation =>
+ withSQLConf(
+ "hoodie.write.concurrency.mode" -> "NON_BLOCKING_CONCURRENCY_CONTROL",
+ "hoodie.datasource.write.operation" -> operation
+ ) {
+ withTempDir { tmp =>
+ withTable(generateTableName) { tableName =>
+ val tablePath = s"""${tmp.getCanonicalPath}/$tableName"""
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'mor',
+ | preCombineField = 'ts',
+ | hoodie.index.type = 'BUCKET',
+ | hoodie.index.bucket.engine = 'SIMPLE',
+ | hoodie.bucket.index.hash.field = 'id',
+ | hoodie.bucket.index.num.buckets = 1)
+ | location '${tablePath}'
+ | """.stripMargin)
+
+ checkExceptionContain(
+ s"""insert into $tableName values
+ | (1, 'a1', 10, 1000)
+ | """.stripMargin)(
+ "Insert overwrite is not supported with non-blocking concurrency
control")
+ }
+ }
+ }
+ }
+ }
+
+ test("Test Insert Overwrite (bulk_insert mode) With Non Blocking Concurrency
Control Is Rejected") {
+ withSQLConf(
+ "hoodie.write.concurrency.mode" -> "NON_BLOCKING_CONCURRENCY_CONTROL",
+ "hoodie.datasource.write.operation" -> "BULK_INSERT"
Review Comment:
This leg discriminates only because of the `equalsIgnoreCase` change: on
master, uppercase `BULK_INSERT` + `insert overwrite` fell through to `Map()` in
`buildHoodieInsertConfig`, `BULKINSERT_OVERWRITE_OPERATION_TYPE` was never set,
and `DatasetBulkInsertCommitActionExecutor` ran a plain commit, so the
overwrite silently became an append. Every existing test spells it lowercase
(`TestInsertTable3.scala:274,362,407,461`), so that fix has no positive
assertion. Could we add an uppercase leg to `TestInsertTable3:405` ("Test
Insert Overwrite Bucket Index Table") asserting the rows are replaced, and call
the fix out in the description?
##########
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:
`BUCKET_RESCALE` is not matched here, but it shares the hazard:
`DatasetBucketRescaleCommitActionExecutor` extends the overwrite executor,
replaces `getAllExistingFileIds`, and the row-writer helper hands out the fixed
NB-CC id (`BucketBulkInsertDataInternalWriterHelper.java:137`), so every bucket
number that exists before and after the rescale collides the same way. Not
verified at runtime. Should this reject any non-null
`BULKINSERT_OVERWRITE_OPERATION_TYPE` under NB-CC, or is the exemption
deliberate? (A `preWrite` backstop would catch it anyway, since that executor
reports `INSERT_OVERWRITE`.)
##########
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 ||
+ operation == WriteOperationType.INSERT_OVERWRITE_TABLE ||
+ rowWriterOverwriteType == WriteOperationType.INSERT_OVERWRITE ||
+ rowWriterOverwriteType == WriteOperationType.INSERT_OVERWRITE_TABLE
+ if (isNonBlockingConcurrencyControl && isInsertOverwrite) {
+ throw new HoodieException("Insert overwrite is not supported with
non-blocking concurrency control")
Review Comment:
nit: this message literal now exists in three places
(`OptionsResolver.java:462`, `BaseSparkBucketIndexBucketInfoGetter.java:41`,
here) and is thrown as `HoodieException` on Spark but
`IllegalArgumentException` on Flink and in the partitioner. Could we hoist it
to one constant (e.g. next to `WriteConcurrencyMode`) and reuse it in all
three, so a future reword cannot drift?
##########
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 ||
+ operation == WriteOperationType.INSERT_OVERWRITE_TABLE ||
+ rowWriterOverwriteType == WriteOperationType.INSERT_OVERWRITE ||
+ rowWriterOverwriteType == WriteOperationType.INSERT_OVERWRITE_TABLE
+ if (isNonBlockingConcurrencyControl && isInsertOverwrite) {
+ throw new HoodieException("Insert overwrite is not supported with
non-blocking concurrency control")
+ }
+ }
+
+ /**
* @param partitions list of partitions that may contain wildcards
Review Comment:
nit: the new scaladoc consumed the summary line of
`resolvePartitionWildcards`, leaving this block headless and indented by three
spaces.
```suggestion
/**
* Resolve wildcards in partitions
*
* @param partitions list of partitions that may contain wildcards
```
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/insert/TestInsertTable5.scala:
##########
@@ -358,4 +358,80 @@ class TestInsertTable5 extends HoodieSparkSqlTestBase {
}
}
}
+
+ test("Test Insert Overwrite With Non Blocking Concurrency Control Is
Rejected") {
+ Seq("INSERT_OVERWRITE", "INSERT_OVERWRITE_TABLE").foreach { operation =>
+ withSQLConf(
+ "hoodie.write.concurrency.mode" -> "NON_BLOCKING_CONCURRENCY_CONTROL",
+ "hoodie.datasource.write.operation" -> operation
+ ) {
+ withTempDir { tmp =>
+ withTable(generateTableName) { tableName =>
+ val tablePath = s"""${tmp.getCanonicalPath}/$tableName"""
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'mor',
+ | preCombineField = 'ts',
+ | hoodie.index.type = 'BUCKET',
+ | hoodie.index.bucket.engine = 'SIMPLE',
+ | hoodie.bucket.index.hash.field = 'id',
+ | hoodie.bucket.index.num.buckets = 1)
+ | location '${tablePath}'
+ | """.stripMargin)
+
+ checkExceptionContain(
+ s"""insert into $tableName values
+ | (1, 'a1', 10, 1000)
+ | """.stripMargin)(
+ "Insert overwrite is not supported with non-blocking concurrency
control")
+ }
+ }
+ }
+ }
+ }
+
+ test("Test Insert Overwrite (bulk_insert mode) With Non Blocking Concurrency
Control Is Rejected") {
+ withSQLConf(
+ "hoodie.write.concurrency.mode" -> "NON_BLOCKING_CONCURRENCY_CONTROL",
+ "hoodie.datasource.write.operation" -> "BULK_INSERT"
+ ) {
+ withTempDir { tmp =>
+ withTable(generateTableName) { tableName =>
+ val tablePath = s"""${tmp.getCanonicalPath}/$tableName"""
+ spark.sql(
+ s"""
+ |create table $tableName (
Review Comment:
nit, feel free to ignore: the two tests repeat this 17-line DDL verbatim,
and bucket-index insert-overwrite tests live in `TestInsertTable3` (`:216,
:272, :405, :574`) rather than here. Could the two collapse into one test in
`TestInsertTable3` looping over `Seq((sqlConf, dml))` with a single local DDL?
(Also a double space before `->` at `:366` and `:404`.)
--
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]
