hudi-agent commented on code in PR #19801:
URL: https://github.com/apache/hudi/pull/19801#discussion_r3969849155
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkStreamingMetadataWriteHandler.java:
##########
@@ -42,26 +42,21 @@ public HoodieData<WriteStatus>
streamWriteToMetadataTable(HoodieTable table, Hoo
Option<HoodieTableMetadataWriter> metadataWriterOpt =
getMetadataWriter(instantTime, table);
ValidationUtils.checkState(metadataWriterOpt.isPresent(),
"Cannot instantiate metadata writer for the table of interest " +
table.getMetaClient().getBasePath());
- return streamWriteToMetadataTable(dataTableWriteStatuses,
metadataWriterOpt.get(), table, instantTime, coalesceDivisorForDataTableWrites);
+ // Spark task retries and speculation are reconciled while the data-table
commit is finalized.
+ // Defer metadata generation until completeStreamingCommit can consume the
resulting committed
+ // HoodieCommitMetadata. This deliberately reuses the HUDI-8564
committed-output path instead
+ // of deriving RLI/SI locations from pre-commit WriteStatus task attempts.
+ return coalesceDataTableWriteStatuses(dataTableWriteStatuses,
coalesceDivisorForDataTableWrites);
Review Comment:
🤖 With this, Spark's "streaming" MDT write no longer streams anything:
RLI/SI now go through `BatchMetadataConversionFunction` at commit, which
re-reads every written base/log file (`convertMetadataToRecordIndexRecords`,
`convertWriteStatsToSecondaryIndexRecords`), while the write handles still
build SI stats/record delegates under the streaming flag that nothing consumes
— and `STREAMING_WRITE_ENABLED`'s doc still says Spark streams. Before going
this far: the data-write RDD is persisted at `WRITE_STATUS_STORAGE_LEVEL`
(BaseSparkCommitActionExecutor:359) and RLI/SI records carry fileId+instant,
not the write token, so a losing attempt's delegates yield identical records —
was the stale-location corruption actually reproduced end-to-end? @nsivabalan
@yihua this looks like it needs an explicit design call.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -997,11 +1002,15 @@ private HoodieData<WriteStatus>
prepareAndWriteToNonStreamingPartitions(HoodieCo
return
convertEngineSpecificDataToHoodieData(secondaryWriteToMetadataTablePartitions(preppedRecords,
instantTime));
}
- private Set<String> getNonStreamingMetadataPartitionsToUpdate() {
+ private Set<String>
getNonStreamingMetadataPartitionsToUpdate(List<HoodieWriteStat>
partialWriteStats) {
+ Set<MetadataPartitionType> streamedPartitionTypes =
partialWriteStats.stream()
Review Comment:
🤖 Inferring 'streamed' from the returned stats changes Flink behavior:
`RecordIndexMapper` emits nothing for in-place updates (and SI nothing when the
secondary key is unchanged), so an update-only commit yields zero RLI/SI write
stats, and RLI/SI now fall back to batch regeneration here — reading the
rewritten base files (COW) and previous+new file slices for SI — where
previously `getStreamingMetadataPartitionsToUpdate()` always excluded them.
Would an explicit signal from the handler (Spark deferred vs. engine streamed)
be safer than inferring from stats?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -980,14 +980,19 @@ public void completeStreamingCommit(String instantTime,
HoodieEngineContext cont
}
List<HoodieWriteStat> allWriteStats = new ArrayList<>(partialWriteStats);
- // update metadata for left over partitions which does not have streaming
writes support.
- allWriteStats.addAll(prepareAndWriteToNonStreamingPartitions(metadata,
instantTime).map(WriteStatus::getStat).collectAsList());
+ maybeInitializeNewFileGroupsForPartitionedRLI(metadata, instantTime);
Review Comment:
🤖 Flink already ran
`maybeInitializeNewFileGroupsForPartitionedRLI(dataPartitions, instant)` inside
`streamWriteToMetadataPartitions` (via `IndexWriteFunction`), and the 'already
exists' check here reads FILES via `metadata.getAllPartitionPaths()`, which
won't contain a new data partition until this same MDT commit lands — so this
re-runs `initializeFileGroups` → `createEmptyFileGroupLogFile` for the same
fileIds/instant. Is that idempotent on the native-HFile log path? Might be
safer to only do this when `partialWriteStats` is empty (i.e. no streaming
phase ran).
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
