Skip to content

perf(index): reuse driver-resolved base files in the bloom key check #20263

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
yihua wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

perf(index): reuse driver-resolved base files in the bloom key check #20263

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 27 additions & 3 deletions hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.core.io.storage.HoodieFileReader;
import org.apache.hudi.core.io.storage.HoodieFileReaderFactory;
import org.apache.hudi.core.io.storage.HoodieIOFactory;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.HoodieIndexException;
Expand All @@ -77,6 +78,7 @@
import org.apache.hudi.metadata.MetadataPartitionType;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
import org.apache.hudi.table.HoodieTable;

import lombok.extern.slf4j.Slf4j;
Expand Down Expand Up @@ -246,14 +248,36 @@ public static <R> HoodieRecord<R> tagRecord(HoodieRecord<R> record, HoodieRecord
public static Collection<Pair<String, Long>> filterKeysFromFile(StoragePath filePath,
Set<String> candidateRecordKeys,
HoodieStorage storage) throws HoodieIndexException {
return filterKeysFromFile(filePath, Option.empty(), candidateRecordKeys, storage);
}

/**
* Given a list of row keys and one base file, return only row keys existing in that file.
*
* @param baseFile - Base file to filter keys from
* @param candidateRecordKeys - Candidate keys to filter
* @param storage - Storage to read the file with
* @return List of pairs of candidate keys and positions that are available in the file
*/
public static Collection<Pair<String, Long>> filterKeysFromFile(HoodieBaseFile baseFile,
Set<String> candidateRecordKeys,
HoodieStorage storage) throws HoodieIndexException {
return filterKeysFromFile(baseFile.getStoragePath(), baseFile.getPathInfoWithKnownSize(), candidateRecordKeys, storage);
}

private static Collection<Pair<String, Long>> filterKeysFromFile(StoragePath filePath,
Option<StoragePathInfo> pathInfo,
Set<String> candidateRecordKeys,
HoodieStorage storage) throws HoodieIndexException {
checkArgument(FSUtils.isBaseFile(filePath));
if (candidateRecordKeys.isEmpty()) {
return Collections.emptyList();
}
log.info("Going to filter {} keys from file {}", candidateRecordKeys.size(), filePath);
try (HoodieFileReader fileReader = HoodieIOFactory.getIOFactory(storage)
.getReaderFactory(HoodieRecordType.AVRO)
.getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, filePath)) {
HoodieFileReaderFactory readerFactory = HoodieIOFactory.getIOFactory(storage).getReaderFactory(HoodieRecordType.AVRO);
try (HoodieFileReader fileReader = pathInfo.isPresent()
? readerFactory.getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, pathInfo.get())
: readerFactory.getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, filePath)) {
// Load all rowKeys from the file, to double-confirm
HoodieTimer timer = HoodieTimer.start();
Set<Pair<String, Long>> fileRowKeys = fileReader.filterRowKeys(candidateRecordKeys);
Expand Down
13 changes: 13 additions & 0 deletions ...i-client-common/src/main/java/org/apache/hudi/index/bloom/BaseHoodieBloomIndexHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,13 +21,15 @@

import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecordLocation;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.table.HoodieTable;

import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

Expand All @@ -53,4 +55,15 @@ public abstract HoodiePairData<HoodieKey, HoodieRecordLocation> findMatchingFile
HoodiePairData<HoodieFileGroupId, String> fileComparisonPairs,
Map<String, List<BloomIndexFileInfo>> partitionToFileInfo,
Map<String, Long> recordsPerPartition);

/**
* Collects the base files carried by the {@link BloomIndexFileInfo}s by file group.
*/
protected static Map<HoodieFileGroupId, HoodieBaseFile> getBaseFilesByFileGroup(
Map<String, List<BloomIndexFileInfo>> partitionToFileInfo) {
Map<HoodieFileGroupId, HoodieBaseFile> baseFiles = new HashMap<>();
partitionToFileInfo.forEach((partitionPath, fileInfos) -> fileInfos.forEach(fileInfo -> fileInfo.getBaseFile()
.ifPresent(baseFile -> baseFiles.put(new HoodieFileGroupId(partitionPath, fileInfo.getFileId()), baseFile))));
return baseFiles;
}
}
39 changes: 35 additions & 4 deletions ...ient/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/BloomIndexFileInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,11 @@

package org.apache.hudi.index.bloom;

import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.util.Option;

import lombok.EqualsAndHashCode;
import lombok.ToString;
import lombok.Value;

