You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/07/01 12:31:08 UTC
(paimon) branch master updated: [core] Read deletion indexes at once to reduce file IO in splits generation (#3646)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 3521f8cc3 [core] Read deletion indexes at once to reduce file IO in splits generation (#3646)
3521f8cc3 is described below
commit 3521f8cc310a7e3cfcdd7758697c413b73db1f9a
Author: Zouxxyy <zo...@alibaba-inc.com>
AuthorDate: Mon Jul 1 20:31:03 2024 +0800
[core] Read deletion indexes at once to reduce file IO in splits generation (#3646)
---
.../DeletionVectorIndexFileMaintainer.java | 2 +-
.../deletionvectors/DeletionVectorsMaintainer.java | 3 +-
.../org/apache/paimon/index/IndexFileHandler.java | 51 ++++++++++++++--------
.../org/apache/paimon/index/PartitionIndex.java | 2 +-
.../table/source/snapshot/SnapshotReaderImpl.java | 41 +++++++++++------
.../paimon/operation/FileStoreCommitTest.java | 6 +--
.../paimon/flink/DynamicBucketTableITCase.java | 2 +-
7 files changed, 70 insertions(+), 37 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVectorIndexFileMaintainer.java b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVectorIndexFileMaintainer.java
index a9c52fb22..eb8202155 100644
--- a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVectorIndexFileMaintainer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVectorIndexFileMaintainer.java
@@ -52,7 +52,7 @@ public class DeletionVectorIndexFileMaintainer {
.map(deletionFile -> new Path(deletionFile.path()).getName())
.distinct()
.collect(Collectors.toList());
- indexFileHandler.scan().stream()
+ indexFileHandler.scanEntries().stream()
.filter(
indexManifestEntry ->
touchedIndexFileNames.contains(
diff --git a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVectorsMaintainer.java b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVectorsMaintainer.java
index 8cc563994..8079d977c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVectorsMaintainer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVectorsMaintainer.java
@@ -148,7 +148,8 @@ public class DeletionVectorsMaintainer {
List<IndexFileMeta> indexFiles =
snapshotId == null
? Collections.emptyList()
- : handler.scan(snapshotId, DELETION_VECTORS_INDEX, partition).stream()
+ : handler.scanEntries(snapshotId, DELETION_VECTORS_INDEX, partition)
+ .stream()
.map(IndexManifestEntry::indexFile)
.collect(Collectors.toList());
Map<String, DeletionVector> deletionVectors =
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
index ae4c6f51b..afafb42ec 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
@@ -28,6 +28,7 @@ import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.IndexManifestFile;
import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.utils.IntIterator;
+import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.PathFactory;
import org.apache.paimon.utils.SnapshotManager;
@@ -35,9 +36,11 @@ import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import static org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
import static org.apache.paimon.index.HashIndexFile.HASH_INDEX;
@@ -69,14 +72,6 @@ public class IndexFileHandler {
return this.deletionVectorsIndex;
}
- public List<IndexManifestEntry> scan() {
- Snapshot snapshot = snapshotManager.latestSnapshot();
- if (snapshot == null || snapshot.indexManifest() == null) {
- return Collections.emptyList();
- }
- return indexManifestFile.read(snapshot.indexManifest());
- }
-
public Optional<IndexFileMeta> scanHashIndex(long snapshotId, BinaryRow partition, int bucket) {
List<IndexFileMeta> result = scan(snapshotId, HASH_INDEX, partition, bucket);
if (result.size() > 1) {
@@ -88,9 +83,8 @@ public class IndexFileHandler {
public List<IndexFileMeta> scan(
long snapshotId, String indexType, BinaryRow partition, int bucket) {
- List<IndexManifestEntry> entries = scan(snapshotId, indexType, partition);
List<IndexFileMeta> result = new ArrayList<>();
- for (IndexManifestEntry file : entries) {
+ for (IndexManifestEntry file : scanEntries(snapshotId, indexType, partition)) {
if (file.bucket() == bucket) {
result.add(file.indexFile());
}
@@ -98,31 +92,54 @@ public class IndexFileHandler {
return result;
}
- public List<IndexManifestEntry> scan(String indexType, BinaryRow partition) {
+ public Map<Pair<BinaryRow, Integer>, List<IndexFileMeta>> scan(
+ long snapshotId, String indexType, Set<BinaryRow> partitions) {
+ Map<Pair<BinaryRow, Integer>, List<IndexFileMeta>> result = new HashMap<>();
+ for (IndexManifestEntry file : scanEntries(snapshotId, indexType, partitions)) {
+ result.computeIfAbsent(Pair.of(file.partition(), file.bucket()), k -> new ArrayList<>())
+ .add(file.indexFile());
+ }
+ return result;
+ }
+
+ public List<IndexManifestEntry> scanEntries() {
+ Snapshot snapshot = snapshotManager.latestSnapshot();
+ if (snapshot == null || snapshot.indexManifest() == null) {
+ return Collections.emptyList();
+ }
+
+ return indexManifestFile.read(snapshot.indexManifest());
+ }
+
+ public List<IndexManifestEntry> scanEntries(String indexType, BinaryRow partition) {
Long snapshot = snapshotManager.latestSnapshotId();
if (snapshot == null) {
return Collections.emptyList();
}
- return scan(snapshot, indexType, partition);
+ return scanEntries(snapshot, indexType, partition);
}
- public List<IndexManifestEntry> scan(long snapshotId, String indexType, BinaryRow partition) {
+ public List<IndexManifestEntry> scanEntries(
+ long snapshotId, String indexType, BinaryRow partition) {
+ return scanEntries(snapshotId, indexType, Collections.singleton(partition));
+ }
+
+ public List<IndexManifestEntry> scanEntries(
+ long snapshotId, String indexType, Set<BinaryRow> partitions) {
Snapshot snapshot = snapshotManager.snapshot(snapshotId);
String indexManifest = snapshot.indexManifest();
if (indexManifest == null) {
return Collections.emptyList();
}
- List<IndexManifestEntry> allFiles = indexManifestFile.read(indexManifest);
List<IndexManifestEntry> result = new ArrayList<>();
- for (IndexManifestEntry file : allFiles) {
+ for (IndexManifestEntry file : indexManifestFile.read(indexManifest)) {
if (file.indexFile().indexType().equals(indexType)
- && file.partition().equals(partition)) {
+ && partitions.contains(file.partition())) {
result.add(file);
}
}
-
return result;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/PartitionIndex.java b/paimon-core/src/main/java/org/apache/paimon/index/PartitionIndex.java
index 79ff72656..bace2c1ac 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/PartitionIndex.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/PartitionIndex.java
@@ -111,7 +111,7 @@ public class PartitionIndex {
long targetBucketRowNumber,
IntPredicate loadFilter,
IntPredicate bucketFilter) {
- List<IndexManifestEntry> files = indexFileHandler.scan(HASH_INDEX, partition);
+ List<IndexManifestEntry> files = indexFileHandler.scanEntries(HASH_INDEX, partition);
Int2ShortHashMap.Builder mapBuilder = Int2ShortHashMap.builder();
Map<Integer, Long> buckets = new HashMap<>();
for (IndexManifestEntry file : files) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
index 6e27a2480..bd9e48e4c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
@@ -270,6 +270,12 @@ public class SnapshotReaderImpl implements SnapshotReader {
SplitGenerator splitGenerator,
Map<BinaryRow, Map<Integer, List<DataFileMeta>>> groupedDataFiles) {
List<DataSplit> splits = new ArrayList<>();
+ // Read deletion indexes at once to reduce file IO
+ Map<Pair<BinaryRow, Integer>, List<IndexFileMeta>> deletionIndexFilesMap =
+ deletionVectors
+ ? indexFileHandler.scan(
+ snapshotId, DELETION_VECTORS_INDEX, groupedDataFiles.keySet())
+ : Collections.emptyMap();
for (Map.Entry<BinaryRow, Map<Integer, List<DataFileMeta>>> entry :
groupedDataFiles.entrySet()) {
BinaryRow partition = entry.getKey();
@@ -287,12 +293,9 @@ public class SnapshotReaderImpl implements SnapshotReader {
isStreaming
? splitGenerator.splitForStreaming(bucketFiles)
: splitGenerator.splitForBatch(bucketFiles);
-
List<IndexFileMeta> deletionIndexFiles =
- deletionVectors
- ? indexFileHandler.scan(
- snapshotId, DELETION_VECTORS_INDEX, partition, bucket)
- : Collections.emptyList();
+ deletionIndexFilesMap.getOrDefault(
+ Pair.of(partition, bucket), Collections.emptyList());
for (SplitGenerator.SplitGroup splitGroup : splitGroups) {
List<DataFileMeta> dataFiles = splitGroup.files;
String bucketPath = pathFactory.bucketPath(partition, bucket).toString();
@@ -350,6 +353,17 @@ public class SnapshotReaderImpl implements SnapshotReader {
(part, bucketMap) ->
buckets.computeIfAbsent(part, k -> new HashSet<>())
.addAll(bucketMap.keySet()));
+ // Read deletion indexes at once to reduce file IO
+ Map<Pair<BinaryRow, Integer>, List<IndexFileMeta>> beforDeletionIndexFilesMap =
+ deletionVectors
+ ? indexFileHandler.scan(
+ beforeSnapshotId, DELETION_VECTORS_INDEX, beforeFiles.keySet())
+ : Collections.emptyMap();
+ Map<Pair<BinaryRow, Integer>, List<IndexFileMeta>> deletionIndexFilesMap =
+ deletionVectors
+ ? indexFileHandler.scan(
+ plan.snapshotId(), DELETION_VECTORS_INDEX, dataFiles.keySet())
+ : Collections.emptyMap();
for (Map.Entry<BinaryRow, Set<Integer>> entry : buckets.entrySet()) {
BinaryRow part = entry.getKey();
@@ -376,15 +390,16 @@ public class SnapshotReaderImpl implements SnapshotReader {
.isStreaming(isStreaming)
.withBucketPath(pathFactory.bucketPath(part, bucket).toString());
if (deletionVectors) {
- List<IndexFileMeta> beforeDeletionIndexes =
- indexFileHandler.scan(
- beforeSnapshotId, DELETION_VECTORS_INDEX, part, bucket);
- List<IndexFileMeta> deletionIndexes =
- indexFileHandler.scan(
- plan.snapshotId(), DELETION_VECTORS_INDEX, part, bucket);
builder.withBeforeDeletionFiles(
- getDeletionFiles(before, beforeDeletionIndexes));
- builder.withDataDeletionFiles(getDeletionFiles(data, deletionIndexes));
+ getDeletionFiles(
+ before,
+ beforDeletionIndexFilesMap.getOrDefault(
+ Pair.of(part, bucket), Collections.emptyList())));
+ builder.withDataDeletionFiles(
+ getDeletionFiles(
+ data,
+ deletionIndexFilesMap.getOrDefault(
+ Pair.of(part, bucket), Collections.emptyList())));
}
splits.add(builder.build());
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index 695b023aa..a60554db2 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -725,7 +725,7 @@ public class FileStoreCommitTest {
// assert part1
List<IndexManifestEntry> part1Index =
- indexFileHandler.scan(snapshot.id(), HASH_INDEX, part1);
+ indexFileHandler.scanEntries(snapshot.id(), HASH_INDEX, part1);
assertThat(part1Index.size()).isEqualTo(2);
IndexManifestEntry indexManifestEntry =
@@ -740,7 +740,7 @@ public class FileStoreCommitTest {
// assert part2
List<IndexManifestEntry> part2Index =
- indexFileHandler.scan(snapshot.id(), HASH_INDEX, part2);
+ indexFileHandler.scanEntries(snapshot.id(), HASH_INDEX, part2);
assertThat(part2Index.size()).isEqualTo(1);
assertThat(part2Index.get(0).bucket()).isEqualTo(2);
assertThat(indexFileHandler.readHashIndexList(part2Index.get(0).indexFile()))
@@ -752,7 +752,7 @@ public class FileStoreCommitTest {
snapshot = store.snapshotManager().latestSnapshot();
// assert update part1
- part1Index = indexFileHandler.scan(snapshot.id(), HASH_INDEX, part1);
+ part1Index = indexFileHandler.scanEntries(snapshot.id(), HASH_INDEX, part1);
assertThat(part1Index.size()).isEqualTo(2);
indexManifestEntry =
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DynamicBucketTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DynamicBucketTableITCase.java
index bb27851f1..30506495f 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DynamicBucketTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DynamicBucketTableITCase.java
@@ -122,7 +122,7 @@ public class DynamicBucketTableITCase extends CatalogITCaseBase {
IndexFileHandler indexFileHandler = table.store().newIndexFileHandler();
List<BinaryRow> partitions = table.newScan().listPartitions();
List<IndexManifestEntry> entries = new ArrayList<>();
- partitions.forEach(p -> entries.addAll(indexFileHandler.scan(HASH_INDEX, p)));
+ partitions.forEach(p -> entries.addAll(indexFileHandler.scanEntries(HASH_INDEX, p)));
Long records =
entries.stream().map(entry -> entry.indexFile().rowCount()).reduce(Long::sum).get();