This is an automated email from the ASF dual-hosted git repository.
akudinkin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new e270924c5eb [HUDI-5534] Optimizing Bloom Index lookup when using Bloom
Filters from Metadata Table (#7642)
e270924c5eb is described below
commit e270924c5eb1548d9022fe3e52d5dd64f61fbdc1
Author: Alexey Kudinkin <alexey.kudin...@gmail.com>
AuthorDate: Fri Jan 27 15:47:12 2023 -0800
[HUDI-5534] Optimizing Bloom Index lookup when using Bloom Filters from
Metadata Table (#7642)
Most recently, while trying to use Metadata Table in Bloom Index it was
resulting in failures due to exhaustion of S3 connection pool no matter how
(reasonably big) we're setting the pool size (we've tested up to 3k
connections).
This PR focuses on optimizing the Bloom Index lookup sequence in case when
it's leveraging Bloom Filter partition in Metadata Table. The premise of this
change is based on the following observations:
Increasing the size of the batch of the requests to MT allows to amortize
the cost of processing it (bigger the batch, lesser the cost).
Having too few partitions in the Bloom Index path however, starts to hurt
parallelism when we actually probe individual files whether they actually
contain target keys or not. Solution to this is to split these 2 in different
stages w/ drastically different parallelism levels: constrain parallelism when
reading from MT (10s of tasks) and keep at the current level for probing
individual files (100s of tasks)
Current way of partitioning records (relying on Spark's default
partitioner) was entailing that every Spark executor with high likelihood will
be opening up (and processing) every file-group of the MT Bloom Filter
partition. To alleviate that same hashing algorithm used by MT should be used
to partition records into Spark's individual partitions, so that we can limit
every task to open no more than 1 file-group in Bloom Filter's partition of MT
To achieve that following changes in Bloom Index sequence (leveraging MT)
are implemented
Bloom Filter probing and actual File Probing are split into 2 separate
operations (so that parallelism of each of them could be controlled
individually)
Requests to MT are replaced to invoke batch APIs
Custom partitioner is introduced AffineBloomIndexFileGroupPartitioner
repartitioning dataset of filenames with corresponding record keys in a way
that is affine w/ MT Bloom Filters' partitioning (allowing us to open no more
than a single file-group per Spark's task)
Additionally, this PR addresses some of the low-hanging performance
optimizations that could considerably improve performance of the Bloom Index
lookup sequence like mapping file-comparison pairs to PairRDD (where key is
file-name, and value is record-key) instead of RDD so that we could:
Do in-partition sorting by filename (to make sure we check all records w/in
the file all at once) w/in a single Spark partition instead of global one
(reducing shuffling as well)
Avoid re-shuffling (by re-mapping from RDD to PairRDD later)
---
.../org/apache/hudi/index/HoodieIndexUtils.java | 5 +-
.../index/bloom/BaseHoodieBloomIndexHelper.java | 5 +-
.../apache/hudi/index/bloom/HoodieBloomIndex.java | 77 +++----
...ion.java => HoodieBloomIndexCheckFunction.java} | 59 +++---
.../hudi/index/bloom/HoodieGlobalBloomIndex.java | 14 +-
.../bloom/ListBasedHoodieBloomIndexHelper.java | 26 +--
.../index/bloom/TestFlinkHoodieBloomIndex.java | 8 +-
.../org/apache/hudi/data/HoodieJavaPairRDD.java | 5 +
.../java/org/apache/hudi/data/HoodieJavaRDD.java | 14 +-
.../bloom/BucketizedBloomCheckPartitioner.java | 18 +-
.../bloom/HoodieBloomFilterProbingResult.java | 35 ++++
.../index/bloom/HoodieBloomIndexCheckFunction.java | 120 -----------
.../index/bloom/HoodieFileProbingFunction.java | 143 +++++++++++++
.../HoodieMetadataBloomFilterProbingFunction.java | 157 ++++++++++++++
.../HoodieMetadataBloomIndexCheckFunction.java | 154 --------------
.../index/bloom/SparkHoodieBloomIndexHelper.java | 229 ++++++++++++++++++---
.../org/apache/spark/sql/hudi/SparkAdapter.scala | 4 +-
.../bloom/TestBucketizedBloomCheckPartitioner.java | 43 ++--
.../hudi/index/bloom/TestHoodieBloomIndex.java | 10 +-
.../index/bloom/TestHoodieGlobalBloomIndex.java | 10 +-
.../org/apache/hudi/common/data/HoodieData.java | 17 +-
.../apache/hudi/common/data/HoodieListData.java | 10 +
.../hudi/common/data/HoodieListPairData.java | 23 +++
.../apache/hudi/common/data/HoodiePairData.java | 11 +-
.../ThrowingConsumer.java} | 31 +--
.../hudi/common/model/HoodieAvroIndexedRecord.java | 2 +-
.../hudi/common/table/HoodieTableMetaClient.java | 4 +
.../common/util/collection/FlatteningIterator.java | 55 +++++
.../common/util/collection/MappingIterator.java | 4 +-
.../apache/hudi/metadata/BaseTableMetadata.java | 53 +++--
.../hudi/metadata/HoodieBackedTableMetadata.java | 34 ++-
.../metadata/HoodieMetadataFileSystemView.java | 5 +-
.../main/java/org/apache/hudi/util/Transient.java | 108 ++++++++++
.../hudi/common/util/collection/TestIterators.java | 48 +++++
.../apache/spark/sql/adapter/Spark2Adapter.scala | 4 +-
35 files changed, 1038 insertions(+), 507 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
index 8a8a03e1b17..2ec46f19a43 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
@@ -62,9 +62,8 @@ public class HoodieIndexUtils {
* @param hoodieTable Instance of {@link HoodieTable} of interest
* @return the list of {@link HoodieBaseFile}
*/
- public static List<HoodieBaseFile> getLatestBaseFilesForPartition(
- final String partition,
- final HoodieTable hoodieTable) {
+ public static List<HoodieBaseFile> getLatestBaseFilesForPartition(String
partition,
+
HoodieTable hoodieTable) {
Option<HoodieInstant> latestCommitTime =
hoodieTable.getMetaClient().getCommitsTimeline()
.filterCompletedInstants().lastInstant();
if (latestCommitTime.isPresent()) {
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/BaseHoodieBloomIndexHelper.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/BaseHoodieBloomIndexHelper.java
index 9430d9bb5e5..f144540ed22 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/BaseHoodieBloomIndexHelper.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/BaseHoodieBloomIndexHelper.java
@@ -19,12 +19,11 @@
package org.apache.hudi.index.bloom;
-import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
+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.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.table.HoodieTable;
@@ -51,7 +50,7 @@ public abstract class BaseHoodieBloomIndexHelper implements
Serializable {
public abstract HoodiePairData<HoodieKey, HoodieRecordLocation>
findMatchingFilesForRecordKeys(
HoodieWriteConfig config, HoodieEngineContext context, HoodieTable
hoodieTable,
HoodiePairData<String, String> partitionRecordKeyPairs,
- HoodieData<Pair<String, HoodieKey>> fileComparisonPairs,
+ HoodiePairData<HoodieFileGroupId, String> fileComparisonPairs,
Map<String, List<BloomIndexFileInfo>> partitionToFileInfo,
Map<String, Long> recordsPerPartition);
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java
index 57d9def9b42..cf067c3a991 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndex.java
@@ -27,6 +27,7 @@ import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.fs.FSUtils;
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.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordLocation;
@@ -35,7 +36,6 @@ import org.apache.hudi.common.util.collection.ImmutablePair;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieIndexConfig;
import org.apache.hudi.config.HoodieWriteConfig;
-import org.apache.hudi.exception.HoodieMetadataException;
import org.apache.hudi.exception.MetadataNotFoundException;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.index.HoodieIndexUtils;
@@ -45,9 +45,9 @@ import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.stream.Collectors;
import java.util.stream.Stream;
import static java.util.stream.Collectors.groupingBy;
@@ -127,7 +127,7 @@ public class HoodieBloomIndex extends HoodieIndex<Object,
Object> {
// Step 3: Obtain a HoodieData, for each incoming record, that already
exists, with the file id,
// that contains it.
- HoodieData<Pair<String, HoodieKey>> fileComparisonPairs =
+ HoodiePairData<HoodieFileGroupId, String> fileComparisonPairs =
explodeRecordsWithFileComparisons(partitionToFileInfo,
partitionRecordKeyPairs);
return bloomIndexHelper.findMatchingFilesForRecordKeys(config, context,
hoodieTable,
@@ -210,34 +210,35 @@ public class HoodieBloomIndex extends HoodieIndex<Object,
Object> {
// also obtain file ranges, if range pruning is enabled
context.setJobStatus(this.getClass().getName(), "Load meta index key
ranges for file slices: " + config.getTableName());
- final String keyField =
hoodieTable.getMetaClient().getTableConfig().getRecordKeyFieldProp();
- return context.flatMap(partitions, partitionName -> {
- // Partition and file name pairs
- List<Pair<String, String>> partitionFileNameList =
HoodieIndexUtils.getLatestBaseFilesForPartition(partitionName,
- hoodieTable).stream().map(baseFile -> Pair.of(partitionName,
baseFile.getFileName()))
- .sorted()
- .collect(toList());
- if (partitionFileNameList.isEmpty()) {
- return Stream.empty();
- }
- try {
- Map<Pair<String, String>, HoodieMetadataColumnStats>
fileToColumnStatsMap =
-
hoodieTable.getMetadataTable().getColumnStats(partitionFileNameList, keyField);
- List<Pair<String, BloomIndexFileInfo>> result = new ArrayList<>();
- for (Map.Entry<Pair<String, String>, HoodieMetadataColumnStats> entry
: fileToColumnStatsMap.entrySet()) {
- result.add(Pair.of(entry.getKey().getLeft(),
- new BloomIndexFileInfo(
- FSUtils.getFileId(entry.getKey().getRight()),
- // NOTE: Here we assume that the type of the primary key
field is string
- (String)
unwrapStatisticValueWrapper(entry.getValue().getMinValue()),
- (String)
unwrapStatisticValueWrapper(entry.getValue().getMaxValue())
- )));
- }
- return result.stream();
- } catch (MetadataNotFoundException me) {
- throw new HoodieMetadataException("Unable to find column range
metadata for partition:" + partitionName, me);
- }
- }, Math.max(partitions.size(), 1));
+ String keyField =
hoodieTable.getMetaClient().getTableConfig().getRecordKeyFieldProp();
+
+ // Partition and file name pairs
+ List<Pair<String, String>> partitionFileNameList =
+ HoodieIndexUtils.getLatestBaseFilesForAllPartitions(partitions,
context, hoodieTable).stream()
+ .map(partitionBaseFilePair ->
Pair.of(partitionBaseFilePair.getLeft(),
partitionBaseFilePair.getRight().getFileName()))
+ .sorted()
+ .collect(toList());
+
+ if (partitionFileNameList.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ Map<Pair<String, String>, HoodieMetadataColumnStats> fileToColumnStatsMap =
+ hoodieTable.getMetadataTable().getColumnStats(partitionFileNameList,
keyField);
+
+ List<Pair<String, BloomIndexFileInfo>> result = new
ArrayList<>(fileToColumnStatsMap.size());
+
+ for (Map.Entry<Pair<String, String>, HoodieMetadataColumnStats> entry :
fileToColumnStatsMap.entrySet()) {
+ result.add(Pair.of(entry.getKey().getLeft(),
+ new BloomIndexFileInfo(
+ FSUtils.getFileId(entry.getKey().getRight()),
+ // NOTE: Here we assume that the type of the primary key field
is string
+ (String)
unwrapStatisticValueWrapper(entry.getValue().getMinValue()),
+ (String)
unwrapStatisticValueWrapper(entry.getValue().getMaxValue())
+ )));
+ }
+
+ return result;
}
@Override
@@ -278,7 +279,7 @@ public class HoodieBloomIndex extends HoodieIndex<Object,
Object> {
* Sub-partition to ensure the records can be looked up against files & also
prune file<=>record comparisons based on
* recordKey ranges in the index info.
*/
- HoodieData<Pair<String, HoodieKey>> explodeRecordsWithFileComparisons(
+ HoodiePairData<HoodieFileGroupId, String> explodeRecordsWithFileComparisons(
final Map<String, List<BloomIndexFileInfo>> partitionToFileIndexInfo,
HoodiePairData<String, String> partitionRecordKeyPairs) {
IndexFileFilter indexFileFilter =
@@ -289,11 +290,13 @@ public class HoodieBloomIndex extends HoodieIndex<Object,
Object> {
String recordKey = partitionRecordKeyPair.getRight();
String partitionPath = partitionRecordKeyPair.getLeft();
- return indexFileFilter.getMatchingFilesAndPartition(partitionPath,
recordKey).stream()
- .map(partitionFileIdPair -> (Pair<String, HoodieKey>) new
ImmutablePair<>(partitionFileIdPair.getRight(),
- new HoodieKey(recordKey, partitionPath)))
- .collect(Collectors.toList());
- }).flatMap(List::iterator);
+ return indexFileFilter.getMatchingFilesAndPartition(partitionPath,
recordKey)
+ .stream()
+ .map(partitionFileIdPair ->
+ new ImmutablePair<>(
+ new HoodieFileGroupId(partitionFileIdPair.getLeft(),
partitionFileIdPair.getRight()), recordKey));
+ })
+ .flatMapToPair(Stream::iterator);
}
/**
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBaseBloomIndexCheckFunction.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndexCheckFunction.java
similarity index 61%
rename from
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBaseBloomIndexCheckFunction.java
rename to
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndexCheckFunction.java
index 80031f4e8f0..52b504e9ab1 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBaseBloomIndexCheckFunction.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndexCheckFunction.java
@@ -19,7 +19,8 @@
package org.apache.hudi.index.bloom;
import org.apache.hudi.client.utils.LazyIterableIterator;
-import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.function.SerializableFunction;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieException;
@@ -28,53 +29,67 @@ import org.apache.hudi.io.HoodieKeyLookupHandle;
import org.apache.hudi.io.HoodieKeyLookupResult;
import org.apache.hudi.table.HoodieTable;
-import java.util.function.Function;
+import java.io.Serializable;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
+import java.util.function.Function;
/**
- * Function performing actual checking of list containing (fileId, hoodieKeys)
against the actual files.
+ * Function accepting a tuple of {@link HoodieFileGroupId} and a record key
and producing
+ * a list of {@link HoodieKeyLookupResult} for every file identified by the
file-group ids
+ *
+ * @param <I> type of the tuple of {@code (HoodieFileGroupId, <record-key>)}.
Note that this is
+ * parameterized as generic such that this code could be reused for
Spark as well
*/
-public class HoodieBaseBloomIndexCheckFunction
- implements Function<Iterator<Pair<String, HoodieKey>>,
Iterator<List<HoodieKeyLookupResult>>> {
+public class HoodieBloomIndexCheckFunction<I>
+ implements Function<Iterator<I>, Iterator<List<HoodieKeyLookupResult>>>,
Serializable {
private final HoodieTable hoodieTable;
private final HoodieWriteConfig config;
- public HoodieBaseBloomIndexCheckFunction(HoodieTable hoodieTable,
HoodieWriteConfig config) {
+ private final SerializableFunction<I, HoodieFileGroupId>
fileGroupIdExtractor;
+ private final SerializableFunction<I, String> recordKeyExtractor;
+
+ public HoodieBloomIndexCheckFunction(HoodieTable hoodieTable,
+ HoodieWriteConfig config,
+ SerializableFunction<I,
HoodieFileGroupId> fileGroupIdExtractor,
+ SerializableFunction<I, String>
recordKeyExtractor) {
this.hoodieTable = hoodieTable;
this.config = config;
+ this.fileGroupIdExtractor = fileGroupIdExtractor;
+ this.recordKeyExtractor = recordKeyExtractor;
}
@Override
- public Iterator<List<HoodieKeyLookupResult>> apply(Iterator<Pair<String,
HoodieKey>> filePartitionRecordKeyTripletItr) {
- return new LazyKeyCheckIterator(filePartitionRecordKeyTripletItr);
+ public Iterator<List<HoodieKeyLookupResult>> apply(Iterator<I>
fileGroupIdRecordKeyPairIterator) {
+ return new LazyKeyCheckIterator(fileGroupIdRecordKeyPairIterator);
}
- class LazyKeyCheckIterator extends LazyIterableIterator<Pair<String,
HoodieKey>, List<HoodieKeyLookupResult>> {
+ protected class LazyKeyCheckIterator extends LazyIterableIterator<I,
List<HoodieKeyLookupResult>> {
private HoodieKeyLookupHandle keyLookupHandle;
- LazyKeyCheckIterator(Iterator<Pair<String, HoodieKey>>
filePartitionRecordKeyTripletItr) {
+ LazyKeyCheckIterator(Iterator<I> filePartitionRecordKeyTripletItr) {
super(filePartitionRecordKeyTripletItr);
}
- @Override
- protected void start() {
- }
-
@Override
protected List<HoodieKeyLookupResult> computeNext() {
+
List<HoodieKeyLookupResult> ret = new ArrayList<>();
try {
// process one file in each go.
while (inputItr.hasNext()) {
- Pair<String, HoodieKey> currentTuple = inputItr.next();
- String fileId = currentTuple.getLeft();
- String partitionPath = currentTuple.getRight().getPartitionPath();
- String recordKey = currentTuple.getRight().getRecordKey();
+ I tuple = inputItr.next();
+
+ HoodieFileGroupId fileGroupId = fileGroupIdExtractor.apply(tuple);
+ String recordKey = recordKeyExtractor.apply(tuple);
+
+ String fileId = fileGroupId.getFileId();
+ String partitionPath = fileGroupId.getPartitionPath();
+
Pair<String, String> partitionPathFilePair = Pair.of(partitionPath,
fileId);
// lazily init state
@@ -100,15 +115,13 @@ public class HoodieBaseBloomIndexCheckFunction
}
} catch (Throwable e) {
if (e instanceof HoodieException) {
- throw e;
+ throw (HoodieException) e;
}
+
throw new HoodieIndexException("Error checking bloom filter index. ",
e);
}
- return ret;
- }
- @Override
- protected void end() {
+ return ret;
}
}
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieGlobalBloomIndex.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieGlobalBloomIndex.java
index 5f2007ea536..b5604312d3f 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieGlobalBloomIndex.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieGlobalBloomIndex.java
@@ -25,6 +25,7 @@ import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.EmptyHoodieRecordPayload;
import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordLocation;
@@ -41,7 +42,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.stream.Collectors;
+import java.util.stream.Stream;
/**
* This filter will only work with hoodie table since it will only load
partitions
@@ -74,7 +75,7 @@ public class HoodieGlobalBloomIndex extends HoodieBloomIndex {
*/
@Override
- HoodieData<Pair<String, HoodieKey>> explodeRecordsWithFileComparisons(
+ HoodiePairData<HoodieFileGroupId, String> explodeRecordsWithFileComparisons(
final Map<String, List<BloomIndexFileInfo>> partitionToFileIndexInfo,
HoodiePairData<String, String> partitionRecordKeyPairs) {
@@ -87,10 +88,11 @@ public class HoodieGlobalBloomIndex extends
HoodieBloomIndex {
String partitionPath = partitionRecordKeyPair.getLeft();
return indexFileFilter.getMatchingFilesAndPartition(partitionPath,
recordKey).stream()
- .map(partitionFileIdPair -> (Pair<String, HoodieKey>) new
ImmutablePair<>(partitionFileIdPair.getRight(),
- new HoodieKey(recordKey, partitionFileIdPair.getLeft())))
- .collect(Collectors.toList());
- }).flatMap(List::iterator);
+ .map(partitionFileIdPair ->
+ new ImmutablePair<>(
+ new HoodieFileGroupId(partitionFileIdPair.getLeft(),
partitionFileIdPair.getRight()), recordKey));
+ })
+ .flatMapToPair(Stream::iterator);
}
/**
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/ListBasedHoodieBloomIndexHelper.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/ListBasedHoodieBloomIndexHelper.java
index cffee5ee740..b47f5cf066c 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/ListBasedHoodieBloomIndexHelper.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/ListBasedHoodieBloomIndexHelper.java
@@ -19,20 +19,20 @@
package org.apache.hudi.index.bloom;
-import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
+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.collection.ImmutablePair;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.io.HoodieKeyLookupResult;
import org.apache.hudi.table.HoodieTable;
-import java.util.ArrayList;
+import java.util.Collection;
import java.util.Comparator;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -56,21 +56,21 @@ public class ListBasedHoodieBloomIndexHelper extends
BaseHoodieBloomIndexHelper
public HoodiePairData<HoodieKey, HoodieRecordLocation>
findMatchingFilesForRecordKeys(
HoodieWriteConfig config, HoodieEngineContext context, HoodieTable
hoodieTable,
HoodiePairData<String, String> partitionRecordKeyPairs,
- HoodieData<Pair<String, HoodieKey>> fileComparisonPairs,
+ HoodiePairData<HoodieFileGroupId, String> fileComparisonPairs,
Map<String, List<BloomIndexFileInfo>> partitionToFileInfo, Map<String,
Long> recordsPerPartition) {
- List<Pair<String, HoodieKey>> fileComparisonPairList =
+ List<Pair<HoodieFileGroupId, String>> fileComparisonPairList =
fileComparisonPairs.collectAsList().stream()
.sorted(Comparator.comparing(Pair::getLeft)).collect(toList());
- List<HoodieKeyLookupResult> keyLookupResults = new ArrayList<>();
- Iterator<List<HoodieKeyLookupResult>> iterator = new
HoodieBaseBloomIndexCheckFunction(
- hoodieTable, config).apply(fileComparisonPairList.iterator());
- while (iterator.hasNext()) {
- keyLookupResults.addAll(iterator.next());
- }
+ List<HoodieKeyLookupResult> keyLookupResults =
+ CollectionUtils.toStream(
+ new HoodieBloomIndexCheckFunction<Pair<HoodieFileGroupId,
String>>(hoodieTable, config, Pair::getLeft, Pair::getRight)
+ .apply(fileComparisonPairList.iterator())
+ )
+ .flatMap(Collection::stream)
+ .filter(lr -> lr.getMatchingRecordKeys().size() > 0)
+ .collect(toList());
- keyLookupResults = keyLookupResults.stream().filter(
- lr -> lr.getMatchingRecordKeys().size() > 0).collect(toList());
return context.parallelize(keyLookupResults).flatMap(lookupResult ->
lookupResult.getMatchingRecordKeys().stream()
.map(recordKey -> new ImmutablePair<>(lookupResult,
recordKey)).iterator()
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/index/bloom/TestFlinkHoodieBloomIndex.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/index/bloom/TestFlinkHoodieBloomIndex.java
index 6dd5a1c27e2..d4b4007bedb 100644
---
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/index/bloom/TestFlinkHoodieBloomIndex.java
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/index/bloom/TestFlinkHoodieBloomIndex.java
@@ -24,6 +24,7 @@ import org.apache.hudi.common.bloom.BloomFilterTypeCode;
import org.apache.hudi.common.data.HoodieListPairData;
import org.apache.hudi.common.engine.EngineType;
import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -186,11 +187,14 @@ public class TestFlinkHoodieBloomIndex extends
HoodieFlinkClientTestHarness {
partitionRecordKeyMap.put(t.getLeft(), recordKeyList);
});
- List<Pair<String, HoodieKey>> comparisonKeyList =
index.explodeRecordsWithFileComparisons(partitionToFileIndexInfo,
HoodieListPairData.lazy(partitionRecordKeyMap)).collectAsList();
+ List<Pair<HoodieFileGroupId, String>> comparisonKeyList =
+ index.explodeRecordsWithFileComparisons(partitionToFileIndexInfo,
HoodieListPairData.lazy(partitionRecordKeyMap)).collectAsList();
assertEquals(10, comparisonKeyList.size());
java.util.Map<String, List<String>> recordKeyToFileComps =
comparisonKeyList.stream()
- .collect(java.util.stream.Collectors.groupingBy(t ->
t.getRight().getRecordKey(), java.util.stream.Collectors.mapping(t ->
t.getLeft(), java.util.stream.Collectors.toList())));
+ .collect(
+ java.util.stream.Collectors.groupingBy(t -> t.getRight(),
+ java.util.stream.Collectors.mapping(t ->
t.getLeft().getFileId(), java.util.stream.Collectors.toList())));
assertEquals(4, recordKeyToFileComps.size());
assertEquals(new java.util.HashSet<>(asList("f1", "f3", "f4")), new
java.util.HashSet<>(recordKeyToFileComps.get("002")));
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/data/HoodieJavaPairRDD.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/data/HoodieJavaPairRDD.java
index 9ec3c4cf715..9019fb43ff0 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/data/HoodieJavaPairRDD.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/data/HoodieJavaPairRDD.java
@@ -121,6 +121,11 @@ public class HoodieJavaPairRDD<K, V> implements
HoodiePairData<K, V> {
tuple -> func.apply(new ImmutablePair<>(tuple._1, tuple._2))));
}
+ @Override
+ public <W> HoodiePairData<K, W> mapValues(SerializableFunction<V, W> func) {
+ return HoodieJavaPairRDD.of(pairRDDData.mapValues(func::apply));
+ }
+
@Override
public <L, W> HoodiePairData<L, W>
mapToPair(SerializablePairFunction<Pair<K, V>, L, W> mapToPairFunc) {
return HoodieJavaPairRDD.of(pairRDDData.mapToPair(pair -> {
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/data/HoodieJavaRDD.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/data/HoodieJavaRDD.java
index ed9613bc15f..6ed3a854962 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/data/HoodieJavaRDD.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/data/HoodieJavaRDD.java
@@ -24,17 +24,16 @@ import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.function.SerializableFunction;
import org.apache.hudi.common.function.SerializablePairFunction;
+import org.apache.hudi.common.util.collection.MappingIterator;
import org.apache.hudi.common.util.collection.Pair;
-
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.storage.StorageLevel;
+import scala.Tuple2;
import java.util.Iterator;
import java.util.List;
-import scala.Tuple2;
-
/**
* Holds a {@link JavaRDD} of objects.
*
@@ -119,9 +118,18 @@ public class HoodieJavaRDD<T> implements HoodieData<T> {
@Override
public <O> HoodieData<O> flatMap(SerializableFunction<T, Iterator<O>> func) {
+ // NOTE: Unrolling this lambda into a method reference results in
[[ClassCastException]]
+ // due to weird interop b/w Scala and Java
return HoodieJavaRDD.of(rddData.flatMap(e -> func.apply(e)));
}
+ @Override
+ public <K, V> HoodiePairData<K, V> flatMapToPair(SerializableFunction<T,
Iterator<? extends Pair<K, V>>> func) {
+ return HoodieJavaPairRDD.of(
+ rddData.flatMapToPair(e ->
+ new MappingIterator<>(func.apply(e), p -> new Tuple2<>(p.getKey(),
p.getValue()))));
+ }
+
@Override
public <K, V> HoodiePairData<K, V> mapToPair(SerializablePairFunction<T, K,
V> func) {
return HoodieJavaPairRDD.of(rddData.mapToPair(input -> {
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/BucketizedBloomCheckPartitioner.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/BucketizedBloomCheckPartitioner.java
index 36710dc02bb..48099220dbf 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/BucketizedBloomCheckPartitioner.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/BucketizedBloomCheckPartitioner.java
@@ -18,6 +18,7 @@
package org.apache.hudi.index.bloom;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.util.NumericUtils;
import org.apache.hudi.common.util.collection.Pair;
@@ -63,7 +64,7 @@ public class BucketizedBloomCheckPartitioner extends
Partitioner {
/**
* Stores the final mapping of a file group to a list of partitions for its
keys.
*/
- private Map<String, List<Integer>> fileGroupToPartitions;
+ private Map<HoodieFileGroupId, List<Integer>> fileGroupToPartitions;
/**
* Create a partitioner that computes a plan based on provided workload
characteristics.
@@ -72,11 +73,11 @@ public class BucketizedBloomCheckPartitioner extends
Partitioner {
* @param fileGroupToComparisons number of expected comparisons per file
group
* @param keysPerBucket maximum number of keys to pack in a single bucket
*/
- public BucketizedBloomCheckPartitioner(int targetPartitions, Map<String,
Long> fileGroupToComparisons,
+ public BucketizedBloomCheckPartitioner(int targetPartitions,
Map<HoodieFileGroupId, Long> fileGroupToComparisons,
int keysPerBucket) {
this.fileGroupToPartitions = new HashMap<>();
- Map<String, Integer> bucketsPerFileGroup = new HashMap<>();
+ Map<HoodieFileGroupId, Integer> bucketsPerFileGroup = new HashMap<>();
// Compute the buckets needed per file group, using simple uniform
distribution
fileGroupToComparisons.forEach((f, c) -> bucketsPerFileGroup.put(f, (int)
Math.ceil((c * 1.0) / keysPerBucket)));
int totalBuckets = bucketsPerFileGroup.values().stream().mapToInt(i ->
i).sum();
@@ -90,9 +91,9 @@ public class BucketizedBloomCheckPartitioner extends
Partitioner {
int minBucketsPerPartition = Math.max((int) Math.floor((1.0 *
totalBuckets) / partitions), 1);
LOG.info(String.format("TotalBuckets %d, min_buckets/partition %d",
totalBuckets, minBucketsPerPartition));
int[] bucketsFilled = new int[partitions];
- Map<String, AtomicInteger> bucketsFilledPerFileGroup = new HashMap<>();
+ Map<HoodieFileGroupId, AtomicInteger> bucketsFilledPerFileGroup = new
HashMap<>();
int partitionIndex = 0;
- for (Map.Entry<String, Integer> e : bucketsPerFileGroup.entrySet()) {
+ for (Map.Entry<HoodieFileGroupId, Integer> e :
bucketsPerFileGroup.entrySet()) {
for (int b = 0; b < Math.max(1, e.getValue() - 1); b++) {
// keep filled counts upto date
bucketsFilled[partitionIndex]++;
@@ -115,7 +116,7 @@ public class BucketizedBloomCheckPartitioner extends
Partitioner {
// PHASE 2 : for remaining unassigned buckets, round robin over partitions
once. Since we withheld 1 bucket from
// each file group uniformly, this remaining is also an uniform mix across
file groups. We just round robin to
// optimize for goal 2.
- for (Map.Entry<String, Integer> e : bucketsPerFileGroup.entrySet()) {
+ for (Map.Entry<HoodieFileGroupId, Integer> e :
bucketsPerFileGroup.entrySet()) {
int remaining = e.getValue() -
bucketsFilledPerFileGroup.get(e.getKey()).intValue();
for (int r = 0; r < remaining; r++) {
// mark this partition against the file group
@@ -142,7 +143,8 @@ public class BucketizedBloomCheckPartitioner extends
Partitioner {
@Override
public int getPartition(Object key) {
- final Pair<String, String> parts = (Pair<String, String>) key;
+ final Pair<HoodieFileGroupId, String> parts = (Pair<HoodieFileGroupId,
String>) key;
+ // TODO replace w/ more performant hash
final long hashOfKey = NumericUtils.getMessageDigestHash("MD5",
parts.getRight());
final List<Integer> candidatePartitions =
fileGroupToPartitions.get(parts.getLeft());
final int idx = (int) Math.floorMod((int) hashOfKey,
candidatePartitions.size());
@@ -150,7 +152,7 @@ public class BucketizedBloomCheckPartitioner extends
Partitioner {
return candidatePartitions.get(idx);
}
- Map<String, List<Integer>> getFileGroupToPartitions() {
+ Map<HoodieFileGroupId, List<Integer>> getFileGroupToPartitions() {
return fileGroupToPartitions;
}
}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieBloomFilterProbingResult.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieBloomFilterProbingResult.java
new file mode 100644
index 00000000000..c124f8b27b8
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieBloomFilterProbingResult.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.index.bloom;
+
+import java.util.List;
+
+class HoodieBloomFilterProbingResult {
+
+ private final List<String> candidateKeys;
+
+ HoodieBloomFilterProbingResult(List<String> candidateKeys) {
+ this.candidateKeys = candidateKeys;
+ }
+
+ public List<String> getCandidateKeys() {
+ return candidateKeys;
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndexCheckFunction.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndexCheckFunction.java
deleted file mode 100644
index e19a429ea72..00000000000
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieBloomIndexCheckFunction.java
+++ /dev/null
@@ -1,120 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.hudi.index.bloom;
-
-import org.apache.hudi.client.utils.LazyIterableIterator;
-import org.apache.hudi.common.model.HoodieKey;
-import org.apache.hudi.common.util.collection.Pair;
-import org.apache.hudi.config.HoodieWriteConfig;
-import org.apache.hudi.exception.HoodieException;
-import org.apache.hudi.exception.HoodieIndexException;
-import org.apache.hudi.io.HoodieKeyLookupHandle;
-import org.apache.hudi.io.HoodieKeyLookupResult;
-import org.apache.hudi.table.HoodieTable;
-
-import org.apache.spark.api.java.function.Function2;
-
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-
-import scala.Tuple2;
-
-/**
- * Function performing actual checking of RDD partition containing (fileId,
hoodieKeys) against the actual files.
- */
-public class HoodieBloomIndexCheckFunction
- implements Function2<Integer, Iterator<Tuple2<String, HoodieKey>>,
Iterator<List<HoodieKeyLookupResult>>> {
-
- private final HoodieTable hoodieTable;
-
- private final HoodieWriteConfig config;
-
- public HoodieBloomIndexCheckFunction(HoodieTable hoodieTable,
HoodieWriteConfig config) {
- this.hoodieTable = hoodieTable;
- this.config = config;
- }
-
- @Override
- public Iterator<List<HoodieKeyLookupResult>> call(Integer partition,
- Iterator<Tuple2<String,
HoodieKey>> filePartitionRecordKeyTripletItr) {
- return new LazyKeyCheckIterator(filePartitionRecordKeyTripletItr);
- }
-
- class LazyKeyCheckIterator extends LazyIterableIterator<Tuple2<String,
HoodieKey>, List<HoodieKeyLookupResult>> {
-
- private HoodieKeyLookupHandle keyLookupHandle;
-
- LazyKeyCheckIterator(Iterator<Tuple2<String, HoodieKey>>
filePartitionRecordKeyTripletItr) {
- super(filePartitionRecordKeyTripletItr);
- }
-
- @Override
- protected void start() {
- }
-
- @Override
- protected List<HoodieKeyLookupResult> computeNext() {
-
- List<HoodieKeyLookupResult> ret = new ArrayList<>();
- try {
- // process one file in each go.
- while (inputItr.hasNext()) {
- Tuple2<String, HoodieKey> currentTuple = inputItr.next();
- String fileId = currentTuple._1;
- String partitionPath = currentTuple._2.getPartitionPath();
- String recordKey = currentTuple._2.getRecordKey();
- Pair<String, String> partitionPathFilePair = Pair.of(partitionPath,
fileId);
-
- // lazily init state
- if (keyLookupHandle == null) {
- keyLookupHandle = new HoodieKeyLookupHandle(config, hoodieTable,
partitionPathFilePair);
- }
-
- // if continue on current file
- if
(keyLookupHandle.getPartitionPathFileIDPair().equals(partitionPathFilePair)) {
- keyLookupHandle.addKey(recordKey);
- } else {
- // do the actual checking of file & break out
- ret.add(keyLookupHandle.getLookupResult());
- keyLookupHandle = new HoodieKeyLookupHandle(config, hoodieTable,
partitionPathFilePair);
- keyLookupHandle.addKey(recordKey);
- break;
- }
- }
-
- // handle case, where we ran out of input, close pending work, update
return val
- if (!inputItr.hasNext()) {
- ret.add(keyLookupHandle.getLookupResult());
- }
- } catch (Throwable e) {
- if (e instanceof HoodieException) {
- throw e;
- }
- throw new HoodieIndexException("Error checking bloom filter index. ",
e);
- }
-
- return ret;
- }
-
- @Override
- protected void end() {
- }
- }
-}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieFileProbingFunction.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieFileProbingFunction.java
new file mode 100644
index 00000000000..0809042c9fb
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieFileProbingFunction.java
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.index.bloom;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.client.utils.LazyIterableIterator;
+import org.apache.hudi.common.config.SerializableConfiguration;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieIndexException;
+import org.apache.hudi.index.HoodieIndexUtils;
+import org.apache.hudi.io.HoodieKeyLookupResult;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.apache.spark.api.java.function.FlatMapFunction;
+import org.apache.spark.broadcast.Broadcast;
+import scala.Tuple2;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Implementation of the function probing filtered in candidate keys provided
in
+ * {@link HoodieBloomFilterProbingResult} w/in corresponding files identified
by {@link HoodieFileGroupId}
+ * to validate whether the record w/ the provided key is indeed persisted in it
+ */
+public class HoodieFileProbingFunction implements
+ FlatMapFunction<Iterator<Tuple2<HoodieFileGroupId,
HoodieBloomFilterProbingResult>>, List<HoodieKeyLookupResult>> {
+
+ private static final Logger LOG =
LogManager.getLogger(HoodieFileProbingFunction.class);
+
+ // Assuming each file bloom filter takes up 512K, sizing the max file count
+ // per batch so that the total fetched bloom filters would not cross 128 MB.
+ private static final long BLOOM_FILTER_CHECK_MAX_FILE_COUNT_PER_BATCH = 256;
+
+ private final Broadcast<HoodieTableFileSystemView> baseFileOnlyViewBroadcast;
+ private final SerializableConfiguration hadoopConf;
+
+ public HoodieFileProbingFunction(Broadcast<HoodieTableFileSystemView>
baseFileOnlyViewBroadcast,
+ SerializableConfiguration hadoopConf) {
+ this.baseFileOnlyViewBroadcast = baseFileOnlyViewBroadcast;
+ this.hadoopConf = hadoopConf;
+ }
+
+ @Override
+ public Iterator<List<HoodieKeyLookupResult>>
call(Iterator<Tuple2<HoodieFileGroupId, HoodieBloomFilterProbingResult>>
tuple2Iterator) throws Exception {
+ return new BloomIndexLazyKeyCheckIterator(tuple2Iterator);
+ }
+
+ private class BloomIndexLazyKeyCheckIterator
+ extends LazyIterableIterator<Tuple2<HoodieFileGroupId,
HoodieBloomFilterProbingResult>, List<HoodieKeyLookupResult>> {
+
+ public BloomIndexLazyKeyCheckIterator(Iterator<Tuple2<HoodieFileGroupId,
HoodieBloomFilterProbingResult>> tuple2Iterator) {
+ super(tuple2Iterator);
+ }
+
+ @Override
+ protected List<HoodieKeyLookupResult> computeNext() {
+ // Partition path and file name pair to list of keys
+ final Map<Pair<String, String>, HoodieBloomFilterProbingResult>
fileToLookupResults = new HashMap<>();
+ final Map<String, HoodieBaseFile> fileIDBaseFileMap = new HashMap<>();
+
+ while (inputItr.hasNext()) {
+ Tuple2<HoodieFileGroupId, HoodieBloomFilterProbingResult> entry =
inputItr.next();
+ final String partitionPath = entry._1.getPartitionPath();
+ final String fileId = entry._1.getFileId();
+
+ if (!fileIDBaseFileMap.containsKey(fileId)) {
+ Option<HoodieBaseFile> baseFile =
+
baseFileOnlyViewBroadcast.getValue().getLatestBaseFile(partitionPath, fileId);
+ if (!baseFile.isPresent()) {
+ throw new HoodieIndexException("Failed to find the base file for
partition: " + partitionPath
+ + ", fileId: " + fileId);
+ }
+
+ fileIDBaseFileMap.put(fileId, baseFile.get());
+ }
+
+ fileToLookupResults.putIfAbsent(Pair.of(partitionPath,
fileIDBaseFileMap.get(fileId).getFileName()), entry._2);
+
+ if (fileToLookupResults.size() >
BLOOM_FILTER_CHECK_MAX_FILE_COUNT_PER_BATCH) {
+ break;
+ }
+ }
+
+ if (fileToLookupResults.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ return fileToLookupResults.entrySet().stream()
+ .map(entry -> {
+ Pair<String, String> partitionPathFileNamePair = entry.getKey();
+ HoodieBloomFilterProbingResult bloomFilterKeyLookupResult =
entry.getValue();
+
+ final String partitionPath = partitionPathFileNamePair.getLeft();
+ final String fileName = partitionPathFileNamePair.getRight();
+ final String fileId = FSUtils.getFileId(fileName);
+ ValidationUtils.checkState(!fileId.isEmpty());
+
+ List<String> candidateRecordKeys =
bloomFilterKeyLookupResult.getCandidateKeys();
+
+ // TODO add assertion that file is checked only once
+
+ final HoodieBaseFile dataFile = fileIDBaseFileMap.get(fileId);
+ List<String> matchingKeys =
HoodieIndexUtils.filterKeysFromFile(new Path(dataFile.getPath()),
+ candidateRecordKeys, hadoopConf.get());
+
+ LOG.debug(
+ String.format("Bloom filter candidates (%d) / false positives
(%d), actual matches (%d)",
+ candidateRecordKeys.size(), candidateRecordKeys.size() -
matchingKeys.size(), matchingKeys.size()));
+
+ return new HoodieKeyLookupResult(fileId, partitionPath,
dataFile.getCommitTime(), matchingKeys);
+ })
+ .collect(Collectors.toList());
+ }
+
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieMetadataBloomFilterProbingFunction.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieMetadataBloomFilterProbingFunction.java
new file mode 100644
index 00000000000..406be816500
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieMetadataBloomFilterProbingFunction.java
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.index.bloom;
+
+import org.apache.hudi.client.utils.LazyIterableIterator;
+import org.apache.hudi.common.bloom.BloomFilter;
+import org.apache.hudi.common.fs.FSUtils;
+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.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.FlatteningIterator;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieIndexException;
+import org.apache.hudi.metadata.HoodieBackedTableMetadata;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.apache.spark.api.java.function.PairFlatMapFunction;
+import org.apache.spark.broadcast.Broadcast;
+import scala.Tuple2;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Implementation of the function that probing Bloom Filters of individual
files verifying
+ * whether particular record key could be stored in the latest file-slice of
the file-group
+ * identified by the {@link HoodieFileGroupId}
+ */
+public class HoodieMetadataBloomFilterProbingFunction implements
+ PairFlatMapFunction<Iterator<Tuple2<HoodieFileGroupId, String>>,
HoodieFileGroupId, HoodieBloomFilterProbingResult> {
+
+ private static final Logger LOG =
LogManager.getLogger(HoodieMetadataBloomFilterProbingFunction.class);
+
+ // Assuming each file bloom filter takes up 512K, sizing the max file count
+ // per batch so that the total fetched bloom filters would not cross 128 MB.
+ private static final long BLOOM_FILTER_CHECK_MAX_FILE_COUNT_PER_BATCH = 256;
+ private final HoodieTable hoodieTable;
+
+ private final Broadcast<HoodieTableFileSystemView> baseFileOnlyViewBroadcast;
+
+ /**
+ * NOTE: It's critical for this ctor to accept {@link HoodieTable} to make
sure that it uses
+ * broadcast-ed instance of {@link HoodieBackedTableMetadata}
internally, instead of
+ * one being serialized and deserialized for _every_ task individually
+ *
+ * NOTE: We pass in broadcasted {@link HoodieTableFileSystemView} to make
sure it's materialized
+ * on executor once
+ */
+ public
HoodieMetadataBloomFilterProbingFunction(Broadcast<HoodieTableFileSystemView>
baseFileOnlyViewBroadcast,
+ HoodieTable hoodieTable) {
+ this.baseFileOnlyViewBroadcast = baseFileOnlyViewBroadcast;
+ this.hoodieTable = hoodieTable;
+ }
+
+ @Override
+ public Iterator<Tuple2<HoodieFileGroupId, HoodieBloomFilterProbingResult>>
call(Iterator<Tuple2<HoodieFileGroupId, String>> tuple2Iterator) throws
Exception {
+ return new FlatteningIterator<>(new
BloomIndexLazyKeyCheckIterator(tuple2Iterator));
+ }
+
+ private class BloomIndexLazyKeyCheckIterator
+ extends LazyIterableIterator<Tuple2<HoodieFileGroupId, String>,
Iterator<Tuple2<HoodieFileGroupId, HoodieBloomFilterProbingResult>>> {
+
+ public BloomIndexLazyKeyCheckIterator(Iterator<Tuple2<HoodieFileGroupId,
String>> tuple2Iterator) {
+ super(tuple2Iterator);
+ }
+
+ @Override
+ protected Iterator<Tuple2<HoodieFileGroupId,
HoodieBloomFilterProbingResult>> computeNext() {
+ // Partition path and file name pair to list of keys
+ final Map<Pair<String, String>, List<HoodieKey>> fileToKeysMap = new
HashMap<>();
+ final Map<String, HoodieBaseFile> fileIDBaseFileMap = new HashMap<>();
+
+ while (inputItr.hasNext()) {
+ Tuple2<HoodieFileGroupId, String> entry = inputItr.next();
+ String partitionPath = entry._1.getPartitionPath();
+ String fileId = entry._1.getFileId();
+
+ if (!fileIDBaseFileMap.containsKey(fileId)) {
+ Option<HoodieBaseFile> baseFile =
baseFileOnlyViewBroadcast.getValue().getLatestBaseFile(partitionPath, fileId);
+ if (!baseFile.isPresent()) {
+ throw new HoodieIndexException("Failed to find the base file for
partition: " + partitionPath
+ + ", fileId: " + fileId);
+ }
+ fileIDBaseFileMap.put(fileId, baseFile.get());
+ }
+
+ fileToKeysMap.computeIfAbsent(Pair.of(partitionPath,
fileIDBaseFileMap.get(fileId).getFileName()),
+ k -> new ArrayList<>()).add(new HoodieKey(entry._2,
partitionPath));
+
+ if (fileToKeysMap.size() >
BLOOM_FILTER_CHECK_MAX_FILE_COUNT_PER_BATCH) {
+ break;
+ }
+ }
+
+ if (fileToKeysMap.isEmpty()) {
+ return Collections.emptyIterator();
+ }
+
+ List<Pair<String, String>> partitionNameFileNameList = new
ArrayList<>(fileToKeysMap.keySet());
+ Map<Pair<String, String>, BloomFilter> fileToBloomFilterMap =
+
hoodieTable.getMetadataTable().getBloomFilters(partitionNameFileNameList);
+
+ return fileToKeysMap.entrySet().stream()
+ .map(entry -> {
+ Pair<String, String> partitionPathFileNamePair = entry.getKey();
+ List<HoodieKey> hoodieKeyList = entry.getValue();
+
+ final String partitionPath = partitionPathFileNamePair.getLeft();
+ final String fileName = partitionPathFileNamePair.getRight();
+ final String fileId = FSUtils.getFileId(fileName);
+ ValidationUtils.checkState(!fileId.isEmpty());
+
+ if (!fileToBloomFilterMap.containsKey(partitionPathFileNamePair)) {
+ throw new HoodieIndexException("Failed to get the bloom filter
for " + partitionPathFileNamePair);
+ }
+ final BloomFilter fileBloomFilter =
fileToBloomFilterMap.get(partitionPathFileNamePair);
+
+ List<String> candidateRecordKeys = new ArrayList<>();
+ hoodieKeyList.forEach(hoodieKey -> {
+ if (fileBloomFilter.mightContain(hoodieKey.getRecordKey())) {
+ candidateRecordKeys.add(hoodieKey.getRecordKey());
+ }
+ });
+
+ LOG.debug(String.format("Total records (%d), bloom filter
candidates (%d)",
+ hoodieKeyList.size(), candidateRecordKeys.size()));
+
+ return Tuple2.apply(new HoodieFileGroupId(partitionPath, fileId),
new HoodieBloomFilterProbingResult(candidateRecordKeys));
+ })
+ .iterator();
+ }
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieMetadataBloomIndexCheckFunction.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieMetadataBloomIndexCheckFunction.java
deleted file mode 100644
index 8a2958eab9d..00000000000
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieMetadataBloomIndexCheckFunction.java
+++ /dev/null
@@ -1,154 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.hudi.index.bloom;
-
-import org.apache.hadoop.fs.Path;
-import org.apache.hudi.client.utils.LazyIterableIterator;
-import org.apache.hudi.common.bloom.BloomFilter;
-import org.apache.hudi.common.fs.FSUtils;
-import org.apache.hudi.common.model.HoodieBaseFile;
-import org.apache.hudi.common.model.HoodieKey;
-import org.apache.hudi.common.util.Option;
-import org.apache.hudi.common.util.ValidationUtils;
-import org.apache.hudi.common.util.collection.Pair;
-import org.apache.hudi.exception.HoodieIndexException;
-import org.apache.hudi.index.HoodieIndexUtils;
-import org.apache.hudi.io.HoodieKeyLookupResult;
-import org.apache.hudi.table.HoodieTable;
-import org.apache.log4j.LogManager;
-import org.apache.log4j.Logger;
-import org.apache.spark.api.java.function.Function2;
-import scala.Tuple2;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
-
-/**
- * Spark Function2 implementation for checking bloom filters for the
- * requested keys from the metadata table index. The bloom filter
- * checking for keys and the actual file verification for the
- * candidate keys is done in an iterative fashion. In each iteration,
- * bloom filters are requested for a batch of partition files and the
- * keys are checked against them.
- */
-public class HoodieMetadataBloomIndexCheckFunction implements
- Function2<Integer, Iterator<Tuple2<String, HoodieKey>>,
Iterator<List<HoodieKeyLookupResult>>> {
-
- private static final Logger LOG =
LogManager.getLogger(HoodieMetadataBloomIndexCheckFunction.class);
-
- // Assuming each file bloom filter takes up 512K, sizing the max file count
- // per batch so that the total fetched bloom filters would not cross 128 MB.
- private static final long BLOOM_FILTER_CHECK_MAX_FILE_COUNT_PER_BATCH = 256;
- private final HoodieTable hoodieTable;
-
- public HoodieMetadataBloomIndexCheckFunction(HoodieTable hoodieTable) {
- this.hoodieTable = hoodieTable;
- }
-
- @Override
- public Iterator<List<HoodieKeyLookupResult>> call(Integer integer,
Iterator<Tuple2<String, HoodieKey>> tuple2Iterator) throws Exception {
- return new BloomIndexLazyKeyCheckIterator(tuple2Iterator);
- }
-
- private class BloomIndexLazyKeyCheckIterator extends
LazyIterableIterator<Tuple2<String, HoodieKey>, List<HoodieKeyLookupResult>> {
- public BloomIndexLazyKeyCheckIterator(Iterator<Tuple2<String, HoodieKey>>
tuple2Iterator) {
- super(tuple2Iterator);
- }
-
- @Override
- protected void start() {
- }
-
- @Override
- protected List<HoodieKeyLookupResult> computeNext() {
- // Partition path and file name pair to list of keys
- final Map<Pair<String, String>, List<HoodieKey>> fileToKeysMap = new
HashMap<>();
- final Map<String, HoodieBaseFile> fileIDBaseFileMap = new HashMap<>();
- final List<HoodieKeyLookupResult> resultList = new ArrayList<>();
-
- while (inputItr.hasNext()) {
- Tuple2<String, HoodieKey> entry = inputItr.next();
- final String partitionPath = entry._2.getPartitionPath();
- final String fileId = entry._1;
- if (!fileIDBaseFileMap.containsKey(fileId)) {
- Option<HoodieBaseFile> baseFile =
hoodieTable.getBaseFileOnlyView().getLatestBaseFile(partitionPath, fileId);
- if (!baseFile.isPresent()) {
- throw new HoodieIndexException("Failed to find the base file for
partition: " + partitionPath
- + ", fileId: " + fileId);
- }
- fileIDBaseFileMap.put(fileId, baseFile.get());
- }
- fileToKeysMap.computeIfAbsent(Pair.of(partitionPath,
fileIDBaseFileMap.get(fileId).getFileName()),
- k -> new ArrayList<>()).add(entry._2);
- if (fileToKeysMap.size() >
BLOOM_FILTER_CHECK_MAX_FILE_COUNT_PER_BATCH) {
- break;
- }
- }
- if (fileToKeysMap.isEmpty()) {
- return Collections.emptyList();
- }
-
- List<Pair<String, String>> partitionNameFileNameList = new
ArrayList<>(fileToKeysMap.keySet());
- Map<Pair<String, String>, BloomFilter> fileToBloomFilterMap =
-
hoodieTable.getMetadataTable().getBloomFilters(partitionNameFileNameList);
-
- final AtomicInteger totalKeys = new AtomicInteger(0);
- fileToKeysMap.forEach((partitionPathFileNamePair, hoodieKeyList) -> {
- final String partitionPath = partitionPathFileNamePair.getLeft();
- final String fileName = partitionPathFileNamePair.getRight();
- final String fileId = FSUtils.getFileId(fileName);
- ValidationUtils.checkState(!fileId.isEmpty());
-
- if (!fileToBloomFilterMap.containsKey(partitionPathFileNamePair)) {
- throw new HoodieIndexException("Failed to get the bloom filter for "
+ partitionPathFileNamePair);
- }
- final BloomFilter fileBloomFilter =
fileToBloomFilterMap.get(partitionPathFileNamePair);
-
- List<String> candidateRecordKeys = new ArrayList<>();
- hoodieKeyList.forEach(hoodieKey -> {
- totalKeys.incrementAndGet();
- if (fileBloomFilter.mightContain(hoodieKey.getRecordKey())) {
- candidateRecordKeys.add(hoodieKey.getRecordKey());
- }
- });
-
- final HoodieBaseFile dataFile = fileIDBaseFileMap.get(fileId);
- List<String> matchingKeys =
- HoodieIndexUtils.filterKeysFromFile(new Path(dataFile.getPath()),
candidateRecordKeys,
- hoodieTable.getHadoopConf());
- LOG.debug(
- String.format("Total records (%d), bloom filter candidates
(%d)/fp(%d), actual matches (%d)",
- hoodieKeyList.size(), candidateRecordKeys.size(),
- candidateRecordKeys.size() - matchingKeys.size(),
matchingKeys.size()));
-
- resultList.add(new HoodieKeyLookupResult(fileId, partitionPath,
dataFile.getCommitTime(), matchingKeys));
- });
- return resultList;
- }
-
- @Override
- protected void end() {
- }
- }
-}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/SparkHoodieBloomIndexHelper.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/SparkHoodieBloomIndexHelper.java
index 5736024dc24..265b0507768 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/SparkHoodieBloomIndexHelper.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/SparkHoodieBloomIndexHelper.java
@@ -19,30 +19,47 @@
package org.apache.hudi.index.bloom;
-import org.apache.hudi.common.data.HoodieData;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hudi.client.common.HoodieSparkEngineContext;
+import org.apache.hudi.common.config.SerializableConfiguration;
import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.BaseFile;
+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.table.view.HoodieTableFileSystemView;
import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.common.util.hash.FileIndexID;
+import org.apache.hudi.common.util.hash.PartitionIndexID;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.data.HoodieJavaPairRDD;
import org.apache.hudi.data.HoodieJavaRDD;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.io.HoodieKeyLookupResult;
import org.apache.hudi.table.HoodieTable;
-
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.apache.spark.Partitioner;
+import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.function.FlatMapFunction;
+import org.apache.spark.broadcast.Broadcast;
+import scala.Tuple2;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
-import scala.Tuple2;
-
+import static
org.apache.hudi.metadata.HoodieMetadataPayload.getBloomFilterIndexKey;
+import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.mapRecordKeyToFileGroupIndex;
import static org.apache.hudi.metadata.MetadataPartitionType.BLOOM_FILTERS;
/**
@@ -55,8 +72,7 @@ public class SparkHoodieBloomIndexHelper extends
BaseHoodieBloomIndexHelper {
private static final SparkHoodieBloomIndexHelper SINGLETON_INSTANCE =
new SparkHoodieBloomIndexHelper();
- private SparkHoodieBloomIndexHelper() {
- }
+ private SparkHoodieBloomIndexHelper() {}
public static SparkHoodieBloomIndexHelper getInstance() {
return SINGLETON_INSTANCE;
@@ -66,42 +82,92 @@ public class SparkHoodieBloomIndexHelper extends
BaseHoodieBloomIndexHelper {
public HoodiePairData<HoodieKey, HoodieRecordLocation>
findMatchingFilesForRecordKeys(
HoodieWriteConfig config, HoodieEngineContext context, HoodieTable
hoodieTable,
HoodiePairData<String, String> partitionRecordKeyPairs,
- HoodieData<Pair<String, HoodieKey>> fileComparisonPairs,
+ HoodiePairData<HoodieFileGroupId, String> fileComparisonPairs,
Map<String, List<BloomIndexFileInfo>> partitionToFileInfo,
Map<String, Long> recordsPerPartition) {
- JavaRDD<Tuple2<String, HoodieKey>> fileComparisonsRDD =
- HoodieJavaRDD.getJavaRDD(fileComparisonPairs)
- .map(pair -> new Tuple2<>(pair.getLeft(), pair.getRight()));
- int inputParallelism =
HoodieJavaPairRDD.getJavaPairRDD(partitionRecordKeyPairs).partitions().size();
- int joinParallelism = Math.max(inputParallelism,
config.getBloomIndexParallelism());
- LOG.info("InputParallelism: ${" + inputParallelism + "}, IndexParallelism:
${"
- + config.getBloomIndexParallelism() + "}");
+ int inputParallelism =
HoodieJavaPairRDD.getJavaPairRDD(partitionRecordKeyPairs).getNumPartitions();
+ int configuredBloomIndexParallelism = config.getBloomIndexParallelism();
+ // NOTE: Target parallelism could be overridden by the config
+ int targetParallelism =
+ configuredBloomIndexParallelism > 0 ? configuredBloomIndexParallelism
: inputParallelism;
+
+ LOG.info(String.format("Input parallelism: %d, Index parallelism: %d",
inputParallelism, targetParallelism));
+
+ JavaPairRDD<HoodieFileGroupId, String> fileComparisonsRDD =
HoodieJavaRDD.getJavaRDD(fileComparisonPairs);
JavaRDD<List<HoodieKeyLookupResult>> keyLookupResultRDD;
+
if (config.getBloomIndexUseMetadata()
&& hoodieTable.getMetaClient().getTableConfig().getMetadataPartitions()
.contains(BLOOM_FILTERS.getPartitionPath())) {
- // Step 1: Sort by file id
- JavaRDD<Tuple2<String, HoodieKey>> sortedFileIdAndKeyPairs =
- fileComparisonsRDD.sortBy(Tuple2::_1, true, joinParallelism);
+ SerializableConfiguration hadoopConf = new
SerializableConfiguration(hoodieTable.getHadoopConf());
+
+ HoodieTableFileSystemView baseFileOnlyView =
+ getBaseFileOnlyView(hoodieTable, partitionToFileInfo.keySet());
+
+ Broadcast<HoodieTableFileSystemView> baseFileOnlyViewBroadcast =
+ ((HoodieSparkEngineContext)
context).getJavaSparkContext().broadcast(baseFileOnlyView);
+
+ // When leveraging MT we're aiming for following goals:
+ // - (G1) All requests to MT are made in batch (ie we're trying to
fetch all the values
+ // for corresponding keys at once)
+ // - (G2) Each task reads no more than just _one_ file-group from the
MT Bloom Filters
+ // partition
+ //
+ // Ta achieve G2, following invariant have to be maintained: Spark
partitions have to be
+ // affine w/ Metadata Table's file-groups, meaning that each Spark
partition holds records
+ // belonging to one and only file-group in MT Bloom Filters partition.
To provide for that
+ // we need to make sure
+ // - Spark's used [[Partitioner]] employs same hashing function as
Metadata Table (as well
+ // as being applied to the same keys as the MT one)
+ // - Make sure that # of partitions is congruent to the # of
file-groups (ie number of Spark
+ // partitions is a multiple of the # of the file-groups).
+ //
+ // Last provision is necessary, so that for every key it's the case
that
+ //
+ // (hash(key) % N) % M = hash(key) % M, iff N % M = 0
+ //
+ // Let's take an example of N = 8 and M = 4 (default # of file-groups
in Bloom Filter
+ // partition). In that case Spark partitions for which `hash(key) %
N` will be either 0
+ // or 4, will map to the same (first) file-group in MT
+ int bloomFilterPartitionFileGroupCount =
+ config.getMetadataConfig().getBloomFilterIndexFileGroupCount();
+ int adjustedTargetParallelism =
+ targetParallelism % bloomFilterPartitionFileGroupCount == 0
+ ? targetParallelism
+ // NOTE: We add 1 to make sure parallelism a) value always stays
positive and b)
+ // {@code targetParallelism <= adjustedTargetParallelism}
+ : (targetParallelism / bloomFilterPartitionFileGroupCount + 1) *
bloomFilterPartitionFileGroupCount;
+
+ AffineBloomIndexFileGroupPartitioner partitioner =
+ new AffineBloomIndexFileGroupPartitioner(baseFileOnlyViewBroadcast,
adjustedTargetParallelism);
+
+ // First, we need to repartition and sort records using
[[AffineBloomIndexFileGroupPartitioner]]
+ // to make sure every Spark task accesses no more than just a single
file-group in MT (allows
+ // us to achieve G2).
+ //
+ // NOTE: Sorting records w/in individual partitions is required to make
sure that we cluster
+ // together keys co-located w/in the MT files (sorted by keys)
+ keyLookupResultRDD =
fileComparisonsRDD.repartitionAndSortWithinPartitions(partitioner)
+ .mapPartitionsToPair(new
HoodieMetadataBloomFilterProbingFunction(baseFileOnlyViewBroadcast,
hoodieTable))
+ // Second, we use [[HoodieFileProbingFunction]] to open actual file
and check whether it
+ // contains the records with candidate keys that were filtered in by
the Bloom Filter
+ .mapPartitions(new
HoodieFileProbingFunction(baseFileOnlyViewBroadcast, hadoopConf), true);
- // Step 2: Use bloom filter to filter and the actual log file to get the
record location
- keyLookupResultRDD = sortedFileIdAndKeyPairs.mapPartitionsWithIndex(
- new HoodieMetadataBloomIndexCheckFunction(hoodieTable), true);
} else if (config.useBloomIndexBucketizedChecking()) {
- Map<String, Long> comparisonsPerFileGroup =
computeComparisonsPerFileGroup(
+ Map<HoodieFileGroupId, Long> comparisonsPerFileGroup =
computeComparisonsPerFileGroup(
config, recordsPerPartition, partitionToFileInfo,
fileComparisonsRDD, context);
- Partitioner partitioner = new
BucketizedBloomCheckPartitioner(joinParallelism, comparisonsPerFileGroup,
+ Partitioner partitioner = new
BucketizedBloomCheckPartitioner(targetParallelism, comparisonsPerFileGroup,
config.getBloomIndexKeysPerBucket());
- keyLookupResultRDD = fileComparisonsRDD.mapToPair(t -> new
Tuple2<>(Pair.of(t._1, t._2.getRecordKey()), t))
+ keyLookupResultRDD = fileComparisonsRDD.mapToPair(t -> new
Tuple2<>(Pair.of(t._1, t._2), t))
.repartitionAndSortWithinPartitions(partitioner)
.map(Tuple2::_2)
- .mapPartitionsWithIndex(new
HoodieBloomIndexCheckFunction(hoodieTable, config), true);
+ .mapPartitions(new HoodieSparkBloomIndexCheckFunction(hoodieTable,
config), true);
} else {
- keyLookupResultRDD = fileComparisonsRDD.sortBy(Tuple2::_1, true,
joinParallelism)
- .mapPartitionsWithIndex(new
HoodieBloomIndexCheckFunction(hoodieTable, config), true);
+ keyLookupResultRDD = fileComparisonsRDD.sortByKey(true,
targetParallelism)
+ .mapPartitions(new HoodieSparkBloomIndexCheckFunction(hoodieTable,
config), true);
}
return HoodieJavaPairRDD.of(keyLookupResultRDD.flatMap(List::iterator)
@@ -115,27 +181,124 @@ public class SparkHoodieBloomIndexHelper extends
BaseHoodieBloomIndexHelper {
/**
* Compute the estimated number of bloom filter comparisons to be performed
on each file group.
*/
- private Map<String, Long> computeComparisonsPerFileGroup(
+ private Map<HoodieFileGroupId, Long> computeComparisonsPerFileGroup(
final HoodieWriteConfig config,
final Map<String, Long> recordsPerPartition,
final Map<String, List<BloomIndexFileInfo>> partitionToFileInfo,
- final JavaRDD<Tuple2<String, HoodieKey>> fileComparisonsRDD,
+ final JavaPairRDD<HoodieFileGroupId, String> fileComparisonsRDD,
final HoodieEngineContext context) {
- Map<String, Long> fileToComparisons;
+ Map<HoodieFileGroupId, Long> fileToComparisons;
if (config.getBloomIndexPruneByRanges()) {
// we will just try exploding the input and then count to determine
comparisons
// FIX(vc): Only do sampling here and extrapolate?
context.setJobStatus(this.getClass().getSimpleName(), "Compute all
comparisons needed between records and files: " + config.getTableName());
- fileToComparisons = fileComparisonsRDD.mapToPair(t -> t).countByKey();
+ fileToComparisons = fileComparisonsRDD.countByKey();
} else {
fileToComparisons = new HashMap<>();
- partitionToFileInfo.forEach((key, value) -> {
- for (BloomIndexFileInfo fileInfo : value) {
+ partitionToFileInfo.forEach((partitionPath, fileInfos) -> {
+ for (BloomIndexFileInfo fileInfo : fileInfos) {
// each file needs to be compared against all the records coming
into the partition
- fileToComparisons.put(fileInfo.getFileId(),
recordsPerPartition.get(key));
+ fileToComparisons.put(
+ new HoodieFileGroupId(partitionPath, fileInfo.getFileId()),
recordsPerPartition.get(partitionPath));
}
});
}
return fileToComparisons;
}
+
+ private static HoodieTableFileSystemView getBaseFileOnlyView(HoodieTable<?,
?, ?, ?> hoodieTable, Collection<String> partitionPaths) {
+ try {
+ List<String> fullPartitionPaths = partitionPaths.stream()
+ .map(partitionPath ->
+ String.format("%s/%s",
hoodieTable.getMetaClient().getBasePathV2(), partitionPath))
+ .collect(Collectors.toList());
+
+ FileStatus[] allFiles =
+
hoodieTable.getMetadataTable().getAllFilesInPartitions(fullPartitionPaths).values().stream()
+ .flatMap(Arrays::stream)
+ .toArray(FileStatus[]::new);
+
+ return new HoodieTableFileSystemView(hoodieTable.getMetaClient(),
hoodieTable.getActiveTimeline(), allFiles);
+ } catch (IOException e) {
+ LOG.error(String.format("Failed to fetch all files for partitions (%s)",
partitionPaths));
+ throw new HoodieIOException("Failed to fetch all files for partitions",
e);
+ }
+ }
+
+ static class AffineBloomIndexFileGroupPartitioner extends Partitioner {
+
+ private final Broadcast<HoodieTableFileSystemView>
latestBaseFilesBroadcast;
+
+ // TODO(HUDI-5619) remove when addressed
+ private final Map<String, Map<String, String>> cachedLatestBaseFileNames =
+ new HashMap<>(16);
+
+ private final int targetPartitions;
+
+ AffineBloomIndexFileGroupPartitioner(Broadcast<HoodieTableFileSystemView>
baseFileOnlyViewBroadcast,
+ int targetPartitions) {
+ this.targetPartitions = targetPartitions;
+ this.latestBaseFilesBroadcast = baseFileOnlyViewBroadcast;
+ }
+
+ @Override
+ public int numPartitions() {
+ return targetPartitions;
+ }
+
+ @Override
+ public int getPartition(Object key) {
+ HoodieFileGroupId partitionFileGroupId = (HoodieFileGroupId) key;
+ String partitionPath = partitionFileGroupId.getPartitionPath();
+ String fileGroupId = partitionFileGroupId.getFileId();
+
+ /*
+ // TODO(HUDI-5619) uncomment when addressed
+ String baseFileName =
+ latestBaseFilesBroadcast.getValue()
+ .getLatestBaseFile(partitionPath, fileGroupId)
+ .orElseThrow(() -> new HoodieException(
+ String.format("File from file-group (%s) not found in
partition path (%s)", fileGroupId, partitionPath)))
+ .getFileName();
+ */
+
+ // NOTE: This is a workaround to alleviate performance impact of needing
to process whole
+ // partition for every file-group being looked up.
+ // See HUDI-5619 for more details
+ String baseFileName =
cachedLatestBaseFileNames.computeIfAbsent(partitionPath, ignored ->
+ latestBaseFilesBroadcast.getValue()
+ .getLatestBaseFiles(partitionPath)
+ .collect(
+ Collectors.toMap(HoodieBaseFile::getFileId,
BaseFile::getFileName)
+ )
+ )
+ .get(fileGroupId);
+
+ if (baseFileName == null) {
+ throw new HoodieException(
+ String.format("File from file-group (%s) not found in partition
path (%s)", fileGroupId, partitionPath));
+ }
+
+ String bloomIndexEncodedKey =
+ getBloomFilterIndexKey(new PartitionIndexID(partitionPath), new
FileIndexID(baseFileName));
+
+ // NOTE: It's crucial that [[targetPartitions]] be congruent w/ the
number of
+ // actual file-groups in the Bloom Index in MT
+ return mapRecordKeyToFileGroupIndex(bloomIndexEncodedKey,
targetPartitions);
+ }
+ }
+
+ public static class HoodieSparkBloomIndexCheckFunction extends
HoodieBloomIndexCheckFunction<Tuple2<HoodieFileGroupId, String>>
+ implements FlatMapFunction<Iterator<Tuple2<HoodieFileGroupId, String>>,
List<HoodieKeyLookupResult>> {
+
+ public HoodieSparkBloomIndexCheckFunction(HoodieTable hoodieTable,
+ HoodieWriteConfig config) {
+ super(hoodieTable, config, t -> t._1, t -> t._2);
+ }
+
+ @Override
+ public Iterator<List<HoodieKeyLookupResult>>
call(Iterator<Tuple2<HoodieFileGroupId, String>>
fileGroupIdRecordKeyPairIterator) {
+ return new LazyKeyCheckIterator(fileGroupIdRecordKeyPairIterator);
+ }
+ }
}
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
index f4edf7fb942..5853b4eb8a8 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
@@ -23,17 +23,17 @@ import org.apache.hadoop.fs.Path
import org.apache.hudi.client.utils.SparkRowSerDe
import org.apache.hudi.common.table.HoodieTableMetaClient
import org.apache.hudi.common.util.TablePathUtils
+import org.apache.spark.sql._
import org.apache.spark.sql.avro.{HoodieAvroDeserializer,
HoodieAvroSchemaConverters, HoodieAvroSerializer}
-import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier}
import org.apache.spark.sql.catalyst.catalog.CatalogTable
import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
Expression, InterpretedPredicate}
import org.apache.spark.sql.catalyst.parser.ParserInterface
import org.apache.spark.sql.catalyst.plans.logical.{Command, LogicalPlan,
SubqueryAlias}
+import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier}
import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.sources.BaseRelation
import org.apache.spark.sql.types.{DataType, StructType}
-import org.apache.spark.sql._
import org.apache.spark.storage.StorageLevel
import java.util.Locale
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/bloom/TestBucketizedBloomCheckPartitioner.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/bloom/TestBucketizedBloomCheckPartitioner.java
index e946450c904..fe178839c0d 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/bloom/TestBucketizedBloomCheckPartitioner.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/bloom/TestBucketizedBloomCheckPartitioner.java
@@ -18,6 +18,7 @@
package org.apache.hudi.index.bloom;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.util.collection.Pair;
import org.junit.jupiter.api.Test;
@@ -36,33 +37,37 @@ public class TestBucketizedBloomCheckPartitioner {
@Test
public void testAssignmentCorrectness() {
- Map<String, Long> fileToComparisons = new HashMap<String, Long>() {
+ HoodieFileGroupId fg1 = new HoodieFileGroupId("p1", "f1");
+ HoodieFileGroupId fg2 = new HoodieFileGroupId("p1", "f2");
+ HoodieFileGroupId fg3 = new HoodieFileGroupId("p1", "f3");
+
+ Map<HoodieFileGroupId, Long> fileToComparisons = new
HashMap<HoodieFileGroupId, Long>() {
{
- put("f1", 40L);
- put("f2", 35L);
- put("f3", 20L);
+ put(fg1, 40L);
+ put(fg2, 35L);
+ put(fg3, 20L);
}
};
BucketizedBloomCheckPartitioner p = new BucketizedBloomCheckPartitioner(4,
fileToComparisons, 10);
- Map<String, List<Integer>> assignments = p.getFileGroupToPartitions();
- assertEquals(4, assignments.get("f1").size(), "f1 should have 4 buckets");
- assertEquals(4, assignments.get("f2").size(), "f2 should have 4 buckets");
- assertEquals(2, assignments.get("f3").size(), "f3 should have 2 buckets");
- assertArrayEquals(new Integer[] {0, 0, 1, 3},
assignments.get("f1").toArray(), "f1 spread across 3 partitions");
- assertArrayEquals(new Integer[] {1, 2, 2, 0},
assignments.get("f2").toArray(), "f2 spread across 3 partitions");
- assertArrayEquals(new Integer[] {3, 1}, assignments.get("f3").toArray(),
"f3 spread across 2 partitions");
+ Map<HoodieFileGroupId, List<Integer>> assignments =
p.getFileGroupToPartitions();
+ assertEquals(4, assignments.get(fg1).size(), "f1 should have 4 buckets");
+ assertEquals(4, assignments.get(fg2).size(), "f2 should have 4 buckets");
+ assertEquals(2, assignments.get(fg3).size(), "f3 should have 2 buckets");
+ assertArrayEquals(new Integer[] {0, 0, 1, 3},
assignments.get(fg1).toArray(), "f1 spread across 3 partitions");
+ assertArrayEquals(new Integer[] {2, 2, 3, 1},
assignments.get(fg2).toArray(), "f2 spread across 3 partitions");
+ assertArrayEquals(new Integer[] {1, 0}, assignments.get(fg3).toArray(),
"f3 spread across 2 partitions");
}
@Test
public void testUniformPacking() {
// evenly distribute 10 buckets/file across 100 partitions
- Map<String, Long> comparisons1 = new HashMap<String, Long>() {
+ Map<HoodieFileGroupId, Long> comparisons1 = new HashMap<HoodieFileGroupId,
Long>() {
{
- IntStream.range(0, 10).forEach(f -> put("f" + f, 100L));
+ IntStream.range(0, 10).forEach(f -> put(new HoodieFileGroupId("p1",
"f" + f), 100L));
}
};
BucketizedBloomCheckPartitioner partitioner = new
BucketizedBloomCheckPartitioner(100, comparisons1, 10);
- Map<String, List<Integer>> assignments =
partitioner.getFileGroupToPartitions();
+ Map<HoodieFileGroupId, List<Integer>> assignments =
partitioner.getFileGroupToPartitions();
assignments.forEach((key, value) -> assertEquals(10, value.size()));
Map<Integer, Long> partitionToNumBuckets =
assignments.entrySet().stream().flatMap(e ->
e.getValue().stream().map(p -> Pair.of(p, e.getKey())))
@@ -72,9 +77,9 @@ public class TestBucketizedBloomCheckPartitioner {
@Test
public void testNumPartitions() {
- Map<String, Long> comparisons1 = new HashMap<String, Long>() {
+ Map<HoodieFileGroupId, Long> comparisons1 = new HashMap<HoodieFileGroupId,
Long>() {
{
- IntStream.range(0, 10).forEach(f -> put("f" + f, 100L));
+ IntStream.range(0, 10).forEach(f -> put(new HoodieFileGroupId("p1",
"f" + f), 100L));
}
};
BucketizedBloomCheckPartitioner p = new
BucketizedBloomCheckPartitioner(10000, comparisons1, 10);
@@ -83,15 +88,15 @@ public class TestBucketizedBloomCheckPartitioner {
@Test
public void testGetPartitions() {
- Map<String, Long> comparisons1 = new HashMap<String, Long>() {
+ Map<HoodieFileGroupId, Long> comparisons1 = new HashMap<HoodieFileGroupId,
Long>() {
{
- IntStream.range(0, 100000).forEach(f -> put("f" + f, 100L));
+ IntStream.range(0, 100000).forEach(f -> put(new
HoodieFileGroupId("p1", "f" + f), 100L));
}
};
BucketizedBloomCheckPartitioner p = new
BucketizedBloomCheckPartitioner(1000, comparisons1, 10);
IntStream.range(0, 100000).forEach(f -> {
- int partition = p.getPartition(Pair.of("f" + f, "value"));
+ int partition = p.getPartition(Pair.of(new HoodieFileGroupId("p1", "f" +
f), "value"));
assertTrue(0 <= partition && partition <= 1000, "partition is out of
range: " + partition);
});
}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/bloom/TestHoodieBloomIndex.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/bloom/TestHoodieBloomIndex.java
index 5be4e4ce624..3c906062c16 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/bloom/TestHoodieBloomIndex.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/bloom/TestHoodieBloomIndex.java
@@ -24,6 +24,7 @@ import org.apache.hudi.common.bloom.BloomFilterFactory;
import org.apache.hudi.common.bloom.BloomFilterTypeCode;
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.WriteOperationType;
@@ -35,7 +36,6 @@ import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieIndexConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.data.HoodieJavaPairRDD;
-import org.apache.hudi.data.HoodieJavaRDD;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.index.HoodieIndexUtils;
import org.apache.hudi.metadata.SparkHoodieBackedTableMetadataWriter;
@@ -259,12 +259,14 @@ public class TestHoodieBloomIndex extends
TestHoodieMetadataBase {
jsc.parallelize(Arrays.asList(new Tuple2<>("2017/10/22", "003"), new
Tuple2<>("2017/10/22", "002"),
new Tuple2<>("2017/10/22", "005"), new Tuple2<>("2017/10/22",
"004"))).mapToPair(t -> t);
- List<Pair<String, HoodieKey>> comparisonKeyList = HoodieJavaRDD.getJavaRDD(
- index.explodeRecordsWithFileComparisons(partitionToFileIndexInfo,
HoodieJavaPairRDD.of(partitionRecordKeyPairRDD))).collect();
+ List<Pair<HoodieFileGroupId, String>> comparisonKeyList =
+ index.explodeRecordsWithFileComparisons(partitionToFileIndexInfo,
HoodieJavaPairRDD.of(partitionRecordKeyPairRDD)).collectAsList();
assertEquals(10, comparisonKeyList.size());
Map<String, List<String>> recordKeyToFileComps = comparisonKeyList.stream()
- .collect(Collectors.groupingBy(t -> t.getRight().getRecordKey(),
Collectors.mapping(Pair::getLeft, Collectors.toList())));
+ .collect(
+ Collectors.groupingBy(t -> t.getRight(),
+ Collectors.mapping(t -> t.getLeft().getFileId(),
Collectors.toList())));
assertEquals(4, recordKeyToFileComps.size());
assertEquals(new HashSet<>(Arrays.asList("f1", "f3", "f4")), new
HashSet<>(recordKeyToFileComps.get("002")));
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/bloom/TestHoodieGlobalBloomIndex.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/bloom/TestHoodieGlobalBloomIndex.java
index 3ad8952feea..2577f9ba284 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/bloom/TestHoodieGlobalBloomIndex.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/bloom/TestHoodieGlobalBloomIndex.java
@@ -21,6 +21,7 @@ package org.apache.hudi.index.bloom;
import org.apache.hudi.client.functional.TestHoodieMetadataBase;
import org.apache.hudi.common.model.EmptyHoodieRecordPayload;
import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.WriteOperationType;
@@ -29,7 +30,6 @@ import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieIndexConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.data.HoodieJavaPairRDD;
-import org.apache.hudi.data.HoodieJavaRDD;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.table.HoodieSparkTable;
import org.apache.hudi.table.HoodieTable;
@@ -201,9 +201,9 @@ public class TestHoodieGlobalBloomIndex extends
TestHoodieMetadataBase {
jsc.parallelize(Arrays.asList(new Tuple2<>("2017/10/21", "003"), new
Tuple2<>("2017/10/22", "002"),
new Tuple2<>("2017/10/22", "005"), new Tuple2<>("2017/10/23",
"004"))).mapToPair(t -> t);
- List<Pair<String, HoodieKey>> comparisonKeyList = HoodieJavaRDD.getJavaRDD(
- index.explodeRecordsWithFileComparisons(partitionToFileIndexInfo,
- HoodieJavaPairRDD.of(partitionRecordKeyPairRDD))).collect();
+ List<Pair<HoodieFileGroupId, String>> comparisonKeyList =
+ index.explodeRecordsWithFileComparisons(partitionToFileIndexInfo,
HoodieJavaPairRDD.of(partitionRecordKeyPairRDD))
+ .collectAsList();
/*
* expecting: f4, HoodieKey { recordKey=003 partitionPath=2017/10/23} f1,
HoodieKey { recordKey=003
@@ -216,7 +216,7 @@ public class TestHoodieGlobalBloomIndex extends
TestHoodieMetadataBase {
assertEquals(10, comparisonKeyList.size());
Map<String, List<String>> recordKeyToFileComps = comparisonKeyList.stream()
- .collect(Collectors.groupingBy(t -> t.getRight().getRecordKey(),
Collectors.mapping(Pair::getKey, Collectors.toList())));
+ .collect(Collectors.groupingBy(t -> t.getRight(), Collectors.mapping(t
-> t.getLeft().getFileId(), Collectors.toList())));
assertEquals(4, recordKeyToFileComps.size());
assertEquals(new HashSet<>(Arrays.asList("f4", "f1", "f3")), new
HashSet<>(recordKeyToFileComps.get("002")));
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieData.java
b/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieData.java
index 1d56e63fad9..9ce0947883f 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieData.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieData.java
@@ -105,9 +105,9 @@ public interface HoodieData<T> extends Serializable {
Iterator<O>> func, boolean preservesPartitioning);
/**
- * Maps every element in the collection into a collection of the new
elements (provided by
- * {@link Iterator}) using provided mapping {@code func}, subsequently
flattening the result
- * (by concatenating) into a single collection
+ * Maps every element in the collection into a collection of the new
elements using provided
+ * mapping {@code func}, subsequently flattening the result (by
concatenating) into a single
+ * collection
*
* This is an intermediate operation
*
@@ -117,6 +117,17 @@ public interface HoodieData<T> extends Serializable {
*/
<O> HoodieData<O> flatMap(SerializableFunction<T, Iterator<O>> func);
+ /**
+ * Maps every element in the collection into a collection of the {@link
Pair}s of new elements
+ * using provided mapping {@code func}, subsequently flattening the result
(by concatenating) into
+ * a single collection
+ *
+ * NOTE: That this operation will convert container from {@link HoodieData}
to {@link HoodiePairData}
+ *
+ * This is an intermediate operation
+ */
+ <K, V> HoodiePairData<K, V> flatMapToPair(SerializableFunction<T, Iterator<?
extends Pair<K, V>>> func);
+
/**
* Maps every element in the collection using provided mapping {@code func}
into a {@link Pair<K, V>}
* of elements {@code K} and {@code V}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieListData.java
b/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieListData.java
index b2a503a85b3..c6287a744e0 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieListData.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieListData.java
@@ -125,6 +125,16 @@ public class HoodieListData<T> extends
HoodieBaseListData<T> implements HoodieDa
return new HoodieListData<>(mappedStream, lazy);
}
+ @Override
+ public <K, V> HoodiePairData<K, V> flatMapToPair(SerializableFunction<T,
Iterator<? extends Pair<K, V>>> func) {
+ Function<T, Iterator<? extends Pair<K, V>>> mapper =
throwingMapWrapper(func);
+ Stream<Pair<K, V>> mappedStream = asStream().flatMap(e ->
+ StreamSupport.stream(
+ Spliterators.spliteratorUnknownSize(mapper.apply(e),
Spliterator.ORDERED), true));
+
+ return new HoodieListPairData<>(mappedStream, lazy);
+ }
+
@Override
public <K, V> HoodiePairData<K, V> mapToPair(SerializablePairFunction<T, K,
V> func) {
Function<T, Pair<K, V>> throwableMapToPairFunc =
throwingMapToPairWrapper(func);
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieListPairData.java
b/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieListPairData.java
index a389649548e..39ce1411575 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieListPairData.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieListPairData.java
@@ -23,15 +23,20 @@ import org.apache.hudi.common.function.SerializableFunction;
import org.apache.hudi.common.function.SerializablePairFunction;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.MappingIterator;
import org.apache.hudi.common.util.collection.Pair;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Spliterator;
+import java.util.Spliterators;
import java.util.function.Function;
import java.util.stream.Collector;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
import static
org.apache.hudi.common.function.FunctionWrapper.throwingMapToPairWrapper;
import static
org.apache.hudi.common.function.FunctionWrapper.throwingMapWrapper;
@@ -136,6 +141,24 @@ public class HoodieListPairData<K, V> extends
HoodieBaseListData<Pair<K, V>> imp
return new HoodieListData<>(asStream().map(uncheckedMapper), lazy);
}
+ @Override
+ public <W> HoodiePairData<K, W> mapValues(SerializableFunction<V, W> func) {
+ Function<V, W> uncheckedMapper = throwingMapWrapper(func);
+ return new HoodieListPairData<>(asStream().map(p -> Pair.of(p.getKey(),
uncheckedMapper.apply(p.getValue()))), lazy);
+ }
+
+ public <W> HoodiePairData<K, W> flatMapValues(SerializableFunction<V,
Iterator<W>> func) {
+ Function<V, Iterator<W>> uncheckedMapper = throwingMapWrapper(func);
+ return new HoodieListPairData<>(asStream().flatMap(p -> {
+ Iterator<W> mappedValuesIterator = uncheckedMapper.apply(p.getValue());
+ Iterator<Pair<K, W>> mappedPairsIterator =
+ new MappingIterator<>(mappedValuesIterator, w -> Pair.of(p.getKey(),
w));
+
+ return StreamSupport.stream(
+ Spliterators.spliteratorUnknownSize(mappedPairsIterator,
Spliterator.ORDERED), true);
+ }), lazy);
+ }
+
@Override
public <L, W> HoodiePairData<L, W>
mapToPair(SerializablePairFunction<Pair<K, V>, L, W> mapToPairFunc) {
return new HoodieListPairData<>(asStream().map(p ->
throwingMapToPairWrapper(mapToPairFunc).apply(p)), lazy);
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/data/HoodiePairData.java
b/hudi-common/src/main/java/org/apache/hudi/common/data/HoodiePairData.java
index 49fa7174da9..1d3622786fd 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/data/HoodiePairData.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/data/HoodiePairData.java
@@ -90,12 +90,17 @@ public interface HoodiePairData<K, V> extends Serializable {
HoodiePairData<K, V> reduceByKey(SerializableBiFunction<V, V, V> combiner,
int parallelism);
/**
- * @param func serializable map function.
- * @param <O> output object type.
- * @return {@link HoodieData<O>} containing the result. Actual execution may
be deferred.
+ * Maps key-value pairs of this {@link HoodiePairData} container leveraging
provided mapper
+ *
+ * NOTE: That this returns {@link HoodieData} and not {@link HoodiePairData}
*/
<O> HoodieData<O> map(SerializableFunction<Pair<K, V>, O> func);
+ /**
+ * Maps values of this {@link HoodiePairData} container leveraging provided
mapper
+ */
+ <W> HoodiePairData<K, W> mapValues(SerializableFunction<V, W> func);
+
/**
* @param mapToPairFunc serializable map function to generate another pair.
* @param <L> new key type.
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/util/collection/MappingIterator.java
b/hudi-common/src/main/java/org/apache/hudi/common/function/ThrowingConsumer.java
similarity index 59%
copy from
hudi-common/src/main/java/org/apache/hudi/common/util/collection/MappingIterator.java
copy to
hudi-common/src/main/java/org/apache/hudi/common/function/ThrowingConsumer.java
index a4655764a97..1cbae113fcb 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/util/collection/MappingIterator.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/function/ThrowingConsumer.java
@@ -16,29 +16,14 @@
* limitations under the License.
*/
-package org.apache.hudi.common.util.collection;
+package org.apache.hudi.common.function;
-import java.util.Iterator;
-import java.util.function.Function;
-
-// TODO java-docs
-public class MappingIterator<I, O> implements Iterator<O> {
-
- protected final Iterator<I> source;
- private final Function<I, O> mapper;
-
- public MappingIterator(Iterator<I> source, Function<I, O> mapper) {
- this.source = source;
- this.mapper = mapper;
- }
+/**
+ * Throwing counterpart of {@link java.util.function.Consumer}
+ */
+@FunctionalInterface
+public interface ThrowingConsumer<T> {
- @Override
- public boolean hasNext() {
- return source.hasNext();
- }
+ void accept(T t) throws Exception;
- @Override
- public O next() {
- return mapper.apply(source.next());
- }
-}
+}
\ No newline at end of file
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieAvroIndexedRecord.java
b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieAvroIndexedRecord.java
index 7d497408e64..a8c695240cd 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieAvroIndexedRecord.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieAvroIndexedRecord.java
@@ -113,7 +113,7 @@ public class HoodieAvroIndexedRecord extends
HoodieRecord<IndexedRecord> {
@Override
public HoodieRecord rewriteRecord(Schema recordSchema, Properties props,
Schema targetSchema) throws IOException {
- GenericRecord record = HoodieAvroUtils.rewriteRecord((GenericRecord) data,
targetSchema);
+ GenericRecord record = HoodieAvroUtils.rewriteRecordWithNewSchema(data,
targetSchema);
return new HoodieAvroIndexedRecord(key, record, operation, metaData);
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
index e7c749a2911..00fa1b97db0 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
@@ -338,6 +338,10 @@ public class HoodieTableMetaClient implements Serializable
{
return hadoopConf.get();
}
+ public SerializableConfiguration getSerializableHadoopConf() {
+ return hadoopConf;
+ }
+
/**
* Get the active instants as a timeline.
*
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/util/collection/FlatteningIterator.java
b/hudi-common/src/main/java/org/apache/hudi/common/util/collection/FlatteningIterator.java
new file mode 100644
index 00000000000..4adf6b5c91f
--- /dev/null
+++
b/hudi-common/src/main/java/org/apache/hudi/common/util/collection/FlatteningIterator.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.util.collection;
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+/**
+ * Iterator flattening source {@link Iterator} holding other {@link Iterator}s
+ */
+public final class FlatteningIterator<T, I extends Iterator<T>> implements
Iterator<T> {
+
+ private final Iterator<I> sourceIterator;
+ private Iterator<T> innerSourceIterator;
+
+ public FlatteningIterator(Iterator<I> source) {
+ this.sourceIterator = source;
+ }
+
+ public boolean hasNext() {
+ while (innerSourceIterator == null || !innerSourceIterator.hasNext()) {
+ if (sourceIterator.hasNext()) {
+ innerSourceIterator = sourceIterator.next();
+ } else {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ @Override
+ public T next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+
+ return innerSourceIterator.next();
+ }
+}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/util/collection/MappingIterator.java
b/hudi-common/src/main/java/org/apache/hudi/common/util/collection/MappingIterator.java
index a4655764a97..24b0961470b 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/util/collection/MappingIterator.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/util/collection/MappingIterator.java
@@ -21,7 +21,9 @@ package org.apache.hudi.common.util.collection;
import java.util.Iterator;
import java.util.function.Function;
-// TODO java-docs
+/**
+ * Iterator mapping elements of the provided source {@link Iterator} from
{@code I} to {@code O}
+ */
public class MappingIterator<I, O> implements Iterator<O> {
protected final Iterator<I> source;
diff --git
a/hudi-common/src/main/java/org/apache/hudi/metadata/BaseTableMetadata.java
b/hudi-common/src/main/java/org/apache/hudi/metadata/BaseTableMetadata.java
index 1034aa31255..f343d6437cf 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/BaseTableMetadata.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/BaseTableMetadata.java
@@ -19,12 +19,12 @@
package org.apache.hudi.metadata;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hudi.avro.model.HoodieMetadataBloomFilter;
import org.apache.hudi.avro.model.HoodieMetadataColumnStats;
import org.apache.hudi.common.bloom.BloomFilter;
import org.apache.hudi.common.bloom.BloomFilterFactory;
import org.apache.hudi.common.config.HoodieMetadataConfig;
-import org.apache.hudi.common.config.SerializableConfiguration;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.engine.HoodieLocalEngineContext;
import org.apache.hudi.common.fs.FSUtils;
@@ -73,11 +73,12 @@ public abstract class BaseTableMetadata implements
HoodieTableMetadata {
private static final Logger LOG =
LogManager.getLogger(BaseTableMetadata.class);
- public static final long MAX_MEMORY_SIZE_IN_BYTES = 1024 * 1024 * 1024;
- public static final int BUFFER_SIZE = 10 * 1024 * 1024;
+ protected static final long MAX_MEMORY_SIZE_IN_BYTES = 1024 * 1024 * 1024;
+ // NOTE: Buffer-size is deliberately set pretty low, since MT internally is
relying
+ // on HFile (serving as persisted binary key-value mapping) to do
caching
+ protected static final int BUFFER_SIZE = 10 * 1024; // 10Kb
protected final transient HoodieEngineContext engineContext;
- protected final SerializableConfiguration hadoopConf;
protected final SerializablePath dataBasePath;
protected final HoodieTableMetaClient dataMetaClient;
protected final Option<HoodieMetadataMetrics> metrics;
@@ -92,9 +93,11 @@ public abstract class BaseTableMetadata implements
HoodieTableMetadata {
protected BaseTableMetadata(HoodieEngineContext engineContext,
HoodieMetadataConfig metadataConfig,
String dataBasePath, String
spillableMapDirectory) {
this.engineContext = engineContext;
- this.hadoopConf = new
SerializableConfiguration(engineContext.getHadoopConf());
this.dataBasePath = new SerializablePath(new CachingPath(dataBasePath));
- this.dataMetaClient =
HoodieTableMetaClient.builder().setConf(hadoopConf.get()).setBasePath(dataBasePath).build();
+ this.dataMetaClient = HoodieTableMetaClient.builder()
+ .setConf(engineContext.getHadoopConf().get())
+ .setBasePath(dataBasePath)
+ .build();
this.spillableMapDirectory = spillableMapDirectory;
this.metadataConfig = metadataConfig;
@@ -123,8 +126,10 @@ public abstract class BaseTableMetadata implements
HoodieTableMetadata {
throw new HoodieMetadataException("Failed to retrieve list of
partition from metadata", e);
}
}
- return new FileSystemBackedTableMetadata(getEngineContext(), hadoopConf,
dataBasePath.toString(),
- metadataConfig.shouldAssumeDatePartitioning()).getAllPartitionPaths();
+
+ FileSystemBackedTableMetadata fileSystemBackedTableMetadata =
+ createFileSystemBackedTableMetadata();
+ return fileSystemBackedTableMetadata.getAllPartitionPaths();
}
/**
@@ -148,8 +153,9 @@ public abstract class BaseTableMetadata implements
HoodieTableMetadata {
}
}
- return new FileSystemBackedTableMetadata(getEngineContext(), hadoopConf,
dataBasePath.toString(), metadataConfig.shouldAssumeDatePartitioning())
- .getAllFilesInPartition(partitionPath);
+ FileSystemBackedTableMetadata fileSystemBackedTableMetadata =
+ createFileSystemBackedTableMetadata();
+ return fileSystemBackedTableMetadata.getAllFilesInPartition(partitionPath);
}
@Override
@@ -168,8 +174,9 @@ public abstract class BaseTableMetadata implements
HoodieTableMetadata {
}
}
- return new FileSystemBackedTableMetadata(getEngineContext(), hadoopConf,
dataBasePath.toString(), metadataConfig.shouldAssumeDatePartitioning())
- .getAllFilesInPartitions(partitions);
+ FileSystemBackedTableMetadata fileSystemBackedTableMetadata =
+ createFileSystemBackedTableMetadata();
+ return fileSystemBackedTableMetadata.getAllFilesInPartitions(partitions);
}
@Override
@@ -205,6 +212,7 @@ public abstract class BaseTableMetadata implements
HoodieTableMetadata {
HoodieTimer timer = HoodieTimer.start();
Set<String> partitionIDFileIDSortedStrings = new TreeSet<>();
Map<String, Pair<String, String>> fileToKeyMap = new HashMap<>();
+ // TODO simplify (no sorting is required)
partitionNameFileNameList.forEach(partitionNameFileNamePair -> {
final String bloomFilterIndexKey =
HoodieMetadataPayload.getBloomFilterIndexKey(
new PartitionIndexID(partitionNameFileNamePair.getLeft()), new
FileIndexID(partitionNameFileNamePair.getRight()));
@@ -227,7 +235,11 @@ public abstract class BaseTableMetadata implements
HoodieTableMetadata {
if (bloomFilterMetadata.isPresent()) {
if (!bloomFilterMetadata.get().getIsDeleted()) {
ValidationUtils.checkState(fileToKeyMap.containsKey(entry.getLeft()));
- final ByteBuffer bloomFilterByteBuffer =
bloomFilterMetadata.get().getBloomFilter();
+ // NOTE: We have to duplicate the [[ByteBuffer]] object here since:
+ // - Reading out [[ByteBuffer]] mutates its state
+ // - [[BloomFilterMetadata]] could be re-used, and hence
have to stay immutable
+ final ByteBuffer bloomFilterByteBuffer =
+ bloomFilterMetadata.get().getBloomFilter().duplicate();
final String bloomFilterType = bloomFilterMetadata.get().getType();
final BloomFilter bloomFilter = BloomFilterFactory.fromString(
StandardCharsets.UTF_8.decode(bloomFilterByteBuffer).toString(),
bloomFilterType);
@@ -332,7 +344,7 @@ public abstract class BaseTableMetadata implements
HoodieTableMetadata {
HoodieMetadataPayload metadataPayload = record.getData();
checkForSpuriousDeletes(metadataPayload, recordKey);
try {
- return metadataPayload.getFileStatuses(hadoopConf.get(),
partitionPath);
+ return metadataPayload.getFileStatuses(getHadoopConf(), partitionPath);
} catch (IOException e) {
throw new HoodieIOException("Failed to extract file-statuses from the
payload", e);
}
@@ -358,7 +370,7 @@ public abstract class BaseTableMetadata implements
HoodieTableMetadata {
getRecordsByKeys(new ArrayList<>(partitionIdToPathMap.keySet()),
MetadataPartitionType.FILES.getPartitionPath());
metrics.ifPresent(m ->
m.updateMetrics(HoodieMetadataMetrics.LOOKUP_FILES_STR, timer.endTimer()));
- FileSystem fs = partitionPaths.get(0).getFileSystem(hadoopConf.get());
+ FileSystem fs = partitionPaths.get(0).getFileSystem(getHadoopConf());
Map<String, FileStatus[]> partitionPathToFilesMap =
partitionIdRecordPairs.parallelStream()
.map(pair -> {
@@ -399,18 +411,27 @@ public abstract class BaseTableMetadata implements
HoodieTableMetadata {
}
}
+ private FileSystemBackedTableMetadata createFileSystemBackedTableMetadata() {
+ return new FileSystemBackedTableMetadata(getEngineContext(),
dataMetaClient.getSerializableHadoopConf(), dataBasePath.toString(),
+ metadataConfig.shouldAssumeDatePartitioning());
+ }
+
protected abstract Option<HoodieRecord<HoodieMetadataPayload>>
getRecordByKey(String key, String partitionName);
public abstract List<Pair<String,
Option<HoodieRecord<HoodieMetadataPayload>>>> getRecordsByKeys(List<String>
key, String partitionName);
protected HoodieEngineContext getEngineContext() {
- return engineContext != null ? engineContext : new
HoodieLocalEngineContext(hadoopConf.get());
+ return engineContext != null ? engineContext : new
HoodieLocalEngineContext(getHadoopConf());
}
public HoodieMetadataConfig getMetadataConfig() {
return metadataConfig;
}
+ protected Configuration getHadoopConf() {
+ return dataMetaClient.getHadoopConf();
+ }
+
protected String getLatestDataInstantTime() {
return
dataMetaClient.getActiveTimeline().filterCompletedInstants().lastInstant()
.map(HoodieInstant::getTimestamp).orElse(SOLO_COMMIT_TIMESTAMP);
diff --git
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
index 09c6b35309a..ecb0da8792d 100644
---
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
+++
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
@@ -24,7 +24,6 @@ import org.apache.hudi.avro.model.HoodieRestoreMetadata;
import org.apache.hudi.avro.model.HoodieRollbackMetadata;
import org.apache.hudi.common.config.HoodieCommonConfig;
import org.apache.hudi.common.config.HoodieMetadataConfig;
-import org.apache.hudi.common.config.SerializableConfiguration;
import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.data.HoodieListData;
import org.apache.hudi.common.engine.HoodieEngineContext;
@@ -53,6 +52,7 @@ import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.HoodieMetadataException;
import org.apache.hudi.exception.TableNotFoundException;
+import org.apache.hudi.util.Transient;
import org.apache.hudi.io.storage.HoodieFileReaderFactory;
import org.apache.hudi.io.storage.HoodieSeekingFileReader;
@@ -90,19 +90,18 @@ public class HoodieBackedTableMetadata extends
BaseTableMetadata {
private static final Logger LOG =
LogManager.getLogger(HoodieBackedTableMetadata.class);
- private static final Schema METADATA_RECORD_SCHEMA =
HoodieMetadataRecord.getClassSchema();
+ private final String metadataBasePath;
- private String metadataBasePath;
- // Metadata table's timeline and metaclient
private HoodieTableMetaClient metadataMetaClient;
private HoodieTableConfig metadataTableConfig;
+
private HoodieTableFileSystemView metadataFileSystemView;
// should we reuse the open file handles, across calls
private final boolean reuse;
// Readers for the latest file slice corresponding to file groups in the
metadata partition
- private final Map<Pair<String, String>, Pair<HoodieSeekingFileReader<?>,
HoodieMetadataLogRecordReader>> partitionReaders =
- new ConcurrentHashMap<>();
+ private final Transient<Map<Pair<String, String>,
Pair<HoodieSeekingFileReader<?>, HoodieMetadataLogRecordReader>>>
partitionReaders =
+ Transient.lazy(ConcurrentHashMap::new);
public HoodieBackedTableMetadata(HoodieEngineContext engineContext,
HoodieMetadataConfig metadataConfig,
String datasetBasePath, String
spillableMapDirectory) {
@@ -113,18 +112,19 @@ public class HoodieBackedTableMetadata extends
BaseTableMetadata {
String datasetBasePath, String
spillableMapDirectory, boolean reuse) {
super(engineContext, metadataConfig, datasetBasePath,
spillableMapDirectory);
this.reuse = reuse;
+ this.metadataBasePath =
HoodieTableMetadata.getMetadataTableBasePath(dataBasePath.toString());
+
initIfNeeded();
}
private void initIfNeeded() {
- this.metadataBasePath =
HoodieTableMetadata.getMetadataTableBasePath(dataBasePath.toString());
if (!isMetadataTableEnabled) {
if (!HoodieTableMetadata.isMetadataTable(metadataBasePath)) {
LOG.info("Metadata table is disabled.");
}
} else if (this.metadataMetaClient == null) {
try {
- this.metadataMetaClient =
HoodieTableMetaClient.builder().setConf(hadoopConf.get()).setBasePath(metadataBasePath).build();
+ this.metadataMetaClient =
HoodieTableMetaClient.builder().setConf(getHadoopConf()).setBasePath(metadataBasePath).build();
this.metadataFileSystemView = getFileSystemView(metadataMetaClient);
this.metadataTableConfig = metadataMetaClient.getTableConfig();
this.isBloomFilterIndexEnabled =
metadataConfig.isBloomFilterIndexEnabled();
@@ -213,14 +213,14 @@ public class HoodieBackedTableMetadata extends
BaseTableMetadata {
return mergedRecords.stream()
.map(keyRecordPair -> keyRecordPair.getValue().orElse(null))
+ .filter(Objects::nonNull)
.iterator();
} catch (IOException ioe) {
throw new HoodieIOException("Error merging records from metadata
table for " + sortedKeyPrefixes.size() + " key : ", ioe);
} finally {
closeReader(readers);
}
- })
- .filter(Objects::nonNull);
+ });
}
@Override
@@ -425,7 +425,7 @@ public class HoodieBackedTableMetadata extends
BaseTableMetadata {
private Pair<HoodieSeekingFileReader<?>, HoodieMetadataLogRecordReader>
getOrCreateReaders(String partitionName, FileSlice slice) {
if (reuse) {
Pair<String, String> key = Pair.of(partitionName, slice.getFileId());
- return partitionReaders.computeIfAbsent(key, ignored ->
openReaders(partitionName, slice));
+ return partitionReaders.get().computeIfAbsent(key, ignored ->
openReaders(partitionName, slice));
} else {
return openReaders(partitionName, slice);
}
@@ -462,7 +462,7 @@ public class HoodieBackedTableMetadata extends
BaseTableMetadata {
if (basefile.isPresent()) {
String baseFilePath = basefile.get().getPath();
baseFileReader = (HoodieSeekingFileReader<?>)
HoodieFileReaderFactory.getReaderFactory(HoodieRecordType.AVRO)
- .getFileReader(hadoopConf.get(), new Path(baseFilePath));
+ .getFileReader(getHadoopConf(), new Path(baseFilePath));
baseFileOpenMs = timer.endTimer();
LOG.info(String.format("Opened metadata base file from %s at instant %s
in %d ms", baseFilePath,
basefile.get().getCommitTime(), baseFileOpenMs));
@@ -596,7 +596,7 @@ public class HoodieBackedTableMetadata extends
BaseTableMetadata {
*/
private synchronized void close(Pair<String, String> partitionFileSlicePair)
{
Pair<HoodieSeekingFileReader<?>, HoodieMetadataLogRecordReader> readers =
- partitionReaders.remove(partitionFileSlicePair);
+ partitionReaders.get().remove(partitionFileSlicePair);
closeReader(readers);
}
@@ -604,10 +604,10 @@ public class HoodieBackedTableMetadata extends
BaseTableMetadata {
* Close and clear all the partitions readers.
*/
private void closePartitionReaders() {
- for (Pair<String, String> partitionFileSlicePair :
partitionReaders.keySet()) {
+ for (Pair<String, String> partitionFileSlicePair :
partitionReaders.get().keySet()) {
close(partitionFileSlicePair);
}
- partitionReaders.clear();
+ partitionReaders.get().clear();
}
private void closeReader(Pair<HoodieSeekingFileReader<?>,
HoodieMetadataLogRecordReader> readers) {
@@ -629,10 +629,6 @@ public class HoodieBackedTableMetadata extends
BaseTableMetadata {
return isMetadataTableEnabled;
}
- public SerializableConfiguration getHadoopConf() {
- return hadoopConf;
- }
-
public HoodieTableMetaClient getMetadataMetaClient() {
return metadataMetaClient;
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataFileSystemView.java
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataFileSystemView.java
index 76b341609de..4d1db7e81d4 100644
---
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataFileSystemView.java
+++
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataFileSystemView.java
@@ -54,9 +54,8 @@ public class HoodieMetadataFileSystemView extends
HoodieTableFileSystemView {
HoodieTableMetaClient metaClient,
HoodieTimeline visibleActiveTimeline,
HoodieMetadataConfig metadataConfig) {
- super(metaClient, visibleActiveTimeline);
- this.tableMetadata = HoodieTableMetadata.create(engineContext,
metadataConfig, metaClient.getBasePath(),
- FileSystemViewStorageConfig.SPILLABLE_DIR.defaultValue(), true);
+ this(metaClient, visibleActiveTimeline,
HoodieTableMetadata.create(engineContext, metadataConfig,
+ metaClient.getBasePath(),
FileSystemViewStorageConfig.SPILLABLE_DIR.defaultValue(), true));
}
/**
diff --git a/hudi-common/src/main/java/org/apache/hudi/util/Transient.java
b/hudi-common/src/main/java/org/apache/hudi/util/Transient.java
new file mode 100644
index 00000000000..0d8f6ad6565
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/util/Transient.java
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.util;
+
+import org.apache.hudi.common.function.SerializableSupplier;
+import org.apache.hudi.common.function.ThrowingConsumer;
+
+import javax.annotation.concurrent.ThreadSafe;
+import java.io.Serializable;
+
+import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
+
+/**
+ * {@link Serializable} counterpart of {@link Lazy}
+ *
+ * @param <T> type of the object being held by {@link Transient}
+ */
+@ThreadSafe
+public class Transient<T> implements Serializable {
+
+ private SerializableSupplier<T> initializer;
+
+ private transient boolean initialized;
+ private transient T ref;
+
+ private Transient(SerializableSupplier<T> initializer) {
+ checkArgument(initializer != null);
+
+ this.initializer = initializer;
+ this.ref = null;
+ this.initialized = false;
+ }
+
+ private Transient(T value, SerializableSupplier<T> initializer) {
+ checkArgument(value != null);
+ checkArgument(initializer != null);
+
+ this.initializer = initializer;
+ this.ref = value;
+ this.initialized = true;
+ }
+
+ public T get() {
+ if (!initialized) {
+ synchronized (this) {
+ if (!initialized) {
+ this.ref = initializer.get();
+ initialized = true;
+ }
+ }
+ }
+
+ return ref;
+ }
+
+ public void reset() {
+ synchronized (this) {
+ this.ref = null;
+ this.initialized = false;
+ }
+ }
+
+ public void destroy(ThrowingConsumer<T> cleaner) throws Exception {
+ synchronized (this) {
+ if (initialized) {
+ cleaner.accept(ref);
+ }
+
+ this.ref = null;
+ this.initialized = false;
+ this.initializer = null;
+ }
+ }
+
+ /**
+ * Creates instance of {@link Transient} by lazily executing provided {@code
initializer},
+ * to instantiate value of type {@link T}. Same initializer will be used to
re-instantiate
+ * the value after original one being dropped during
serialization/deserialization cycle
+ */
+ public static <T> Transient<T> lazy(SerializableSupplier<T> initializer) {
+ return new Transient<>(initializer);
+ }
+
+ /**
+ * Creates instance of {@link Transient} by eagerly setting it to provided
{@code value},
+ * while given {@code initializer} will be used to re-instantiate the value
after original
+ * one being dropped during serialization/deserialization cycle
+ */
+ public static <T> Transient<T> eager(T value, SerializableSupplier<T>
initializer) {
+ return new Transient<>(value, initializer);
+ }
+}
\ No newline at end of file
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/util/collection/TestIterators.java
b/hudi-common/src/test/java/org/apache/hudi/common/util/collection/TestIterators.java
new file mode 100644
index 00000000000..6e9d43388b1
--- /dev/null
+++
b/hudi-common/src/test/java/org/apache/hudi/common/util/collection/TestIterators.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.util.collection;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.util.CollectionUtils.toStream;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class TestIterators {
+
+ @Test
+ public void testFlatteningIterator() {
+ List<List<Integer>> listOfList =
+ Arrays.asList(
+ Arrays.asList(0),
+ Arrays.asList(1, 2),
+ Collections.emptyList(),
+ Arrays.asList(3, 4, 5)
+ );
+
+ List<Integer> flattenedList =
+ toStream(new FlatteningIterator<>(new
MappingIterator<>(listOfList.iterator(), List::iterator)))
+ .collect(Collectors.toList());
+
+ assertEquals(Arrays.asList(0, 1, 2, 3, 4, 5), flattenedList);
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark2/src/main/scala/org/apache/spark/sql/adapter/Spark2Adapter.scala
b/hudi-spark-datasource/hudi-spark2/src/main/scala/org/apache/spark/sql/adapter/Spark2Adapter.scala
index da53f00e697..c4d7c1ff5b0 100644
---
a/hudi-spark-datasource/hudi-spark2/src/main/scala/org/apache/spark/sql/adapter/Spark2Adapter.scala
+++
b/hudi-spark-datasource/hudi-spark2/src/main/scala/org/apache/spark/sql/adapter/Spark2Adapter.scala
@@ -33,13 +33,11 @@ import org.apache.spark.sql.catalyst.parser.ParserInterface
import org.apache.spark.sql.catalyst.plans.logical.{Command, DeleteFromTable,
LogicalPlan}
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat,
Spark24HoodieParquetFileFormat}
+import org.apache.spark.sql.execution.vectorized.MutableColumnarRow
import org.apache.spark.sql.hudi.SparkAdapter
import org.apache.spark.sql.hudi.parser.HoodieSpark2ExtendedSqlParser
import org.apache.spark.sql.sources.BaseRelation
import org.apache.spark.sql.types.{DataType, StructType}
-import org.apache.spark.sql._
-import org.apache.spark.sql.execution.vectorized.MutableColumnarRow
-import org.apache.spark.sql.sources.BaseRelation
import org.apache.spark.storage.StorageLevel
import org.apache.spark.storage.StorageLevel._
