You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by cz...@apache.org on 2022/10/10 08:43:03 UTC
[flink-table-store] branch master updated: [FLINK-29293] Introduce a MergeFunction for full compaction
This is an automated email from the ASF dual-hosted git repository.
czweng 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 c2666d04 [FLINK-29293] Introduce a MergeFunction for full compaction
c2666d04 is described below
commit c2666d04c488555fecdc06ed83c7ae2b1339a072
Author: tsreaper <ts...@gmail.com>
AuthorDate: Mon Oct 10 16:42:58 2022 +0800
[FLINK-29293] Introduce a MergeFunction for full compaction
This closes #312.
---
.../store/file/mergetree/SortBufferMemTable.java | 14 +-
.../compact/FullChangelogMergeFunctionWrapper.java | 187 +++++++++++++++++++++
.../file/mergetree/compact/MergeFunction.java | 6 +-
...ergeFunction.java => MergeFunctionWrapper.java} | 16 +-
...elper.java => ReducerMergeFunctionWrapper.java} | 33 ++--
.../file/mergetree/compact/SortMergeReader.java | 10 +-
.../FullChangelogMergeFunctionWrapperTestBase.java | 156 +++++++++++++++++
.../mergetree/compact/MergeFunctionTestUtils.java | 14 ++
...va => ReducerMergeFunctionWrapperTestBase.java} | 40 ++---
9 files changed, 420 insertions(+), 56 deletions(-)
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/SortBufferMemTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/SortBufferMemTable.java
index 1f472d00..5d442126 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/SortBufferMemTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/SortBufferMemTable.java
@@ -32,7 +32,7 @@ import org.apache.flink.table.store.file.KeyValue;
import org.apache.flink.table.store.file.KeyValueSerializer;
import org.apache.flink.table.store.file.memory.sort.BinaryInMemorySortBuffer;
import org.apache.flink.table.store.file.mergetree.compact.MergeFunction;
-import org.apache.flink.table.store.file.mergetree.compact.MergeFunctionHelper;
+import org.apache.flink.table.store.file.mergetree.compact.ReducerMergeFunctionWrapper;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
@@ -124,7 +124,7 @@ public class SortBufferMemTable implements MemTable {
private class MergeIterator implements Iterator<KeyValue> {
private final MutableObjectIterator<BinaryRowData> kvIter;
private final Comparator<RowData> keyComparator;
- private final MergeFunctionHelper mergeFunctionHelper;
+ private final ReducerMergeFunctionWrapper mergeFunctionWrapper;
// previously read kv
private KeyValueSerializer previous;
@@ -142,7 +142,7 @@ public class SortBufferMemTable implements MemTable {
MergeFunction<KeyValue> mergeFunction) {
this.kvIter = kvIter;
this.keyComparator = keyComparator;
- this.mergeFunctionHelper = new MergeFunctionHelper(mergeFunction);
+ this.mergeFunctionWrapper = new ReducerMergeFunctionWrapper(mergeFunction);
int totalFieldCount = keyType.getFieldCount() + 2 + valueType.getFieldCount();
this.previous = new KeyValueSerializer(keyType, valueType);
@@ -180,8 +180,8 @@ public class SortBufferMemTable implements MemTable {
if (previousRow == null) {
return;
}
- mergeFunctionHelper.reset();
- mergeFunctionHelper.add(previous.getReusedKv());
+ mergeFunctionWrapper.reset();
+ mergeFunctionWrapper.add(previous.getReusedKv());
while (readOnce()) {
if (keyComparator.compare(
@@ -189,10 +189,10 @@ public class SortBufferMemTable implements MemTable {
!= 0) {
break;
}
- mergeFunctionHelper.add(current.getReusedKv());
+ mergeFunctionWrapper.add(current.getReusedKv());
swapSerializers();
}
- result = mergeFunctionHelper.getResult();
+ result = mergeFunctionWrapper.getResult();
} while (result == null);
}
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
new file mode 100644
index 00000000..293fd01d
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/FullChangelogMergeFunctionWrapper.java
@@ -0,0 +1,187 @@
+/*
+ * 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.store.file.KeyValue;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nullable;
+
+/**
+ * Wrapper for {@link MergeFunction}s to produce changelog during a full compaction.
+ *
+ * <p>This wrapper can only be used in {@link SortMergeReader} because
+ *
+ * <ul>
+ * <li>This wrapper does not copy {@link KeyValue}s. As {@link KeyValue}s are reused by readers
+ * this may cause issues in other readers.
+ * <li>{@link KeyValue}s with the same key come from different inner readers in {@link
+ * SortMergeReader}, so there is no issue related to object reuse.
+ * </ul>
+ */
+public class FullChangelogMergeFunctionWrapper
+ implements MergeFunctionWrapper<FullChangelogMergeFunctionWrapper.Result> {
+
+ private final MergeFunction<KeyValue> mergeFunction;
+ private final int maxLevel;
+
+ // only full compaction will write files into maxLevel, see UniversalCompaction class
+ private transient KeyValue topLevelKv;
+ private transient KeyValue initialKv;
+ private transient boolean isInitialized;
+
+ private transient Result reusedResult;
+ private transient KeyValue reusedBefore;
+ private transient KeyValue reusedAfter;
+
+ public FullChangelogMergeFunctionWrapper(MergeFunction<KeyValue> mergeFunction, int maxLevel) {
+ Preconditions.checkArgument(
+ !(mergeFunction instanceof ValueCountMergeFunction),
+ "Value count merge function does not need to produce changelog from full compaction. "
+ + "Please set changelog producer to 'input'.");
+ this.mergeFunction = mergeFunction;
+ this.maxLevel = maxLevel;
+ }
+
+ @Override
+ public void reset() {
+ mergeFunction.reset();
+
+ topLevelKv = null;
+ initialKv = null;
+ isInitialized = false;
+ }
+
+ @Override
+ public void add(KeyValue kv) {
+ if (maxLevel == kv.level()) {
+ Preconditions.checkState(
+ topLevelKv == null, "Top level key-value already exists! This is unexpected.");
+ topLevelKv = kv;
+ }
+
+ if (initialKv == null) {
+ initialKv = kv;
+ } else {
+ if (!isInitialized) {
+ merge(initialKv);
+ isInitialized = true;
+ }
+ merge(kv);
+ }
+ }
+
+ private void merge(KeyValue kv) {
+ mergeFunction.add(kv);
+ }
+
+ @Override
+ public Result getResult() {
+ if (reusedResult == null) {
+ reusedResult = new Result();
+ reusedBefore = new KeyValue();
+ reusedAfter = new KeyValue();
+ }
+
+ if (isInitialized) {
+ KeyValue merged = mergeFunction.getResult();
+ if (topLevelKv == null) {
+ if (merged != null && isAdd(merged)) {
+ reusedResult.setChangelog(null, replace(reusedAfter, RowKind.INSERT, merged));
+ } else {
+ reusedResult.setChangelog(null, null);
+ }
+ } else {
+ if (merged != null && isAdd(merged)) {
+ reusedResult.setChangelog(
+ replace(reusedBefore, RowKind.UPDATE_BEFORE, topLevelKv),
+ replace(reusedAfter, RowKind.UPDATE_AFTER, merged));
+ } else {
+ reusedResult.setChangelog(
+ replace(reusedBefore, RowKind.DELETE, topLevelKv), null);
+ }
+ }
+ return reusedResult.setResult(merged);
+ } else {
+ if (topLevelKv == null && isAdd(initialKv)) {
+ reusedResult.setChangelog(null, replace(reusedAfter, RowKind.INSERT, initialKv));
+ } else {
+ // either topLevelKv is not null, but there is only one kv,
+ // so topLevelKv must be the only kv, which means there is no change
+ //
+ // or initialKv is not an ADD kv, so no new key is added
+ reusedResult.setChangelog(null, null);
+ }
+ return reusedResult.setResult(initialKv);
+ }
+ }
+
+ private KeyValue replace(KeyValue reused, RowKind valueKind, KeyValue from) {
+ return reused.replace(from.key(), from.sequenceNumber(), valueKind, from.value());
+ }
+
+ private boolean isAdd(KeyValue kv) {
+ return kv.valueKind() == RowKind.INSERT || kv.valueKind() == RowKind.UPDATE_AFTER;
+ }
+
+ /** Changelog and final result for the same key. */
+ public static class Result {
+
+ @Nullable private KeyValue before;
+ @Nullable private KeyValue after;
+ @Nullable private KeyValue result;
+
+ private Result() {}
+
+ private void setChangelog(@Nullable KeyValue before, @Nullable KeyValue after) {
+ this.before = before;
+ this.after = after;
+ }
+
+ private Result setResult(@Nullable KeyValue result) {
+ this.result = result;
+ return this;
+ }
+
+ /**
+ * Previous full compaction result for this key. Null if this key does not appear in the
+ * previous result or its value remains unchanged.
+ */
+ @Nullable
+ public KeyValue before() {
+ return before;
+ }
+
+ /**
+ * Latest full compaction result for this key. Null if this key does not appear in the
+ * latest result or its value remains unchanged.
+ */
+ @Nullable
+ public KeyValue after() {
+ return after;
+ }
+
+ /** Latest full compaction result (result of merge function) for this key. */
+ @Nullable
+ public KeyValue result() {
+ return result;
+ }
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunction.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunction.java
index 125d2768..95eda7fd 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunction.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunction.java
@@ -24,7 +24,11 @@ import javax.annotation.Nullable;
import java.io.Serializable;
-/** Merge function to merge multiple {@link KeyValue}s. */
+/**
+ * Merge function to merge multiple {@link KeyValue}s.
+ *
+ * @param <T> result type
+ */
public interface MergeFunction<T> extends Serializable {
/** Reset the merge function to its default state. */
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunction.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionWrapper.java
similarity index 68%
copy from flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunction.java
copy to flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionWrapper.java
index 125d2768..baf9b948 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunction.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionWrapper.java
@@ -22,21 +22,17 @@ import org.apache.flink.table.store.file.KeyValue;
import javax.annotation.Nullable;
-import java.io.Serializable;
-
-/** Merge function to merge multiple {@link KeyValue}s. */
-public interface MergeFunction<T> extends Serializable {
+/**
+ * A wrapper for {@link MergeFunction}, which adds new functionalities or optimizations.
+ *
+ * @param <T> result type
+ */
+public interface MergeFunctionWrapper<T> {
- /** Reset the merge function to its default state. */
void reset();
- /** Add the given {@link KeyValue} to the merge function. */
void add(KeyValue kv);
- /** Get current merged value. Return null if this merged result should be skipped. */
@Nullable
T getResult();
-
- /** Create a new merge function object with the same functionality as this one. */
- MergeFunction<T> copy();
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionHelper.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/ReducerMergeFunctionWrapper.java
similarity index 64%
rename from flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionHelper.java
rename to flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/ReducerMergeFunctionWrapper.java
index f82b8b2f..2e6f23f7 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionHelper.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/ReducerMergeFunctionWrapper.java
@@ -20,44 +20,55 @@ package org.apache.flink.table.store.file.mergetree.compact;
import org.apache.flink.table.store.file.KeyValue;
-/** Helper functions for the interaction with {@link MergeFunction}. */
-public class MergeFunctionHelper {
+/**
+ * Wrapper for {@link MergeFunction}s which works like a reducer.
+ *
+ * <p>A reducer is a type of function. If there is only one input the result is equal to that input;
+ * Otherwise the result is calculated by merging all the inputs in some way.
+ *
+ * <p>This wrapper optimize the wrapped {@link MergeFunction}. If there is only one input, the input
+ * will be stored and the inner merge function will not be called, thus saving some computing time.
+ */
+public class ReducerMergeFunctionWrapper implements MergeFunctionWrapper<KeyValue> {
private final MergeFunction<KeyValue> mergeFunction;
- private KeyValue initialKV;
- private boolean isInitialized;
+ private transient KeyValue initialKv;
+ private transient boolean isInitialized;
- public MergeFunctionHelper(MergeFunction<KeyValue> mergeFunction) {
+ public ReducerMergeFunctionWrapper(MergeFunction<KeyValue> mergeFunction) {
this.mergeFunction = mergeFunction;
}
/** Resets the {@link MergeFunction} helper to its default state. */
+ @Override
public void reset() {
- initialKV = null;
+ initialKv = null;
mergeFunction.reset();
isInitialized = false;
}
/** Adds the given {@link KeyValue} to the {@link MergeFunction} helper. */
+ @Override
public void add(KeyValue kv) {
- if (initialKV == null) {
- initialKV = kv;
+ if (initialKv == null) {
+ initialKv = kv;
} else {
if (!isInitialized) {
- merge(initialKV);
+ merge(initialKv);
isInitialized = true;
}
merge(kv);
}
}
- protected void merge(KeyValue kv) {
+ private void merge(KeyValue kv) {
mergeFunction.add(kv);
}
/** Get current value of the {@link MergeFunction} helper. */
+ @Override
public KeyValue getResult() {
- return isInitialized ? mergeFunction.getResult() : initialKV;
+ return isInitialized ? mergeFunction.getResult() : initialKv;
}
}
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 d981e3c5..d9215f86 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
@@ -42,7 +42,7 @@ public class SortMergeReader implements RecordReader<KeyValue> {
private final List<RecordReader<KeyValue>> nextBatchReaders;
private final Comparator<RowData> userKeyComparator;
- private final MergeFunctionHelper mergeFunctionHelper;
+ private final MergeFunctionWrapper<KeyValue> mergeFunctionWrapper;
private final PriorityQueue<Element> minHeap;
private final List<Element> polled;
@@ -53,7 +53,7 @@ public class SortMergeReader implements RecordReader<KeyValue> {
MergeFunction<KeyValue> mergeFunction) {
this.nextBatchReaders = new ArrayList<>(readers);
this.userKeyComparator = userKeyComparator;
- this.mergeFunctionHelper = new MergeFunctionHelper(mergeFunction);
+ this.mergeFunctionWrapper = new ReducerMergeFunctionWrapper(mergeFunction);
this.minHeap =
new PriorityQueue<>(
@@ -130,7 +130,7 @@ public class SortMergeReader implements RecordReader<KeyValue> {
if (!hasMore) {
return null;
}
- KeyValue result = mergeFunctionHelper.getResult();
+ KeyValue result = mergeFunctionWrapper.getResult();
if (result != null) {
return result;
}
@@ -163,7 +163,7 @@ public class SortMergeReader implements RecordReader<KeyValue> {
return false;
}
- mergeFunctionHelper.reset();
+ mergeFunctionWrapper.reset();
RowData key =
Preconditions.checkNotNull(minHeap.peek(), "Min heap is empty. This is a bug.")
.kv
@@ -177,7 +177,7 @@ public class SortMergeReader implements RecordReader<KeyValue> {
break;
}
minHeap.poll();
- mergeFunctionHelper.add(element.kv);
+ mergeFunctionWrapper.add(element.kv);
polled.add(element);
}
return true;
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
new file mode 100644
index 00000000..4e2c606c
--- /dev/null
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/FullChangelogMergeFunctionWrapperTestBase.java
@@ -0,0 +1,156 @@
+/*
+ * 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.store.file.KeyValue;
+import org.apache.flink.types.RowKind;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.flink.table.store.file.io.DataFileTestUtils.row;
+
+/** Tests for {@link FullChangelogMergeFunctionWrapper}. */
+public abstract class FullChangelogMergeFunctionWrapperTestBase {
+
+ private static final int MAX_LEVEL = 3;
+
+ protected FullChangelogMergeFunctionWrapper wrapper;
+
+ protected abstract MergeFunction<KeyValue> createMergeFunction();
+
+ @BeforeEach
+ public void beforeEach() {
+ wrapper = new FullChangelogMergeFunctionWrapper(createMergeFunction(), MAX_LEVEL);
+ }
+
+ private static final List<List<KeyValue>> INPUT_KVS =
+ Arrays.asList(
+ // only 1 insert record, not from top level
+ Collections.singletonList(
+ new KeyValue().replace(row(1), 1, RowKind.INSERT, row(1)).setLevel(0)),
+ // only 1 delete record, not from top level
+ Collections.singletonList(
+ new KeyValue().replace(row(2), 2, RowKind.DELETE, row(0)).setLevel(0)),
+ // only 1 insert record, from top level
+ Collections.singletonList(
+ new KeyValue()
+ .replace(row(3), 3, RowKind.INSERT, row(3))
+ .setLevel(MAX_LEVEL)),
+ // multiple records, none from top level
+ Arrays.asList(
+ new KeyValue().replace(row(4), 4, RowKind.INSERT, row(3)).setLevel(0),
+ new KeyValue().replace(row(4), 5, RowKind.INSERT, row(-3)).setLevel(0)),
+ Arrays.asList(
+ new KeyValue().replace(row(5), 6, RowKind.INSERT, row(3)).setLevel(0),
+ new KeyValue().replace(row(5), 7, RowKind.DELETE, row(3)).setLevel(0)),
+ // multiple records, one from top level
+ Arrays.asList(
+ new KeyValue()
+ .replace(row(6), 8, RowKind.INSERT, row(3))
+ .setLevel(MAX_LEVEL),
+ new KeyValue().replace(row(6), 9, RowKind.INSERT, row(-3)).setLevel(0)),
+ Arrays.asList(
+ new KeyValue()
+ .replace(row(7), 10, RowKind.INSERT, row(3))
+ .setLevel(MAX_LEVEL),
+ new KeyValue()
+ .replace(row(7), 11, RowKind.DELETE, row(3))
+ .setLevel(0)));
+
+ protected abstract KeyValue getExpectedBefore(int idx);
+
+ protected abstract KeyValue getExpectedAfter(int idx);
+
+ protected abstract KeyValue getExpectedResult(int idx);
+
+ @Test
+ public void testFullChangelogMergeFunctionWrapper() {
+ for (int i = 0; i < INPUT_KVS.size(); i++) {
+ wrapper.reset();
+ List<KeyValue> kvs = INPUT_KVS.get(i);
+ kvs.forEach(kv -> wrapper.add(kv));
+ FullChangelogMergeFunctionWrapper.Result actualResult = wrapper.getResult();
+ MergeFunctionTestUtils.assertKvEquals(getExpectedBefore(i), actualResult.before());
+ MergeFunctionTestUtils.assertKvEquals(getExpectedAfter(i), actualResult.after());
+ MergeFunctionTestUtils.assertKvEquals(getExpectedResult(i), actualResult.result());
+ }
+ }
+
+ /**
+ * Tests for {@link FullChangelogMergeFunctionWrapper} with {@link DeduplicateMergeFunction}.
+ */
+ public static class WithDeduplicateMergeFunctionTest
+ extends FullChangelogMergeFunctionWrapperTestBase {
+
+ private static final List<KeyValue> EXPECTED_BEFORE =
+ Arrays.asList(
+ null,
+ null,
+ null,
+ null,
+ null,
+ new KeyValue().replace(row(6), 8, RowKind.UPDATE_BEFORE, row(3)),
+ new KeyValue().replace(row(7), 10, RowKind.DELETE, row(3)));
+
+ private static final List<KeyValue> EXPECTED_AFTER =
+ Arrays.asList(
+ new KeyValue().replace(row(1), 1, RowKind.INSERT, row(1)),
+ null,
+ null,
+ new KeyValue().replace(row(4), 5, RowKind.INSERT, row(-3)),
+ null,
+ new KeyValue().replace(row(6), 9, RowKind.UPDATE_AFTER, row(-3)),
+ null);
+
+ 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)),
+ 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)),
+ new KeyValue().replace(row(6), 9, RowKind.INSERT, row(-3)),
+ new KeyValue().replace(row(7), 11, RowKind.DELETE, row(3)));
+
+ @Override
+ protected MergeFunction<KeyValue> createMergeFunction() {
+ return new DeduplicateMergeFunction();
+ }
+
+ @Override
+ protected KeyValue getExpectedBefore(int idx) {
+ return EXPECTED_BEFORE.get(idx);
+ }
+
+ @Override
+ protected KeyValue getExpectedAfter(int idx) {
+ return EXPECTED_AFTER.get(idx);
+ }
+
+ @Override
+ protected KeyValue getExpectedResult(int idx) {
+ return EXPECTED_RESULT.get(idx);
+ }
+ }
+}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionTestUtils.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionTestUtils.java
index 4ff9b083..3324212d 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionTestUtils.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionTestUtils.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.store.file.mergetree.compact;
+import org.apache.flink.table.store.file.KeyValue;
import org.apache.flink.table.store.file.utils.ReusingTestData;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.Preconditions;
@@ -26,6 +27,8 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** Test utils for {@link MergeFunction}s. */
public class MergeFunctionTestUtils {
@@ -65,4 +68,15 @@ public class MergeFunctionTestUtils {
}
return expected;
}
+
+ public static void assertKvEquals(KeyValue expected, KeyValue actual) {
+ if (expected == null) {
+ assertThat(actual).isNull();
+ } else {
+ assertThat(actual.key()).isEqualTo(expected.key());
+ assertThat(actual.sequenceNumber()).isEqualTo(expected.sequenceNumber());
+ assertThat(actual.valueKind()).isEqualTo(expected.valueKind());
+ assertThat(actual.value()).isEqualTo(expected.value());
+ }
+ }
}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionHelperTestBase.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/ReducerMergeFunctionWrapperTestBase.java
similarity index 72%
rename from flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionHelperTestBase.java
rename to flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/ReducerMergeFunctionWrapperTestBase.java
index 9a2bdb81..73d7ba7c 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionHelperTestBase.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/ReducerMergeFunctionWrapperTestBase.java
@@ -35,12 +35,11 @@ import java.util.stream.Stream;
import static org.apache.flink.table.store.file.io.DataFileTestUtils.row;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-/** Tests for {@link MergeFunctionHelper}. */
-public abstract class MergeFunctionHelperTestBase {
+/** Tests for {@link ReducerMergeFunctionWrapper}. */
+public abstract class ReducerMergeFunctionWrapperTestBase {
- protected MergeFunctionHelper mergeFunctionHelper;
+ protected ReducerMergeFunctionWrapper wrapper;
protected abstract MergeFunction<KeyValue> createMergeFunction();
@@ -48,22 +47,19 @@ public abstract class MergeFunctionHelperTestBase {
@BeforeEach
void setUp() {
- mergeFunctionHelper = new MergeFunctionHelper(createMergeFunction());
+ wrapper = new ReducerMergeFunctionWrapper(createMergeFunction());
}
- @MethodSource("provideMergedKeyValues")
+ @MethodSource("provideKeyValuesToMerge")
@ParameterizedTest
- public void testMergeFunctionHelper(List<KeyValue> kvs) {
+ public void testReducerMergeFunctionWrapper(List<KeyValue> kvs) {
KeyValue expectedKv = getExpected(kvs);
- kvs.forEach(kv -> mergeFunctionHelper.add(kv));
- KeyValue mergedKv = mergeFunctionHelper.getResult();
- assertEquals(expectedKv.key(), mergedKv.key());
- assertEquals(expectedKv.sequenceNumber(), mergedKv.sequenceNumber());
- assertEquals(expectedKv.valueKind(), mergedKv.valueKind());
- assertEquals(expectedKv.value(), mergedKv.value());
+ kvs.forEach(kv -> wrapper.add(kv));
+ KeyValue mergedKv = wrapper.getResult();
+ MergeFunctionTestUtils.assertKvEquals(expectedKv, mergedKv);
}
- public static Stream<Arguments> provideMergedKeyValues() {
+ public static Stream<Arguments> provideKeyValuesToMerge() {
return Stream.of(
Arguments.of(
Collections.singletonList(
@@ -78,8 +74,9 @@ public abstract class MergeFunctionHelperTestBase {
new KeyValue().replace(row(3), 5, RowKind.INSERT, row(2)))));
}
- /** Tests for {@link MergeFunctionHelper} with {@link DeduplicateMergeFunction}. */
- public static class WithDeduplicateMergeFunctionTest extends MergeFunctionHelperTestBase {
+ /** Tests for {@link ReducerMergeFunctionWrapper} with {@link DeduplicateMergeFunction}. */
+ public static class WithDeduplicateMergeFunctionTest
+ extends ReducerMergeFunctionWrapperTestBase {
@Override
protected MergeFunction<KeyValue> createMergeFunction() {
@@ -92,8 +89,9 @@ public abstract class MergeFunctionHelperTestBase {
}
}
- /** Tests for {@link MergeFunctionHelper} with {@link ValueCountMergeFunction}. */
- public static class WithValueRecordMergeFunctionTest extends MergeFunctionHelperTestBase {
+ /** Tests for {@link ReducerMergeFunctionWrapper} with {@link ValueCountMergeFunction}. */
+ public static class WithValueRecordMergeFunctionTest
+ extends ReducerMergeFunctionWrapperTestBase {
@Override
protected MergeFunction<KeyValue> createMergeFunction() {
@@ -122,11 +120,9 @@ public abstract class MergeFunctionHelperTestBase {
@Test
public void testIllegalInput() {
- mergeFunctionHelper.add(new KeyValue().replace(null, RowKind.INSERT, row(1)));
+ wrapper.add(new KeyValue().replace(null, RowKind.INSERT, row(1)));
assertThatThrownBy(
- () ->
- mergeFunctionHelper.add(
- new KeyValue().replace(null, RowKind.DELETE, row(1))))
+ () -> wrapper.add(new KeyValue().replace(null, RowKind.DELETE, row(1))))
.hasMessageContaining(
"In value count mode, only insert records come. This is a bug. Please file an issue.");
}