import java.io.Serializable;
Expand All @@ -35,16 +40,42 @@ public class BloomIndexFileInfo implements Serializable {

String maxRecordKey;

/**
* The latest base file of the file group, when the loader resolved it. It is not part of the
* equality, which covers the file group and its key range.
*/
@EqualsAndHashCode.Exclude
@ToString.Exclude
Option<HoodieBaseFile> baseFile;

public BloomIndexFileInfo(String fileId, String minRecordKey, String maxRecordKey) {
this(fileId, minRecordKey, maxRecordKey, Option.empty());
}

public BloomIndexFileInfo(String fileId) {
this(fileId, null, null, Option.empty());
}

public BloomIndexFileInfo(HoodieBaseFile baseFile, String minRecordKey, String maxRecordKey) {
this(baseFile.getFileId(), minRecordKey, maxRecordKey, Option.of(baseFile));
}

public BloomIndexFileInfo(HoodieBaseFile baseFile) {
this(baseFile.getFileId(), null, null, Option.of(baseFile));
}

private BloomIndexFileInfo(String fileId, String minRecordKey, String maxRecordKey, Option<HoodieBaseFile> baseFile) {
this.fileId = fileId;
this.minRecordKey = minRecordKey;
this.maxRecordKey = maxRecordKey;
this.baseFile = baseFile;
}

public BloomIndexFileInfo(String fileId) {
this.fileId = fileId;
this.minRecordKey = null;
this.maxRecordKey = null;
/**
* Returns this file info without the base file, for when only the file group and its key range are needed.
*/
public BloomIndexFileInfo withoutBaseFile() {
return baseFile.isPresent() ? new BloomIndexFileInfo(fileId, minRecordKey, maxRecordKey) : this;
}

public boolean hasKeyRanges() {
Expand Down
20 changes: 9 additions & 11 deletions ...client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -172,13 +172,14 @@ List<Pair<String, BloomIndexFileInfo>> loadColumnRangesFromFiles(

context.setJobStatus(this.getClass().getName(), "Obtain key ranges for file slices (range pruning=on): " + config.getTableName());
List<Pair<String, BloomIndexFileInfo>> result = context.map(partitionPathFileIDList, pf -> {
HoodieBaseFile baseFile = pf.getValue().getValue();
try {
HoodieRangeInfoHandle rangeInfoHandle = new HoodieRangeInfoHandle(config, hoodieTable, Pair.of(pf.getKey(), pf.getValue().getKey()));
String[] minMaxKeys = rangeInfoHandle.getMinMaxKeys(pf.getValue().getValue());
return Pair.of(pf.getKey(), new BloomIndexFileInfo(pf.getValue().getKey(), minMaxKeys[0], minMaxKeys[1]));
String[] minMaxKeys = rangeInfoHandle.getMinMaxKeys(baseFile);
return Pair.of(pf.getKey(), new BloomIndexFileInfo(baseFile, minMaxKeys[0], minMaxKeys[1]));
} catch (MetadataNotFoundException me) {
log.warn("Unable to find range metadata in file :{}", pf);
return Pair.of(pf.getKey(), new BloomIndexFileInfo(pf.getValue().getKey()));
return Pair.of(pf.getKey(), new BloomIndexFileInfo(baseFile));
}
}, Math.max(partitionPathFileIDList.size(), 1));
context.clearJobStatus();
Expand All @@ -195,12 +196,9 @@ List<Pair<String, BloomIndexFileInfo>> loadColumnRangesFromFiles(
*/
private List<Pair<String, BloomIndexFileInfo>> getFileInfoForLatestBaseFiles(
List<String> partitions, final HoodieEngineContext context, final HoodieTable hoodieTable) {
List<Pair<String, String>> partitionPathFileIDList = getLatestBaseFilesForAllPartitions(partitions, context,
hoodieTable).stream()
.map(pair -> Pair.of(pair.getKey(), pair.getValue().getFileId()))
return getLatestBaseFilesForAllPartitions(partitions, context, hoodieTable).stream()
.map(pair -> Pair.of(pair.getKey(), new BloomIndexFileInfo(pair.getValue())))
.collect(toList());
return partitionPathFileIDList.stream()
.map(pf -> Pair.of(pf.getKey(), new BloomIndexFileInfo(pf.getValue()))).collect(toList());
}

/**
Expand All @@ -221,11 +219,11 @@ protected List<Pair<String, BloomIndexFileInfo>> loadColumnRangesFromMetaIndex(
List<Pair<String, HoodieBaseFile>> baseFilesForAllPartitions = HoodieIndexUtils.getLatestBaseFilesForAllPartitions(partitions, context, hoodieTable);
// Partition and file name pairs
List<Pair<String, String>> partitionFileNameList = new ArrayList<>(baseFilesForAllPartitions.size());
Map<Pair<String, String>, String> partitionAndFileNameToFileId = new HashMap<>(baseFilesForAllPartitions.size(), 1);
Map<Pair<String, String>, HoodieBaseFile> partitionAndFileNameToBaseFile = new HashMap<>(baseFilesForAllPartitions.size(), 1);
baseFilesForAllPartitions.forEach(pair -> {
Pair<String, String> partitionAndFileName = Pair.of(pair.getKey(), pair.getValue().getFileName());
partitionFileNameList.add(partitionAndFileName);
partitionAndFileNameToFileId.put(partitionAndFileName, pair.getValue().getFileId());
partitionAndFileNameToBaseFile.put(partitionAndFileName, pair.getValue());
});

if (partitionFileNameList.isEmpty()) {
Expand All @@ -241,7 +239,7 @@ protected List<Pair<String, BloomIndexFileInfo>> loadColumnRangesFromMetaIndex(
ValueMetadata valueMetadata = ValueMetadata.getValueMetadata(entry.getValue().getValueType());
result.add(Pair.of(entry.getKey().getLeft(),
new BloomIndexFileInfo(
partitionAndFileNameToFileId.get(entry.getKey()),
partitionAndFileNameToBaseFile.get(entry.getKey()),
// NOTE: Here we assume that the type of the primary key field is string
valueMetadata.unwrapValue(entry.getValue().getMinValue()).toString(),
valueMetadata.unwrapValue(entry.getValue().getMaxValue()).toString()
Expand Down
19 changes: 16 additions & 3 deletions ...lient-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndexCheckFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,9 @@
package org.apache.hudi.index.bloom;

import org.apache.hudi.common.function.SerializableFunction;
import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieException;
Expand Down Expand Up @@ -48,22 +50,33 @@ public class HoodieBloomIndexCheckFunction<I>

private final SerializableFunction<I, HoodieFileGroupId> fileGroupIdExtractor;
private final SerializableFunction<I, String> recordKeyExtractor;
private final SerializableFunction<HoodieFileGroupId, Option<HoodieBaseFile>> baseFileLookup;

/**
* @param baseFileLookup returns the latest base file of a file group when the caller already has it, so the
* lookup handle does not ask the file system view for it
*/
public HoodieBloomIndexCheckFunction(HoodieTable hoodieTable,
HoodieWriteConfig config,
SerializableFunction<I, HoodieFileGroupId> fileGroupIdExtractor,
SerializableFunction<I, String> recordKeyExtractor) {
SerializableFunction<I, String> recordKeyExtractor,
SerializableFunction<HoodieFileGroupId, Option<HoodieBaseFile>> baseFileLookup) {
this.hoodieTable = hoodieTable;
this.config = config;
this.fileGroupIdExtractor = fileGroupIdExtractor;
this.recordKeyExtractor = recordKeyExtractor;
this.baseFileLookup = baseFileLookup;
}

@Override
public Iterator<HoodieKeyLookupResult> apply(Iterator<I> fileGroupIdRecordKeyPairIterator) {
return new LazyKeyCheckIterator(fileGroupIdRecordKeyPairIterator);
}

private HoodieKeyLookupHandle newKeyLookupHandle(HoodieFileGroupId fileGroupId, Pair<String, String> partitionPathFilePair) throws Exception {
return new HoodieKeyLookupHandle(config, hoodieTable, partitionPathFilePair, baseFileLookup.apply(fileGroupId));
}

protected class LazyKeyCheckIterator implements Iterator<HoodieKeyLookupResult> {

private final Iterator<I> filePartitionRecordKeyPairItr;
Expand Down Expand Up @@ -95,7 +108,7 @@ public HoodieKeyLookupResult next() {

// lazily init state
if (keyLookupHandle == null) {
keyLookupHandle = new HoodieKeyLookupHandle(config, hoodieTable, partitionPathFilePair);
keyLookupHandle = newKeyLookupHandle(fileGroupId, partitionPathFilePair);
}

// if continue on current file
Expand All @@ -104,7 +117,7 @@ public HoodieKeyLookupResult next() {
} else {
// do the actual checking of file & break out
HoodieKeyLookupResult result = keyLookupHandle.getLookupResult();
keyLookupHandle = new HoodieKeyLookupHandle(config, hoodieTable, partitionPathFilePair);
keyLookupHandle = newKeyLookupHandle(fileGroupId, partitionPathFilePair);
keyLookupHandle.addKey(recordKey);
return result;
}
Expand Down
6 changes: 5 additions & 1 deletion ...ent-common/src/main/java/org/apache/hudi/index/bloom/ListBasedHoodieBloomIndexHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,12 @@

import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecordLocation;
import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.ImmutablePair;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
Expand Down Expand Up @@ -62,9 +64,11 @@ public HoodiePairData<HoodieKey, HoodieRecordLocation> findMatchingFilesForRecor
fileComparisonPairs.collectAsList().stream()
.sorted(Comparator.comparing(Pair::getLeft)).collect(toList());

Map<HoodieFileGroupId, HoodieBaseFile> baseFiles = getBaseFilesByFileGroup(partitionToFileInfo);
List<HoodieKeyLookupResult> keyLookupResults =
CollectionUtils.toStream(
new HoodieBloomIndexCheckFunction<Pair<HoodieFileGroupId, String>>(hoodieTable, config, Pair::getLeft, Pair::getRight)
new HoodieBloomIndexCheckFunction<Pair<HoodieFileGroupId, String>>(hoodieTable, config, Pair::getLeft, Pair::getRight,
fileGroupId -> Option.ofNullable(baseFiles.get(fileGroupId)))
.apply(fileComparisonPairList.iterator())
)
.filter(lr -> !lr.getMatchingRecordKeysAndPositions().isEmpty())
Expand Down
6 changes: 5 additions & 1 deletion ...udi-client-common/src/main/java/org/apache/hudi/index/bloom/ListBasedIndexFileFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
* Simple implementation of {@link IndexFileFilter}. Sequentially goes through every index file in a given partition to
Expand All @@ -39,7 +40,10 @@ class ListBasedIndexFileFilter implements IndexFileFilter {
* @param partitionToFileIndexInfo Map of partition to List of {@link BloomIndexFileInfo}
*/
ListBasedIndexFileFilter(final Map<String, List<BloomIndexFileInfo>> partitionToFileIndexInfo) {
this.partitionToFileIndexInfo = partitionToFileIndexInfo;
// the filter is shipped with every task that pairs records with candidate files, which needs only the key ranges
this.partitionToFileIndexInfo = partitionToFileIndexInfo.entrySet().stream()
.collect(Collectors.toMap(Map.Entry::getKey,
entry -> entry.getValue().stream().map(BloomIndexFileInfo::withoutBaseFile).collect(Collectors.toList())));
}

@Override
Expand Down
7 changes: 6 additions & 1 deletion ...ent/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieKeyLocationFetchHandle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.core.io.storage.HoodieIOFactory;
import org.apache.hudi.keygen.BaseKeyGenerator;
import org.apache.hudi.storage.StoragePathInfo;
import org.apache.hudi.table.HoodieTable;

/**
Expand All @@ -52,7 +53,11 @@ public HoodieKeyLocationFetchHandle(HoodieWriteConfig config, HoodieTable<T, I,
private ClosableIterator<Pair<HoodieKey, Long>> fetchRecordKeysWithPositions(HoodieBaseFile baseFile) {
FileFormatUtils fileFormatUtils = HoodieIOFactory.getIOFactory(hoodieTable.getStorage())
.getFileFormatUtils(baseFile.getStoragePath());
return fileFormatUtils.fetchRecordKeysWithPositions(hoodieTable.getStorage(), baseFile.getStoragePath(), keyGeneratorOpt, Option.of(partitionPathBaseFilePair.getKey()));
Option<StoragePathInfo> pathInfo = baseFile.getPathInfoWithKnownSize();
Option<String> partitionPath = Option.of(partitionPathBaseFilePair.getKey());
return pathInfo.isPresent()
? fileFormatUtils.fetchRecordKeysWithPositions(hoodieTable.getStorage(), pathInfo.get(), keyGeneratorOpt, partitionPath)
: fileFormatUtils.fetchRecordKeysWithPositions(hoodieTable.getStorage(), baseFile.getStoragePath(), keyGeneratorOpt, partitionPath);
}

public ClosableIterator<Pair<HoodieKey, HoodieRecordLocation>> locations() {
Expand Down
Loading
Loading