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 reject this combination at the Flink option/planner boundary,
before table initialization? Keep this shared pipeline guard for
HoodieFlinkStreamer and the other callers, as discussed in the earlier review.
However, HoodieTableSink.getSinkRuntimeProvider already calls
StreamerUtil.initTableFromClientIfNecessary before reaching here; with
incremental job-graph initialization or partition-level simple bucket indexing,
that can create the table before the invalid options are rejected. I would call
OptionsResolver.checkNonBlockingConcurrencyControl from
HoodieTableFactory.sanityCheck for explicitly configured write.operation, and
reject overwrite == true && NBCC at the beginning of
HoodieTableSink.applyOverwrite, before mutating the sink state. The factory
check alone is insufficient for SQL INSERT OVERWRITE: write.operation can still
be the default upsert there, and applyOverwrite changes it later. This
preserves coverage of all pipeline callers while letting table/SQL u
sers fail before initialization. Factory and sink-level tests could pin both
rejection points.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java:
##########
@@ -586,6 +587,12 @@ public void preWrite(String instantTime,
WriteOperationType writeOperationType,
*/
public void preWrite(String instantTime, WriteOperationType
writeOperationType,
HoodieTableMetaClient metaClient,
Option<HoodieData<HoodieRecord<T>>> recordsOpt) {
+ // Engine-agnostic backstop rejecting insert overwrite under non-blocking
concurrency control.
+ // Complements the engine-specific guards by covering entry points that
bypass them (e.g. Hudi
+ // Streamer, which reaches the write client directly).
+ WriteConcurrencyMode.checkInsertOverwriteSupported(
+ config.getWriteConcurrencyMode().isNonBlockingConcurrencyControl(),
+ WriteOperationType.isOverwrite(writeOperationType));
Review Comment:
Could we add an early Spark HoodieStreamer option check alongside this
backstop? StreamSync.writeToSinkAndDoMetaSync calls startCommit before
writeToSink dispatches to insertOverwrite/insertOverwriteTable, and those
client methods call initTable before reaching preWrite. This guard prevents the
overwrite, but the invalid NBCC configuration has already caused client
initialization and commit-instant creation, unlike the Spark SQL guard.
StreamSync.getHoodieClientConfigAndWriterSchema already has the built
HoodieWriteConfig and cfg.operation; calling the shared validator immediately
after builder.build() would reject both overwrite variants before constructing
the write client. Please use cfg.operation rather than assuming the datasource
operation property is present. Keep this preWrite check as protection for
direct client callers, and add a Streamer test asserting that rejection creates
no new timeline instant.
--
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]
