danny0405 commented on code in PR #13449:
URL: https://github.com/apache/hudi/pull/13449#discussion_r2155921073
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -1134,26 +1138,48 @@ public void startCommit(String instantTime) {
@Override
public HoodieData<WriteStatus>
streamWriteToMetadataPartitions(HoodieData<WriteStatus> writeStatus, String
instantTime) {
- List<MetadataPartitionType> partitionsToTag = new
ArrayList<>(enabledPartitionTypes);
- partitionsToTag.remove(FILES);
- partitionsToTag.retainAll(STREAMING_WRITES_SUPPORTED_PARTITIONS);
- if (partitionsToTag.isEmpty()) {
+ Pair<List<MetadataPartitionType>, Set<String>> streamingMDTPartitionsPair
= getStreamingMetadataPartitionsToUpdate();
+ List<MetadataPartitionType> mdtPartitionsToTag =
streamingMDTPartitionsPair.getLeft();
+ Set<String> mdtPartitionPathsToTag = streamingMDTPartitionsPair.getRight();
+
+ if (mdtPartitionPathsToTag.isEmpty()) {
return engineContext.emptyHoodieData();
}
HoodieData<HoodieRecord> untaggedRecords = writeStatus.flatMap(
- new
MetadataIndexGenerator.WriteStatusBasedMetadataIndexMapper(partitionsToTag,
dataWriteConfig));
+ new
MetadataIndexGenerator.WriteStatusBasedMetadataIndexMapper(mdtPartitionsToTag,
dataWriteConfig));
// tag records w/ location
Pair<List<HoodieFileGroupId>, HoodieData<HoodieRecord>>
hoodieFileGroupsToUpdateAndTaggedMdtRecords =
tagRecordsWithLocationForStreamingWrites(untaggedRecords,
-
partitionsToTag.stream().map(MetadataPartitionType::getPartitionPath).collect(Collectors.toSet()));
+ mdtPartitionPathsToTag);
// write partial writes to MDT table (for those partitions where streaming
writes are enabled)
HoodieData<WriteStatus> writeStatusCollection =
convertEngineSpecificDataToHoodieData(streamWriteToMetadataTable(hoodieFileGroupsToUpdateAndTaggedMdtRecords,
instantTime));
// dag not yet de-referenced. do not invoke any action on
writeStatusCollection yet.
return writeStatusCollection;
}
+ private Pair<List<MetadataPartitionType>, Set<String>>
getStreamingMetadataPartitionsToUpdate() {
+ List<MetadataPartitionType> mdtPartitionsToTag = new
ArrayList<>(enabledPartitionTypes);
+ mdtPartitionsToTag.remove(FILES);
Review Comment:
instead of removing, can we initialize an empty list here.
--
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]
