voonhous commented on code in PR #19869:
URL: https://github.com/apache/hudi/pull/19869#discussion_r3994960837
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java:
##########
@@ -478,11 +511,78 @@ private HoodieData<HoodieRecord>
getRecordIndexAdditionalUpserts(
} else if (operationType == WriteOperationType.DELETE_PARTITION) {
// all records from the target partition(s) to be deleted from RLI
return getRecordIndexReplacedRecords((HoodieReplaceCommitMetadata)
commitMetadata, fsView);
+ } else if (commitMetadata instanceof HoodieReplaceCommitMetadata &&
WriteOperationType.isUnknown(operationType)) {
+ // a replace commit without a known operation type registers files
written outside Hudi. The replaced file groups
+ // are dropped without their records being rewritten under the same key,
so the records of the replaced base files
+ // are deleted from RLI unless this commit wrote the same key again.
+ HoodieReplaceCommitMetadata replaceCommitMetadata =
(HoodieReplaceCommitMetadata) commitMetadata;
+ if (!dataTableMetaClient.getTableConfig().hasRecordKey()) {
+ checkReplacedFileGroupsAreNotWritten(replaceCommitMetadata);
+ }
+ HoodiePairData<HoodieKey, HoodieRecord> replacedRecordsByKey =
getRecordIndexReplacedFileGroupRecords(replaceCommitMetadata, fsView)
+ .mapToPair(record -> Pair.of(record.getKey(), record));
+ HoodiePairData<HoodieKey, HoodieRecord> writtenRecordsByKey =
updatesFromWriteStatuses
+ .mapToPair(record -> Pair.of(record.getKey(), record));
+ return replacedRecordsByKey.leftOuterJoin(writtenRecordsByKey)
+ .values()
+ .filter(replacedRecordAndRewrite ->
!replacedRecordAndRewrite.getRight().isPresent())
+ .map(Pair::getLeft);
} else {
return engineContext.emptyHoodieData();
}
}
+ /**
+ * Fails when the given commit writes a file group it replaces. For a table
without record keys, the file id of a
+ * file is its path below the partition, so such a commit registers a file
again under its own name: the previous
+ * content is gone, the keys to delete would be read from the new content,
and the file system view hides a replaced
+ * file group even when the same commit writes it again. Only a commit that
writes other file ids can be indexed.
+ */
+ private void
checkReplacedFileGroupsAreNotWritten(HoodieReplaceCommitMetadata
replaceCommitMetadata) {
+ replaceCommitMetadata.getPartitionToReplaceFileIds().forEach((partition,
replacedFileIds) -> {
+ Set<String> writtenFileIds =
replaceCommitMetadata.getPartitionToWriteStats().getOrDefault(partition,
Collections.emptyList()).stream()
+ .map(HoodieWriteStat::getFileId).collect(Collectors.toSet());
+ List<String> rewrittenFileIds =
replacedFileIds.stream().filter(writtenFileIds::contains).collect(Collectors.toList());
+ checkState(rewrittenFileIds.isEmpty(), "Table " +
dataTableMetaClient.getBasePath() + " has no record key, so a commit cannot
write the file "
+ + "groups it replaces in partition " + partition + ", because their
rows are keyed by file path and position: " + rewrittenFileIds);
+ });
+ }
+
+ /**
+ * Reads the record keys of the latest base file of every file group
replaced by the given commit and
+ * returns a delete record for each of them. The caller keeps the keys that
the same commit writes again.
+ */
+ private HoodieData<HoodieRecord>
getRecordIndexReplacedFileGroupRecords(HoodieReplaceCommitMetadata
replaceCommitMetadata, Lazy<HoodieTableFileSystemView> fsView) {
+ List<Pair<String, HoodieBaseFile>> replacedBaseFiles =
replaceCommitMetadata.getPartitionToReplaceFileIds().entrySet().stream()
+ .flatMap(partitionAndFileIds -> partitionAndFileIds.getValue().stream()
+ .map(fileId -> {
+ Option<HoodieBaseFile> baseFile =
fsView.get().getLatestBaseFile(partitionAndFileIds.getKey(), fileId);
Review Comment:
**major:** One correction for #19886 item 4: on master the
`INSERT_OVERWRITE` / `INSERT_OVERWRITE_TABLE` / `DELETE_PARTITION` arms above
(:493-513) did not share this gap silently, they crashed. `readCommitMetadata`
deserializes into the base class (both SerDes honor the requested `clazz`,
neither reads the action), `operationType` survives, and the blind cast at
master :463 throws a ClassCastException. With `TimelineUtils.getCommitMetadata`
the cast succeeds and the arms run on the post-replace view, so a keyed table
now gets a silently stale RLI where it used to fail loudly. Could item 4 record
that, and could the description's "no behavior change for tables that carry a
record key" carry the caveat? Follow-up either way.
--
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]
