This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 99ff847df [core] provide changelog-producer.row-deduplicate to
deduplicate same change (#1100)
99ff847df is described below
commit 99ff847df40cd10fc96e38eb80f6b7b5ac9fdf92
Author: liming.1018 <[email protected]>
AuthorDate: Tue May 9 16:08:49 2023 +0800
[core] provide changelog-producer.row-deduplicate to deduplicate same
change (#1100)
---
.../shortcodes/generated/core_configuration.html | 6 ++
.../main/java/org/apache/paimon/CoreOptions.java | 11 +++
.../java/org/apache/paimon/KeyValueFileStore.java | 4 +
.../compact/ChangelogMergeTreeRewriter.java | 9 ++-
.../compact/FullChangelogMergeFunctionWrapper.java | 20 ++++-
.../FullChangelogMergeTreeCompactRewriter.java | 16 +++-
.../LookupChangelogMergeFunctionWrapper.java | 16 +++-
.../compact/LookupMergeTreeCompactRewriter.java | 17 ++++-
.../paimon/operation/KeyValueFileStoreWrite.java | 11 ++-
.../paimon/utils/ValueComparatorSupplier.java | 46 ++++++++++++
.../FullChangelogMergeFunctionWrapperTestBase.java | 70 ++++++++++++++---
.../LookupChangelogMergeFunctionWrapperTest.java | 87 ++++++++++++++++++++--
.../flink/FullCompactionFileStoreITCase.java | 35 +++++++--
.../paimon/flink/LookupChangelogWithAggITCase.java | 30 +++++++-
.../flink/source/TestChangelogDataReadWrite.java | 1 +
15 files changed, 335 insertions(+), 44 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html
b/docs/layouts/shortcodes/generated/core_configuration.html
index 25c057926..418149ae3 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -32,6 +32,12 @@
<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><li>"full-compaction": Generate
changelog files with each full compaction.</li><li>"lookup": Generate changelog
files through 'lookup' before committing the d [...]
</tr>
+ <tr>
+ <td><h5>changelog-producer.row-deduplicate</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td><p>Boolean</p></td>
+ <td>Whether to generate -U, +U changelog for the same record. This
configuration is only valid for the changelog-producer is lookup or
full-compaction.</td>
+ </tr>
<tr>
<td><h5>commit.force-compact</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git a/paimon-core/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-core/src/main/java/org/apache/paimon/CoreOptions.java
index d5cb05528..ad7ccda11 100644
--- a/paimon-core/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-core/src/main/java/org/apache/paimon/CoreOptions.java
@@ -348,6 +348,13 @@ public class CoreOptions implements Serializable {
+ "This changelog file keeps the details
of data changes, "
+ "it can be read directly during stream
reads.");
+ public static final ConfigOption<Boolean>
CHANGELOG_PRODUCER_ROW_DEDUPLICATE =
+ key("changelog-producer.row-deduplicate")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to generate -U, +U changelog for the same
record. This configuration is only valid for the changelog-producer is lookup
or full-compaction.");
+
@Immutable
public static final ConfigOption<String> SEQUENCE_FIELD =
key("sequence.field")
@@ -816,6 +823,10 @@ public class CoreOptions implements Serializable {
return options.get(CHANGELOG_PRODUCER);
}
+ public boolean changelogRowDeduplicate() {
+ return options.get(CHANGELOG_PRODUCER_ROW_DEDUPLICATE);
+ }
+
public boolean scanPlanSortPartition() {
return options.get(SCAN_PLAN_SORT_PARTITION);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
index e1e7819b5..079b00ace 100644
--- a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
@@ -30,6 +30,7 @@ import org.apache.paimon.schema.KeyValueFieldsExtractor;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.KeyComparatorSupplier;
+import org.apache.paimon.utils.ValueComparatorSupplier;
import java.util.Comparator;
import java.util.function.Supplier;
@@ -44,6 +45,7 @@ public class KeyValueFileStore extends
AbstractFileStore<KeyValue> {
private final RowType valueType;
private final KeyValueFieldsExtractor keyValueFieldsExtractor;
private final Supplier<Comparator<InternalRow>> keyComparatorSupplier;
+ private final Supplier<Comparator<InternalRow>> valueComparatorSupplier;
private final MergeFunctionFactory<KeyValue> mfFactory;
public KeyValueFileStore(
@@ -64,6 +66,7 @@ public class KeyValueFileStore extends
AbstractFileStore<KeyValue> {
this.keyValueFieldsExtractor = keyValueFieldsExtractor;
this.mfFactory = mfFactory;
this.keyComparatorSupplier = new KeyComparatorSupplier(keyType);
+ this.valueComparatorSupplier = new ValueComparatorSupplier(valueType);
}
@Override
@@ -101,6 +104,7 @@ public class KeyValueFileStore extends
AbstractFileStore<KeyValue> {
keyType,
valueType,
keyComparatorSupplier,
+ valueComparatorSupplier,
mfFactory,
pathFactory(),
snapshotManager(),
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/ChangelogMergeTreeRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/ChangelogMergeTreeRewriter.java
index fbfa81d61..71c017c2c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/ChangelogMergeTreeRewriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/ChangelogMergeTreeRewriter.java
@@ -39,13 +39,20 @@ import java.util.List;
/** A {@link MergeTreeCompactRewriter} which produces changelog files for the
compaction. */
public abstract class ChangelogMergeTreeRewriter extends
MergeTreeCompactRewriter {
+ protected final Comparator<InternalRow> valueComparator;
+ protected final boolean changelogRowDeduplicate;
+
public ChangelogMergeTreeRewriter(
KeyValueFileReaderFactory readerFactory,
KeyValueFileWriterFactory writerFactory,
Comparator<InternalRow> keyComparator,
MergeFunctionFactory<KeyValue> mfFactory,
- SortEngine sortEngine) {
+ SortEngine sortEngine,
+ Comparator<InternalRow> valueComparator,
+ boolean changelogRowDeduplicate) {
super(readerFactory, writerFactory, keyComparator, mfFactory,
sortEngine);
+ this.valueComparator = valueComparator;
+ this.changelogRowDeduplicate = changelogRowDeduplicate;
}
protected abstract boolean rewriteChangelog(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeFunctionWrapper.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeFunctionWrapper.java
index 53214a0af..d6ee05717 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeFunctionWrapper.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeFunctionWrapper.java
@@ -19,9 +19,12 @@
package org.apache.paimon.mergetree.compact;
import org.apache.paimon.KeyValue;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.utils.Preconditions;
+import java.util.Comparator;
+
/**
* Wrapper for {@link MergeFunction}s to produce changelog during a full
compaction.
*
@@ -38,6 +41,8 @@ public class FullChangelogMergeFunctionWrapper implements
MergeFunctionWrapper<C
private final MergeFunction<KeyValue> mergeFunction;
private final int maxLevel;
+ private final Comparator<InternalRow> valueComparator;
+ private final boolean changelogRowDeduplicate;
// only full compaction will write files into maxLevel, see
UniversalCompaction class
private KeyValue topLevelKv;
@@ -48,13 +53,19 @@ public class FullChangelogMergeFunctionWrapper implements
MergeFunctionWrapper<C
private final KeyValue reusedBefore = new KeyValue();
private final KeyValue reusedAfter = new KeyValue();
- public FullChangelogMergeFunctionWrapper(MergeFunction<KeyValue>
mergeFunction, int maxLevel) {
+ public FullChangelogMergeFunctionWrapper(
+ MergeFunction<KeyValue> mergeFunction,
+ int maxLevel,
+ Comparator<InternalRow> valueComparator,
+ boolean changelogRowDeduplicate) {
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;
+ this.valueComparator = valueComparator;
+ this.changelogRowDeduplicate = changelogRowDeduplicate;
}
@Override
@@ -99,12 +110,13 @@ public class FullChangelogMergeFunctionWrapper implements
MergeFunctionWrapper<C
reusedResult.addChangelog(replace(reusedAfter,
RowKind.INSERT, merged));
}
} else {
- if (merged != null && isAdd(merged)) {
+ if (merged == null || !isAdd(merged)) {
+ reusedResult.addChangelog(replace(reusedBefore,
RowKind.DELETE, topLevelKv));
+ } else if (!changelogRowDeduplicate
+ || valueComparator.compare(topLevelKv.value(),
merged.value()) != 0) {
reusedResult
.addChangelog(replace(reusedBefore,
RowKind.UPDATE_BEFORE, topLevelKv))
.addChangelog(replace(reusedAfter,
RowKind.UPDATE_AFTER, merged));
- } else {
- reusedResult.addChangelog(replace(reusedBefore,
RowKind.DELETE, topLevelKv));
}
}
return reusedResult.setResultIfNotRetract(merged);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeTreeCompactRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeTreeCompactRewriter.java
index 9e9d8b0a7..4ab8ced68 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeTreeCompactRewriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeTreeCompactRewriter.java
@@ -42,8 +42,17 @@ public class FullChangelogMergeTreeCompactRewriter extends
ChangelogMergeTreeRew
KeyValueFileWriterFactory writerFactory,
Comparator<InternalRow> keyComparator,
MergeFunctionFactory<KeyValue> mfFactory,
- SortEngine sortEngine) {
- super(readerFactory, writerFactory, keyComparator, mfFactory,
sortEngine);
+ SortEngine sortEngine,
+ Comparator<InternalRow> valueComparator,
+ boolean changelogRowDeduplicate) {
+ super(
+ readerFactory,
+ writerFactory,
+ keyComparator,
+ mfFactory,
+ sortEngine,
+ valueComparator,
+ changelogRowDeduplicate);
this.maxLevel = maxLevel;
}
@@ -66,7 +75,8 @@ public class FullChangelogMergeTreeCompactRewriter extends
ChangelogMergeTreeRew
@Override
protected MergeFunctionWrapper<ChangelogResult> createMergeWrapper(int
outputLevel) {
- return new FullChangelogMergeFunctionWrapper(mfFactory.create(),
maxLevel);
+ return new FullChangelogMergeFunctionWrapper(
+ mfFactory.create(), maxLevel, valueComparator,
changelogRowDeduplicate);
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapper.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapper.java
index e2ee56346..08771fc47 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapper.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapper.java
@@ -22,6 +22,7 @@ import org.apache.paimon.KeyValue;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.types.RowKind;
+import java.util.Comparator;
import java.util.function.Function;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -50,10 +51,14 @@ public class LookupChangelogMergeFunctionWrapper implements
MergeFunctionWrapper
private final ChangelogResult reusedResult = new ChangelogResult();
private final KeyValue reusedBefore = new KeyValue();
private final KeyValue reusedAfter = new KeyValue();
+ private final Comparator<InternalRow> valueComparator;
+ private final boolean changelogRowDeduplicate;
public LookupChangelogMergeFunctionWrapper(
MergeFunctionFactory<KeyValue> mergeFunctionFactory,
- Function<InternalRow, KeyValue> lookup) {
+ Function<InternalRow, KeyValue> lookup,
+ Comparator<InternalRow> valueComparator,
+ boolean changelogRowDeduplicate) {
MergeFunction<KeyValue> mergeFunction = mergeFunctionFactory.create();
checkArgument(
mergeFunction instanceof LookupMergeFunction,
@@ -62,6 +67,8 @@ public class LookupChangelogMergeFunctionWrapper implements
MergeFunctionWrapper
this.mergeFunction = (LookupMergeFunction) mergeFunction;
this.mergeFunction2 = mergeFunctionFactory.create();
this.lookup = lookup;
+ this.valueComparator = valueComparator;
+ this.changelogRowDeduplicate = changelogRowDeduplicate;
}
@Override
@@ -114,12 +121,13 @@ public class LookupChangelogMergeFunctionWrapper
implements MergeFunctionWrapper
reusedResult.addChangelog(replaceAfter(RowKind.INSERT, after));
}
} else {
- if (isAdd(after)) {
+ if (!isAdd(after)) {
+ reusedResult.addChangelog(replaceBefore(RowKind.DELETE,
before));
+ } else if (!changelogRowDeduplicate
+ || valueComparator.compare(before.value(), after.value())
!= 0) {
reusedResult
.addChangelog(replaceBefore(RowKind.UPDATE_BEFORE,
before))
.addChangelog(replaceAfter(RowKind.UPDATE_AFTER,
after));
- } else {
- reusedResult.addChangelog(replaceBefore(RowKind.DELETE,
before));
}
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriter.java
index e799fe73d..98994fe5b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriter.java
@@ -46,8 +46,17 @@ public class LookupMergeTreeCompactRewriter extends
ChangelogMergeTreeRewriter {
KeyValueFileWriterFactory writerFactory,
Comparator<InternalRow> keyComparator,
MergeFunctionFactory<KeyValue> mfFactory,
- SortEngine sortEngine) {
- super(readerFactory, writerFactory, keyComparator, mfFactory,
sortEngine);
+ SortEngine sortEngine,
+ Comparator<InternalRow> valueComparator,
+ boolean changelogRowDeduplicate) {
+ super(
+ readerFactory,
+ writerFactory,
+ keyComparator,
+ mfFactory,
+ sortEngine,
+ valueComparator,
+ changelogRowDeduplicate);
this.lookupLevels = lookupLevels;
}
@@ -85,7 +94,9 @@ public class LookupMergeTreeCompactRewriter extends
ChangelogMergeTreeRewriter {
} catch (IOException e) {
throw new UncheckedIOException(e);
}
- });
+ },
+ valueComparator,
+ changelogRowDeduplicate);
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
index b346168a4..9e58598f9 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
@@ -71,6 +71,7 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
private final KeyValueFileReaderFactory.Builder readerFactoryBuilder;
private final KeyValueFileWriterFactory.Builder writerFactoryBuilder;
private final Supplier<Comparator<InternalRow>> keyComparatorSupplier;
+ private final Supplier<Comparator<InternalRow>> valueComparatorSupplier;
private final MergeFunctionFactory<KeyValue> mfFactory;
private final CoreOptions options;
private final FileIO fileIO;
@@ -85,6 +86,7 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
RowType keyType,
RowType valueType,
Supplier<Comparator<InternalRow>> keyComparatorSupplier,
+ Supplier<Comparator<InternalRow>> valueComparatorSupplier,
MergeFunctionFactory<KeyValue> mfFactory,
FileStorePathFactory pathFactory,
SnapshotManager snapshotManager,
@@ -115,6 +117,7 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
pathFactory,
options.targetFileSize());
this.keyComparatorSupplier = keyComparatorSupplier;
+ this.valueComparatorSupplier = valueComparatorSupplier;
this.mfFactory = mfFactory;
this.options = options;
}
@@ -211,7 +214,9 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
writerFactory,
keyComparator,
mfFactory,
- options.sortEngine());
+ options.sortEngine(),
+ valueComparatorSupplier.get(),
+ options.changelogRowDeduplicate());
case LOOKUP:
LookupLevels lookupLevels = createLookupLevels(levels,
readerFactory);
return new LookupMergeTreeCompactRewriter(
@@ -220,7 +225,9 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
writerFactory,
keyComparator,
mfFactory,
- options.sortEngine());
+ options.sortEngine(),
+ valueComparatorSupplier.get(),
+ options.changelogRowDeduplicate());
default:
return new MergeTreeCompactRewriter(
readerFactory,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/ValueComparatorSupplier.java
b/paimon-core/src/main/java/org/apache/paimon/utils/ValueComparatorSupplier.java
new file mode 100644
index 000000000..4a08c6628
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/utils/ValueComparatorSupplier.java
@@ -0,0 +1,46 @@
+/*
+ * 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.paimon.utils;
+
+import org.apache.paimon.codegen.CodeGenUtils;
+import org.apache.paimon.codegen.GeneratedClass;
+import org.apache.paimon.codegen.RecordComparator;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.types.RowType;
+
+import java.util.Comparator;
+import java.util.function.Supplier;
+
+/** A {@link Supplier} that returns the comparator for the file store value. */
+public class ValueComparatorSupplier implements
SerializableSupplier<Comparator<InternalRow>> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final GeneratedClass<RecordComparator> genRecordComparator;
+
+ public ValueComparatorSupplier(RowType keyType) {
+ genRecordComparator =
+ CodeGenUtils.generateRecordComparator(keyType.getFieldTypes(),
"valueComparator");
+ }
+
+ @Override
+ public RecordComparator get() {
+ return
genRecordComparator.newInstance(ValueComparatorSupplier.class.getClassLoader());
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/FullChangelogMergeFunctionWrapperTestBase.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/FullChangelogMergeFunctionWrapperTestBase.java
index 1f6a1cf60..6f562295a 100644
---
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/FullChangelogMergeFunctionWrapperTestBase.java
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/FullChangelogMergeFunctionWrapperTestBase.java
@@ -19,6 +19,7 @@
package org.apache.paimon.mergetree.compact;
import org.apache.paimon.KeyValue;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.types.RowKind;
import org.junit.jupiter.api.BeforeEach;
@@ -27,6 +28,7 @@ import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Comparator;
import java.util.List;
import static org.apache.paimon.io.DataFileTestUtils.row;
@@ -36,13 +38,20 @@ public abstract class
FullChangelogMergeFunctionWrapperTestBase {
private static final int MAX_LEVEL = 3;
+ private static final Comparator<InternalRow> COMPARATOR =
+ Comparator.comparingInt(o -> o.getInt(0));
+
protected FullChangelogMergeFunctionWrapper wrapper;
protected abstract MergeFunction<KeyValue> createMergeFunction();
+ protected abstract boolean changelogRowDeduplicate();
+
@BeforeEach
public void beforeEach() {
- wrapper = new FullChangelogMergeFunctionWrapper(createMergeFunction(),
MAX_LEVEL);
+ wrapper =
+ new FullChangelogMergeFunctionWrapper(
+ createMergeFunction(), MAX_LEVEL, COMPARATOR,
changelogRowDeduplicate());
}
private static final List<List<KeyValue>> INPUT_KVS =
@@ -82,6 +91,13 @@ public abstract class
FullChangelogMergeFunctionWrapperTestBase {
.setLevel(MAX_LEVEL),
new KeyValue()
.replace(row(7), 13,
RowKind.UPDATE_BEFORE, row(3))
+ .setLevel(0)),
+ Arrays.asList(
+ new KeyValue()
+ .replace(row(8), 14, RowKind.INSERT,
row(3))
+ .setLevel(MAX_LEVEL),
+ new KeyValue()
+ .replace(row(8), 15, RowKind.INSERT,
row(3))
.setLevel(0)));
protected abstract KeyValue getExpectedBefore(int idx);
@@ -112,10 +128,10 @@ public abstract class
FullChangelogMergeFunctionWrapperTestBase {
/**
* Tests for {@link FullChangelogMergeFunctionWrapper} with {@link
DeduplicateMergeFunction}.
*/
- public static class WithDeduplicateMergeFunctionTest
+ public abstract static class WithDeduplicateMergeFunctionTestBase
extends FullChangelogMergeFunctionWrapperTestBase {
- private static final List<KeyValue> EXPECTED_BEFORE =
+ private final List<KeyValue> expectedBefore =
Arrays.asList(
null,
null,
@@ -124,9 +140,13 @@ public abstract class
FullChangelogMergeFunctionWrapperTestBase {
null,
new KeyValue().replace(row(6), 8,
RowKind.UPDATE_BEFORE, row(3)),
new KeyValue().replace(row(7), 10, RowKind.DELETE,
row(3)),
- new KeyValue().replace(row(7), 12, RowKind.DELETE,
row(3)));
+ new KeyValue().replace(row(7), 12, RowKind.DELETE,
row(3)),
+ changelogRowDeduplicate()
+ ? null
+ : new KeyValue()
+ .replace(row(8), 14,
RowKind.UPDATE_BEFORE, row(3)));
- private static final List<KeyValue> EXPECTED_AFTER =
+ private final List<KeyValue> expectedAfter =
Arrays.asList(
new KeyValue().replace(row(1), 1, RowKind.INSERT,
row(1)),
null,
@@ -135,9 +155,12 @@ public abstract class
FullChangelogMergeFunctionWrapperTestBase {
null,
new KeyValue().replace(row(6), 9,
RowKind.UPDATE_AFTER, row(-3)),
null,
- null);
+ null,
+ changelogRowDeduplicate()
+ ? null
+ : new KeyValue().replace(row(8), 15,
RowKind.UPDATE_AFTER, row(3)));
- private static final List<KeyValue> EXPECTED_RESULT =
+ private final List<KeyValue> expectedResult =
Arrays.asList(
new KeyValue().replace(row(1), 1, RowKind.INSERT,
row(1)),
null,
@@ -146,7 +169,8 @@ public abstract class
FullChangelogMergeFunctionWrapperTestBase {
null,
new KeyValue().replace(row(6), 9, RowKind.INSERT,
row(-3)),
null,
- null);
+ null,
+ new KeyValue().replace(row(8), 15, RowKind.INSERT,
row(3)));
@Override
protected MergeFunction<KeyValue> createMergeFunction() {
@@ -155,17 +179,41 @@ public abstract class
FullChangelogMergeFunctionWrapperTestBase {
@Override
protected KeyValue getExpectedBefore(int idx) {
- return EXPECTED_BEFORE.get(idx);
+ return expectedBefore.get(idx);
}
@Override
protected KeyValue getExpectedAfter(int idx) {
- return EXPECTED_AFTER.get(idx);
+ return expectedAfter.get(idx);
}
@Override
protected KeyValue getExpectedResult(int idx) {
- return EXPECTED_RESULT.get(idx);
+ return expectedResult.get(idx);
+ }
+ }
+
+ /**
+ * Tests for {@link WithDeduplicateMergeFunctionTestBase} with changelog
deduplication disabled.
+ */
+ public static class WithoutChangelogRowDeduplicateMergeFunctionTest
+ extends WithDeduplicateMergeFunctionTestBase {
+
+ @Override
+ protected boolean changelogRowDeduplicate() {
+ return false;
+ }
+ }
+
+ /**
+ * Tests for {@link WithDeduplicateMergeFunctionTestBase} with changelog
deduplication enabled.
+ */
+ public static class WithChangelogRowDeduplicateMergeFunctionTest
+ extends WithDeduplicateMergeFunctionTestBase {
+
+ @Override
+ protected boolean changelogRowDeduplicate() {
+ return true;
}
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java
index 1d1511010..84700cf73 100644
---
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java
@@ -26,8 +26,10 @@ import
org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
import org.apache.paimon.mergetree.compact.aggregate.FieldSumAgg;
import org.apache.paimon.types.DataTypes;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -42,13 +44,19 @@ import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link LookupChangelogMergeFunctionWrapper}. */
public class LookupChangelogMergeFunctionWrapperTest {
- @Test
- public void testDeduplicate() {
+ private static final Comparator<InternalRow> COMPARATOR =
+ Comparator.comparingInt(o -> o.getInt(0));
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testDeduplicate(boolean changelogRowDeduplicate) {
Map<InternalRow, KeyValue> highLevel = new HashMap<>();
LookupChangelogMergeFunctionWrapper function =
new LookupChangelogMergeFunctionWrapper(
LookupMergeFunction.wrap(DeduplicateMergeFunction.factory()),
- highLevel::get);
+ highLevel::get,
+ COMPARATOR,
+ changelogRowDeduplicate);
// Without level-0
function.reset();
@@ -142,10 +150,53 @@ public class LookupChangelogMergeFunctionWrapperTest {
assertThat(kv).isNotNull();
assertThat(kv.valueKind()).isEqualTo(DELETE);
assertThat(kv.value().getInt(0)).isEqualTo(2);
+
+ // With level-0 'insert' record, with level-x (x > 0) same record
+ function.reset();
+ function.add(new KeyValue().replace(row(1), 2, INSERT,
row(2)).setLevel(0));
+ function.add(new KeyValue().replace(row(1), 2, INSERT,
row(2)).setLevel(2));
+ result = function.getResult();
+ assertThat(result).isNotNull();
+ changelogs = result.changelogs();
+ if (changelogRowDeduplicate) {
+ assertThat(changelogs).isEmpty();
+ } else {
+ assertThat(changelogs).hasSize(2);
+ assertThat(changelogs.get(0).valueKind()).isEqualTo(UPDATE_BEFORE);
+ assertThat(changelogs.get(0).value().getInt(0)).isEqualTo(2);
+ assertThat(changelogs.get(1).valueKind()).isEqualTo(UPDATE_AFTER);
+ assertThat(changelogs.get(1).value().getInt(0)).isEqualTo(2);
+ }
+ kv = result.result();
+ assertThat(kv).isNotNull();
+ assertThat(kv.valueKind()).isEqualTo(INSERT);
+ assertThat(kv.value().getInt(0)).isEqualTo(2);
+
+ // With level-0 'insert' record, without level-x (x > 0) record, query
success
+ function.reset();
+ highLevel.put(row(1), new KeyValue().replace(row(1), 2, INSERT,
row(2)).setLevel(2));
+ function.add(new KeyValue().replace(row(1), 2, INSERT,
row(2)).setLevel(0));
+ result = function.getResult();
+ assertThat(result).isNotNull();
+ changelogs = result.changelogs();
+ if (changelogRowDeduplicate) {
+ assertThat(changelogs).isEmpty();
+ } else {
+ assertThat(changelogs).hasSize(2);
+ assertThat(changelogs.get(0).valueKind()).isEqualTo(UPDATE_BEFORE);
+ assertThat(changelogs.get(0).value().getInt(0)).isEqualTo(2);
+ assertThat(changelogs.get(1).valueKind()).isEqualTo(UPDATE_AFTER);
+ assertThat(changelogs.get(1).value().getInt(0)).isEqualTo(2);
+ }
+ kv = result.result();
+ assertThat(kv).isNotNull();
+ assertThat(kv.valueKind()).isEqualTo(INSERT);
+ assertThat(kv.value().getInt(0)).isEqualTo(2);
}
- @Test
- public void testSum() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testSum(boolean changlogRowDeduplicate) {
LookupChangelogMergeFunctionWrapper function =
new LookupChangelogMergeFunctionWrapper(
LookupMergeFunction.wrap(
@@ -157,7 +208,9 @@ public class LookupChangelogMergeFunctionWrapperTest {
new FieldAggregator[] {
new
FieldSumAgg(DataTypes.INT())
})),
- key -> null);
+ key -> null,
+ COMPARATOR,
+ changlogRowDeduplicate);
// Without level-0
function.reset();
@@ -203,5 +256,25 @@ public class LookupChangelogMergeFunctionWrapperTest {
kv = result.result();
assertThat(kv).isNotNull();
assertThat(kv.value().getInt(0)).isEqualTo(4);
+
+ // With level-0 record, with the result is not changed
+ function.reset();
+ function.add(new KeyValue().replace(row(1), 1, INSERT,
row(2)).setLevel(1));
+ function.add(new KeyValue().replace(row(1), 2, INSERT,
row(0)).setLevel(0));
+ result = function.getResult();
+ assertThat(result).isNotNull();
+ changelogs = result.changelogs();
+ if (changlogRowDeduplicate) {
+ assertThat(changelogs).isEmpty();
+ } else {
+ assertThat(changelogs).hasSize(2);
+ assertThat(changelogs.get(0).valueKind()).isEqualTo(UPDATE_BEFORE);
+ assertThat(changelogs.get(0).value().getInt(0)).isEqualTo(2);
+ assertThat(changelogs.get(1).valueKind()).isEqualTo(UPDATE_AFTER);
+ assertThat(changelogs.get(1).value().getInt(0)).isEqualTo(2);
+ }
+ kv = result.result();
+ assertThat(kv).isNotNull();
+ assertThat(kv.value().getInt(0)).isEqualTo(2);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FullCompactionFileStoreITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FullCompactionFileStoreITCase.java
index 031ee2089..ec3e68ce1 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FullCompactionFileStoreITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FullCompactionFileStoreITCase.java
@@ -24,6 +24,8 @@ import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
@@ -77,14 +79,19 @@ public class FullCompactionFileStoreITCase extends
CatalogITCaseBase {
Row.of("1", "2", "3"), Row.of("4", "5", "6"),
Row.of("7", "8", "9"));
}
- @Test
- public void testUpdate() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testUpdate(boolean changelogRowDeduplicate) throws Exception {
+ sql(
+ "ALTER TABLE %s SET ('changelog-producer.row-deduplicate' =
'%s')",
+ table, changelogRowDeduplicate);
+
BlockingIterator<Row, Row> iterator =
BlockingIterator.of(streamSqlIter("SELECT * FROM %s", table));
- sql("INSERT INTO %s VALUES ('1', '2', '3')", table);
- assertThat(iterator.collect(1))
- .containsExactlyInAnyOrder(Row.ofKind(RowKind.INSERT, "1",
"2", "3"));
+ sql("INSERT INTO %s VALUES ('1', '2', '3'), ('4', '5', '1')", table);
+ assertThat(iterator.collect(2))
+ .containsExactlyInAnyOrder(Row.of("1", "2", "3"), Row.of("4",
"5", "1"));
sql("INSERT INTO %s VALUES ('1', '4', '5')", table);
assertThat(iterator.collect(2))
@@ -92,6 +99,24 @@ public class FullCompactionFileStoreITCase extends
CatalogITCaseBase {
Row.ofKind(RowKind.UPDATE_BEFORE, "1", "2", "3"),
Row.ofKind(RowKind.UPDATE_AFTER, "1", "4", "5"));
+ // insert duplicate record
+ for (int i = 1; i < 5; i++) {
+ sql("INSERT INTO %s VALUES ('1', '4', '5'), ('4', '5', '%s')",
table, i + 1);
+ if (changelogRowDeduplicate) {
+ assertThat(iterator.collect(2))
+ .containsExactlyInAnyOrder(
+ Row.ofKind(RowKind.UPDATE_BEFORE, "4", "5",
String.valueOf(i)),
+ Row.ofKind(RowKind.UPDATE_AFTER, "4", "5",
String.valueOf(i + 1)));
+ } else {
+ assertThat(iterator.collect(4))
+ .containsExactlyInAnyOrder(
+ Row.ofKind(RowKind.UPDATE_BEFORE, "1", "4",
"5"),
+ Row.ofKind(RowKind.UPDATE_AFTER, "1", "4",
"5"),
+ Row.ofKind(RowKind.UPDATE_BEFORE, "4", "5",
String.valueOf(i)),
+ Row.ofKind(RowKind.UPDATE_AFTER, "4", "5",
String.valueOf(i + 1)));
+ }
+ }
+
iterator.close();
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupChangelogWithAggITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupChangelogWithAggITCase.java
index 95bd81f89..a0481eb9d 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupChangelogWithAggITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupChangelogWithAggITCase.java
@@ -22,21 +22,25 @@ import org.apache.paimon.utils.BlockingIterator;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import static org.assertj.core.api.Assertions.assertThat;
/** Test Lookup changelog producer with aggregation tables. */
public class LookupChangelogWithAggITCase extends CatalogITCaseBase {
- @Test
- public void testMultipleCompaction() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testMultipleCompaction(boolean changelogRowDeduplicate) throws
Exception {
sql(
"CREATE TABLE T (k INT PRIMARY KEY NOT ENFORCED, v INT) WITH ("
+ "'bucket'='3', "
+ "'changelog-producer'='lookup', "
+ + "'changelog-producer.row-deduplicate'='%s', "
+ "'merge-engine'='aggregation', "
- + "'fields.v.aggregate-function'='sum')");
+ + "'fields.v.aggregate-function'='sum')",
+ changelogRowDeduplicate);
BlockingIterator<Row, Row> iterator = streamSqlBlockIter("SELECT *
FROM T");
sql("INSERT INTO T VALUES (1, 1), (2, 2)");
@@ -52,6 +56,24 @@ public class LookupChangelogWithAggITCase extends
CatalogITCaseBase {
Row.ofKind(RowKind.UPDATE_AFTER, 2, 2 * (i + 1)));
}
+ for (int i = 5; i < 10; i++) {
+ // insert (1,0) to keep the result of sum unchanged.
+ sql("INSERT INTO T VALUES (1, 0), (2, 2)");
+ if (changelogRowDeduplicate) {
+ assertThat(iterator.collect(2))
+ .containsExactlyInAnyOrder(
+ Row.ofKind(RowKind.UPDATE_BEFORE, 2, 2 * i),
+ Row.ofKind(RowKind.UPDATE_AFTER, 2, 2 * (i +
1)));
+ } else {
+ assertThat(iterator.collect(4))
+ .containsExactlyInAnyOrder(
+ Row.ofKind(RowKind.UPDATE_BEFORE, 1, 5),
+ Row.ofKind(RowKind.UPDATE_BEFORE, 2, 2 * i),
+ Row.ofKind(RowKind.UPDATE_AFTER, 1, 5),
+ Row.ofKind(RowKind.UPDATE_AFTER, 2, 2 * (i +
1)));
+ }
+ }
+
iterator.close();
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
index 499951cd3..864d86883 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
@@ -177,6 +177,7 @@ public class TestChangelogDataReadWrite {
KEY_TYPE,
VALUE_TYPE,
() -> COMPARATOR,
+ () -> COMPARATOR,
DeduplicateMergeFunction.factory(),
pathFactory,
snapshotManager,
