This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit e2af6db031d902a23c5ec5a437e6c8101957db95 Author: voonhous <[email protected]> AuthorDate: Tue Jun 16 11:39:49 2026 +0800 refactor(metadata): Replace misused stream reduce with a plain for-loop (#18532) - HoodieTableMetadataUtil.convertMetadataToFilesPartitionRecords aggregated per-partition write stats via writeStats.stream().reduce(new HashMap<>(), accumulator, CollectionUtils::combine). - The "identity" is a mutable HashMap that the accumulator mutates in place - a misuse of Stream.reduce. - It only works because the stream is sequential and the method runs on the driver (HoodieMetadataWriteUtils then wraps the result via context.parallelize(..., 1)). - A plain for-loop expresses the same aggregation directly and is idiomatic for mutable-accumulation sequential code. - No behavior change. No measurable perf impact - readability/idiom cleanup. (cherry picked from commit dcfe9d403ecd9ad7f782e6aad4560a2727f04cf7) --- .../hudi/metadata/HoodieTableMetadataUtil.java | 48 +++++++++++----------- 1 file changed, 23 insertions(+), 25 deletions(-) diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java index 23f20fa4efea..4823f041bae6 100644 --- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java +++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java @@ -480,31 +480,29 @@ public class HoodieTableMetadataUtil { String partitionStatName = entry.getKey(); List<HoodieWriteStat> writeStats = entry.getValue(); - HashMap<String, Long> updatedFilesToSizesMapping = - writeStats.stream().reduce(new HashMap<>(writeStats.size()), - (map, stat) -> { - String pathWithPartition = stat.getPath(); - if (pathWithPartition == null) { - // Empty partition - log.warn("Unable to find path in write stat to update metadata table {}", stat); - return map; - } - - String fileName = FSUtils.getFileName(pathWithPartition, partitionStatName); - - // Since write-stats are coming in no particular order, if the same - // file have previously been appended to w/in the txn, we simply pick max - // of the sizes as reported after every write, since file-sizes are - // monotonically increasing (ie file-size never goes down, unless deleted) - map.merge(fileName, stat.getFileSizeInBytes(), Math::max); - - Map<String, Long> cdcPathAndSizes = stat.getCdcStats(); - if (cdcPathAndSizes != null && !cdcPathAndSizes.isEmpty()) { - cdcPathAndSizes.forEach((key, value) -> map.put(FSUtils.getFileName(key, partitionStatName), value)); - } - return map; - }, - CollectionUtils::combine); + HashMap<String, Long> updatedFilesToSizesMapping = new HashMap<>(writeStats.size()); + for (HoodieWriteStat stat : writeStats) { + String pathWithPartition = stat.getPath(); + if (pathWithPartition == null) { + // Empty partition + log.warn("Unable to find path in write stat to update metadata table {}", stat); + continue; + } + + String fileName = FSUtils.getFileName(pathWithPartition, partitionStatName); + + // Since write-stats are coming in no particular order, if the same + // file have previously been appended to w/in the txn, we simply pick max + // of the sizes as reported after every write, since file-sizes are + // monotonically increasing (ie file-size never goes down, unless deleted) + updatedFilesToSizesMapping.merge(fileName, stat.getFileSizeInBytes(), Math::max); + + Map<String, Long> cdcPathAndSizes = stat.getCdcStats(); + if (cdcPathAndSizes != null && !cdcPathAndSizes.isEmpty()) { + cdcPathAndSizes.forEach((key, value) -> + updatedFilesToSizesMapping.put(FSUtils.getFileName(key, partitionStatName), value)); + } + } newFileCount.add(updatedFilesToSizesMapping.size()); return HoodieMetadataPayload.createPartitionFilesRecord(partitionStatName, updatedFilesToSizesMapping,
