danny0405 commented on code in PR #19575:
URL: https://github.com/apache/hudi/pull/19575#discussion_r3877807797
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java:
##########
@@ -146,8 +147,16 @@ public boolean commit(String instantTime,
JavaRDD<WriteStatus> rawWriteStatuses,
// when streaming writes are enabled, writeStatuses is a mix of data
table write status and mdt write status
List<HoodieWriteStat> dataTableHoodieWriteStats =
slimWriteStatsList.stream().filter(entry ->
!entry.isMetadataTable()).map(SlimWriteStats::getWriteStat).collect(Collectors.toList());
List<HoodieWriteStat> partialMetadataTableWriteStats =
slimWriteStatsList.stream().filter(entry ->
entry.isMetadataTable).map(SlimWriteStats::getWriteStat).collect(Collectors.toList());
- return commitStats(instantTime, new
TableWriteStats(dataTableHoodieWriteStats, partialMetadataTableWriteStats),
extraMetadata, commitActionType, partitionToReplacedFileIds, extraPreCommitFunc,
- false, Option.of(table));
+ boolean committed = commitStats(instantTime, new
TableWriteStats(dataTableHoodieWriteStats, partialMetadataTableWriteStats),
extraMetadata, commitActionType, partitionToReplacedFileIds,
+ extraPreCommitFunc, false, Option.of(table));
+ // Publish only if this write tagged. requiresTagging is the same
predicate BaseWriteHelper uses
+ // to decide whether to look up at all, so publishing mirrors
collecting: an INSERT or a
+ // BULK_INSERT never collects, and must not publish counters an earlier
write left behind.
+ if (committed && getOperationType() != null
Review Comment:
Agree with moving this into the shared `postCommit` path; that would also
cover writes landed through the public `commitStats` overloads.
One caveat: `getIndex().requiresTagging(getOperationType())` is not
equivalent to whether this write performed an RLI lookup. For an `INSERT` with
`hoodie.datasource.write.insert.drop.duplicates=true` or a drop/fail duplicate
policy, `HoodieSparkSqlWriter.handleInsertDuplicates` calls
`DataSourceUtils.handleDuplicates` -> `SparkRDDReadClient.tagLocation`, so the
instrumented RLI lookup fills the accumulator.
`HoodieIndex.requiresTagging(INSERT)` is false, however, so this guard skips
publication and the next tagging write clears those counters.
Could the lifecycle track whether this write actually resolved/used the
lookup registry, rather than infer that from operation type? A regression test
with INSERT duplicate handling enabled would pin this case.
--
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]
