danny0405 commented on code in PR #13603:
URL: https://github.com/apache/hudi/pull/13603#discussion_r2234855395
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -1689,40 +1804,75 @@ protected Pair<HoodieData<HoodieRecord>,
List<HoodieFileGroupId>> tagRecordsWith
try (HoodieTableFileSystemView fsView =
HoodieTableMetadataUtil.getFileSystemViewForMetadataTable(metadataMetaClient)) {
List<HoodieFileGroupId> hoodieFileGroupIdList = new ArrayList<>();
for (Map.Entry<String, HoodieData<HoodieRecord>> entry :
partitionRecordsMap.entrySet()) {
- final String partitionName = entry.getKey();
+ final String partitionPath = entry.getKey();
HoodieData<HoodieRecord> records = entry.getValue();
+ boolean isPartitionedRLI = Objects.equals(partitionPath,
RECORD_INDEX.getPartitionPath()) &&
dataWriteConfig.isPartitionedRecordIndexEnabled();
List<FileSlice> fileSlices =
-
HoodieTableMetadataUtil.getPartitionLatestFileSlices(metadataMetaClient,
Option.ofNullable(fsView), partitionName);
- if (fileSlices.isEmpty()) {
- // scheduling or initialising of INDEX only initializes the file
group and not add commit
- // so if there are no committed file slices, look for inflight slices
- ValidationUtils.checkState(isInitializing ||
dataMetaClient.getTableConfig().getMetadataPartitionsInflight().contains(partitionName),
- String.format("Partition %s should be part of inflight metadata
partitions here %s", partitionName,
dataMetaClient.getTableConfig().getMetadataPartitionsInflight()));
- fileSlices =
getPartitionLatestFileSlicesIncludingInflight(metadataMetaClient,
Option.ofNullable(fsView), partitionName);
+
HoodieTableMetadataUtil.getPartitionLatestFileSlices(metadataMetaClient,
Option.ofNullable(fsView), partitionPath);
+ // scheduling of INDEX only initializes the file group and not add
commit
+ // so if there are no committed file slices, look for inflight slices
+ if (isPartitionedRLI || fileSlices.isEmpty()) {
+ // For isPartitionedRLI, new partitions added to the data table will
cause new filegroups that are not yet commited
+ // therefore, we always need to look for inflight filegroups
+ ValidationUtils.checkState(isInitializing ||
dataMetaClient.getTableConfig().getMetadataPartitionsInflight().contains(partitionPath)
|| isPartitionedRLI,
+ String.format("Partition %s should be part of inflight metadata
partitions here %s", partitionPath,
dataMetaClient.getTableConfig().getMetadataPartitionsInflight()));
+ fileSlices =
getPartitionLatestFileSlicesIncludingInflight(metadataMetaClient,
Option.ofNullable(fsView), partitionPath);
}
- final int fileGroupCount = fileSlices.size();
- ValidationUtils.checkArgument(fileGroupCount > 0,
String.format("FileGroup count for MDT partition %s should be > 0",
partitionName));
- hoodieFileGroupIdList.addAll(fileSlices.stream().map(fileSlice -> new
HoodieFileGroupId(partitionName,
fileSlice.getFileId())).collect(Collectors.toList()));
-
- List<FileSlice> finalFileSlices = fileSlices;
- HoodieIndexVersion indexVersion =
existingIndexVersionOrDefault(partitionName, dataMetaClient);
-
- // Determine key format once per partition to avoid repeated checks
- SerializableBiFunction<String, Integer, Integer> mappingFunction =
MetadataPartitionType.fromPartitionPath(partitionName).getFileGroupMappingFunction(indexVersion);
- HoodieData<HoodieRecord> rddSinglePartitionRecords = records.map(r -> {
- FileSlice slice =
finalFileSlices.get(mappingFunction.apply(r.getRecordKey(), fileGroupCount));
- r.unseal();
- r.setCurrentLocation(new
HoodieRecordLocation(slice.getBaseInstantTime(), slice.getFileId()));
- r.seal();
- return r;
- });
-
- allPartitionRecords =
allPartitionRecords.union(rddSinglePartitionRecords);
+ hoodieFileGroupIdList.addAll(fileSlices.stream().map(fileSlice -> new
HoodieFileGroupId(partitionPath,
fileSlice.getFileId())).collect(Collectors.toList()));
+ SerializableFunction<HoodieRecord, HoodieRecord> recordTagger =
getRecordTagger(partitionPath, fileSlices);
+ allPartitionRecords =
allPartitionRecords.union(records.map(recordTagger));
}
return Pair.of(allPartitionRecords, hoodieFileGroupIdList);
}
}
+ private SerializableFunction<HoodieRecord, HoodieRecord>
getRecordTagger(String partitionPath, List<FileSlice> fileSlices) {
+ HoodieIndexVersion indexVersion =
existingIndexVersionOrDefault(partitionPath, dataMetaClient);
+ MetadataPartitionType partitionType =
MetadataPartitionType.fromPartitionPath(partitionPath);
+ // Determine key format once per partition to avoid repeated checks
+ SerializableBiFunction<String, Integer, Integer> mappingFunction =
partitionType.getFileGroupMappingFunction(indexVersion);
+ if (partitionType == RECORD_INDEX &&
dataWriteConfig.isPartitionedRecordIndexEnabled()) {
+ return getRecordTaggerPartitionedRLI(fileSlices, mappingFunction);
+ }
+ final int fileGroupCount = fileSlices.size();
+ ValidationUtils.checkArgument(fileGroupCount > 0, String.format("FileGroup
count for MDT partition %s should be > 0", partitionPath));
+ return r -> {
+ FileSlice slice = fileSlices.get(mappingFunction.apply(r.getRecordKey(),
fileGroupCount));
+ r.unseal();
+ r.setCurrentLocation(new
HoodieRecordLocation(slice.getBaseInstantTime(), slice.getFileId()));
+ r.seal();
+ return r;
+ };
+ }
+
+ /**
+ * Add the location for each incoming record in the partitioned record level
index. Hashing needs to be done within the data table partition and
+ * the number of filegroups can also vary by partition. Therefore, we need
to lookup and group the filegroups by partition to understand the number
+ * of indexes we can hash into
+ *
+ * @param fileSlices latest fileslices in the mdt rli partition
+ * @return the records with the correct location set
+ */
+ private SerializableFunction<HoodieRecord, HoodieRecord>
getRecordTaggerPartitionedRLI(List<FileSlice> fileSlices,
SerializableBiFunction<String, Integer, Integer> mappingFunction) {
+ final Map<String, List<HoodieRecordLocation>> fileSlicesPerHudiPartition =
new HashMap<>();
+ fileSlices.forEach(s ->
fileSlicesPerHudiPartition.computeIfAbsent(HoodieTableMetadataUtil.getDataTablePartitionNameFromFileGroupName(s.getFileId()),
x -> new ArrayList<>())
+ .add(new HoodieRecordLocation(s.getBaseInstantTime(), s.getFileId())));
+ return r -> {
+ String partitionPath;
+ if (r.getData() instanceof EmptyHoodieRecordPayloadWithPartition) {
+ partitionPath = ((EmptyHoodieRecordPayloadWithPartition)
r.getData()).getPartitionPath();
+ } else {
+ partitionPath = ((HoodieMetadataPayload)
r.getData()).getRecordGlobalLocation().getPartitionPath();
Review Comment:
`.getRecordGlobalLocation().getPartitionPath()` is not efficient, we may
need to add a new API on `HoodieMetadataPayload` to fetch the partition path
directly from `recordIndexMetadata`.
--
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]
