You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2022/10/27 11:14:54 UTC
[flink-table-store] branch master updated: [FLINK-29294] Introduce a CompactRewriter for full compaction
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/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new 08211f0e [FLINK-29294] Introduce a CompactRewriter for full compaction
08211f0e is described below
commit 08211f0e190ca8dcce288d93b72350a2b1185ff6
Author: tsreaper <ts...@gmail.com>
AuthorDate: Thu Oct 27 19:14:49 2022 +0800
[FLINK-29294] Introduce a CompactRewriter for full compaction
This closes #332
---
.../shortcodes/generated/core_configuration.html | 2 +-
.../file/mergetree/MergeTreeBenchmark.java | 42 ++++---
.../file/mergetree/MergeTreeReaderBenchmark.java | 4 +-
.../org/apache/flink/table/store/CoreOptions.java | 4 +-
.../table/store/file/append/AppendOnlyWriter.java | 34 ++++--
.../table/store/file/compact/CompactResult.java | 45 ++++++-
.../store/file/mergetree/DropDeleteReader.java | 77 ++++++++++++
...{MergeTreeReader.java => MergeTreeReaders.java} | 86 ++++---------
.../store/file/mergetree/MergeTreeWriter.java | 49 +++++---
...tRewriter.java => AbstractCompactRewriter.java} | 22 +++-
.../file/mergetree/compact/CompactRewriter.java | 6 +-
.../compact/FullChangelogMergeFunctionWrapper.java | 11 +-
.../FullChangelogMergeTreeCompactRewriter.java | 134 +++++++++++++++++++++
.../compact/MergeTreeCompactRewriter.java | 72 +++++++++++
.../mergetree/compact/MergeTreeCompactTask.java | 41 ++-----
.../file/mergetree/compact/SortMergeReader.java | 27 ++---
.../file/operation/AbstractFileStoreScan.java | 6 +
.../table/store/file/operation/FileStoreWrite.java | 3 +-
.../file/operation/KeyValueFileStoreRead.java | 74 +++++++-----
.../file/operation/KeyValueFileStoreWrite.java | 36 +++---
.../flink/table/store/file/utils/RecordWriter.java | 5 +-
.../table/store/file/mergetree/MergeTreeTest.java | 51 ++++----
.../FullChangelogMergeFunctionWrapperTestBase.java | 6 +-
.../compact/MergeTreeCompactManagerTest.java | 23 +++-
.../mergetree/compact/SortMergeReaderTestBase.java | 5 +-
.../store/file/operation/FileStoreCommitTest.java | 66 +++++++++-
.../store/file/operation/TestCommitThread.java | 31 +++++
.../table/ChangelogWithKeyFileStoreTableTest.java | 81 ++++++++++++-
28 files changed, 782 insertions(+), 261 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html
index d6200a94..8f630e02 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -30,7 +30,7 @@
<td><h5>changelog-producer</h5></td>
<td style="word-wrap: break-word;">none</td>
<td><p>Enum</p></td>
- <td>Whether to double write to a changelog file. This changelog file keeps the details of data changes, it can be read directly during stream reads.<br /><br />Possible values:<ul><li>"none": No changelog file.</li><li>"input": Double write to a changelog file when flushing memory table, the changelog is from input.</li></ul></td>
+ <td>Whether to double write to a changelog file. This changelog file keeps the details of data changes, it can be read directly during stream reads.<br /><br />Possible values:<ul><li>"none": No changelog file.</li><li>"input": Double write to a changelog file when flushing memory table, the changelog is from input.</li><li>"full-compaction": Generate changelog files with each full compaction.</li></ul></td>
</tr>
<tr>
<td><h5>commit.force-compact</h5></td>
diff --git a/flink-table-store-benchmark/flink-table-store-micro-benchmarks/src/main/java/org/apache/flink/table/store/benchmark/file/mergetree/MergeTreeBenchmark.java b/flink-table-store-benchmark/flink-table-store-micro-benchmarks/src/main/java/org/apache/flink/table/store/benchmark/file/mergetree/MergeTreeBenchmark.java
index dd9e17de..5c6f3be5 100644
--- a/flink-table-store-benchmark/flink-table-store-micro-benchmarks/src/main/java/org/apache/flink/table/store/benchmark/file/mergetree/MergeTreeBenchmark.java
+++ b/flink-table-store-benchmark/flink-table-store-micro-benchmarks/src/main/java/org/apache/flink/table/store/benchmark/file/mergetree/MergeTreeBenchmark.java
@@ -27,14 +27,17 @@ import org.apache.flink.table.store.CoreOptions;
import org.apache.flink.table.store.benchmark.config.ConfigUtil;
import org.apache.flink.table.store.benchmark.config.FileBenchmarkOptions;
import org.apache.flink.table.store.file.KeyValue;
+import org.apache.flink.table.store.file.compact.CompactResult;
import org.apache.flink.table.store.file.io.DataFileMeta;
import org.apache.flink.table.store.file.io.KeyValueFileReaderFactory;
import org.apache.flink.table.store.file.io.KeyValueFileWriterFactory;
import org.apache.flink.table.store.file.io.RollingFileWriter;
import org.apache.flink.table.store.file.memory.HeapMemorySegmentPool;
import org.apache.flink.table.store.file.mergetree.Levels;
-import org.apache.flink.table.store.file.mergetree.MergeTreeReader;
+import org.apache.flink.table.store.file.mergetree.MergeTreeReaders;
import org.apache.flink.table.store.file.mergetree.MergeTreeWriter;
+import org.apache.flink.table.store.file.mergetree.SortedRun;
+import org.apache.flink.table.store.file.mergetree.compact.AbstractCompactRewriter;
import org.apache.flink.table.store.file.mergetree.compact.CompactRewriter;
import org.apache.flink.table.store.file.mergetree.compact.CompactStrategy;
import org.apache.flink.table.store.file.mergetree.compact.DeduplicateMergeFunction;
@@ -182,21 +185,7 @@ public class MergeTreeBenchmark {
options.sortedRunSizeRatio(),
options.numSortedRunCompactionTrigger(),
options.maxSortedRunNum());
- CompactRewriter rewriter =
- (outputLevel, dropDelete, sections) -> {
- RollingFileWriter<KeyValue, DataFileMeta> writer =
- compactWriterFactory.createRollingMergeTreeFileWriter(outputLevel);
- writer.write(
- new RecordReaderIterator<>(
- new MergeTreeReader(
- sections,
- dropDelete,
- compactReaderFactory,
- comparator,
- new DeduplicateMergeFunction())));
- writer.close();
- return writer.result();
- };
+ CompactRewriter rewriter = new TestCompactRewriter();
return new MergeTreeCompactManager(
compactExecutor,
new Levels(comparator, files, options.numLevels()),
@@ -266,4 +255,25 @@ public class MergeTreeBenchmark {
value = null;
}
}
+
+ private class TestCompactRewriter extends AbstractCompactRewriter {
+
+ @Override
+ public CompactResult rewrite(
+ int outputLevel, boolean dropDelete, List<List<SortedRun>> sections)
+ throws Exception {
+ RollingFileWriter<KeyValue, DataFileMeta> writer =
+ compactWriterFactory.createRollingMergeTreeFileWriter(outputLevel);
+ writer.write(
+ new RecordReaderIterator<>(
+ MergeTreeReaders.readerForMergeTree(
+ sections,
+ dropDelete,
+ compactReaderFactory,
+ comparator,
+ new DeduplicateMergeFunction())));
+ writer.close();
+ return new CompactResult(extractFilesFromSections(sections), writer.result());
+ }
+ }
}
diff --git a/flink-table-store-benchmark/flink-table-store-micro-benchmarks/src/main/java/org/apache/flink/table/store/benchmark/file/mergetree/MergeTreeReaderBenchmark.java b/flink-table-store-benchmark/flink-table-store-micro-benchmarks/src/main/java/org/apache/flink/table/store/benchmark/file/mergetree/MergeTreeReaderBenchmark.java
index 62dd8646..80ef6e7c 100644
--- a/flink-table-store-benchmark/flink-table-store-micro-benchmarks/src/main/java/org/apache/flink/table/store/benchmark/file/mergetree/MergeTreeReaderBenchmark.java
+++ b/flink-table-store-benchmark/flink-table-store-micro-benchmarks/src/main/java/org/apache/flink/table/store/benchmark/file/mergetree/MergeTreeReaderBenchmark.java
@@ -20,7 +20,7 @@ package org.apache.flink.table.store.benchmark.file.mergetree;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.store.file.KeyValue;
-import org.apache.flink.table.store.file.mergetree.MergeTreeReader;
+import org.apache.flink.table.store.file.mergetree.MergeTreeReaders;
import org.apache.flink.table.store.file.mergetree.compact.DeduplicateMergeFunction;
import org.apache.flink.table.store.file.mergetree.compact.IntervalPartition;
import org.apache.flink.table.store.file.utils.RecordReader;
@@ -95,7 +95,7 @@ public class MergeTreeReaderBenchmark extends MergeTreeBenchmark {
@BenchmarkMode(Mode.SingleShotTime)
public long scanAll() throws Exception {
try (RecordReader<KeyValue> reader =
- new MergeTreeReader(
+ MergeTreeReaders.readerForMergeTree(
new IntervalPartition(compactedFiles, comparator).partition(),
true,
readerFactory,
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java
index 45360b54..4bedffd8 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java
@@ -648,7 +648,9 @@ public class CoreOptions implements Serializable {
INPUT(
"input",
- "Double write to a changelog file when flushing memory table, the changelog is from input.");
+ "Double write to a changelog file when flushing memory table, the changelog is from input."),
+
+ FULL_COMPACTION("full-compaction", "Generate changelog files with each full compaction.");
private final String value;
private final String description;
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/append/AppendOnlyWriter.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/append/AppendOnlyWriter.java
index 8c6c9635..5dedfc0e 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/append/AppendOnlyWriter.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/append/AppendOnlyWriter.java
@@ -53,6 +53,7 @@ public class AppendOnlyWriter implements RecordWriter<RowData> {
private final DataFilePathFactory pathFactory;
private final CompactManager compactManager;
private final boolean forceCompact;
+ private final List<DataFileMeta> newFiles;
private final List<DataFileMeta> compactBefore;
private final List<DataFileMeta> compactAfter;
private final LongCounter seqNumCounter;
@@ -75,9 +76,11 @@ public class AppendOnlyWriter implements RecordWriter<RowData> {
this.pathFactory = pathFactory;
this.compactManager = compactManager;
this.forceCompact = forceCompact;
+ this.newFiles = new ArrayList<>();
this.compactBefore = new ArrayList<>();
this.compactAfter = new ArrayList<>();
this.seqNumCounter = new LongCounter(maxSequenceNumber + 1);
+
this.writer = createRollingRowWriter();
}
@@ -92,29 +95,33 @@ public class AppendOnlyWriter implements RecordWriter<RowData> {
@Override
public void fullCompaction() throws Exception {
- submitCompaction(true);
+ flushWriter(true);
}
@Override
public CommitIncrement prepareCommit(boolean endOnfInput) throws Exception {
- List<DataFileMeta> newFiles = new ArrayList<>();
+ flushWriter(false);
+ boolean blocking = endOnfInput || forceCompact;
+ trySyncLatestCompaction(blocking);
+ return drainIncrement();
+ }
+
+ private void flushWriter(boolean forcedFullCompaction) throws Exception {
+ List<DataFileMeta> flushedFiles = new ArrayList<>();
if (writer != null) {
writer.close();
- newFiles.addAll(writer.result());
+ flushedFiles.addAll(writer.result());
// Reopen the writer to accept further records.
seqNumCounter.resetLocal();
- seqNumCounter.add(getMaxSequenceNumber(newFiles) + 1);
+ seqNumCounter.add(getMaxSequenceNumber(flushedFiles) + 1);
writer = createRollingRowWriter();
}
- // add new generated files
- newFiles.forEach(compactManager::addNewFile);
- submitCompaction(false);
- boolean blocking = endOnfInput || forceCompact;
- trySyncLatestCompaction(blocking);
-
- return drainIncrement(newFiles);
+ // add new generated files
+ flushedFiles.forEach(compactManager::addNewFile);
+ submitCompaction(forcedFullCompaction);
+ newFiles.addAll(flushedFiles);
}
@Override
@@ -156,15 +163,16 @@ public class AppendOnlyWriter implements RecordWriter<RowData> {
});
}
- private CommitIncrement drainIncrement(List<DataFileMeta> newFiles) {
+ private CommitIncrement drainIncrement() {
NewFilesIncrement newFilesIncrement =
- new NewFilesIncrement(newFiles, Collections.emptyList());
+ new NewFilesIncrement(new ArrayList<>(newFiles), Collections.emptyList());
CompactIncrement compactIncrement =
new CompactIncrement(
new ArrayList<>(compactBefore),
new ArrayList<>(compactAfter),
Collections.emptyList());
+ newFiles.clear();
compactBefore.clear();
compactAfter.clear();
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/compact/CompactResult.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/compact/CompactResult.java
index 3928e0a4..896b2637 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/compact/CompactResult.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/compact/CompactResult.java
@@ -20,12 +20,51 @@ package org.apache.flink.table.store.file.compact;
import org.apache.flink.table.store.file.io.DataFileMeta;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
/** Result of compaction. */
-public interface CompactResult {
+public class CompactResult {
- List<DataFileMeta> before();
+ private final List<DataFileMeta> before;
+ private final List<DataFileMeta> after;
+ private final List<DataFileMeta> changelog;
- List<DataFileMeta> after();
+ public CompactResult() {
+ this(Collections.emptyList(), Collections.emptyList());
+ }
+
+ public CompactResult(DataFileMeta before, DataFileMeta after) {
+ this(Collections.singletonList(before), Collections.singletonList(after));
+ }
+
+ public CompactResult(List<DataFileMeta> before, List<DataFileMeta> after) {
+ this(before, after, Collections.emptyList());
+ }
+
+ public CompactResult(
+ List<DataFileMeta> before, List<DataFileMeta> after, List<DataFileMeta> changelog) {
+ this.before = new ArrayList<>(before);
+ this.after = new ArrayList<>(after);
+ this.changelog = new ArrayList<>(changelog);
+ }
+
+ public List<DataFileMeta> before() {
+ return before;
+ }
+
+ public List<DataFileMeta> after() {
+ return after;
+ }
+
+ public List<DataFileMeta> changelog() {
+ return changelog;
+ }
+
+ public void merge(CompactResult that) {
+ before.addAll(that.before);
+ after.addAll(that.after);
+ changelog.addAll(that.changelog);
+ }
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/DropDeleteReader.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/DropDeleteReader.java
new file mode 100644
index 00000000..6d4ae403
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/DropDeleteReader.java
@@ -0,0 +1,77 @@
+/*
+ * 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.flink.table.store.file.mergetree;
+
+import org.apache.flink.table.store.file.KeyValue;
+import org.apache.flink.table.store.file.utils.RecordReader;
+import org.apache.flink.types.RowKind;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * A {@link RecordReader} which drops {@link KeyValue} of {@link RowKind#DELETE} kind from the
+ * wrapped reader.
+ */
+public class DropDeleteReader implements RecordReader<KeyValue> {
+
+ private final RecordReader<KeyValue> reader;
+
+ public DropDeleteReader(RecordReader<KeyValue> reader) {
+ this.reader = reader;
+ }
+
+ @Nullable
+ @Override
+ public RecordIterator<KeyValue> readBatch() throws IOException {
+ RecordIterator<KeyValue> batch = reader.readBatch();
+
+ if (batch == null) {
+ return null;
+ }
+
+ return new RecordIterator<KeyValue>() {
+ @Override
+ public KeyValue next() throws IOException {
+ while (true) {
+ KeyValue kv = batch.next();
+ if (kv == null) {
+ return null;
+ }
+
+ if (kv.valueKind() == RowKind.INSERT
+ || kv.valueKind() == RowKind.UPDATE_AFTER) {
+ return kv;
+ }
+ }
+ }
+
+ @Override
+ public void releaseBatch() {
+ batch.releaseBatch();
+ }
+ };
+ }
+
+ @Override
+ public void close() throws IOException {
+ reader.close();
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeReader.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeReaders.java
similarity index 56%
rename from flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeReader.java
rename to flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeReaders.java
index 444d10ce..3bb991c9 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeReader.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeReaders.java
@@ -23,108 +23,66 @@ import org.apache.flink.table.store.file.KeyValue;
import org.apache.flink.table.store.file.io.DataFileMeta;
import org.apache.flink.table.store.file.io.KeyValueFileReaderFactory;
import org.apache.flink.table.store.file.mergetree.compact.ConcatRecordReader;
-import org.apache.flink.table.store.file.mergetree.compact.ConcatRecordReader.ReaderSupplier;
import org.apache.flink.table.store.file.mergetree.compact.MergeFunction;
+import org.apache.flink.table.store.file.mergetree.compact.MergeFunctionWrapper;
+import org.apache.flink.table.store.file.mergetree.compact.ReducerMergeFunctionWrapper;
import org.apache.flink.table.store.file.mergetree.compact.SortMergeReader;
import org.apache.flink.table.store.file.utils.RecordReader;
-import org.apache.flink.types.RowKind;
-
-import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
-/** A {@link RecordReader} to read merge tree sections. */
-public class MergeTreeReader implements RecordReader<KeyValue> {
-
- private final RecordReader<KeyValue> reader;
+/** Utility class to create commonly used {@link RecordReader}s for merge trees. */
+public class MergeTreeReaders {
- private final boolean dropDelete;
+ private MergeTreeReaders() {}
- public MergeTreeReader(
+ public static RecordReader<KeyValue> readerForMergeTree(
List<List<SortedRun>> sections,
boolean dropDelete,
KeyValueFileReaderFactory readerFactory,
Comparator<RowData> userKeyComparator,
MergeFunction<KeyValue> mergeFunction)
throws IOException {
- this.dropDelete = dropDelete;
-
- List<ReaderSupplier<KeyValue>> readers = new ArrayList<>();
+ List<ConcatRecordReader.ReaderSupplier<KeyValue>> readers = new ArrayList<>();
for (List<SortedRun> section : sections) {
readers.add(
() ->
readerForSection(
- section, readerFactory, userKeyComparator, mergeFunction));
- }
- this.reader = ConcatRecordReader.create(readers);
- }
-
- public MergeTreeReader(boolean dropDelete, List<ReaderSupplier<KeyValue>> sectionReaders)
- throws IOException {
- this.dropDelete = dropDelete;
- this.reader = ConcatRecordReader.create(sectionReaders);
- }
-
- @Nullable
- @Override
- public RecordIterator<KeyValue> readBatch() throws IOException {
- RecordIterator<KeyValue> batch = reader.readBatch();
-
- if (!dropDelete) {
- return batch;
+ section,
+ readerFactory,
+ userKeyComparator,
+ new ReducerMergeFunctionWrapper(mergeFunction)));
}
-
- if (batch == null) {
- return null;
+ RecordReader<KeyValue> reader = ConcatRecordReader.create(readers);
+ if (dropDelete) {
+ reader = new DropDeleteReader(reader);
}
-
- return new RecordIterator<KeyValue>() {
- @Override
- public KeyValue next() throws IOException {
- while (true) {
- KeyValue kv = batch.next();
- if (kv == null) {
- return null;
- }
-
- if (kv.valueKind() == RowKind.INSERT
- || kv.valueKind() == RowKind.UPDATE_AFTER) {
- return kv;
- }
- }
- }
-
- @Override
- public void releaseBatch() {
- batch.releaseBatch();
- }
- };
- }
-
- @Override
- public void close() throws IOException {
- reader.close();
+ return reader;
}
public static RecordReader<KeyValue> readerForSection(
List<SortedRun> section,
KeyValueFileReaderFactory readerFactory,
Comparator<RowData> userKeyComparator,
- MergeFunction<KeyValue> mergeFunction)
+ MergeFunctionWrapper<KeyValue> mergeFunctionWrapper)
throws IOException {
List<RecordReader<KeyValue>> readers = new ArrayList<>();
for (SortedRun run : section) {
readers.add(readerForRun(run, readerFactory));
}
- return SortMergeReader.create(readers, userKeyComparator, mergeFunction);
+ if (readers.size() == 1) {
+ return readers.get(0);
+ } else {
+ return new SortMergeReader<>(readers, userKeyComparator, mergeFunctionWrapper);
+ }
}
public static RecordReader<KeyValue> readerForRun(
SortedRun run, KeyValueFileReaderFactory readerFactory) throws IOException {
- List<ReaderSupplier<KeyValue>> readers = new ArrayList<>();
+ List<ConcatRecordReader.ReaderSupplier<KeyValue>> readers = new ArrayList<>();
for (DataFileMeta file : run.files()) {
readers.add(() -> readerFactory.createRecordReader(file.fileName(), file.level()));
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeWriter.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeWriter.java
index 7f0adfcc..ba953277 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeWriter.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeWriter.java
@@ -37,7 +37,6 @@ import org.apache.flink.table.store.file.utils.RecordWriter;
import org.apache.flink.table.types.logical.RowType;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
@@ -63,9 +62,10 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
private final ChangelogProducer changelogProducer;
private final LinkedHashSet<DataFileMeta> newFiles;
+ private final LinkedHashSet<DataFileMeta> newFilesChangelog;
private final LinkedHashMap<String, DataFileMeta> compactBefore;
private final LinkedHashSet<DataFileMeta> compactAfter;
- private final LinkedHashSet<DataFileMeta> changelogFiles;
+ private final LinkedHashSet<DataFileMeta> compactChangelog;
private long newSequenceNumber;
private WriteBuffer writeBuffer;
@@ -95,9 +95,10 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
this.changelogProducer = changelogProducer;
this.newFiles = new LinkedHashSet<>();
+ this.newFilesChangelog = new LinkedHashSet<>();
this.compactBefore = new LinkedHashMap<>();
this.compactAfter = new LinkedHashSet<>();
- this.changelogFiles = new LinkedHashSet<>();
+ this.compactChangelog = new LinkedHashSet<>();
}
private long newSequenceNumber() {
@@ -129,7 +130,7 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
: kv.sequenceNumber();
boolean success = writeBuffer.put(sequenceNumber, kv.valueKind(), kv.key(), kv.value());
if (!success) {
- flushWriteBuffer();
+ flushWriteBuffer(false);
success = writeBuffer.put(sequenceNumber, kv.valueKind(), kv.key(), kv.value());
if (!success) {
throw new RuntimeException("Mem table is too small to hold a single element.");
@@ -139,7 +140,7 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
@Override
public void fullCompaction() throws Exception {
- submitCompaction(true);
+ flushWriteBuffer(true);
}
@Override
@@ -151,11 +152,11 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
public void flushMemory() throws Exception {
boolean success = writeBuffer.flushMemory();
if (!success) {
- flushWriteBuffer();
+ flushWriteBuffer(false);
}
}
- private void flushWriteBuffer() throws Exception {
+ private void flushWriteBuffer(boolean forcedFullCompaction) throws Exception {
if (writeBuffer.size() > 0) {
if (compactManager.shouldWaitCompaction()) {
// stop writing, wait for compaction finished
@@ -183,7 +184,7 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
}
if (changelogWriter != null) {
- changelogFiles.addAll(changelogWriter.result());
+ newFilesChangelog.addAll(changelogWriter.result());
}
for (DataFileMeta fileMeta : dataWriter.result()) {
@@ -192,13 +193,13 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
}
writeBuffer.clear();
- submitCompaction(false);
}
+ submitCompaction(forcedFullCompaction);
}
@Override
public CommitIncrement prepareCommit(boolean endOfInput) throws Exception {
- flushWriteBuffer();
+ flushWriteBuffer(false);
boolean blocking = endOfInput || commitForceCompact;
trySyncLatestCompaction(blocking);
return drainIncrement();
@@ -211,17 +212,19 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
private CommitIncrement drainIncrement() {
NewFilesIncrement newFilesIncrement =
- new NewFilesIncrement(new ArrayList<>(newFiles), new ArrayList<>(changelogFiles));
+ new NewFilesIncrement(
+ new ArrayList<>(newFiles), new ArrayList<>(newFilesChangelog));
CompactIncrement compactIncrement =
new CompactIncrement(
new ArrayList<>(compactBefore.values()),
new ArrayList<>(compactAfter),
- Collections.emptyList());
+ new ArrayList<>(compactChangelog));
newFiles.clear();
+ newFilesChangelog.clear();
compactBefore.clear();
compactAfter.clear();
- changelogFiles.clear();
+ compactChangelog.clear();
return new CommitIncrement() {
@Override
@@ -255,6 +258,7 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
}
}
compactAfter.addAll(result.after());
+ compactChangelog.addAll(result.changelog());
}
private void submitCompaction(boolean forcedFullCompaction) throws Exception {
@@ -275,6 +279,13 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
// delete temporary files
List<DataFileMeta> delete = new ArrayList<>(newFiles);
+ newFiles.clear();
+
+ for (DataFileMeta file : newFilesChangelog) {
+ writerFactory.deleteFile(file.fileName());
+ }
+ newFilesChangelog.clear();
+
for (DataFileMeta file : compactAfter) {
// upgrade file is required by previous snapshot, so we should ensure that this file is
// not the output of upgraded.
@@ -282,14 +293,16 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
delete.add(file);
}
}
- for (DataFileMeta file : delete) {
+
+ compactAfter.clear();
+
+ for (DataFileMeta file : compactChangelog) {
writerFactory.deleteFile(file.fileName());
}
- newFiles.clear();
- compactAfter.clear();
- for (DataFileMeta file : changelogFiles) {
+ compactChangelog.clear();
+
+ for (DataFileMeta file : delete) {
writerFactory.deleteFile(file.fileName());
}
- changelogFiles.clear();
}
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/CompactRewriter.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/AbstractCompactRewriter.java
similarity index 57%
copy from flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/CompactRewriter.java
copy to flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/AbstractCompactRewriter.java
index 02c7fe5e..9881b0a5 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/CompactRewriter.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/AbstractCompactRewriter.java
@@ -18,15 +18,27 @@
package org.apache.flink.table.store.file.mergetree.compact;
+import org.apache.flink.table.store.file.compact.CompactResult;
import org.apache.flink.table.store.file.io.DataFileMeta;
import org.apache.flink.table.store.file.mergetree.SortedRun;
+import java.util.Collection;
import java.util.List;
+import java.util.stream.Collectors;
-/** Rewrite sections to the files. */
-@FunctionalInterface
-public interface CompactRewriter {
+/** Common implementation of {@link CompactRewriter}. */
+public abstract class AbstractCompactRewriter implements CompactRewriter {
- List<DataFileMeta> rewrite(int outputLevel, boolean dropDelete, List<List<SortedRun>> sections)
- throws Exception;
+ @Override
+ public CompactResult upgrade(int outputLevel, DataFileMeta file) throws Exception {
+ return new CompactResult(file, file.upgrade(outputLevel));
+ }
+
+ protected static List<DataFileMeta> extractFilesFromSections(List<List<SortedRun>> sections) {
+ return sections.stream()
+ .flatMap(Collection::stream)
+ .map(SortedRun::files)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ }
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/CompactRewriter.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/CompactRewriter.java
index 02c7fe5e..d2b62bd1 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/CompactRewriter.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/CompactRewriter.java
@@ -18,15 +18,17 @@
package org.apache.flink.table.store.file.mergetree.compact;
+import org.apache.flink.table.store.file.compact.CompactResult;
import org.apache.flink.table.store.file.io.DataFileMeta;
import org.apache.flink.table.store.file.mergetree.SortedRun;
import java.util.List;
/** Rewrite sections to the files. */
-@FunctionalInterface
public interface CompactRewriter {
- List<DataFileMeta> rewrite(int outputLevel, boolean dropDelete, List<List<SortedRun>> sections)
+ CompactResult rewrite(int outputLevel, boolean dropDelete, List<List<SortedRun>> sections)
throws Exception;
+
+ CompactResult upgrade(int outputLevel, DataFileMeta file) throws Exception;
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/FullChangelogMergeFunctionWrapper.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/FullChangelogMergeFunctionWrapper.java
index 293fd01d..4a1be778 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/FullChangelogMergeFunctionWrapper.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/FullChangelogMergeFunctionWrapper.java
@@ -156,7 +156,11 @@ public class FullChangelogMergeFunctionWrapper
}
private Result setResult(@Nullable KeyValue result) {
- this.result = result;
+ if (result != null && result.valueKind() != RowKind.DELETE) {
+ this.result = result;
+ } else {
+ this.result = null;
+ }
return this;
}
@@ -178,7 +182,10 @@ public class FullChangelogMergeFunctionWrapper
return after;
}
- /** Latest full compaction result (result of merge function) for this key. */
+ /**
+ * Latest full compaction result (result of merge function) for this key. Null if the merged
+ * result is null or is of DELETE kind.
+ */
@Nullable
public KeyValue result() {
return result;
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/FullChangelogMergeTreeCompactRewriter.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/FullChangelogMergeTreeCompactRewriter.java
new file mode 100644
index 00000000..516cf53d
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/FullChangelogMergeTreeCompactRewriter.java
@@ -0,0 +1,134 @@
+/*
+ * 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.flink.table.store.file.mergetree.compact;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.store.file.KeyValue;
+import org.apache.flink.table.store.file.compact.CompactResult;
+import org.apache.flink.table.store.file.io.DataFileMeta;
+import org.apache.flink.table.store.file.io.KeyValueFileReaderFactory;
+import org.apache.flink.table.store.file.io.KeyValueFileWriterFactory;
+import org.apache.flink.table.store.file.io.RollingFileWriter;
+import org.apache.flink.table.store.file.mergetree.MergeTreeReaders;
+import org.apache.flink.table.store.file.mergetree.SortedRun;
+import org.apache.flink.table.store.file.utils.RecordReader;
+import org.apache.flink.table.store.file.utils.RecordReaderIterator;
+import org.apache.flink.util.Preconditions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+
+/** A {@link MergeTreeCompactRewriter} which produces changelog files for each full compaction. */
+public class FullChangelogMergeTreeCompactRewriter extends MergeTreeCompactRewriter {
+
+ private final int maxLevel;
+
+ public FullChangelogMergeTreeCompactRewriter(
+ int maxLevel,
+ KeyValueFileReaderFactory readerFactory,
+ KeyValueFileWriterFactory writerFactory,
+ Comparator<RowData> keyComparator,
+ MergeFunction<KeyValue> mergeFunction) {
+ super(readerFactory, writerFactory, keyComparator, mergeFunction);
+ this.maxLevel = maxLevel;
+ }
+
+ @Override
+ public CompactResult rewrite(
+ int outputLevel, boolean dropDelete, List<List<SortedRun>> sections) throws Exception {
+ if (outputLevel == maxLevel) {
+ Preconditions.checkArgument(
+ dropDelete,
+ "Delete records should be dropped from result of full compaction. This is unexpected.");
+ return rewriteFullCompaction(sections);
+ } else {
+ return rewriteCompaction(outputLevel, dropDelete, sections);
+ }
+ }
+
+ private CompactResult rewriteFullCompaction(List<List<SortedRun>> sections) throws Exception {
+ List<ConcatRecordReader.ReaderSupplier<FullChangelogMergeFunctionWrapper.Result>>
+ sectionReaders = new ArrayList<>();
+ for (List<SortedRun> section : sections) {
+ sectionReaders.add(
+ () -> {
+ List<RecordReader<KeyValue>> runReaders = new ArrayList<>();
+ for (SortedRun run : section) {
+ runReaders.add(MergeTreeReaders.readerForRun(run, readerFactory));
+ }
+ return new SortMergeReader<>(
+ runReaders,
+ keyComparator,
+ new FullChangelogMergeFunctionWrapper(
+ mergeFunction.copy(), maxLevel));
+ });
+ }
+
+ RecordReaderIterator<FullChangelogMergeFunctionWrapper.Result> iterator = null;
+ RollingFileWriter<KeyValue, DataFileMeta> compactFileWriter = null;
+ RollingFileWriter<KeyValue, DataFileMeta> changelogFileWriter = null;
+
+ try {
+ iterator = new RecordReaderIterator<>(ConcatRecordReader.create(sectionReaders));
+ compactFileWriter = writerFactory.createRollingMergeTreeFileWriter(maxLevel);
+ changelogFileWriter = writerFactory.createRollingChangelogFileWriter(maxLevel);
+
+ while (iterator.hasNext()) {
+ FullChangelogMergeFunctionWrapper.Result result = iterator.next();
+ if (result.result() != null) {
+ compactFileWriter.write(result.result());
+ }
+ if (result.before() != null) {
+ changelogFileWriter.write(result.before());
+ }
+ if (result.after() != null) {
+ changelogFileWriter.write(result.after());
+ }
+ }
+ } finally {
+ if (iterator != null) {
+ iterator.close();
+ }
+ if (compactFileWriter != null) {
+ compactFileWriter.close();
+ }
+ if (changelogFileWriter != null) {
+ changelogFileWriter.close();
+ }
+ }
+
+ return new CompactResult(
+ extractFilesFromSections(sections),
+ compactFileWriter.result(),
+ changelogFileWriter.result());
+ }
+
+ @Override
+ public CompactResult upgrade(int outputLevel, DataFileMeta file) throws Exception {
+ if (outputLevel == maxLevel) {
+ return rewriteFullCompaction(
+ Collections.singletonList(
+ Collections.singletonList(SortedRun.fromSingle(file))));
+ } else {
+ return super.upgrade(outputLevel, file);
+ }
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeTreeCompactRewriter.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeTreeCompactRewriter.java
new file mode 100644
index 00000000..0d277008
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeTreeCompactRewriter.java
@@ -0,0 +1,72 @@
+/*
+ * 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.flink.table.store.file.mergetree.compact;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.store.file.KeyValue;
+import org.apache.flink.table.store.file.compact.CompactResult;
+import org.apache.flink.table.store.file.io.DataFileMeta;
+import org.apache.flink.table.store.file.io.KeyValueFileReaderFactory;
+import org.apache.flink.table.store.file.io.KeyValueFileWriterFactory;
+import org.apache.flink.table.store.file.io.RollingFileWriter;
+import org.apache.flink.table.store.file.mergetree.MergeTreeReaders;
+import org.apache.flink.table.store.file.mergetree.SortedRun;
+import org.apache.flink.table.store.file.utils.RecordReader;
+import org.apache.flink.table.store.file.utils.RecordReaderIterator;
+
+import java.util.Comparator;
+import java.util.List;
+
+/** Default {@link CompactRewriter} for merge trees. */
+public class MergeTreeCompactRewriter extends AbstractCompactRewriter {
+
+ protected final KeyValueFileReaderFactory readerFactory;
+ protected final KeyValueFileWriterFactory writerFactory;
+ protected final Comparator<RowData> keyComparator;
+ protected final MergeFunction<KeyValue> mergeFunction;
+
+ public MergeTreeCompactRewriter(
+ KeyValueFileReaderFactory readerFactory,
+ KeyValueFileWriterFactory writerFactory,
+ Comparator<RowData> keyComparator,
+ MergeFunction<KeyValue> mergeFunction) {
+ this.readerFactory = readerFactory;
+ this.writerFactory = writerFactory;
+ this.keyComparator = keyComparator;
+ this.mergeFunction = mergeFunction;
+ }
+
+ @Override
+ public CompactResult rewrite(
+ int outputLevel, boolean dropDelete, List<List<SortedRun>> sections) throws Exception {
+ return rewriteCompaction(outputLevel, dropDelete, sections);
+ }
+
+ protected CompactResult rewriteCompaction(
+ int outputLevel, boolean dropDelete, List<List<SortedRun>> sections) throws Exception {
+ RollingFileWriter<KeyValue, DataFileMeta> writer =
+ writerFactory.createRollingMergeTreeFileWriter(outputLevel);
+ RecordReader<KeyValue> sectionsReader =
+ MergeTreeReaders.readerForMergeTree(
+ sections, dropDelete, readerFactory, keyComparator, mergeFunction.copy());
+ writer.write(new RecordReaderIterator<>(sectionsReader));
+ writer.close();
+ return new CompactResult(extractFilesFromSections(sections), writer.result());
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeTreeCompactTask.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeTreeCompactTask.java
index 4b09e358..49974a49 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeTreeCompactTask.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeTreeCompactTask.java
@@ -64,8 +64,7 @@ public class MergeTreeCompactTask extends CompactTask {
@Override
protected CompactResult doCompact(List<DataFileMeta> inputs) throws Exception {
List<List<SortedRun>> candidate = new ArrayList<>();
- List<DataFileMeta> compactBefore = new ArrayList<>();
- List<DataFileMeta> compactAfter = new ArrayList<>();
+ CompactResult result = new CompactResult();
// Checking the order and compacting adjacent and contiguous files
// Note: can't skip an intermediate file to compact, this will destroy the overall
@@ -85,24 +84,14 @@ public class MergeTreeCompactTask extends CompactTask {
candidate.add(singletonList(SortedRun.fromSingle(file)));
} else {
// Large file appear, rewrite previous and upgrade it
- rewrite(candidate, compactBefore, compactAfter);
- upgrade(file, compactBefore, compactAfter);
+ rewrite(candidate, result);
+ upgrade(file, result);
}
}
}
}
- rewrite(candidate, compactBefore, compactAfter);
- return new CompactResult() {
- @Override
- public List<DataFileMeta> before() {
- return compactBefore;
- }
-
- @Override
- public List<DataFileMeta> after() {
- return compactAfter;
- }
- };
+ rewrite(candidate, result);
+ return result;
}
@Override
@@ -113,20 +102,15 @@ public class MergeTreeCompactTask extends CompactTask {
super.logMetric(startMillis, compactBefore, compactAfter), upgradeFilesNum);
}
- private void upgrade(
- DataFileMeta file, List<DataFileMeta> compactBefore, List<DataFileMeta> compactAfter) {
+ private void upgrade(DataFileMeta file, CompactResult toUpdate) throws Exception {
if (file.level() != outputLevel) {
- compactBefore.add(file);
- compactAfter.add(file.upgrade(outputLevel));
+ CompactResult upgradeResult = rewriter.upgrade(outputLevel, file);
+ toUpdate.merge(upgradeResult);
upgradeFilesNum++;
}
}
- private void rewrite(
- List<List<SortedRun>> candidate,
- List<DataFileMeta> compactBefore,
- List<DataFileMeta> compactAfter)
- throws Exception {
+ private void rewrite(List<List<SortedRun>> candidate, CompactResult toUpdate) throws Exception {
if (candidate.isEmpty()) {
return;
}
@@ -136,15 +120,14 @@ public class MergeTreeCompactTask extends CompactTask {
return;
} else if (section.size() == 1) {
for (DataFileMeta file : section.get(0).files()) {
- upgrade(file, compactBefore, compactAfter);
+ upgrade(file, toUpdate);
}
candidate.clear();
return;
}
}
- candidate.forEach(runs -> runs.forEach(run -> compactBefore.addAll(run.files())));
- List<DataFileMeta> result = rewriter.rewrite(outputLevel, dropDelete, candidate);
- compactAfter.addAll(result);
+ CompactResult rewriteResult = rewriter.rewrite(outputLevel, dropDelete, candidate);
+ toUpdate.merge(rewriteResult);
candidate.clear();
}
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReader.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReader.java
index d9215f86..920de180 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReader.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReader.java
@@ -38,22 +38,22 @@ import java.util.PriorityQueue;
*
* <p>NOTE: {@link KeyValue}s from the same {@link RecordReader} must not contain the same key.
*/
-public class SortMergeReader implements RecordReader<KeyValue> {
+public class SortMergeReader<T> implements RecordReader<T> {
private final List<RecordReader<KeyValue>> nextBatchReaders;
private final Comparator<RowData> userKeyComparator;
- private final MergeFunctionWrapper<KeyValue> mergeFunctionWrapper;
+ private final MergeFunctionWrapper<T> mergeFunctionWrapper;
private final PriorityQueue<Element> minHeap;
private final List<Element> polled;
- protected SortMergeReader(
+ public SortMergeReader(
List<RecordReader<KeyValue>> readers,
Comparator<RowData> userKeyComparator,
- MergeFunction<KeyValue> mergeFunction) {
+ MergeFunctionWrapper<T> mergeFunctionWrapper) {
this.nextBatchReaders = new ArrayList<>(readers);
this.userKeyComparator = userKeyComparator;
- this.mergeFunctionWrapper = new ReducerMergeFunctionWrapper(mergeFunction);
+ this.mergeFunctionWrapper = mergeFunctionWrapper;
this.minHeap =
new PriorityQueue<>(
@@ -67,18 +67,9 @@ public class SortMergeReader implements RecordReader<KeyValue> {
this.polled = new ArrayList<>();
}
- public static RecordReader<KeyValue> create(
- List<RecordReader<KeyValue>> readers,
- Comparator<RowData> userKeyComparator,
- MergeFunction<KeyValue> mergeFunction) {
- return readers.size() == 1
- ? readers.get(0)
- : new SortMergeReader(readers, userKeyComparator, mergeFunction);
- }
-
@Nullable
@Override
- public RecordIterator<KeyValue> readBatch() throws IOException {
+ public RecordIterator<T> readBatch() throws IOException {
for (RecordReader<KeyValue> reader : nextBatchReaders) {
while (true) {
RecordIterator<KeyValue> iterator = reader.readBatch();
@@ -119,18 +110,18 @@ public class SortMergeReader implements RecordReader<KeyValue> {
}
/** The iterator iterates on {@link SortMergeReader}. */
- private class SortMergeIterator implements RecordIterator<KeyValue> {
+ private class SortMergeIterator implements RecordIterator<T> {
private boolean released = false;
@Override
- public KeyValue next() throws IOException {
+ public T next() throws IOException {
while (true) {
boolean hasMore = nextImpl();
if (!hasMore) {
return null;
}
- KeyValue result = mergeFunctionWrapper.getResult();
+ T result = mergeFunctionWrapper.getResult();
if (result != null) {
return result;
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreScan.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreScan.java
index 275dede4..58257440 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreScan.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreScan.java
@@ -288,6 +288,12 @@ public abstract class AbstractFileStoreScan implements FileStoreScan {
String.format(
"Incremental scan does not accept %s snapshot",
snapshot.commitKind()));
+ case FULL_COMPACTION:
+ if (snapshot.changelogManifestList() == null) {
+ return Collections.emptyList();
+ } else {
+ return manifestList.read(snapshot.changelogManifestList());
+ }
case NONE:
if (snapshot.commitKind() == Snapshot.CommitKind.APPEND) {
return manifestList.read(snapshot.deltaManifestList());
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreWrite.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreWrite.java
index 7f79abef..b40d4d64 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreWrite.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreWrite.java
@@ -55,7 +55,8 @@ public interface FileStoreWrite<T> {
void write(BinaryRowData partition, int bucket, T data) throws Exception;
/**
- * Compact data stored in given partition and bucket.
+ * Compact data stored in given partition and bucket. Note that compaction process is only
+ * submitted and may not be completed when the method returns.
*
* @param partition the partition to compact
* @param bucket the bucket to compact
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java
index 6914fbcb..dd2c9830 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java
@@ -22,11 +22,14 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.store.file.KeyValue;
import org.apache.flink.table.store.file.io.DataFileMeta;
import org.apache.flink.table.store.file.io.KeyValueFileReaderFactory;
-import org.apache.flink.table.store.file.mergetree.MergeTreeReader;
+import org.apache.flink.table.store.file.mergetree.DropDeleteReader;
+import org.apache.flink.table.store.file.mergetree.MergeTreeReaders;
import org.apache.flink.table.store.file.mergetree.SortedRun;
import org.apache.flink.table.store.file.mergetree.compact.ConcatRecordReader;
import org.apache.flink.table.store.file.mergetree.compact.IntervalPartition;
import org.apache.flink.table.store.file.mergetree.compact.MergeFunction;
+import org.apache.flink.table.store.file.mergetree.compact.MergeFunctionWrapper;
+import org.apache.flink.table.store.file.mergetree.compact.ReducerMergeFunctionWrapper;
import org.apache.flink.table.store.file.predicate.Predicate;
import org.apache.flink.table.store.file.schema.SchemaManager;
import org.apache.flink.table.store.file.schema.TableSchema;
@@ -118,7 +121,16 @@ public class KeyValueFileStoreRead implements FileStoreRead<KeyValue> {
pkFilters.add(sub);
}
}
- // for section which does not have key range overlap, push down value filters too
+ // Consider this case:
+ // Denote (seqNumber, key, value) as a record. We have two overlapping runs in a section:
+ // * First run: (1, k1, 100), (2, k2, 200)
+ // * Second run: (3, k1, 10), (4, k2, 20)
+ // If we push down filter "value >= 100" for this section, only the first run will be read,
+ // and the second run is lost. This will produce incorrect results.
+ //
+ // So for sections with overlapping runs, we only push down key filters.
+ // For sections with only one run, as each key only appears once, it is OK to push down
+ // value filters.
filtersForNonOverlappedSection = allFilters;
filtersForOverlappedSection = valueCountMode ? allFilters : pkFilters;
return this;
@@ -143,41 +155,43 @@ public class KeyValueFileStoreRead implements FileStoreRead<KeyValue> {
}
return ConcatRecordReader.create(suppliers);
} else {
- // in this case merge tree should merge records with same key
- // Do not project key in MergeTreeReader.
- MergeTreeReader reader = new MergeTreeReader(true, createSectionReaders(split));
+ // Sections are read by SortMergeReader, which sorts and merges records by keys.
+ // So we cannot project keys or else the sorting will be incorrect.
+ KeyValueFileReaderFactory overlappedSectionFactory =
+ readerFactoryBuilder.build(
+ split.partition(), split.bucket(), false, filtersForOverlappedSection);
+ KeyValueFileReaderFactory nonOverlappedSectionFactory =
+ readerFactoryBuilder.build(
+ split.partition(),
+ split.bucket(),
+ false,
+ filtersForNonOverlappedSection);
+
+ List<ConcatRecordReader.ReaderSupplier<KeyValue>> sectionReaders = new ArrayList<>();
+ MergeFunctionWrapper<KeyValue> mergeFuncWrapper =
+ new ReducerMergeFunctionWrapper(mergeFunction.copy());
+ for (List<SortedRun> section :
+ new IntervalPartition(split.files(), keyComparator).partition()) {
+ sectionReaders.add(
+ () ->
+ MergeTreeReaders.readerForSection(
+ section,
+ section.size() > 1
+ ? overlappedSectionFactory
+ : nonOverlappedSectionFactory,
+ keyComparator,
+ mergeFuncWrapper));
+ }
+ DropDeleteReader reader =
+ new DropDeleteReader(ConcatRecordReader.create(sectionReaders));
- // project key using ProjectKeyRecordReader
+ // Project results from SortMergeReader using ProjectKeyRecordReader.
return keyProjectedFields == null
? reader
: new ProjectKeyRecordReader(reader, keyProjectedFields);
}
}
- private List<ConcatRecordReader.ReaderSupplier<KeyValue>> createSectionReaders(
- DataSplit split) {
- List<ConcatRecordReader.ReaderSupplier<KeyValue>> sectionReaders = new ArrayList<>();
- MergeFunction<KeyValue> mergeFunc = mergeFunction.copy();
- for (List<SortedRun> section :
- new IntervalPartition(split.files(), keyComparator).partition()) {
- KeyValueFileReaderFactory readerFactory =
- createReaderFactory(split, section.size() > 1);
- sectionReaders.add(
- () ->
- MergeTreeReader.readerForSection(
- section, readerFactory, keyComparator, mergeFunc));
- }
- return sectionReaders;
- }
-
- private KeyValueFileReaderFactory createReaderFactory(DataSplit split, boolean overlapped) {
- return readerFactoryBuilder.build(
- split.partition(),
- split.bucket(),
- false,
- overlapped ? filtersForOverlappedSection : filtersForNonOverlappedSection);
- }
-
private Optional<String> changelogFile(DataFileMeta fileMeta) {
for (String file : fileMeta.extraFiles()) {
if (file.startsWith(CHANGELOG_FILE_PREFIX)) {
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreWrite.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreWrite.java
index 89b038e3..1e7b5200 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreWrite.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreWrite.java
@@ -27,18 +27,17 @@ import org.apache.flink.table.store.file.compact.NoopCompactManager;
import org.apache.flink.table.store.file.io.DataFileMeta;
import org.apache.flink.table.store.file.io.KeyValueFileReaderFactory;
import org.apache.flink.table.store.file.io.KeyValueFileWriterFactory;
-import org.apache.flink.table.store.file.io.RollingFileWriter;
import org.apache.flink.table.store.file.mergetree.Levels;
-import org.apache.flink.table.store.file.mergetree.MergeTreeReader;
import org.apache.flink.table.store.file.mergetree.MergeTreeWriter;
import org.apache.flink.table.store.file.mergetree.compact.CompactRewriter;
import org.apache.flink.table.store.file.mergetree.compact.CompactStrategy;
+import org.apache.flink.table.store.file.mergetree.compact.FullChangelogMergeTreeCompactRewriter;
import org.apache.flink.table.store.file.mergetree.compact.MergeFunction;
import org.apache.flink.table.store.file.mergetree.compact.MergeTreeCompactManager;
+import org.apache.flink.table.store.file.mergetree.compact.MergeTreeCompactRewriter;
import org.apache.flink.table.store.file.mergetree.compact.UniversalCompaction;
import org.apache.flink.table.store.file.schema.SchemaManager;
import org.apache.flink.table.store.file.utils.FileStorePathFactory;
-import org.apache.flink.table.store.file.utils.RecordReaderIterator;
import org.apache.flink.table.store.file.utils.RecordWriter;
import org.apache.flink.table.store.file.utils.SnapshotManager;
import org.apache.flink.table.types.logical.RowType;
@@ -148,7 +147,7 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> {
return new NoopCompactManager();
} else {
Comparator<RowData> keyComparator = keyComparatorSupplier.get();
- CompactRewriter rewriter = compactRewriter(partition, bucket, keyComparator);
+ CompactRewriter rewriter = createRewriter(partition, bucket, keyComparator);
return new MergeTreeCompactManager(
compactExecutor,
levels,
@@ -160,22 +159,21 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> {
}
}
- private CompactRewriter compactRewriter(
+ private MergeTreeCompactRewriter createRewriter(
BinaryRowData partition, int bucket, Comparator<RowData> keyComparator) {
+ KeyValueFileReaderFactory readerFactory = readerFactoryBuilder.build(partition, bucket);
KeyValueFileWriterFactory writerFactory = writerFactoryBuilder.build(partition, bucket);
- return (outputLevel, dropDelete, sections) -> {
- RollingFileWriter<KeyValue, DataFileMeta> writer =
- writerFactory.createRollingMergeTreeFileWriter(outputLevel);
- writer.write(
- new RecordReaderIterator<>(
- new MergeTreeReader(
- sections,
- dropDelete,
- readerFactoryBuilder.build(partition, bucket),
- keyComparator,
- mergeFunction.copy())));
- writer.close();
- return writer.result();
- };
+
+ if (options.changelogProducer() == CoreOptions.ChangelogProducer.FULL_COMPACTION) {
+ return new FullChangelogMergeTreeCompactRewriter(
+ options.numLevels() - 1,
+ readerFactory,
+ writerFactory,
+ keyComparator,
+ mergeFunction);
+ } else {
+ return new MergeTreeCompactRewriter(
+ readerFactory, writerFactory, keyComparator, mergeFunction);
+ }
}
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/RecordWriter.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/RecordWriter.java
index 4acc1dd0..3c9e6946 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/RecordWriter.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/RecordWriter.java
@@ -33,7 +33,10 @@ public interface RecordWriter<T> {
/** Add a key-value element to the writer. */
void write(T record) throws Exception;
- /** Compact all files related to the writer. */
+ /**
+ * Compact all files related to the writer. Note that compaction process is only submitted and
+ * may not be completed when the method returns.
+ */
void fullCompaction() throws Exception;
/**
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/MergeTreeTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/MergeTreeTest.java
index c5fa6a1f..b846f565 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/MergeTreeTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/MergeTreeTest.java
@@ -28,13 +28,14 @@ import org.apache.flink.table.data.binary.BinaryRowDataUtil;
import org.apache.flink.table.store.CoreOptions;
import org.apache.flink.table.store.CoreOptions.ChangelogProducer;
import org.apache.flink.table.store.file.KeyValue;
+import org.apache.flink.table.store.file.compact.CompactResult;
import org.apache.flink.table.store.file.format.FlushingFileFormat;
import org.apache.flink.table.store.file.io.DataFileMeta;
import org.apache.flink.table.store.file.io.KeyValueFileReaderFactory;
import org.apache.flink.table.store.file.io.KeyValueFileWriterFactory;
import org.apache.flink.table.store.file.io.RollingFileWriter;
import org.apache.flink.table.store.file.memory.HeapMemorySegmentPool;
-import org.apache.flink.table.store.file.mergetree.compact.CompactRewriter;
+import org.apache.flink.table.store.file.mergetree.compact.AbstractCompactRewriter;
import org.apache.flink.table.store.file.mergetree.compact.CompactStrategy;
import org.apache.flink.table.store.file.mergetree.compact.DeduplicateMergeFunction;
import org.apache.flink.table.store.file.mergetree.compact.IntervalPartition;
@@ -75,7 +76,7 @@ import java.util.stream.Collectors;
import static java.util.Collections.singletonList;
import static org.assertj.core.api.Assertions.assertThat;
-/** Tests for {@link MergeTreeReader} and {@link MergeTreeWriter}. */
+/** Tests for {@link MergeTreeReaders} and {@link MergeTreeWriter}. */
public class MergeTreeTest {
@TempDir java.nio.file.Path tempDir;
@@ -259,7 +260,7 @@ public class MergeTreeTest {
false,
128,
null,
- createCompactManager(writerFactory, service, files),
+ createCompactManager(service, files),
maxSequenceNumber,
comparator,
new DeduplicateMergeFunction(),
@@ -272,30 +273,13 @@ public class MergeTreeTest {
}
private MergeTreeCompactManager createCompactManager(
- KeyValueFileWriterFactory writerFactory,
- ExecutorService compactExecutor,
- List<DataFileMeta> files) {
+ ExecutorService compactExecutor, List<DataFileMeta> files) {
CompactStrategy strategy =
new UniversalCompaction(
options.maxSizeAmplificationPercent(),
options.sortedRunSizeRatio(),
options.numSortedRunCompactionTrigger(),
options.maxSortedRunNum());
- CompactRewriter rewriter =
- (outputLevel, dropDelete, sections) -> {
- RollingFileWriter<KeyValue, DataFileMeta> writer =
- compactWriterFactory.createRollingMergeTreeFileWriter(outputLevel);
- writer.write(
- new RecordReaderIterator<>(
- new MergeTreeReader(
- sections,
- dropDelete,
- compactReaderFactory,
- comparator,
- new DeduplicateMergeFunction())));
- writer.close();
- return writer.result();
- };
return new MergeTreeCompactManager(
compactExecutor,
new Levels(comparator, files, options.numLevels()),
@@ -303,7 +287,7 @@ public class MergeTreeTest {
comparator,
options.targetFileSize(),
options.numSortedRunStopTrigger(),
- rewriter);
+ new TestRewriter());
}
private void mergeCompacted(
@@ -377,7 +361,7 @@ public class MergeTreeTest {
private List<TestRecord> readAll(List<DataFileMeta> files, boolean dropDelete)
throws Exception {
RecordReader<KeyValue> reader =
- new MergeTreeReader(
+ MergeTreeReaders.readerForMergeTree(
new IntervalPartition(files, comparator).partition(),
dropDelete,
readerFactory,
@@ -411,6 +395,27 @@ public class MergeTreeTest {
return records;
}
+ private class TestRewriter extends AbstractCompactRewriter {
+
+ @Override
+ public CompactResult rewrite(
+ int outputLevel, boolean dropDelete, List<List<SortedRun>> sections)
+ throws Exception {
+ RollingFileWriter<KeyValue, DataFileMeta> writer =
+ writerFactory.createRollingMergeTreeFileWriter(outputLevel);
+ RecordReader<KeyValue> sectionsReader =
+ MergeTreeReaders.readerForMergeTree(
+ sections,
+ dropDelete,
+ compactReaderFactory,
+ comparator,
+ new DeduplicateMergeFunction());
+ writer.write(new RecordReaderIterator<>(sectionsReader));
+ writer.close();
+ return new CompactResult(extractFilesFromSections(sections), writer.result());
+ }
+ }
+
private static class TestRecord {
private final RowKind kind;
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/FullChangelogMergeFunctionWrapperTestBase.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/FullChangelogMergeFunctionWrapperTestBase.java
index 4e2c606c..0fdceb8b 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/FullChangelogMergeFunctionWrapperTestBase.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/FullChangelogMergeFunctionWrapperTestBase.java
@@ -126,12 +126,12 @@ public abstract class FullChangelogMergeFunctionWrapperTestBase {
private static final List<KeyValue> EXPECTED_RESULT =
Arrays.asList(
new KeyValue().replace(row(1), 1, RowKind.INSERT, row(1)),
- new KeyValue().replace(row(2), 2, RowKind.DELETE, row(0)),
+ null,
new KeyValue().replace(row(3), 3, RowKind.INSERT, row(3)),
new KeyValue().replace(row(4), 5, RowKind.INSERT, row(-3)),
- new KeyValue().replace(row(5), 7, RowKind.DELETE, row(3)),
+ null,
new KeyValue().replace(row(6), 9, RowKind.INSERT, row(-3)),
- new KeyValue().replace(row(7), 11, RowKind.DELETE, row(3)));
+ null);
@Override
protected MergeFunction<KeyValue> createMergeFunction() {
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/MergeTreeCompactManagerTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/MergeTreeCompactManagerTest.java
index ace9965b..3f4f5161 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/MergeTreeCompactManagerTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/MergeTreeCompactManagerTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.store.file.mergetree.compact;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.store.file.compact.CompactResult;
import org.apache.flink.table.store.file.compact.CompactUnit;
import org.apache.flink.table.store.file.io.DataFileMeta;
import org.apache.flink.table.store.file.io.DataFileTestUtils;
@@ -203,7 +204,7 @@ public class MergeTreeCompactManagerTest {
comparator,
2,
Integer.MAX_VALUE,
- testRewriter(expectedDropDelete));
+ new TestRewriter(expectedDropDelete));
manager.triggerCompaction(false);
manager.getCompactionResult(true);
List<LevelMinMax> outputs =
@@ -219,8 +220,18 @@ public class MergeTreeCompactManagerTest {
return (numLevels, runs) -> Optional.of(CompactUnit.fromLevelRuns(numLevels - 1, runs));
}
- private CompactRewriter testRewriter(boolean expectedDropDelete) {
- return (outputLevel, dropDelete, sections) -> {
+ private static class TestRewriter extends AbstractCompactRewriter {
+
+ private final boolean expectedDropDelete;
+
+ private TestRewriter(boolean expectedDropDelete) {
+ this.expectedDropDelete = expectedDropDelete;
+ }
+
+ @Override
+ public CompactResult rewrite(
+ int outputLevel, boolean dropDelete, List<List<SortedRun>> sections)
+ throws Exception {
assertThat(dropDelete).isEqualTo(expectedDropDelete);
int minKey = Integer.MAX_VALUE;
int maxKey = Integer.MIN_VALUE;
@@ -242,8 +253,10 @@ public class MergeTreeCompactManagerTest {
}
}
}
- return Collections.singletonList(newFile(outputLevel, minKey, maxKey, maxSequence));
- };
+ return new CompactResult(
+ extractFilesFromSections(sections),
+ Collections.singletonList(newFile(outputLevel, minKey, maxKey, maxSequence)));
+ }
}
private static class LevelMinMax {
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReaderTestBase.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReaderTestBase.java
index 800d927e..ff85a342 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReaderTestBase.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReaderTestBase.java
@@ -36,7 +36,10 @@ public abstract class SortMergeReaderTestBase extends CombiningRecordReaderTestB
@Override
protected RecordReader<KeyValue> createRecordReader(List<TestReusingRecordReader> readers) {
- return new SortMergeReader(new ArrayList<>(readers), KEY_COMPARATOR, createMergeFunction());
+ return new SortMergeReader<>(
+ new ArrayList<>(readers),
+ KEY_COMPARATOR,
+ new ReducerMergeFunctionWrapper(createMergeFunction()));
}
@Test
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreCommitTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreCommitTest.java
index 09d819c9..21f08f32 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreCommitTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreCommitTest.java
@@ -92,14 +92,28 @@ public class FileStoreCommitTest {
}
@ParameterizedTest
- @CsvSource({"false,NONE", "false,INPUT", "true,NONE", "true,INPUT"})
+ @CsvSource({
+ "false,NONE",
+ "false,INPUT",
+ "true,NONE",
+ "true,INPUT",
+ "false,FULL_COMPACTION",
+ "true,FULL_COMPACTION"
+ })
public void testSingleCommitUser(boolean failing, String changelogProducer) throws Exception {
testRandomConcurrentNoConflict(
1, failing, CoreOptions.ChangelogProducer.valueOf(changelogProducer));
}
@ParameterizedTest
- @CsvSource({"false,NONE", "false,INPUT", "true,NONE", "true,INPUT"})
+ @CsvSource({
+ "false,NONE",
+ "false,INPUT",
+ "true,NONE",
+ "true,INPUT",
+ "false,FULL_COMPACTION",
+ "true,FULL_COMPACTION"
+ })
public void testManyCommitUsersNoConflict(boolean failing, String changelogProducer)
throws Exception {
testRandomConcurrentNoConflict(
@@ -109,7 +123,14 @@ public class FileStoreCommitTest {
}
@ParameterizedTest
- @CsvSource({"false,NONE", "false,INPUT", "true,NONE", "true,INPUT"})
+ @CsvSource({
+ "false,NONE",
+ "false,INPUT",
+ "true,NONE",
+ "true,INPUT",
+ "false,FULL_COMPACTION",
+ "true,FULL_COMPACTION"
+ })
public void testManyCommitUsersWithConflict(boolean failing, String changelogProducer)
throws Exception {
testRandomConcurrentWithConflict(
@@ -256,6 +277,45 @@ public class FileStoreCommitTest {
Map<BinaryRowData, BinaryRowData> actualChangelogMap = store.toKvMap(actualChangelog);
logData(() -> kvMapToKvList(actualChangelogMap), "actual changelog map");
assertThat(actualChangelogMap).isEqualTo(expected);
+
+ if (changelogProducer == CoreOptions.ChangelogProducer.FULL_COMPACTION) {
+ validateFullChangelog(actualChangelog);
+ }
+ }
+ }
+
+ private void validateFullChangelog(List<KeyValue> changelog) {
+ Map<BinaryRowData, KeyValue> kvMap = new HashMap<>();
+ Map<BinaryRowData, RowKind> kindMap = new HashMap<>();
+ for (KeyValue kv : changelog) {
+ BinaryRowData key = TestKeyValueGenerator.KEY_SERIALIZER.toBinaryRow(kv.key()).copy();
+ switch (kv.valueKind()) {
+ case INSERT:
+ assertThat(kvMap).doesNotContainKey(key);
+ if (kindMap.containsKey(key)) {
+ assertThat(kindMap.get(key)).isEqualTo(RowKind.DELETE);
+ }
+ kvMap.put(key, kv);
+ kindMap.put(key, RowKind.INSERT);
+ break;
+ case UPDATE_AFTER:
+ assertThat(kvMap).doesNotContainKey(key);
+ assertThat(kindMap.get(key)).isEqualTo(RowKind.UPDATE_BEFORE);
+ kvMap.put(key, kv);
+ kindMap.put(key, RowKind.UPDATE_AFTER);
+ break;
+ case UPDATE_BEFORE:
+ case DELETE:
+ assertThat(kvMap).containsKey(key);
+ assertThat(kv.value()).isEqualTo(kvMap.get(key).value());
+ assertThat(kindMap.get(key)).isIn(RowKind.INSERT, RowKind.UPDATE_AFTER);
+ kvMap.remove(key);
+ kindMap.put(key, kv.valueKind());
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ "Unknown value kind " + kv.valueKind().name());
+ }
}
}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/TestCommitThread.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/TestCommitThread.java
index a08bda5e..fbedd536 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/TestCommitThread.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/TestCommitThread.java
@@ -36,9 +36,11 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -60,6 +62,7 @@ public class TestCommitThread extends Thread {
private final Map<BinaryRowData, List<KeyValue>> data;
private final Map<BinaryRowData, List<KeyValue>> result;
private final Map<BinaryRowData, MergeTreeWriter> writers;
+ private final Set<BinaryRowData> writtenPartitions;
private final AbstractFileStoreWrite<KeyValue> write;
private final FileStoreCommit commit;
@@ -77,6 +80,7 @@ public class TestCommitThread extends Thread {
this.data = data;
this.result = new HashMap<>();
this.writers = new HashMap<>();
+ this.writtenPartitions = new HashSet<>();
if (LOG.isDebugEnabled()) {
LOG.debug(
@@ -109,6 +113,7 @@ public class TestCommitThread extends Thread {
throw new RuntimeException(e);
}
}
+ doFinalCompact();
for (MergeTreeWriter writer : writers.values()) {
try {
@@ -153,6 +158,31 @@ public class TestCommitThread extends Thread {
Collections.emptyMap()));
}
+ private void doFinalCompact() {
+ while (true) {
+ try {
+ ManifestCommittable committable =
+ new ManifestCommittable(String.valueOf(new Random().nextLong()));
+ for (BinaryRowData partition : writtenPartitions) {
+ MergeTreeWriter writer =
+ writers.computeIfAbsent(partition, p -> createWriter(p, false));
+ writer.fullCompaction();
+ RecordWriter.CommitIncrement inc = writer.prepareCommit(true);
+ committable.addFileCommittable(
+ new FileCommittable(
+ partition, 0, inc.newFilesIncrement(), inc.compactIncrement()));
+ }
+ commit.commit(committable, Collections.emptyMap());
+ break;
+ } catch (Exception e) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Failed to do final compact because of exception, try again", e);
+ }
+ writers.clear();
+ }
+ }
+ }
+
private void runWithRetry(ManifestCommittable committable, Runnable runnable) {
boolean shouldCheckFilter = false;
while (true) {
@@ -178,6 +208,7 @@ public class TestCommitThread extends Thread {
List<KeyValue> changes = new ArrayList<>();
BinaryRowData partition = pickData(changes);
result.computeIfAbsent(partition, p -> new ArrayList<>()).addAll(changes);
+ writtenPartitions.add(partition);
if (LOG.isDebugEnabled()) {
LOG.debug(
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
index b0d96233..c710cbc6 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
@@ -167,7 +167,7 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
}
@Test
- public void testStreamingChangelog() throws Exception {
+ public void testStreamingInputChangelog() throws Exception {
FileStoreTable table =
createFileStoreTable(
conf -> conf.set(CoreOptions.CHANGELOG_PRODUCER, ChangelogProducer.INPUT));
@@ -198,6 +198,85 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
"+U 1|10|102|binary|varbinary");
}
+ @Test
+ public void testStreamingFullChangelog() throws Exception {
+ FileStoreTable table =
+ createFileStoreTable(
+ conf ->
+ conf.set(
+ CoreOptions.CHANGELOG_PRODUCER,
+ ChangelogProducer.FULL_COMPACTION));
+ TableWrite write = table.newWrite();
+ TableCommit commit = table.newCommit("user");
+
+ write.write(rowData(1, 10, 110L));
+ write.write(rowData(1, 20, 120L));
+ write.write(rowData(2, 10, 210L));
+ write.write(rowData(2, 20, 220L));
+ write.write(rowDataWithKind(RowKind.DELETE, 2, 10, 210L));
+ write.compact(binaryRow(1), 0);
+ write.compact(binaryRow(2), 0);
+ commit.commit("0", write.prepareCommit(true));
+
+ List<Split> splits = table.newScan().withIncremental(true).plan().splits();
+ TableRead read = table.newRead();
+ assertThat(getResult(read, splits, binaryRow(1), 0, CHANGELOG_ROW_TO_STRING))
+ .containsExactlyInAnyOrder(
+ "+I 1|10|110|binary|varbinary", "+I 1|20|120|binary|varbinary");
+ assertThat(getResult(read, splits, binaryRow(2), 0, CHANGELOG_ROW_TO_STRING))
+ .containsExactlyInAnyOrder("+I 2|20|220|binary|varbinary");
+
+ write.write(rowData(1, 30, 130L));
+ write.write(rowData(1, 40, 140L));
+ write.write(rowData(2, 30, 230L));
+ write.write(rowData(2, 40, 240L));
+ commit.commit("1", write.prepareCommit(true));
+
+ write.write(rowDataWithKind(RowKind.DELETE, 1, 40, 140L));
+ write.write(rowData(2, 40, 241L));
+ write.compact(binaryRow(1), 0);
+ write.compact(binaryRow(2), 0);
+ commit.commit("2", write.prepareCommit(true));
+
+ splits = table.newScan().withIncremental(true).plan().splits();
+ assertThat(getResult(read, splits, binaryRow(1), 0, CHANGELOG_ROW_TO_STRING))
+ .containsExactlyInAnyOrder("+I 1|30|130|binary|varbinary");
+ assertThat(getResult(read, splits, binaryRow(2), 0, CHANGELOG_ROW_TO_STRING))
+ .containsExactlyInAnyOrder(
+ "+I 2|30|230|binary|varbinary", "+I 2|40|241|binary|varbinary");
+
+ write.write(rowData(1, 20, 121L));
+ write.write(rowData(1, 30, 131L));
+ write.write(rowData(2, 30, 231L));
+ commit.commit("3", write.prepareCommit(true));
+
+ write.write(rowDataWithKind(RowKind.DELETE, 1, 20, 121L));
+ write.write(rowData(1, 30, 132L));
+ write.write(rowData(1, 40, 141L));
+ write.write(rowDataWithKind(RowKind.DELETE, 2, 20, 220L));
+ write.write(rowData(2, 20, 221L));
+ write.write(rowDataWithKind(RowKind.DELETE, 2, 20, 221L));
+ write.write(rowData(2, 40, 242L));
+ write.compact(binaryRow(1), 0);
+ write.compact(binaryRow(2), 0);
+ commit.commit("4", write.prepareCommit(true));
+
+ splits = table.newScan().withIncremental(true).plan().splits();
+ assertThat(getResult(read, splits, binaryRow(1), 0, CHANGELOG_ROW_TO_STRING))
+ .containsExactlyInAnyOrder(
+ "-D 1|20|120|binary|varbinary",
+ "-U 1|30|130|binary|varbinary",
+ "+U 1|30|132|binary|varbinary",
+ "+I 1|40|141|binary|varbinary");
+ assertThat(getResult(read, splits, binaryRow(2), 0, CHANGELOG_ROW_TO_STRING))
+ .containsExactlyInAnyOrder(
+ "-D 2|20|220|binary|varbinary",
+ "-U 2|30|230|binary|varbinary",
+ "+U 2|30|231|binary|varbinary",
+ "-U 2|40|241|binary|varbinary",
+ "+U 2|40|242|binary|varbinary");
+ }
+
@Test
public void testStreamingChangelogCompatibility02() throws Exception {
// already contains 2 commits