You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/08/27 15:19:26 UTC
(paimon) branch master updated: [core] Fix deletion vectors mode with custom sequence field (#4075)
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/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new fb862df2e [core] Fix deletion vectors mode with custom sequence field (#4075)
fb862df2e is described below
commit fb862df2e88f79145a88abfd299ff177bcab17cf
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Tue Aug 27 23:19:21 2024 +0800
[core] Fix deletion vectors mode with custom sequence field (#4075)
---
.../compact/LookupMergeTreeCompactRewriter.java | 8 ++++++--
.../apache/paimon/operation/KeyValueFileStoreWrite.java | 6 ++++--
.../org/apache/paimon/flink/DeletionVectorITCase.java | 17 +++++++++++++++++
3 files changed, 27 insertions(+), 4 deletions(-)
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 ca7069455..328015eb8 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
@@ -18,6 +18,7 @@
package org.apache.paimon.mergetree.compact;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.CoreOptions.MergeEngine;
import org.apache.paimon.KeyValue;
import org.apache.paimon.codegen.RecordEqualiser;
@@ -52,6 +53,7 @@ public class LookupMergeTreeCompactRewriter<T> extends ChangelogMergeTreeRewrite
private final LookupLevels<T> lookupLevels;
private final MergeFunctionWrapperFactory<T> wrapperFactory;
+ private final boolean noSequenceField;
@Nullable private final DeletionVectorsMaintainer dvMaintainer;
public LookupMergeTreeCompactRewriter(
@@ -66,7 +68,8 @@ public class LookupMergeTreeCompactRewriter<T> extends ChangelogMergeTreeRewrite
MergeSorter mergeSorter,
MergeFunctionWrapperFactory<T> wrapperFactory,
boolean produceChangelog,
- @Nullable DeletionVectorsMaintainer dvMaintainer) {
+ @Nullable DeletionVectorsMaintainer dvMaintainer,
+ CoreOptions options) {
super(
maxLevel,
mergeEngine,
@@ -81,6 +84,7 @@ public class LookupMergeTreeCompactRewriter<T> extends ChangelogMergeTreeRewrite
this.dvMaintainer = dvMaintainer;
this.lookupLevels = lookupLevels;
this.wrapperFactory = wrapperFactory;
+ this.noSequenceField = options.sequenceField().isEmpty();
}
@Override
@@ -114,7 +118,7 @@ public class LookupMergeTreeCompactRewriter<T> extends ChangelogMergeTreeRewrite
// DEDUPLICATE retains the latest records as the final result, so merging has no impact on
// it at all.
- if (mergeEngine == MergeEngine.DEDUPLICATE) {
+ if (mergeEngine == MergeEngine.DEDUPLICATE && noSequenceField) {
return CHANGELOG_NO_REWRITE;
}
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 9f4570cf2..e0018a1ea 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
@@ -318,7 +318,8 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> {
? new PositionedKeyValueProcessor(
valueType,
lookupStrategy.produceChangelog
- || mergeEngine != DEDUPLICATE)
+ || mergeEngine != DEDUPLICATE
+ || !options.sequenceField().isEmpty())
: new KeyValueProcessor(valueType);
wrapperFactory =
new LookupMergeFunctionWrapperFactory<>(
@@ -339,7 +340,8 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> {
mergeSorter,
wrapperFactory,
lookupStrategy.produceChangelog,
- dvMaintainer);
+ dvMaintainer,
+ options);
} else {
return new MergeTreeCompactRewriter(
readerFactory,
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DeletionVectorITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DeletionVectorITCase.java
index 9f27f9405..832cdf286 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DeletionVectorITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DeletionVectorITCase.java
@@ -245,4 +245,21 @@ public class DeletionVectorITCase extends CatalogITCaseBase {
}
}
}
+
+ @ParameterizedTest
+ @ValueSource(strings = {"none", "lookup"})
+ public void testBatchReadDVTableWithSequenceField(String changelogProducer) {
+ sql(
+ String.format(
+ "CREATE TABLE T (id INT PRIMARY KEY NOT ENFORCED, sequence INT, name STRING) "
+ + "WITH ('deletion-vectors.enabled' = 'true', 'sequence.field' = 'sequence', 'changelog-producer' = '%s')",
+ changelogProducer));
+
+ sql("INSERT INTO T VALUES (1, 1, '1'), (2, 1, '2')");
+ sql("INSERT INTO T VALUES (1, 2, '1_1'), (2, 2, '2_1')");
+ sql("INSERT INTO T VALUES (1, 3, '1_2'), (2, 1, '2_2')");
+
+ assertThat(batchSql("SELECT * FROM T"))
+ .containsExactlyInAnyOrder(Row.of(1, 3, "1_2"), Row.of(2, 2, "2_1"));
+ }
}