danny0405 commented on code in PR #19791:
URL: https://github.com/apache/hudi/pull/19791#discussion_r3891546793
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java:
##########
@@ -500,6 +500,9 @@ public static DataStream<HoodieFlinkInternalRow>
rowDataToHoodieRecord(Configura
public static DataStream<RowData> hoodieStreamWrite(Configuration conf,
RowType rowType,
DataStream<HoodieFlinkInternalRow> dataStream) {
+ // Validate the finalized write operation (after
overwrite/static-partition resolution) at the
+ // single point all production callers (table sink, flink streamer, sink
v2) funnel through.
+ OptionsResolver.checkNonBlockingConcurrencyControl(conf);
Review Comment:
Could we also call
`OptionsResolver.checkNonBlockingConcurrencyControl(conf)` at the very
beginning of `HoodieTableSink.getSinkRuntimeProvider()`, before returning the
`DataStreamSinkProviderAdapter` lambda?
By that point, the planner has applied `applyOverwrite()` and
`applyStaticPartition()`, so `conf` contains the resolved operation. This gives
the Table/SQL path a single early validation point for both explicit
`write.operation` options and SQL `INSERT OVERWRITE`, without adding separate
checks in `HoodieTableFactory.sanityCheck()` and `applyOverwrite()`.
The placement before the lambda matters: the lambda performs client-ID setup
and calls `StreamerUtil.initTableFromClientIfNecessary()` before reaching this
pipeline guard. With incremental job-graph initialization or partition-level
simple bucket indexing, the latter can create the table before the invalid
combination is rejected.
Please keep this shared pipeline guard as well, since `HoodieFlinkStreamer`
and direct pipeline callers bypass `HoodieTableSink`. A sink-level test could
assert that `getSinkRuntimeProvider()` itself rejects the resolved overwrite
operation under NBCC, without invoking the returned provider.
Updated to recommend this single sink-level check instead of my earlier
factory-plus-`applyOverwrite()` suggestion.
--
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]
