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 2dbb3d089 [core] Support ignore delete/update_before records in
deduplicate engine (#2501)
2dbb3d089 is described below
commit 2dbb3d08971201f8985c1958b98d493eb010678b
Author: yuzelin <[email protected]>
AuthorDate: Wed Dec 13 18:50:13 2023 +0800
[core] Support ignore delete/update_before records in deduplicate engine
(#2501)
---
.../shortcodes/generated/core_configuration.html | 6 ++++++
.../main/java/org/apache/paimon/CoreOptions.java | 6 ++++++
.../compact/DeduplicateMergeFunction.java | 25 +++++++++++++++++++---
.../apache/paimon/table/PrimaryKeyTableUtils.java | 2 +-
.../FullChangelogMergeFunctionWrapperTestBase.java | 2 +-
.../ReducerMergeFunctionWrapperTestBase.java | 2 +-
.../mergetree/compact/SortMergeReaderTestBase.java | 2 +-
.../cdc/kafka/KafkaCanalSyncTableActionITCase.java | 23 +++++++++++++-------
.../apache/paimon/flink/BatchFileStoreITCase.java | 18 ++++++++++++++++
.../paimon/flink/ContinuousFileStoreITCase.java | 16 ++++++++++++++
10 files changed, 87 insertions(+), 15 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html
b/docs/layouts/shortcodes/generated/core_configuration.html
index 82a8fc798..1c3c747f6 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -134,6 +134,12 @@ under the License.
<td>Duration</td>
<td>The TTL in rocksdb index for cross partition upsert (primary
keys not contain all partition fields), this can avoid maintaining too many
indexes and lead to worse and worse performance, but please note that this may
also cause data duplication.</td>
</tr>
+ <tr>
+ <td><h5>deduplicate.ignore-delete</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to ignore delete records in deduplicate mode.</td>
+ </tr>
<tr>
<td><h5>dynamic-bucket.assigner-parallelism</h5></td>
<td style="word-wrap: break-word;">(none)</td>
diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
index 4274ea8d2..90f74ad2a 100644
--- a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
@@ -232,6 +232,12 @@ public class CoreOptions implements Serializable {
.defaultValue(MergeEngine.DEDUPLICATE)
.withDescription("Specify the merge engine for table with
primary key.");
+ public static final ConfigOption<Boolean> DEDUPLICATE_IGNORE_DELETE =
+ key("deduplicate.ignore-delete")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Whether to ignore delete records in
deduplicate mode.");
+
public static final ConfigOption<Boolean> PARTIAL_UPDATE_IGNORE_DELETE =
key("partial-update.ignore-delete")
.booleanType()
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/DeduplicateMergeFunction.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/DeduplicateMergeFunction.java
index e05e2a33c..b062ce9b6 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/DeduplicateMergeFunction.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/DeduplicateMergeFunction.java
@@ -18,7 +18,9 @@
package org.apache.paimon.mergetree.compact;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.KeyValue;
+import org.apache.paimon.options.Options;
import javax.annotation.Nullable;
@@ -28,9 +30,13 @@ import javax.annotation.Nullable;
*/
public class DeduplicateMergeFunction implements MergeFunction<KeyValue> {
+ private final boolean ignoreDelete;
+
private KeyValue latestKv;
- protected DeduplicateMergeFunction() {}
+ protected DeduplicateMergeFunction(boolean ignoreDelete) {
+ this.ignoreDelete = ignoreDelete;
+ }
@Override
public void reset() {
@@ -39,6 +45,9 @@ public class DeduplicateMergeFunction implements
MergeFunction<KeyValue> {
@Override
public void add(KeyValue kv) {
+ if (ignoreDelete && kv.valueKind().isRetract()) {
+ return;
+ }
latestKv = kv;
}
@@ -49,16 +58,26 @@ public class DeduplicateMergeFunction implements
MergeFunction<KeyValue> {
}
public static MergeFunctionFactory<KeyValue> factory() {
- return new Factory();
+ return new Factory(new Options());
+ }
+
+ public static MergeFunctionFactory<KeyValue> factory(Options options) {
+ return new Factory(options);
}
private static class Factory implements MergeFunctionFactory<KeyValue> {
private static final long serialVersionUID = 1L;
+ private final Options options;
+
+ private Factory(Options options) {
+ this.options = options;
+ }
+
@Override
public MergeFunction<KeyValue> create(@Nullable int[][] projection) {
- return new DeduplicateMergeFunction();
+ return new
DeduplicateMergeFunction(options.get(CoreOptions.DEDUPLICATE_IGNORE_DELETE));
}
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java
b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java
index 727664da2..72ea4b456 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java
@@ -64,7 +64,7 @@ public class PrimaryKeyTableUtils {
switch (mergeEngine) {
case DEDUPLICATE:
- return DeduplicateMergeFunction.factory();
+ return DeduplicateMergeFunction.factory(conf);
case PARTIAL_UPDATE:
return PartialUpdateMergeFunction.factory(conf, rowType,
tableSchema.primaryKeys());
case AGGREGATE:
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 70e361f87..88bba7722 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
@@ -173,7 +173,7 @@ public abstract class
FullChangelogMergeFunctionWrapperTestBase {
@Override
protected MergeFunction<KeyValue> createMergeFunction() {
- return new DeduplicateMergeFunction();
+ return DeduplicateMergeFunction.factory().create();
}
@Override
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/ReducerMergeFunctionWrapperTestBase.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/ReducerMergeFunctionWrapperTestBase.java
index f729e023b..33aa97004 100644
---
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/ReducerMergeFunctionWrapperTestBase.java
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/ReducerMergeFunctionWrapperTestBase.java
@@ -77,7 +77,7 @@ public abstract class ReducerMergeFunctionWrapperTestBase {
@Override
protected MergeFunction<KeyValue> createMergeFunction() {
- return new DeduplicateMergeFunction();
+ return DeduplicateMergeFunction.factory().create();
}
@Override
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/SortMergeReaderTestBase.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/SortMergeReaderTestBase.java
index 876280125..6895815cd 100644
---
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/SortMergeReaderTestBase.java
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/SortMergeReaderTestBase.java
@@ -107,7 +107,7 @@ public abstract class SortMergeReaderTestBase extends
CombiningRecordReaderTestB
@Override
protected MergeFunction<KeyValue> createMergeFunction() {
- return new DeduplicateMergeFunction();
+ return DeduplicateMergeFunction.factory().create();
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
index 237b4366b..62d87a733 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.action.cdc.kafka;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.FileSystemCatalogOptions;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataType;
@@ -971,10 +972,11 @@ public class KafkaCanalSyncTableActionITCase extends
KafkaActionITCaseBase {
.containsExactlyEntriesOf(Collections.singletonMap("table-key", "table-value"));
}
- @Test
+ @ParameterizedTest(name = "ignore-delete = {0}")
+ @ValueSource(booleans = {true, false})
@Timeout(60)
- public void testCDCOperations() throws Exception {
- final String topic = "event-insert";
+ public void testCDCOperations(boolean ignoreDelete) throws Exception {
+ final String topic = "event-insert" + UUID.randomUUID();
createTestTopic(topic, 1, 1);
// ---------- Write the Canal json into Kafka -------------------
@@ -984,8 +986,11 @@ public class KafkaCanalSyncTableActionITCase extends
KafkaActionITCaseBase {
kafkaConfig.put(VALUE_FORMAT.key(), "canal-json");
kafkaConfig.put(TOPIC.key(), topic);
+ Map<String, String> tableConfig = getBasicTableConfig();
+ tableConfig.put(CoreOptions.DEDUPLICATE_IGNORE_DELETE.key(),
String.valueOf(ignoreDelete));
+
KafkaSyncTableAction action =
-
syncTableActionBuilder(kafkaConfig).withTableConfig(getBasicTableConfig()).build();
+
syncTableActionBuilder(kafkaConfig).withTableConfig(tableConfig).build();
runActionWithDefaultEnv(action);
FileStoreTable table = getFileStoreTable(tableName);
@@ -1035,10 +1040,12 @@ public class KafkaCanalSyncTableActionITCase extends
KafkaActionITCaseBase {
// For the DELETE operation
List<String> expectedDelete =
- Arrays.asList(
- "+I[1, 2, second, NULL, NULL, NULL, NULL]",
- "+I[1, 9, nine, 90000000000, 99999.999, [110, 105,
110, 101, 46, 98, 105, 110], 9.9]",
- "+I[2, 4, four, NULL, NULL, NULL, NULL]");
+ ignoreDelete
+ ? expectedUpdate
+ : Arrays.asList(
+ "+I[1, 2, second, NULL, NULL, NULL, NULL]",
+ "+I[1, 9, nine, 90000000000, 99999.999, [110,
105, 110, 101, 46, 98, 105, 110], 9.9]",
+ "+I[2, 4, four, NULL, NULL, NULL, NULL]");
waitForResult(expectedDelete, table, rowType, primaryKeys);
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
index fae1f5acf..ac68dcad7 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
@@ -20,6 +20,7 @@ package org.apache.paimon.flink;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.BlockingIterator;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
@@ -368,4 +369,21 @@ public class BatchFileStoreITCase extends
CatalogITCaseBase {
sql("INSERT INTO R_T VALUES (1, 2, '-D')");
assertThat(sql("SELECT * FROM R_T")).isEmpty();
}
+
+ @Test
+ public void testIgnoreDelete() throws Exception {
+ sql(
+ "CREATE TABLE ignore_delete (pk INT PRIMARY KEY NOT ENFORCED,
v STRING) "
+ + "WITH ('deduplicate.ignore-delete' = 'true')");
+ BlockingIterator<Row, Row> iterator = streamSqlBlockIter("SELECT *
FROM ignore_delete");
+
+ sql("INSERT INTO ignore_delete VALUES (1, 'A'), (2, 'B')");
+ sql("DELETE FROM ignore_delete WHERE pk = 1");
+ sql("INSERT INTO ignore_delete VALUES (1, 'B')");
+
+ assertThat(iterator.collect(2))
+ .containsExactlyInAnyOrder(
+ Row.ofKind(RowKind.INSERT, 1, "B"),
Row.ofKind(RowKind.INSERT, 2, "B"));
+ iterator.close();
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ContinuousFileStoreITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ContinuousFileStoreITCase.java
index eda44b0bc..b056c9e86 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ContinuousFileStoreITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ContinuousFileStoreITCase.java
@@ -505,4 +505,20 @@ public class ContinuousFileStoreITCase extends
CatalogITCaseBase {
Row.of(22, 202L, "ccc", 2, 2, "b"));
iterator.close();
}
+
+ @Test
+ public void testIgnoreDelete() {
+ sql(
+ "CREATE TABLE ignore_delete (pk INT PRIMARY KEY NOT ENFORCED,
v STRING) "
+ + "WITH ('deduplicate.ignore-delete' = 'true')");
+
+ sql("INSERT INTO ignore_delete VALUES (1, 'A')");
+ assertThat(sql("SELECT * FROM
ignore_delete")).containsExactly(Row.of(1, "A"));
+
+ sql("DELETE FROM ignore_delete WHERE pk = 1");
+ assertThat(sql("SELECT * FROM
ignore_delete")).containsExactly(Row.of(1, "A"));
+
+ sql("INSERT INTO ignore_delete VALUES (1, 'B')");
+ assertThat(sql("SELECT * FROM
ignore_delete")).containsExactly(Row.of(1, "B"));
+ }
}
