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:54:18 UTC
[incubator-paimon] branch master updated: [flink] Avoid unexpected Number of L0 File (#1351)
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/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b13791d20 [flink] Avoid unexpected Number of L0 File (#1351)
b13791d20 is described below
commit b13791d20c74310c5e268235df46b3d3e3b33028
Author: wgcn <10...@qq.com>
AuthorDate: Tue Jun 20 10:54:14 2023 +0800
[flink] Avoid unexpected Number of L0 File (#1351)
---
.../paimon/append/AppendOnlyCompactManager.java | 7 +-
.../paimon/compact/CompactFutureManager.java | 12 +-
.../org/apache/paimon/compact/CompactManager.java | 4 +-
.../apache/paimon/compact/NoopCompactManager.java | 7 +-
.../apache/paimon/mergetree/MergeTreeWriter.java | 7 +-
.../mergetree/compact/MergeTreeCompactManager.java | 7 +-
.../apache/paimon/mergetree/MergeTreeTestBase.java | 131 ++++++++++++++++++++-
7 files changed, 164 insertions(+), 11 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyCompactManager.java b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyCompactManager.java
index b598ca181..cc05d65db 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyCompactManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyCompactManager.java
@@ -103,7 +103,12 @@ public class AppendOnlyCompactManager extends CompactFutureManager {
}
@Override
- public boolean shouldWaitCompaction() {
+ public boolean shouldWaitForLatestCompaction() {
+ return false;
+ }
+
+ @Override
+ public boolean shouldWaitForPreparingCheckpoint() {
return false;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/compact/CompactFutureManager.java b/paimon-core/src/main/java/org/apache/paimon/compact/CompactFutureManager.java
index 3a2a1a199..2de966ac5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/compact/CompactFutureManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/compact/CompactFutureManager.java
@@ -18,6 +18,8 @@
package org.apache.paimon.compact;
+import org.apache.paimon.annotation.VisibleForTesting;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -48,15 +50,21 @@ public abstract class CompactFutureManager implements CompactManager {
if (blocking || taskFuture.isDone()) {
CompactResult result;
try {
- result = taskFuture.get();
+ result = obtainCompactResult();
} catch (CancellationException e) {
taskFuture = null;
return Optional.empty();
+ } finally {
+ taskFuture = null;
}
- taskFuture = null;
return Optional.of(result);
}
}
return Optional.empty();
}
+
+ @VisibleForTesting
+ protected CompactResult obtainCompactResult() throws InterruptedException, ExecutionException {
+ return taskFuture.get();
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/compact/CompactManager.java b/paimon-core/src/main/java/org/apache/paimon/compact/CompactManager.java
index c9f3d6f59..c8c89d3c3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/compact/CompactManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/compact/CompactManager.java
@@ -29,7 +29,9 @@ import java.util.concurrent.ExecutionException;
public interface CompactManager extends Closeable {
/** Should wait compaction finish. */
- boolean shouldWaitCompaction();
+ boolean shouldWaitForLatestCompaction();
+
+ boolean shouldWaitForPreparingCheckpoint();
/** Add a new file. */
void addNewFile(DataFileMeta file);
diff --git a/paimon-core/src/main/java/org/apache/paimon/compact/NoopCompactManager.java b/paimon-core/src/main/java/org/apache/paimon/compact/NoopCompactManager.java
index db3ae9595..bf1b2c79c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/compact/NoopCompactManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/compact/NoopCompactManager.java
@@ -34,7 +34,12 @@ public class NoopCompactManager implements CompactManager {
public NoopCompactManager() {}
@Override
- public boolean shouldWaitCompaction() {
+ public boolean shouldWaitForLatestCompaction() {
+ return false;
+ }
+
+ @Override
+ public boolean shouldWaitForPreparingCheckpoint() {
return false;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
index 443ba2f50..dfe22f2fe 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
@@ -184,7 +184,7 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
private void flushWriteBuffer(boolean waitForLatestCompaction, boolean forcedFullCompaction)
throws Exception {
if (writeBuffer.size() > 0) {
- if (compactManager.shouldWaitCompaction()) {
+ if (compactManager.shouldWaitForLatestCompaction()) {
waitForLatestCompaction = true;
}
@@ -227,7 +227,10 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
@Override
public CommitIncrement prepareCommit(boolean waitCompaction) throws Exception {
flushWriteBuffer(waitCompaction, false);
- trySyncLatestCompaction(waitCompaction || commitForceCompact);
+ trySyncLatestCompaction(
+ waitCompaction
+ || commitForceCompact
+ || compactManager.shouldWaitForPreparingCheckpoint());
return drainIncrement();
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
index 8fd41ce4d..c49ca0764 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
@@ -71,10 +71,15 @@ public class MergeTreeCompactManager extends CompactFutureManager {
}
@Override
- public boolean shouldWaitCompaction() {
+ public boolean shouldWaitForLatestCompaction() {
return levels.numberOfSortedRuns() > numSortedRunStopTrigger;
}
+ @Override
+ public boolean shouldWaitForPreparingCheckpoint() {
+ return levels.numberOfSortedRuns() > numSortedRunStopTrigger + 1;
+ }
+
@Override
public void addNewFile(DataFileMeta file) {
levels.addLevel0File(file);
diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java
index 5538928e9..7cc4c79c7 100644
--- a/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java
+++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java
@@ -37,6 +37,7 @@ import org.apache.paimon.io.KeyValueFileWriterFactory;
import org.apache.paimon.io.RollingFileWriter;
import org.apache.paimon.memory.HeapMemorySegmentPool;
import org.apache.paimon.mergetree.compact.AbstractCompactRewriter;
+import org.apache.paimon.mergetree.compact.CompactRewriter;
import org.apache.paimon.mergetree.compact.CompactStrategy;
import org.apache.paimon.mergetree.compact.DeduplicateMergeFunction;
import org.apache.paimon.mergetree.compact.IntervalPartition;
@@ -55,10 +56,12 @@ import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.CommitIncrement;
+import org.apache.paimon.utils.ExceptionUtils;
import org.apache.paimon.utils.FileStorePathFactory;
-import org.apache.paimon.utils.RecordWriter;
+import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -78,6 +81,7 @@ import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
@@ -99,7 +103,7 @@ public abstract class MergeTreeTestBase {
private KeyValueFileReaderFactory compactReaderFactory;
private KeyValueFileWriterFactory writerFactory;
private KeyValueFileWriterFactory compactWriterFactory;
- private RecordWriter<KeyValue> writer;
+ private MergeTreeWriter writer;
private SortEngine sortEngine;
@BeforeEach
@@ -240,6 +244,97 @@ public abstract class MergeTreeTestBase {
assertRecords(expected);
}
+ @Test
+ public void testPrepareCommitWaitCompaction() throws Exception {
+
+ List<DataFileMeta> newFiles = generateDataFileToCommit();
+
+ writer = createMergeTreeWriter(newFiles);
+ writeBatch(2);
+ CommitIncrement increment = writer.prepareCommit(false);
+
+ Assertions.assertEquals(1, increment.newFilesIncrement().newFiles().size());
+
+ Assertions.assertEquals(11, increment.compactIncrement().compactBefore().size());
+ Assertions.assertEquals(1, increment.compactIncrement().compactAfter().size());
+ }
+
+ @Nullable
+ private List<DataFileMeta> generateDataFileToCommit() throws Exception {
+ List<DataFileMeta> newFiles = new ArrayList<>();
+
+ // append successsfully but compaction fail
+ for (int i = 0; i < 10; i++) {
+ List<DataFileMeta> dataFileMetas = new ArrayList<>();
+ MergeTreeCompactManager compactManager =
+ createCompactManager(service, new ArrayList<>());
+ writer = createMergeTreeWriter(dataFileMetas, compactManager);
+ writeBatch(200);
+ newFiles.addAll(writer.dataFiles());
+ }
+ return newFiles;
+ }
+
+ @Test
+ public void testPrepareCommitRecycleReference() throws Exception {
+
+ List<DataFileMeta> dataFileMetas = generateDataFileToCommit();
+
+ MockFailResultCompactionManager mockFailResultCompactionManager =
+ new MockFailResultCompactionManager(
+ service,
+ new Levels(comparator, dataFileMetas, options.numLevels()),
+ new UniversalCompaction(
+ options.maxSizeAmplificationPercent(),
+ options.sortedRunSizeRatio(),
+ options.numSortedRunCompactionTrigger(),
+ options.maxSortedRunNum()),
+ comparator,
+ options.targetFileSize(),
+ options.numSortedRunStopTrigger(),
+ new TestRewriter());
+ writer = createMergeTreeWriter(dataFileMetas, mockFailResultCompactionManager);
+
+ writeBatch(2);
+
+ // When PrepareCommit receives an exception, trigger the close method to simulate the
+ // lifecycle of Flink Task.
+ RunnableWithException computeEngineCall =
+ () -> {
+ try {
+ try {
+ writer.prepareCommit(false);
+ } catch (Throwable e) {
+ throw new IOException("mock", e);
+ }
+ } catch (Throwable e) {
+ try {
+ writer.close();
+ } catch (Throwable ex) {
+ e.addSuppressed(ex);
+ }
+ throw e;
+ }
+ };
+
+ Throwable ex = null;
+ try {
+ computeEngineCall.run();
+ } catch (Throwable e) {
+ ex = e;
+ }
+
+ Assertions.assertNotNull(ex);
+ Assertions.assertTrue(
+ ExceptionUtils.findThrowable(ex, ExecutionException.class).isPresent());
+ Assertions.assertFalse(
+ ExceptionUtils.stringifyException(ex).contains("CIRCULAR REFERENCE"));
+ }
+
+ interface RunnableWithException {
+ void run() throws Exception;
+ }
+
@ParameterizedTest
@ValueSource(longs = {1, 1024 * 1024})
public void testCloseUpgrade(long targetFileSize) throws Exception {
@@ -322,6 +417,11 @@ public abstract class MergeTreeTestBase {
}
private MergeTreeWriter createMergeTreeWriter(List<DataFileMeta> files) {
+ return createMergeTreeWriter(files, createCompactManager(service, files));
+ }
+
+ private MergeTreeWriter createMergeTreeWriter(
+ List<DataFileMeta> files, MergeTreeCompactManager compactManager) {
long maxSequenceNumber =
files.stream().map(DataFileMeta::maxSequenceNumber).max(Long::compare).orElse(-1L);
MergeTreeWriter writer =
@@ -329,7 +429,7 @@ public abstract class MergeTreeTestBase {
false,
128,
null,
- createCompactManager(service, files),
+ compactManager,
maxSequenceNumber,
comparator,
DeduplicateMergeFunction.factory().create(),
@@ -360,6 +460,31 @@ public abstract class MergeTreeTestBase {
new TestRewriter());
}
+ static class MockFailResultCompactionManager extends MergeTreeCompactManager {
+ public MockFailResultCompactionManager(
+ ExecutorService executor,
+ Levels levels,
+ CompactStrategy strategy,
+ Comparator<InternalRow> keyComparator,
+ long minFileSize,
+ int numSortedRunStopTrigger,
+ CompactRewriter rewriter) {
+ super(
+ executor,
+ levels,
+ strategy,
+ keyComparator,
+ minFileSize,
+ numSortedRunStopTrigger,
+ rewriter);
+ }
+
+ protected CompactResult obtainCompactResult() throws ExecutionException {
+ OutOfMemoryError outOfMemoryError = new OutOfMemoryError();
+ throw new ExecutionException("mock", outOfMemoryError);
+ }
+ }
+
private void mergeCompacted(
Set<String> newFileNames,
List<DataFileMeta> compactedFiles,