the-other-tim-brown commented on code in PR #13300:
URL: https://github.com/apache/hudi/pull/13300#discussion_r2098706974
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/KeyBasedFileGroupRecordBuffer.java:
##########
@@ -77,6 +79,11 @@ public void processDataBlock(HoodieDataBlock dataBlock,
Option<KeySpec> keySpecO
try (ClosableIterator<T> recordIterator =
recordsIteratorSchemaPair.getLeft()) {
while (recordIterator.hasNext()) {
T nextRecord = recordIterator.next();
+ String recordKey = readerContext.getRecordKey(nextRecord, schema);
+ // The filter is given but record key does not pass the filter.
+ if (useFilter && !evalFilter.get().apply(recordKey)) {
Review Comment:
instead of requiring a `useFilter` arg, can we just have a filter that
always returns true?
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodieFileGroupReader.java:
##########
@@ -497,6 +509,11 @@ public Builder<T> withAllowInflightInstants(boolean
allowInflightInstants) {
return this;
}
+ public Builder<T> withFilter(Option<Predicate> filter) {
Review Comment:
It looks like this is only applied to the keys. If that is the intention
then the naming should reflect that
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java:
##########
@@ -210,38 +222,85 @@ public HoodieData<HoodieRecord<HoodieMetadataPayload>>
getRecordsByKeyPrefixes(L
getEngineContext().parallelize(partitionFileSlices))
.flatMap(
(SerializableFunction<FileSlice,
Iterator<HoodieRecord<HoodieMetadataPayload>>>) fileSlice -> {
- // NOTE: Since this will be executed by executors, we can't
access previously cached
- // readers, and therefore have to always open new ones
- Pair<HoodieSeekingFileReader<?>, HoodieMetadataLogRecordReader>
readers =
- openReaders(partitionName, fileSlice);
- try {
- List<Long> timings = new ArrayList<>();
-
- HoodieSeekingFileReader<?> baseFileReader = readers.getKey();
- HoodieMetadataLogRecordReader logRecordScanner =
readers.getRight();
-
- if (baseFileReader == null && logRecordScanner == null) {
- // TODO: what do we do if both does not exist? should we
throw an exception and let caller do the fallback ?
- return Collections.emptyIterator();
- }
+ return getByKeyPrefixesWithFileGroupReader(fileSlice,
sortedKeyPrefixes, partitionName);
+ });
+ }
- boolean fullKeys = false;
+ private Iterator<HoodieRecord<HoodieMetadataPayload>>
getByKeyPrefixesWithFileGroupReader(FileSlice fileSlice,
+
List<String> sortedKeyPrefixes,
+
String partitionName) throws IOException {
+ 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();
+ HoodieFileGroupReader<IndexedRecord> fileGroupReader = getFileGroupReader(
+ metadataMetaClient.getTableConfig(),
+ latestMetadataInstantTime,
+ fileSlice,
+ schema,
+ schema,
+ metadataMetaClient,
+ new TypedProperties(),
+ Option.of(transformKeyPrefixesToPredicate(sortedKeyPrefixes)));
+ fileGroupReader.setInstantRange(Option.of(instantRange));
+ ClosableIterator<IndexedRecord> it = fileGroupReader.getClosableIterator();
+ return new HoodieRecordIterator(it, partitionName, sortedKeyPrefixes);
+ }
+
+ private Predicate transformKeysToPredicate(List<String> keys) {
+ List<Expression> right = keys.stream().map(k ->
Literal.from(k)).collect(Collectors.toList());
+ return Predicates.in(null, right);
+ }
+
+ private Predicate transformKeyPrefixesToPredicate(List<String> keyPrefixes) {
+ List<Expression> right = keyPrefixes.stream().map(kp ->
Literal.from(kp)).collect(Collectors.toList());
+ return Predicates.startsWithAny(null, right);
+ }
+
+ public static class HoodieRecordIterator implements
Iterator<HoodieRecord<HoodieMetadataPayload>> {
+ private final ClosableIterator<IndexedRecord> baseIterator;
+ private final String partitionName;
+ private final List<String> sortedKeyPrefixes;
+ private GenericRecord metadataRecord;
+
+ public HoodieRecordIterator(ClosableIterator<IndexedRecord> baseIterator,
String partitionName, List<String> sortedKeyPrefixes) {
+ this.baseIterator = baseIterator;
+ this.partitionName = partitionName;
+ this.sortedKeyPrefixes = sortedKeyPrefixes;
+ }
- Map<String, HoodieRecord<HoodieMetadataPayload>> logRecords =
- readLogRecords(logRecordScanner, sortedKeyPrefixes,
fullKeys, timings);
+ @Override
+ public boolean hasNext() {
+ if (metadataRecord != null) {
+ return true;
+ }
- Map<String, HoodieRecord<HoodieMetadataPayload>> mergedRecords
=
- readFromBaseAndMergeWithLogRecords(baseFileReader,
sortedKeyPrefixes, fullKeys, logRecords, timings, partitionName);
+ if (baseIterator.hasNext()) {
+ metadataRecord = (GenericRecord) baseIterator.next();
Review Comment:
Should this just call `hasNext`?
--
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]
