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/06 02:19:10 UTC
(paimon) branch release-0.8 updated: [core] Read deletion indexes at once to reduce file IO in splits generation
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/release-0.8 by this push:
new 237a53071 [core] Read deletion indexes at once to reduce file IO in splits generation
237a53071 is described below
commit 237a5307127319c6b8395cf397d34d6da715b9ba
Author: Jingsong <ji...@gmail.com>
AuthorDate: Sat Jul 6 10:18:55 2024 +0800
[core] Read deletion indexes at once to reduce file IO in splits generation
---
.../org/apache/paimon/index/IndexFileHandler.java | 24 ++++++++++++
.../table/source/snapshot/SnapshotReaderImpl.java | 45 ++++++++++++++++------
2 files changed, 58 insertions(+), 11 deletions(-)
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 fdc2fe754..417d0205d 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
@@ -34,10 +34,12 @@ import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.LinkedHashMap;
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;
@@ -108,6 +110,28 @@ public class IndexFileHandler {
return result;
}
+ public Map<Pair<BinaryRow, Integer>, List<IndexFileMeta>> scanPartitions(
+ long snapshotId, String indexType, Set<BinaryRow> partitions) {
+ Map<Pair<BinaryRow, Integer>, List<IndexFileMeta>> result = new HashMap<>();
+ Snapshot snapshot = snapshotManager.snapshot(snapshotId);
+ String indexManifest = snapshot.indexManifest();
+ if (indexManifest == null) {
+ return Collections.emptyMap();
+ }
+
+ List<IndexManifestEntry> allFiles = indexManifestFile.read(indexManifest);
+ for (IndexManifestEntry file : allFiles) {
+ if (file.indexFile().indexType().equals(indexType)
+ && partitions.contains(file.partition())) {
+ result.computeIfAbsent(
+ Pair.of(file.partition(), file.bucket()), k -> new ArrayList<>())
+ .add(file.indexFile());
+ }
+ }
+
+ return result;
+ }
+
public Path filePath(IndexFileMeta file) {
return pathFactory.toPath(file.fileName());
}
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 aae2ccfc9..f5da6a9fa 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
@@ -253,6 +253,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.scanPartitions(
+ snapshotId, DELETION_VECTORS_INDEX, groupedDataFiles.keySet())
+ : Collections.emptyMap();
for (Map.Entry<BinaryRow, Map<Integer, List<DataFileMeta>>> entry :
groupedDataFiles.entrySet()) {
BinaryRow partition = entry.getKey();
@@ -271,12 +277,15 @@ public class SnapshotReaderImpl implements SnapshotReader {
? splitGenerator.splitForStreaming(bucketFiles)
: splitGenerator.splitForBatch(bucketFiles);
+ List<IndexFileMeta> deletionIndexFiles =
+ deletionIndexFilesMap.getOrDefault(
+ Pair.of(partition, bucket), Collections.emptyList());
+ if (deletionIndexFiles.size() > 1) {
+ throw new IllegalArgumentException(
+ "Find multiple index files for one bucket: " + deletionIndexFiles);
+ }
IndexFileMeta deletionIndexFile =
- deletionVectors
- ? indexFileHandler
- .scan(snapshotId, DELETION_VECTORS_INDEX, partition, bucket)
- .orElse(null)
- : null;
+ deletionIndexFiles.isEmpty() ? null : deletionIndexFiles.get(0);
for (SplitGenerator.SplitGroup splitGroup : splitGroups) {
List<DataFileMeta> dataFiles = splitGroup.files;
String bucketPath = pathFactory.bucketPath(partition, bucket).toString();
@@ -335,6 +344,18 @@ public class SnapshotReaderImpl implements SnapshotReader {
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.scanPartitions(
+ beforeSnapshotId, DELETION_VECTORS_INDEX, beforeFiles.keySet())
+ : Collections.emptyMap();
+ Map<Pair<BinaryRow, Integer>, List<IndexFileMeta>> deletionIndexFilesMap =
+ deletionVectors
+ ? indexFileHandler.scanPartitions(
+ plan.snapshotId(), DELETION_VECTORS_INDEX, dataFiles.keySet())
+ : Collections.emptyMap();
+
for (Map.Entry<BinaryRow, Set<Integer>> entry : buckets.entrySet()) {
BinaryRow part = entry.getKey();
for (Integer bucket : entry.getValue()) {
@@ -360,14 +381,16 @@ public class SnapshotReaderImpl implements SnapshotReader {
.isStreaming(isStreaming)
.withBucketPath(pathFactory.bucketPath(part, bucket).toString());
if (deletionVectors) {
+ List<IndexFileMeta> beforeDeletionIndexes =
+ beforDeletionIndexFilesMap.getOrDefault(
+ Pair.of(part, bucket), Collections.emptyList());
IndexFileMeta beforeDeletionIndex =
- indexFileHandler
- .scan(beforeSnapshotId, DELETION_VECTORS_INDEX, part, bucket)
- .orElse(null);
+ beforeDeletionIndexes.isEmpty() ? null : beforeDeletionIndexes.get(0);
+ List<IndexFileMeta> deletionIndexes =
+ deletionIndexFilesMap.getOrDefault(
+ Pair.of(part, bucket), Collections.emptyList());
IndexFileMeta deletionIndex =
- indexFileHandler
- .scan(plan.snapshotId(), DELETION_VECTORS_INDEX, part, bucket)
- .orElse(null);
+ deletionIndexes.isEmpty() ? null : deletionIndexes.get(0);
builder.withBeforeDeletionFiles(getDeletionFiles(before, beforeDeletionIndex));
builder.withDataDeletionFiles(getDeletionFiles(data, deletionIndex));
}