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 2023/06/20 02:36:30 UTC
[incubator-paimon] 07/09: [core] Skip more unwanted merges for Manifest full compaction (#1268)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.4
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit 4473585c3eb2008775d26f7c4fc9890f4eddab98
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Wed May 31 16:58:10 2023 +0800
[core] Skip more unwanted merges for Manifest full compaction (#1268)
---
.../org/apache/paimon/manifest/ManifestEntry.java | 8 +-
.../apache/paimon/manifest/ManifestFileMeta.java | 169 ++++++++++++++-------
.../paimon/manifest/ManifestFileMetaTest.java | 46 +++---
.../manifest/NoPartitionManifestFileMetaTest.java | 9 +-
4 files changed, 149 insertions(+), 83 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntry.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntry.java
index 84af10819..3f2927315 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntry.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntry.java
@@ -168,6 +168,9 @@ public class ManifestEntry {
public final int level;
public final String fileName;
+ /* Cache the hash code for the string */
+ private Integer hash;
+
private Identifier(BinaryRow partition, int bucket, int level, String fileName) {
this.partition = partition;
this.bucket = bucket;
@@ -189,7 +192,10 @@ public class ManifestEntry {
@Override
public int hashCode() {
- return Objects.hash(partition, bucket, level, fileName);
+ if (hash == null) {
+ hash = Objects.hash(partition, bucket, level, fileName);
+ }
+ return hash;
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java
index 89f74f1b2..e4e7f707b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java
@@ -19,6 +19,7 @@
package org.apache.paimon.manifest;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.manifest.ManifestEntry.Identifier;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.stats.BinaryTableStats;
@@ -44,7 +45,8 @@ import java.util.stream.Collectors;
/** Metadata of a manifest file. */
public class ManifestFileMeta {
- private static final Logger LOGGER = LoggerFactory.getLogger(ManifestFileMeta.class);
+
+ private static final Logger LOG = LoggerFactory.getLogger(ManifestFileMeta.class);
private final String fileName;
private final long fileSize;
@@ -137,45 +139,32 @@ public class ManifestFileMeta {
* <p>NOTE: This method is atomic.
*/
public static List<ManifestFileMeta> merge(
- List<ManifestFileMeta> metas,
+ List<ManifestFileMeta> input,
ManifestFile manifestFile,
long suggestedMetaSize,
int suggestedMinMetaCount,
long manifestFullCompactionSize,
RowType partitionType) {
- List<ManifestFileMeta> result = new ArrayList<>();
// these are the newly created manifest files, clean them up if exception occurs
List<ManifestFileMeta> newMetas = new ArrayList<>();
- List<ManifestFileMeta> candidates = new ArrayList<>();
- long totalSize = 0;
try {
- List<ManifestFileMeta> manifestFileMetas =
+ Optional<List<ManifestFileMeta>> fullCompacted =
tryFullCompaction(
- metas,
+ input,
newMetas,
manifestFile,
- partitionType,
suggestedMetaSize,
- manifestFullCompactionSize);
- // merge existing manifests first
- for (ManifestFileMeta manifest : manifestFileMetas) {
- totalSize += manifest.fileSize;
- candidates.add(manifest);
- if (totalSize >= suggestedMetaSize) {
- // reach suggested file size, perform merging and produce new file
- mergeCandidates(candidates, manifestFile, result, newMetas);
- candidates.clear();
- totalSize = 0;
- }
- }
-
- // merge the last bit of manifests if there are too many
- if (candidates.size() >= suggestedMinMetaCount) {
- mergeCandidates(candidates, manifestFile, result, newMetas);
- } else {
- result.addAll(candidates);
- }
+ manifestFullCompactionSize,
+ partitionType);
+ return fullCompacted.orElseGet(
+ () ->
+ tryMinorCompaction(
+ input,
+ newMetas,
+ manifestFile,
+ suggestedMetaSize,
+ suggestedMinMetaCount));
} catch (Throwable e) {
// exception occurs, clean up and rethrow
for (ManifestFileMeta manifest : newMetas) {
@@ -183,7 +172,35 @@ public class ManifestFileMeta {
}
throw e;
}
+ }
+ private static List<ManifestFileMeta> tryMinorCompaction(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newMetas,
+ ManifestFile manifestFile,
+ long suggestedMetaSize,
+ int suggestedMinMetaCount) {
+ List<ManifestFileMeta> result = new ArrayList<>();
+ List<ManifestFileMeta> candidates = new ArrayList<>();
+ long totalSize = 0;
+ // merge existing small manifest files
+ for (ManifestFileMeta manifest : input) {
+ totalSize += manifest.fileSize;
+ candidates.add(manifest);
+ if (totalSize >= suggestedMetaSize) {
+ // reach suggested file size, perform merging and produce new file
+ mergeCandidates(candidates, manifestFile, result, newMetas);
+ candidates.clear();
+ totalSize = 0;
+ }
+ }
+
+ // merge the last bit of manifests if there are too many
+ if (candidates.size() >= suggestedMinMetaCount) {
+ mergeCandidates(candidates, manifestFile, result, newMetas);
+ } else {
+ result.addAll(candidates);
+ }
return result;
}
@@ -197,7 +214,7 @@ public class ManifestFileMeta {
return;
}
- Map<ManifestEntry.Identifier, ManifestEntry> map = new LinkedHashMap<>();
+ Map<Identifier, ManifestEntry> map = new LinkedHashMap<>();
for (ManifestFileMeta manifest : candidates) {
ManifestEntry.mergeEntries(manifestFile.read(manifest.fileName), map);
}
@@ -208,51 +225,53 @@ public class ManifestFileMeta {
}
}
- public static List<ManifestFileMeta> tryFullCompaction(
+ public static Optional<List<ManifestFileMeta>> tryFullCompaction(
List<ManifestFileMeta> inputs,
List<ManifestFileMeta> newMetas,
ManifestFile manifestFile,
- RowType partitionType,
long suggestedMetaSize,
- long sizeTrigger) {
+ long sizeTrigger,
+ RowType partitionType) {
+ // 1. should trigger full compaction
+
List<ManifestFileMeta> base = new ArrayList<>();
int totalManifestSize = 0;
- int i;
- for (i = 0; i < inputs.size(); i++) {
- ManifestFileMeta manifestFileMeta = inputs.get(i);
- if (manifestFileMeta.numDeletedFiles == 0
- && manifestFileMeta.fileSize >= suggestedMetaSize) {
- base.add(manifestFileMeta);
- totalManifestSize += manifestFileMeta.fileSize;
+ int i = 0;
+ for (; i < inputs.size(); i++) {
+ ManifestFileMeta file = inputs.get(i);
+ if (file.numDeletedFiles == 0 && file.fileSize >= suggestedMetaSize) {
+ base.add(file);
+ totalManifestSize += file.fileSize;
} else {
break;
}
}
List<ManifestFileMeta> delta = new ArrayList<>();
- for (; i < inputs.size(); i++) {
- ManifestFileMeta manifestFileMeta = inputs.get(i);
- delta.add(manifestFileMeta);
- totalManifestSize += manifestFileMeta.fileSize;
- }
-
long deltaDeleteFileNum = 0;
long totalDeltaFileSize = 0;
- for (ManifestFileMeta manifestFileMeta : delta) {
- deltaDeleteFileNum += manifestFileMeta.numDeletedFiles();
- totalDeltaFileSize += manifestFileMeta.fileSize();
+ for (; i < inputs.size(); i++) {
+ ManifestFileMeta file = inputs.get(i);
+ delta.add(file);
+ totalManifestSize += file.fileSize;
+ deltaDeleteFileNum += file.numDeletedFiles();
+ totalDeltaFileSize += file.fileSize();
}
if (totalDeltaFileSize < sizeTrigger) {
- return inputs;
+ return Optional.empty();
}
- LOGGER.info(
- "Start Manifest File Full Compaction,pick the number of delete file: {} ,total manifest file size: {}",
+ // 2. do full compaction
+
+ LOG.info(
+ "Start Manifest File Full Compaction, pick the number of delete file: {}, total manifest file size: {}",
deltaDeleteFileNum,
totalManifestSize);
- Map<ManifestEntry.Identifier, ManifestEntry> deltaMerged = new LinkedHashMap<>();
+ // 2.1. try to skip base files by partition filter
+
+ Map<Identifier, ManifestEntry> deltaMerged = new LinkedHashMap<>();
for (ManifestFileMeta manifest : delta) {
ManifestEntry.mergeEntries(manifestFile.read(manifest.fileName), deltaMerged);
}
@@ -269,7 +288,7 @@ public class ManifestFileMeta {
FieldStatsArraySerializer fieldStatsArraySerializer =
new FieldStatsArraySerializer(partitionType);
for (; j < base.size(); j++) {
- // todo optimize this to binary search.
+ // TODO: optimize this to binary search.
ManifestFileMeta file = base.get(j);
if (predicate.test(
file.numAddedFiles + file.numDeletedFiles,
@@ -280,19 +299,53 @@ public class ManifestFileMeta {
}
}
} else {
- // There is no DELETE Entry in Delta,Base don't need compaction
+ // There is no DELETE Entry in Delta, Base don't need compaction
j = base.size();
result.addAll(base);
}
}
- Map<ManifestEntry.Identifier, ManifestEntry> fullMerged = new LinkedHashMap<>();
+ // 2.2. try to skip base files by reading entries
+
+ Set<Identifier> deleteEntries = new HashSet<>();
+ deltaMerged.forEach(
+ (k, v) -> {
+ if (v.kind() == FileKind.DELETE) {
+ deleteEntries.add(k);
+ }
+ });
+
+ Map<Identifier, ManifestEntry> fullMerged = new LinkedHashMap<>();
+ for (; j < base.size(); j++) {
+ ManifestFileMeta file = base.get(j);
+ ManifestEntry.mergeEntries(manifestFile.read(file.fileName), fullMerged);
+ boolean contains = false;
+ for (Identifier identifier : deleteEntries) {
+ if (fullMerged.containsKey(identifier)) {
+ contains = true;
+ break;
+ }
+ }
+ if (contains) {
+ // already read this file into fullMerged
+ j++;
+ break;
+ } else {
+ fullMerged.clear();
+ result.add(file);
+ }
+ }
+
+ // 2.3. merge base files
+
for (; j < base.size(); j++) {
ManifestFileMeta manifestFileMeta = base.get(j);
ManifestEntry.mergeEntries(manifestFile.read(manifestFileMeta.fileName), fullMerged);
}
ManifestEntry.mergeEntries(deltaMerged.values(), fullMerged);
+ // 2.4. write new manifest files
+
if (!fullMerged.isEmpty()) {
List<ManifestFileMeta> merged =
manifestFile.write(new ArrayList<>(fullMerged.values()));
@@ -300,11 +353,11 @@ public class ManifestFileMeta {
newMetas.addAll(merged);
}
- return result;
+ return Optional.of(result);
}
private static Set<BinaryRow> computeDeletePartitions(
- Map<ManifestEntry.Identifier, ManifestEntry> deltaMerged) {
+ Map<Identifier, ManifestEntry> deltaMerged) {
Set<BinaryRow> partitions = new HashSet<>();
for (ManifestEntry manifestEntry : deltaMerged.values()) {
if (manifestEntry.kind() == FileKind.DELETE) {
@@ -319,12 +372,12 @@ public class ManifestFileMeta {
RowType partitionType, Set<BinaryRow> partitions) {
Optional<Predicate> predicateOpt;
if (!partitions.isEmpty()) {
- RowDataToObjectArrayConverter rowDataToObjectArrayConverter =
+ RowDataToObjectArrayConverter rowArrayConverter =
new RowDataToObjectArrayConverter(partitionType);
List<Predicate> predicateList =
partitions.stream()
- .map(rowDataToObjectArrayConverter::createEqualPredicate)
+ .map(rowArrayConverter::createEqualPredicate)
.collect(Collectors.toList());
predicateOpt = Optional.of(PredicateBuilder.or(predicateList));
} else {
diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
index 54452cb9b..7ff95be36 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
@@ -28,6 +28,7 @@ import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.RepeatedTest;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -37,6 +38,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
+import java.util.Optional;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
@@ -131,7 +133,7 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
testCleanUp(input, 0L);
}
- @RepeatedTest(3)
+ @Test
public void testMerge() {
List<ManifestFileMeta> input = createBaseManifestFileMetas(true);
// delta with delete apply parititon 1,2
@@ -147,7 +149,7 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
assertEquivalentEntries(input, merged);
}
- @RepeatedTest(3)
+ @Test
public void testMergeWithoutDelta() {
// base
@@ -172,7 +174,7 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
assertEquivalentEntries(input1, merged1);
}
- @RepeatedTest(3)
+ @Test
public void testMergeWithoutBase() {
List<ManifestFileMeta> input = new ArrayList<>();
addDeltaManifests(input, true);
@@ -181,7 +183,7 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
assertEquivalentEntries(input, merged);
}
- @RepeatedTest(3)
+ @Test
public void testMergeWithoutDeleteFile() {
// entries are All ADD.
List<ManifestFileMeta> input = new ArrayList<>();
@@ -207,18 +209,17 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
assertEquivalentEntries(input, merged);
}
- @RepeatedTest(3)
+ @Test
public void testTriggerFullCompaction() {
-
- List<ManifestEntry> entrys = new ArrayList<>();
+ List<ManifestEntry> entries = new ArrayList<>();
for (int i = 0; i < 16; i++) {
- entrys.add(makeEntry(true, String.valueOf(i)));
+ entries.add(makeEntry(true, String.valueOf(i)));
}
List<ManifestFileMeta> input = new ArrayList<>();
// base manifest
- input.add(makeManifest(entrys.toArray(new ManifestEntry[0])));
+ input.add(makeManifest(entries.toArray(new ManifestEntry[0])));
// delta manifest
input.add(makeManifest(makeEntry(true, "A"), makeEntry(true, "B"), makeEntry(true, "C")));
@@ -234,18 +235,18 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
// no trigger for delta size
List<ManifestFileMeta> newMetas2 = new ArrayList<>();
- List<ManifestFileMeta> manifestFileMetas =
+ Optional<List<ManifestFileMeta>> fullCompacted =
ManifestFileMeta.tryFullCompaction(
- input, newMetas2, manifestFile, getPartitionType(), 500, Long.MAX_VALUE);
- List<ManifestFileMeta> expected2 = new ArrayList<>(input);
- assertThat(manifestFileMetas).hasSameElementsAs(expected2);
+ input, newMetas2, manifestFile, 500, Long.MAX_VALUE, getPartitionType());
+ assertThat(fullCompacted).isEmpty();
assertThat(newMetas2).isEmpty();
// trigger full compaction
List<ManifestFileMeta> newMetas3 = new ArrayList<>();
- List<ManifestFileMeta> mergedMainfest =
+ List<ManifestFileMeta> merged =
ManifestFileMeta.tryFullCompaction(
- input, newMetas3, manifestFile, getPartitionType(), 500, 100);
+ input, newMetas3, manifestFile, 500, 100, getPartitionType())
+ .get();
List<String> entryFileNameExptected = new ArrayList<>();
entryFileNameExptected.add("ADD-G");
@@ -253,23 +254,24 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
entryFileNameExptected.add("ADD-" + i);
}
- containSameEntryFile(mergedMainfest, entryFileNameExptected);
+ containSameEntryFile(merged, entryFileNameExptected);
assertThat(newMetas3).isNotEmpty();
}
- @RepeatedTest(3)
- public void testMultiParitionFullCompaction() {
+ @Test
+ public void testMultiPartitionsFullCompaction() {
List<ManifestFileMeta> input = createBaseManifestFileMetas(true);
- // delta with delete apply parititon 1,2
+ // delta with delete apply partition 1,2
addDeltaManifests(input, true);
List<ManifestFileMeta> newMetas = new ArrayList<>();
- List<ManifestFileMeta> mergedMainfest =
+ List<ManifestFileMeta> mergedManifest =
ManifestFileMeta.tryFullCompaction(
- input, newMetas, manifestFile, getPartitionType(), 500, 100);
+ input, newMetas, manifestFile, 500, 100, getPartitionType())
+ .get();
List<String> expected = Lists.newArrayList("ADD-C2", "ADD-D2", "ADD-G");
HashMap<Integer, Integer> partitionAndFileLimit =
@@ -286,7 +288,7 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
}
}
- containSameEntryFile(mergedMainfest, expected);
+ containSameEntryFile(mergedManifest, expected);
}
private void createData(
diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/NoPartitionManifestFileMetaTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/NoPartitionManifestFileMetaTest.java
index 116de95e5..acdc78991 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/NoPartitionManifestFileMetaTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/NoPartitionManifestFileMetaTest.java
@@ -21,11 +21,13 @@ package org.apache.paimon.manifest;
import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.RepeatedTest;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.util.List;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** Test {@link ManifestFile}. for table without partition */
public class NoPartitionManifestFileMetaTest extends ManifestFileMetaTestBase {
private final RowType noPartitionType = RowType.of();
@@ -38,7 +40,7 @@ public class NoPartitionManifestFileMetaTest extends ManifestFileMetaTestBase {
manifestFile = createManifestFile(tempDir.toString());
}
- @RepeatedTest(3)
+ @Test
public void testMerge() {
List<ManifestFileMeta> input = createBaseManifestFileMetas(false);
addDeltaManifests(input, false);
@@ -46,6 +48,9 @@ public class NoPartitionManifestFileMetaTest extends ManifestFileMetaTestBase {
List<ManifestFileMeta> merged =
ManifestFileMeta.merge(input, manifestFile, 500, 3, 200, getPartitionType());
assertEquivalentEntries(input, merged);
+
+ // the first one is not deleted, it should not be merged
+ assertThat(merged.get(0)).isSameAs(input.get(0));
}
@Override