hudi-agent commented on code in PR #19869:
URL: https://github.com/apache/hudi/pull/19869#discussion_r3961511572
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -197,6 +207,12 @@ public static <T> HoodieData<HoodieRecord>
convertWriteStatsToSecondaryIndexReco
return records.iterator();
});
+ if (commitMetadata instanceof HoodieReplaceCommitMetadata) {
Review Comment:
🤖 This branch is unreachable for a replace commit that only drops files:
`SecondaryIndexer.getSecondaryIndexUpdates` returns early when
`allWriteStats.isEmpty()`, while the RLI side
(`getRecordIndexAdditionalUpserts`) still emits deletes for the replaced
groups. An XTable sync of a source `DELETE` that removes a whole data file
without adding one would then leave SI entries pointing at record keys that no
longer exist in RLI. Could the early return in `SecondaryIndexer` be relaxed
for `HoodieReplaceCommitMetadata` with a non-empty `partitionToReplaceFileIds`?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -319,17 +380,32 @@ public boolean hasNext() {
while (recordIterator.hasNext()) {
T record = recordIterator.next();
Object secondaryKey =
readerContext.getRecordContext().getValue(record, requestedSchema,
secondaryKeyField);
- nextValidRecord = Pair.of(
- readerContext.getRecordContext().getRecordKey(record,
requestedSchema),
- secondaryKey == null ? null : secondaryKey.toString()
- );
+ nextValidRecord = Pair.of(getRecordKey(record), secondaryKey == null
? null : secondaryKey.toString());
+ rowPosition++;
return true;
}
// If no valid records are found
return false;
}
+ private String getRecordKey(T record) {
+ Object recordKey;
+ if (hasRecordKeyMetaField) {
+ recordKey = readerContext.getRecordContext().getValue(record,
requestedSchema, RECORD_KEY_METADATA_FIELD);
+ } else if (hasRecordKeyFields) {
+ recordKey = readerContext.getRecordContext().getRecordKey(record,
requestedSchema);
+ } else {
+ recordKey = null;
Review Comment:
🤖 The positional-key fallback kicks in for any null key, not just for
external files. `TableSchemaResolver.getTableSchema()` includes meta fields for
every `MetaFieldsMode != NONE`, so on a keyed table in a selective mode (e.g.
COMMIT_TIME_ONLY, where `_hoodie_record_key` is physically null)
`hasRecordKeyMetaField` is true, `getValue` returns null, and every row
silently gets `path_pos` as its SI key instead of failing as it did before.
Would it be safer to only generate a key when `!hasRecordKeyMetaField &&
!hasRecordKeyFields`, and otherwise fail loudly on a null key?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/metadata/BaseFileRecordParsingUtils.java:
##########
@@ -167,10 +171,38 @@ public static Map<RecordStatus, List<String>>
getRecordKeyStatuses(String basePa
}
}
+ /**
+ * 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.
+ * @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) {
+ return getRecordKeysFromBaseFile(storage, basePath, dataFilePath).stream()
+ .map(recordKey ->
HoodieMetadataPayload.createRecordIndexDelete(recordKey, partition,
isPartitionedRLI))
+ .iterator();
+ }
+
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);
+ // a file written outside Hudi is recorded with an external file marker
that is not part of the name on storage.
+ String filePathInPartition =
ExternalFilePathUtil.getFilePathInPartition(fileName);
+ StoragePath dataFilePath = new StoragePath(basePath,
StringUtils.isNullOrEmpty(partition) ? filePathInPartition : (partition +
StoragePath.SEPARATOR) + filePathInPartition);
+ return getRecordKeysFromBaseFile(storage, basePath, dataFilePath);
+ }
+
+ private static Set<String> getRecordKeysFromBaseFile(HoodieStorage storage,
String basePath, StoragePath dataFilePath) {
FileFormatUtils fileFormatUtils =
HoodieIOFactory.getIOFactory(storage).getFileFormatUtils(HoodieFileFormat.PARQUET);
- return fileFormatUtils.readRowKeys(storage, dataFilePath);
+ return fileFormatUtils.readRowKeys(storage, dataFilePath, new
StoragePath(basePath));
Review Comment:
🤖 This now always passes `basePath`, so `ParquetUtils.filterParquetRowKeys`
will generate `path_pos` keys for any Hudi-written base file whose
`_hoodie_record_key` column is null (e.g. selective `MetaFieldsMode`s), where
it previously failed with an NPE. Would it make sense to only pass the base
path when `ExternalFilePathUtil.isExternallyCreatedFile(fileName)` (and for the
replaced-file variant, when the resolved file is external), so a missing key in
a Hudi-written file still fails loudly?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java:
##########
@@ -319,17 +380,32 @@ public boolean hasNext() {
while (recordIterator.hasNext()) {
T record = recordIterator.next();
Object secondaryKey =
readerContext.getRecordContext().getValue(record, requestedSchema,
secondaryKeyField);
- nextValidRecord = Pair.of(
- readerContext.getRecordContext().getRecordKey(record,
requestedSchema),
- secondaryKey == null ? null : secondaryKey.toString()
- );
+ nextValidRecord = Pair.of(getRecordKey(record), secondaryKey == null
? null : secondaryKey.toString());
+ rowPosition++;
Review Comment:
🤖 The SI key here depends on `HoodieFileGroupReader` emitting every row of
the base file in file order so that `rowPosition` lines up with the
`rowPosition` counted by `AvroParquetReader` in
`ParquetUtils.filterParquetRowKeys` for RLI. Is there anything in the file
group reader path (instant-range filtering, delete handling, engine-specific
readers) that could drop or reorder rows for a key-less base file? If the two
counters ever diverge, SI would resolve to the wrong RLI entry with no error.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java:
##########
@@ -478,11 +478,48 @@ 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 &&
operationType != WriteOperationType.CLUSTER) {
+ // a replace commit that is neither a table service nor an overwrite,
e.g. files written outside Hudi being
+ // registered in the table. 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 them again.
Review Comment:
🤖 nit: this chained mapToPair/leftOuterJoin/values/filter/map is doing a lot
in one expression (dedupe replaced-file-group deletes against records rewritten
in the same commit) - might be worth breaking it into named intermediate
variables so the intent is easier to follow at a glance.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java:
##########
@@ -478,11 +478,48 @@ 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 &&
operationType != WriteOperationType.CLUSTER) {
+ // a replace commit that is neither a table service nor an overwrite,
e.g. files written outside Hudi being
+ // registered in the table. 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 them again.
+ return
getRecordIndexReplacedFileGroupRecords((HoodieReplaceCommitMetadata)
commitMetadata, fsView)
+ .mapToPair(r -> Pair.of(r.getKey(), r))
+ .leftOuterJoin(updatesFromWriteStatuses.mapToPair(r ->
Pair.of(r.getKey(), r)))
Review Comment:
🤖 nit: `p` is a pretty generic name for this Pair of (replaced record,
matching write-status option) - since this method's whole point is
distinguishing replaced-but-unwritten records from rewritten ones, a more
descriptive name would help readability here.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
