voonhous commented on code in PR #19869:
URL: https://github.com/apache/hudi/pull/19869#discussion_r3994963800
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -105,12 +117,7 @@ public static <T> HoodieData<HoodieRecord>
convertWriteStatsToSecondaryIndexReco
throw new HoodieIOException("Secondary index cannot support logs having
inserts with current offering. Please disable secondary index.");
}
- HoodieSchema tableSchema;
- try {
- tableSchema = tryResolveSchemaForTable(dataMetaClient).get();
- } catch (Exception e) {
- throw new HoodieException("Failed to get latest schema for " +
dataMetaClient.getBasePath(), e);
- }
+ HoodieSchema tableSchema = resolveTableSchema(dataMetaClient,
commitMetadata);
Map<String, List<HoodieWriteStat>> writeStatsByFileId =
allWriteStats.stream().collect(Collectors.groupingBy(HoodieWriteStat::getFileId));
Review Comment:
**major:** `groupingBy(HoodieWriteStat::getFileId)` spans every partition,
and a keyless file id is `[prefix/]name` (`ExternalFilePathUtil:203`), so
registering `p1/000000_0` and `p2/000000_0` in one commit puts both stats in
one group and the size-1 `checkArgument` at line 146 fails the commit; RLI
accepts the same commit. Hive-style sources hit this on the first sync. Could
we group by `(partition, fileId)` in a follow-up, with a two-partition case in
`TestExternalFileRecordAndSecondaryIndex`?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -245,35 +327,36 @@ public static <T> HoodieData<HoodieRecord>
readSecondaryKeysFromFileSlices(Hoodi
return engineContext.emptyHoodieData();
}
final int parallelism = Math.min(fileSlices.size(),
secondaryIndexMaxParallelism);
- final StoragePath basePath = metaClient.getBasePath();
- HoodieSchema tableSchema;
- try {
- tableSchema = new TableSchemaResolver(metaClient).getTableSchema();
- } catch (Exception e) {
- throw new HoodieException("Failed to get latest schema for " +
metaClient.getBasePath(), e);
- }
ReaderContextFactory<T> readerContextFactory =
engineContext.getReaderContextFactory(metaClient);
engineContext.setJobStatus(activeModule, "Secondary Index: reading
secondary keys from " + fileSlices.size() + " file slices");
HoodieFileFormat baseFileFormat =
metaClient.getTableConfig().getBaseFileFormat();
+ // a file slice without a base file has only log files, whose schema is
the table schema. It is resolved once here
+ // rather than in every task, because resolving it loads the timeline
+ Option<HoodieSchema> tableSchema = fileSlices.stream().anyMatch(slice ->
!slice.getFileSlice().getBaseFile().isPresent())
+ ? Option.of(resolveTableSchema(metaClient))
+ : Option.empty();
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));
- HoodieSchema readerSchema;
- if (dataFilePath.isPresent()) {
- readerSchema = HoodieIOFactory.getIOFactory(metaClient.getStorage())
- .getFileFormatUtils(baseFileFormat)
- .readSchema(metaClient.getStorage(), dataFilePath.get());
- } else {
- readerSchema = tableSchema;
- }
+ // 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 = dataFilePath.isPresent()
+ ?
HoodieIOFactory.getIOFactory(metaClient.getStorage()).getFileFormatUtils(baseFileFormat).readSchema(metaClient.getStorage(),
dataFilePath.get())
+ : tableSchema.get();
Review Comment:
**major:** This branch is live for the first time: master :261 wrapped the
base-file lookup in `Option.ofNullable(...orElseGet(null))`, which calls
`.get()` on a null Supplier for any log-only file slice (bucket-index MOR,
Flink append), so SI initialize crashed there since #14045 and the
`tableSchema` arm was dead. The refactor fixes it, but no test reaches it:
`testSecondaryIndexRecordGenerationForMOR` inserts first, so every slice has a
base file. Could a follow-up add a log-only slice there, and could the
description mention the fix so it can be backported?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -197,6 +204,15 @@ public static <T> HoodieData<HoodieRecord>
convertWriteStatsToSecondaryIndexReco
return records.iterator();
});
+ if (commitMetadata instanceof HoodieReplaceCommitMetadata &&
WriteOperationType.isUnknown(commitMetadata.getOperationType())
+ && !dataMetaClient.getTableConfig().hasRecordKey()) {
+ // a replace commit without a known operation type registers files
written outside Hudi and drops the replaced
+ // file groups without rewriting their records under the same key
+ secondaryIndexRecords =
secondaryIndexRecords.union(convertReplacedFileGroupsToSecondaryIndexRecords(
+ (HoodieReplaceCommitMetadata) commitMetadata,
writeStatsByFileId.keySet(), instantTime, indexDefinition, metadataConfig,
Review Comment:
**major:** `writeStatsByFileId.keySet()` is not partition-qualified, so a
group replaced in partition A is skipped at line 266 whenever partition B
writes the same file id in the same commit.
`checkReplacedFileGroupsAreNotWritten` (`BaseRecordIndexer:541-544`) is
per-partition and lets that commit through, so RLI deletes the rows and SI
keeps them. No wrong results (the phantom keys fail the RLI join), but nothing
reclaims them. Grouping by `(partition, fileId)` above fixes this too; could it
ride the same follow-up?
--
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]
