vinishjail97 commented on code in PR #19869:
URL: https://github.com/apache/hudi/pull/19869#discussion_r3994340336
##########
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)) {
Review Comment:
Fixed in 5a5a8b78. The branch now reads `commitMetadata instanceof
HoodieReplaceCommitMetadata && WriteOperationType.isUnknown(operationType) &&
!dataTableMetaClient.getTableConfig().hasRecordKey()`, so a keyed table falls
through to `emptyHoodieData()` as it did on master, and the `generateRecordKeys
= false` leg of `getRecordIndexReplacedFileGroupRecords` is unreachable. The
inner `if (!hasRecordKey())` around `checkReplacedFileGroupsAreNotWritten` is
now unconditional.
I took the same gate on both SI sites, because gating only RLI would let SI
delete the entries of a replaced group that RLI still keeps:
`SecondaryIndexer.getSecondaryIndexUpdates` adds `!hasRecordKey()` to
`dropsReplacedFileGroups`, and
`SecondaryIndexRecordGenerationUtils.convertWriteStatsToSecondaryIndexRecords`
adds it to the union of `convertReplacedFileGroupsToSecondaryIndexRecords`.
`TestRecordIndexer.testBuildUpdateKeepsReplacedFileGroupsOfTableWithRecordKeys`
asserts that a keyed table produces no records and never touches the file
system view.
`TestSecondaryIndexer.testBuildUpdateForReplaceCommitFromExternalWriterWithoutWriteStats`
is now parameterized over the replace map and `hasRecordKey`, and the keyed
case expects an empty result.
##########
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:
Agreed, and I moved it to #19886 rather than into this PR. I added the chain
you traced to the issue: `RunIndexActionExecutor` builds the catch-up writer on
a timeline where the replace commit is complete,
`AbstractTableFileSystemView.resetFileGroupsReplaced` hides the groups, and
both `getLatestBaseFile` and `getLatestMergedFileSliceBeforeOrOn` return empty,
so a replayed registration commit deletes nothing. The proposed fix there is
the `fsView.getReplacedFileGroupsBeforeOrOn(instantTime, partition)` fallback
on both lookups, with a functional test over a completed external-file replace
commit.
The issue now says that async catch-up over registration commits is not
covered yet, and that the `TimelineUtils.getCommitMetadata` change in this PR
only lets the catch-up tasks read the operation type. Thanks for calling out
that this reads as delivering it.
##########
hudi-common/src/main/java/org/apache/hudi/metadata/BaseFileRecordParsingUtils.java:
##########
@@ -167,10 +196,43 @@ public static Map<RecordStatus, List<String>>
getRecordKeyStatuses(String basePa
}
}
- private static Set<String> getRecordKeysFromBaseFile(HoodieStorage storage,
String basePath, String partition, String fileName) {
- StoragePath dataFilePath = new StoragePath(basePath,
StringUtils.isNullOrEmpty(partition) ? fileName : (partition +
StoragePath.SEPARATOR) + fileName);
+ /**
+ * Generates RLI Metadata delete records for every record key in the given
base file.
+ * Used when a file group is replaced by a commit that does not rewrite its
records, for example a replace commit
+ * that registers files written outside Hudi.
+ *
+ * @param basePath base path of the table.
+ * @param partition partition of the base file.
+ * @param dataFilePath path of the base file on storage.
+ * @param storage instance of {@link HoodieStorage}.
+ * @param isPartitionedRLI whether the record index is partitioned.
+ * @param generateRecordKeys whether the table carries no record key, so
that every row is keyed by the file path
+ * relative to the table base path and the row
position.
+ * @return Iterator of delete {@link HoodieRecord}s for RLI Metadata
partition.
+ */
+ public static Iterator<HoodieRecord>
generateRLIMetadataHoodieRecordsForReplacedBaseFile(String basePath,
+
String partition,
+
StoragePath dataFilePath,
+
HoodieStorage storage,
+
boolean isPartitionedRLI,
+
boolean generateRecordKeys) {
+ return getRecordKeysFromBaseFile(storage, basePath, dataFilePath,
generateRecordKeys).stream()
Review Comment:
Taking it as a follow-up under #19886, thanks. The issue now lists the four
materializations you counted, `ParquetUtils.filterRowKeys` collecting a
`Set<Pair>`, `FileFormatUtils.readRowKeys` mapping it to a `Set<String>`, the
`prevFileName == null` branch copying it into an `ArrayList`, and the caller
building the record list, against initialize streaming the same data through
`getRowKeyIterator`, with the fix being to stream both branches and keep the
set only for the previous-file diff.
--
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]
