danny0405 commented on code in PR #19575:
URL: https://github.com/apache/hudi/pull/19575#discussion_r3870637479
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/SparkMetadataTableGlobalRecordLevelIndex.java:
##########
@@ -181,21 +185,27 @@ public boolean isImplicitWithStorage() {
*/
private static class RecordIndexFileGroupLookupFunction implements
PairFlatMapFunction<Iterator<String>, String, HoodieRecordGlobalLocation> {
private final HoodieTable hoodieTable;
+ // Null when no counters should be collected; see
RecordIndexLookupMetrics#resolveRegistry.
+ private final Registry lookupMetrics;
- public RecordIndexFileGroupLookupFunction(HoodieTable hoodieTable) {
+ public RecordIndexFileGroupLookupFunction(HoodieTable hoodieTable,
Registry lookupMetrics) {
this.hoodieTable = hoodieTable;
+ this.lookupMetrics = lookupMetrics;
}
@Override
public Iterator<Tuple2<String, HoodieRecordGlobalLocation>>
call(Iterator<String> recordKeyIterator) {
List<String> keysToLookup = new ArrayList<>();
recordKeyIterator.forEachRemaining(keysToLookup::add);
+ HoodieTimer shardTimer = HoodieTimer.start();
// recordIndexInfo object only contains records that are present in
record_index.
HoodiePairData<String, HoodieRecordGlobalLocation> recordIndexData =
hoodieTable.getTableMetadata().readRecordIndexLocationsWithKeys(HoodieListData.eager(keysToLookup));
try {
List<Pair<String, HoodieRecordGlobalLocation>> recordIndexInfo =
HoodieDataUtils.dedupeAndCollectAsList(recordIndexData);
+ RecordIndexLookupMetrics.recordShardLookup(lookupMetrics, keysToLookup,
+
recordIndexInfo.stream().map(Pair::getKey).collect(Collectors.toSet()),
shardTimer.endTimer());
Review Comment:
[P1] Avoid disabled-path hit-set allocation
`lookupMetrics` is null by default, but Java evaluates this argument before
`recordShardLookup` can take its null fast path. Every global-RLI shard
therefore still traverses all hits and allocates an O(hits) `HashSet` when the
feature is disabled. For large high-hit upserts this makes an opt-in metric a
default-path CPU/memory regression. Could we guard this block with `if
(lookupMetrics != null)` or defer construction of the set into the helper?
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java:
##########
@@ -146,8 +147,14 @@ 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));
+ if (committed) {
+ // The commit landed, so the counters the executors collected for it
can be reported and
+ // released. A commit that never lands publishes nothing.
+ ExecutorMetrics.publishAndRelease(config, metrics);
Review Comment:
[P2] Publish from the actual successful-commit boundary
This hook runs only through `SparkRDDWriteClient.commit`, but the inherited
public `commitStats` overloads can also land a write directly. A manual RLI
upsert committed through `commitStats` leaves its counters unpublished and
unreleased; a following insert with no lookup can then publish those old counts
as its own. Could publication be attached to the shared successful
`commitStats` boundary, with a direct-`commitStats` regression test, so all
supported commit entry points have the same lifecycle?
--
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]
