voonhous commented on code in PR #19869:
URL: https://github.com/apache/hudi/pull/19869#discussion_r3976033506
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/BaseIndexer.java:
##########
@@ -71,4 +75,15 @@ public void postInitialization(HoodieTableMetaClient
metadataMetaClient, HoodieD
public List<IndexPartitionAndRecords> buildRestore(IndexRestoreContext
context) {
return Collections.emptyList();
}
+
+ /**
+ * Fails when a clustering commit reaches a table without record keys. The
record index and the secondary index
+ * key the rows of such a table by file path and row position, and
clustering rewrites the rows into new files,
+ * so neither index could follow them. Every other operation type either
keeps the keys or registers new files.
+ */
+ protected void checkClusteringKeepsRecordKeys(HoodieCommitMetadata
commitMetadata) {
Review Comment:
**blocker:** This guard runs only on the batch metadata path, and a Spark
`CLUSTER` commit bypasses that path under the default
`hoodie.metadata.streaming.write.enabled=true`, so it never fires. The replaced
groups' positional entries are then never deleted either, so after clustering
SI/RLI resolve to a dead fileId and a query on the indexed column returns zero
rows. `TestExternalFileRecordAndSecondaryIndex:227` pins streaming off and
replays the commit instead of clustering, so it cannot catch this.
Our earlier ask said "both indexers", which is where this landed, so the
placement is on us. Could the check move to a point both paths share, e.g.
`HoodieBackedTableMetadataWriter` ahead of the streaming/batch fork, with the
functional case running a real clustering under the Spark default?
<details>
<summary>Chain (by reading, not executed end to end)</summary>
- `HoodieMetadataConfig:1445` defaults streaming writes on for Spark, table
version >= 8
- `SparkRDDTableServiceClient.writeToMetadataTable:106` routes to the
streaming handler on that flag alone
- `HoodieBackedTableMetadataWriter.completeStreamingCommit:990` runs
`buildUpdate` only for `getNonStreamingMetadataPartitionsToUpdate()`; RLI and
SI are the streaming partitions (`:872`, `:915`)
- `BaseRecordIndexer.getRecordIndexAdditionalUpserts` (batch only) is the
only place replaced groups' entries are deleted
- `RecordIndexMapper:78` still passes the raw config encoding rather than
`getFileIdEncoding`
- Read side: `RecordLevelIndexSupport.filterCandidateFiles` matches on live
slice fileIds, so a stale fileId yields an empty candidate set and
`HoodieFileIndex` prunes every slice
</details>
##########
hudi-common/src/main/java/org/apache/hudi/common/util/ExternalFilePathUtil.java:
##########
@@ -143,7 +175,12 @@ private static String getOriginalFileName(String fileName)
{
*/
public static StoragePath getFullPathOfPartition(StoragePath parent, String
fileName) {
return getExternalFileGroupPrefix(fileName)
- .map(prefix -> new StoragePath(parent.toString().substring(0,
parent.toString().length() - prefix.length() - 1)))
+ .map(prefix -> {
+ String parentPath = parent.toString();
+ checkArgument(parentPath.endsWith(StoragePath.SEPARATOR + prefix),
+ "External file " + fileName + " carries the file group prefix "
+ prefix + " but its parent " + parentPath + " does not end with it");
Review Comment:
**minor:** Not blocking. This `checkArgument` builds its message eagerly on
every call, and `getFullPathOfPartition` sits on the per-base-file view-build
path (`HoodieBaseFile:86`, `AbstractTableFileSystemView:180`), so every
prefixed external file pays the concatenation on each view build. Could we use
the `Supplier<String>` overload (`ValidationUtils:49`) so the message is only
built when the check fails?
```suggestion
checkArgument(parentPath.endsWith(StoragePath.SEPARATOR + prefix),
() -> "External file " + fileName + " carries the file group
prefix " + prefix + " but its parent " + parentPath + " does not end with it");
```
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -245,35 +326,32 @@ 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();
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);
+ // a file slice without a base file has only log files, whose schema is
the table schema
+ HoodieSchema readerSchema = dataFilePath.isPresent()
+ ?
HoodieIOFactory.getIOFactory(metaClient.getStorage()).getFileFormatUtils(baseFileFormat).readSchema(metaClient.getStorage(),
dataFilePath.get())
+ : resolveTableSchema(metaClient);
Review Comment:
**minor:** Not blocking. Resolving the schema inside the lambda answers the
laziness ask, but it now runs `new
TableSchemaResolver(metaClient).getTableSchema()` once per log-only file slice
on the executors, where b10d8167 ran it once on the driver; a MOR table with
many base-file-less slices pays that many timeline loads. Could we resolve it
once on the driver, guarded by `fileSlices.stream().anyMatch(s ->
!s.getFileSlice().getBaseFile().isPresent())`, and capture the result in the
closure?
--
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]
