vinishjail97 commented on code in PR #19869:
URL: https://github.com/apache/hudi/pull/19869#discussion_r3974592809
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java:
##########
@@ -478,11 +500,51 @@ 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 9085866. Both catch-up tasks read the instant through
`TimelineUtils.getCommitMetadata`, and `TestIndexingCatchupTask` replays a
replace commit and a commit through each task and verifies the metadata writer
receives the `HoodieReplaceCommitMetadata`.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -258,7 +318,8 @@ public static <T> HoodieData<HoodieRecord>
readSecondaryKeysFromFileSlices(Hoodi
return engineContext.parallelize(fileSlices,
parallelism).flatMap(partitionAndBaseFile -> {
final String partition = partitionAndBaseFile.getPartitionPath();
final FileSlice fileSlice = partitionAndBaseFile.getFileSlice();
- Option<StoragePath> dataFilePath =
Option.ofNullable(fileSlice.getBaseFile().map(baseFile ->
FSUtils.getAbsoluteFilePath(basePath, partition,
baseFile.getFileName())).orElseGet(null));
+ // the storage path keeps the directory prefix of a file written outside
Hudi, which its file name alone loses
+ Option<StoragePath> dataFilePath =
fileSlice.getBaseFile().map(HoodieBaseFile::getStoragePath);
HoodieSchema readerSchema;
if (dataFilePath.isPresent()) {
readerSchema = HoodieIOFactory.getIOFactory(metaClient.getStorage())
Review Comment:
Fixed in 9085866. The schema is resolved inside the task only for a file
slice without a base file, so initializing the index on a keyless table no
longer touches `TableSchemaResolver`.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java:
##########
@@ -478,11 +500,51 @@ 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.
+ HoodiePairData<HoodieKey, HoodieRecord> replacedRecordsByKey =
getRecordIndexReplacedFileGroupRecords((HoodieReplaceCommitMetadata)
commitMetadata, 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();
}
}
+ /**
+ * 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.
+ */
+ 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 ->
fsView.get().getLatestBaseFile(partitionAndFileIds.getKey(), fileId))
+ .filter(Option::isPresent)
Review Comment:
Added a `log.warn` with partition and file id at both sites in 9085866.
##########
hudi-common/src/main/java/org/apache/hudi/common/util/ExternalFilePathUtil.java:
##########
@@ -93,6 +93,32 @@ public static boolean isExternallyCreatedFile(String
fileName) {
return fileName.endsWith(EXTERNAL_FILE_SUFFIX);
}
+ /**
+ * Returns the path of a base file relative to its partition, as the file
exists on storage.
+ * For an external file name, the commit time and the external file marker
are stripped and the file group
+ * prefix, if any, is restored. For example,
"data.parquet_123_fg%3Dbucket-0_hudiext" returns "bucket-0/data.parquet".
+ * A file name that was not created externally is returned as is.
+ *
+ * @param fileName The file name as recorded in the commit metadata
+ * @return The path of the file relative to its partition
+ */
+ public static String getFilePathInPartition(String fileName) {
Review Comment:
Fixed in 9085866. `getOriginalFileName` throws a `HoodieException` naming
the file when no commit time precedes the marker, and `getFullPathOfPartition`
checks that the parent ends with the prefix. Both have tests.
--
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]
