You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "yihua (via GitHub)" <gi...@apache.org> on 2025/07/11 17:30:04 UTC
Re: [PR] [HUDI-9235] Read MDT through FG reader [hudi]
yihua commented on code in PR #13300:
URL: https://github.com/apache/hudi/pull/13300#discussion_r2201410390
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java:
##########
@@ -292,180 +325,58 @@ private static ArrayList<ArrayList<String>> partitionKeysByFileSlices(List<Strin
return partitionedKeys;
}
- /**
- * Lookup list of keys from a single file slice.
- *
- * @param partitionName Name of the partition
- * @param keys The list of keys to lookup
- * @param fileSlice The file slice to read
- * @return A {@code Map} of key name to {@code HoodieRecord} for the keys which were found in the file slice
- */
- private Map<String, HoodieRecord<HoodieMetadataPayload>> lookupKeysFromFileSlice(String partitionName, List<String> keys, FileSlice fileSlice) {
- Pair<HoodieSeekingFileReader<?>, HoodieMetadataLogRecordReader> readers = getOrCreateReaders(partitionName, fileSlice);
- try {
- HoodieSeekingFileReader<?> baseFileReader = readers.getKey();
- HoodieMetadataLogRecordReader logRecordScanner = readers.getRight();
- if (baseFileReader == null && logRecordScanner == null) {
- return Collections.emptyMap();
- }
-
- // Sort it here once so that we don't need to sort individually for base file and for each individual log files.
- List<String> sortedKeys = new ArrayList<>(keys);
- Collections.sort(sortedKeys);
- boolean fullKeys = true;
- List<Long> timings = new ArrayList<>(1);
- Map<String, HoodieRecord<HoodieMetadataPayload>> logRecords = readLogRecords(logRecordScanner, sortedKeys, fullKeys, timings);
- return readFromBaseAndMergeWithLogRecords(baseFileReader, sortedKeys, fullKeys, logRecords, timings, partitionName);
- } catch (IOException ioe) {
- throw new HoodieIOException("Error merging records from metadata table for " + keys.size() + " key : ", ioe);
- } finally {
- if (!reuse) {
- closeReader(readers);
+ private Map<String, HoodieRecord<HoodieMetadataPayload>> lookupKeys(String partitionName,
+ List<String> keys,
+ FileSlice fileSlice) {
+ // Sort it here once so that we don't need to sort individually for base file and for each individual log files.
+ List<String> sortedKeys = new ArrayList<>(keys);
+ // So we use the natural order to sort.
+ Collections.sort(sortedKeys);
+ Option<HoodieInstant> latestMetadataInstant =
+ metadataMetaClient.getActiveTimeline().filterCompletedInstants().lastInstant();
+ String latestMetadataInstantTime =
+ latestMetadataInstant.map(HoodieInstant::requestedTime).orElse(SOLO_COMMIT_TIMESTAMP);
+ Schema schema = HoodieAvroUtils.addMetadataFields(HoodieMetadataRecord.getClassSchema());
+ // Only those log files which have a corresponding completed instant on the dataset should be read
+ // This is because the metadata table is updated before the dataset instants are committed.
+ Set<String> validInstantTimestamps = getValidInstantTimestamps();
+ InstantRange instantRange = InstantRange.builder()
+ .rangeType(InstantRange.RangeType.EXACT_MATCH)
+ .explicitInstants(validInstantTimestamps).build();
+ HoodieReaderContext<IndexedRecord> readerContext = new HoodieAvroReaderContext(
+ storageConf,
+ metadataMetaClient.getTableConfig(),
+ Option.of(instantRange),
+ Option.of(transformKeysToPredicate(sortedKeys)));
+ try (HoodieFileGroupReader<IndexedRecord> fileGroupReader = HoodieFileGroupReader.<IndexedRecord>newBuilder()
+ .withReaderContext(readerContext)
+ .withHoodieTableMetaClient(metadataMetaClient)
+ .withLatestCommitTime(latestMetadataInstantTime)
+ .withFileSlice(fileSlice)
+ .withDataSchema(schema)
+ .withRequestedSchema(schema)
+ .withProps(buildFileGroupReaderProperties(metadataConfig))
+ .withStart(0)
+ .withLength(Long.MAX_VALUE)
+ .withShouldUseRecordPosition(false)
+ .build();
+ ClosableIterator<IndexedRecord> it = fileGroupReader.getClosableIterator()) {
+ Map<String, HoodieRecord<HoodieMetadataPayload>> records = new HashMap<>();
+ while (it.hasNext()) {
+ GenericRecord metadataRecord = (GenericRecord) it.next();
+ HoodieMetadataPayload payload = new HoodieMetadataPayload(Option.of(metadataRecord));
+ String rowKey = payload.key != null
+ ? payload.key : ((GenericRecord) metadataRecord).get(KEY_FIELD_NAME).toString();
+ HoodieKey hoodieKey = new HoodieKey(rowKey, partitionName);
+ records.put(rowKey, new HoodieAvroRecord<>(hoodieKey, payload));
}
- }
- }
-
- private Map<String, HoodieRecord<HoodieMetadataPayload>> readLogRecords(HoodieMetadataLogRecordReader logRecordReader,
- List<String> sortedKeys,
- boolean fullKey,
- List<Long> timings) {
- HoodieTimer timer = HoodieTimer.start();
-
- if (logRecordReader == null) {
- timings.add(timer.endTimer());
- return Collections.emptyMap();
- }
-
- try {
- return fullKey ? logRecordReader.getRecordsByKeys(sortedKeys) : logRecordReader.getRecordsByKeyPrefixes(sortedKeys);
- } finally {
- timings.add(timer.endTimer());
- }
- }
-
- private Map<String, HoodieRecord<HoodieMetadataPayload>> readFromBaseAndMergeWithLogRecords(HoodieSeekingFileReader<?> reader,
- List<String> sortedKeys,
- boolean fullKeys,
- Map<String, HoodieRecord<HoodieMetadataPayload>> logRecords,
- List<Long> timings,
- String partitionName) throws IOException {
- HoodieTimer timer = HoodieTimer.start();
-
- if (reader == null) {
- // No base file at all
- timings.add(timer.endTimer());
- return logRecords;
- }
-
- HoodieTimer readTimer = HoodieTimer.start();
-
- Map<String, HoodieRecord<HoodieMetadataPayload>> records =
- fetchBaseFileRecordsByKeys(reader, sortedKeys, fullKeys, partitionName);
-
- metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.BASEFILE_READ_STR, readTimer.endTimer()));
-
- // Iterate over all provided log-records, merging them into existing records
- logRecords.values().forEach(logRecord ->
- records.merge(
- logRecord.getRecordKey(),
- logRecord,
- (oldRecord, newRecord) -> {
- HoodieMetadataPayload mergedPayload = newRecord.getData().preCombine(oldRecord.getData());
- return mergedPayload.isDeleted() ? null : new HoodieAvroRecord<>(oldRecord.getKey(), mergedPayload);
- }
- ));
-
- timings.add(timer.endTimer());
- return records;
- }
-
- @SuppressWarnings("unchecked")
- private Map<String, HoodieRecord<HoodieMetadataPayload>> fetchBaseFileRecordsByKeys(HoodieSeekingFileReader reader,
- List<String> sortedKeys,
- boolean fullKeys,
- String partitionName) throws IOException {
- Map<String, HoodieRecord<HoodieMetadataPayload>> result;
- try (ClosableIterator<HoodieRecord<?>> records = fullKeys
- ? reader.getRecordsByKeysIterator(sortedKeys)
- : reader.getRecordsByKeyPrefixIterator(sortedKeys)) {
- result = toStream(records)
- .map(record -> {
- GenericRecord data = (GenericRecord) record.getData();
- // populateMetaFields is hardcoded to false for the metadata table so key must be extracted from the `key` field
- String recordKey = (String) data.get(HoodieMetadataPayload.KEY_FIELD_NAME);
- return Pair.of(recordKey, composeRecord(data, recordKey, partitionName));
- })
- .collect(Collectors.toMap(Pair::getKey, Pair::getValue));
- }
- return result;
- }
-
- private HoodieRecord<HoodieMetadataPayload> composeRecord(GenericRecord avroRecord, String recordKey, String partitionName) {
- return new HoodieAvroRecord<>(new HoodieKey(recordKey, partitionName),
- new HoodieMetadataPayload(avroRecord, 0L), null);
- }
-
- /**
- * Create a file reader and the record scanner for a given partition and file slice
- * if readers are not already available.
- *
- * @param partitionName - Partition name
- * @param slice - The file slice to open readers for
- * @return File reader and the record scanner pair for the requested file slice
- */
- private Pair<HoodieSeekingFileReader<?>, HoodieMetadataLogRecordReader> getOrCreateReaders(String partitionName, FileSlice slice) {
- if (reuse) {
- Pair<String, String> key = Pair.of(partitionName, slice.getFileId());
- return partitionReaders.get().computeIfAbsent(key, ignored -> openReaders(partitionName, slice));
- } else {
- return openReaders(partitionName, slice);
- }
Review Comment:
Is the `reuse` behavior handled in a different way though the FGReader-based MDT reading?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org