voonhous commented on code in PR #19869:
URL: https://github.com/apache/hudi/pull/19869#discussion_r3987612759
##########
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:** This lookup returns empty for a file group whose replace commit
is complete in the view's timeline, which is the state async index catch-up
runs in, so a replayed registration commit deletes nothing here (the new warn
fires) and the replaced groups' RLI/SI entries survive. The sync path is
unaffected because `commitStats` writes the MDT before `saveAsComplete`;
`TestIndexingCatchupTask` is mock-only. By reading, not executed. Could the
lookup fall back to `fsView.getReplacedFileGroupsBeforeOrOn(instantTime,
partition)` when `getLatestBaseFile` is empty (same for the SI slice lookup),
with a functional test running the indexer over a completed external-file
replace commit?
Can be a follow-up under #19886, since the `INSERT_OVERWRITE` arm has the
same catch-up gap on master today. If it goes there, could the description say
that async catch-up over registration commits is not covered yet, so the
`TimelineUtils.getCommitMetadata` change is not read as delivering it?
<details><summary>Chain</summary>
`RunIndexActionExecutor:142/183` builds the catch-up writer after
`getInstantsToCatchup` picked completed instants; the writer's fresh meta
client (`HoodieBackedTableMetadataWriter:200`) loads that timeline;
`getMetadataView` (`:241`) builds the view on it; `resetFileGroupsReplaced`
(`AbstractTableFileSystemView:275-278`) marks the groups from
`getCompletedReplaceTimeline`; `getLatestBaseFile` (`:826`) returns empty for
them, and `getLatestMergedFileSliceBeforeOrOn` hides them the same way on the
SI side.
</details>
##########
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:
**major (perf):** The keyless update path materializes every key of a base
file three to four times in one task: `ParquetUtils.filterRowKeys` collects a
`Set<Pair>` from `getRowKeyIterator` (`:140-151`),
`FileFormatUtils.readRowKeys` maps it to a second `Set<String>` (`:164-167`),
the `prevFileName == null` branch copies it into an `ArrayList` (`:161`), and
the caller builds the record list (`:87-100`), while initialize streams the
same data through `getRowKeyIterator` (`BaseRecordIndexer:282`). Registered
files are conversion output, typically 10^7 rows at 50-70 chars per key. Could
this branch and the `prevFileName == null` branch stream through
`getRowKeyIterator`, keeping the set only for the previous-file diff?
Fine as a follow-up under #19886: no wrong results, and keyed tables have
always had this shape.
##########
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:
**major:** Only the safety check below is gated on `!hasRecordKey()`; the
delete generation runs for keyed tables too, where master returned
`emptyHoodieData()` for this shape. A keyed table's replace commit with a null
or `UNKNOWN` operation type now reads the real record keys of every replaced
base file (`:575` passes `generateRecordKeys = false`, a leg the only test
stubs with `any()`) and emits deletes, which contradicts the description's "No
behavior change for tables that carry a record key"; `UNKNOWN` is also an
existing sentinel (`MetadataConversionUtils:360`,
`ListingBasedRollbackStrategy:360`). Could the branch condition include
`!hasRecordKey()`? That one line restores master's behavior for keyed tables
and makes the untested leg unreachable, which is why I would take it in this PR
rather than in #19886; owning the keyed case instead, with a test and a
description update, is fine too but is more work than the gate.
```suggestion
} else if (commitMetadata instanceof HoodieReplaceCommitMetadata &&
WriteOperationType.isUnknown(operationType)
&& !dataTableMetaClient.getTableConfig().hasRecordKey()) {
```
--
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]
