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 a2892b159 [core] Refactor logDedupEqualSupplier creation and names
a2892b159 is described below
commit a2892b1599f191327f80b85a27b4a1a3ab1a5731
Author: Jingsong <[email protected]>
AuthorDate: Wed Sep 11 13:16:14 2024 +0800
[core] Refactor logDedupEqualSupplier creation and names
---
.../main/java/org/apache/paimon/KeyValueFileStore.java | 8 ++++----
.../compact/FullChangelogMergeFunctionWrapper.java | 11 +++++------
.../compact/FullChangelogMergeTreeCompactRewriter.java | 10 +++-------
.../compact/LookupChangelogMergeFunctionWrapper.java | 9 +++------
.../compact/LookupMergeTreeCompactRewriter.java | 8 ++------
.../apache/paimon/operation/KeyValueFileStoreWrite.java | 12 +++++-------
.../org/apache/paimon/utils/ValueEqualiserSupplier.java | 17 ++++++++---------
.../mergetree/ChangelogMergeTreeRewriterTest.java | 3 +--
.../FullChangelogMergeFunctionWrapperTestBase.java | 6 ++++--
.../LookupChangelogMergeFunctionWrapperTest.java | 10 +++-------
10 files changed, 38 insertions(+), 56 deletions(-)
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 d9ca0e175..88fdd01ea 100644
--- a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
@@ -93,11 +93,11 @@ public class KeyValueFileStore extends
AbstractFileStore<KeyValue> {
this.keyValueFieldsExtractor = keyValueFieldsExtractor;
this.mfFactory = mfFactory;
this.keyComparatorSupplier = new KeyComparatorSupplier(keyType);
- List<String> ignoreFields =
options.changelogRowDeduplicateIgnoreFields();
+ List<String> logDedupIgnoreFields =
options.changelogRowDeduplicateIgnoreFields();
this.logDedupEqualSupplier =
- options.changelogRowDeduplicate() && !ignoreFields.isEmpty()
- ? ValueEqualiserSupplier.fromIgnoreFields(valueType,
ignoreFields)
- : new ValueEqualiserSupplier(valueType);
+ options.changelogRowDeduplicate()
+ ? ValueEqualiserSupplier.fromIgnoreFields(valueType,
logDedupIgnoreFields)
+ : () -> null;
this.tableName = tableName;
}
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 9d5dab1ae..35491380a 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
@@ -23,6 +23,8 @@ import org.apache.paimon.codegen.RecordEqualiser;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.utils.Preconditions;
+import javax.annotation.Nullable;
+
/**
* Wrapper for {@link MergeFunction}s to produce changelog during a full
compaction.
*
@@ -39,8 +41,7 @@ public class FullChangelogMergeFunctionWrapper implements
MergeFunctionWrapper<C
private final MergeFunction<KeyValue> mergeFunction;
private final int maxLevel;
- private final RecordEqualiser valueEqualiser;
- private final boolean changelogRowDeduplicate;
+ @Nullable private final RecordEqualiser valueEqualiser;
// only full compaction will write files into maxLevel, see
UniversalCompaction class
private KeyValue topLevelKv;
@@ -54,12 +55,10 @@ public class FullChangelogMergeFunctionWrapper implements
MergeFunctionWrapper<C
public FullChangelogMergeFunctionWrapper(
MergeFunction<KeyValue> mergeFunction,
int maxLevel,
- RecordEqualiser valueEqualiser,
- boolean changelogRowDeduplicate) {
+ @Nullable RecordEqualiser valueEqualiser) {
this.mergeFunction = mergeFunction;
this.maxLevel = maxLevel;
this.valueEqualiser = valueEqualiser;
- this.changelogRowDeduplicate = changelogRowDeduplicate;
}
@Override
@@ -106,7 +105,7 @@ public class FullChangelogMergeFunctionWrapper implements
MergeFunctionWrapper<C
} else {
if (!merged.isAdd()) {
reusedResult.addChangelog(replace(reusedBefore,
RowKind.DELETE, topLevelKv));
- } else if (!changelogRowDeduplicate
+ } else if (valueEqualiser == null
|| !valueEqualiser.equals(topLevelKv.value(),
merged.value())) {
reusedResult
.addChangelog(replace(reusedBefore,
RowKind.UPDATE_BEFORE, topLevelKv))
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 56a13dfc7..0304819e9 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,7 @@ import static
org.apache.paimon.mergetree.compact.ChangelogMergeTreeRewriter.Upg
/** A {@link MergeTreeCompactRewriter} which produces changelog files for each
full compaction. */
public class FullChangelogMergeTreeCompactRewriter extends
ChangelogMergeTreeRewriter {
- private final RecordEqualiser valueEqualiser;
- private final boolean changelogRowDeduplicate;
+ @Nullable private final RecordEqualiser valueEqualiser;
public FullChangelogMergeTreeCompactRewriter(
int maxLevel,
@@ -54,8 +53,7 @@ public class FullChangelogMergeTreeCompactRewriter extends
ChangelogMergeTreeRew
@Nullable FieldsComparator userDefinedSeqComparator,
MergeFunctionFactory<KeyValue> mfFactory,
MergeSorter mergeSorter,
- RecordEqualiser valueEqualiser,
- boolean changelogRowDeduplicate) {
+ @Nullable RecordEqualiser valueEqualiser) {
super(
maxLevel,
mergeEngine,
@@ -68,7 +66,6 @@ public class FullChangelogMergeTreeCompactRewriter extends
ChangelogMergeTreeRew
true,
false);
this.valueEqualiser = valueEqualiser;
- this.changelogRowDeduplicate = changelogRowDeduplicate;
}
@Override
@@ -90,8 +87,7 @@ public class FullChangelogMergeTreeCompactRewriter extends
ChangelogMergeTreeRew
@Override
protected MergeFunctionWrapper<ChangelogResult> createMergeWrapper(int
outputLevel) {
- return new FullChangelogMergeFunctionWrapper(
- mfFactory.create(), maxLevel, valueEqualiser,
changelogRowDeduplicate);
+ return new FullChangelogMergeFunctionWrapper(mfFactory.create(),
maxLevel, valueEqualiser);
}
@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 2d0be73fc..f182c7f4a 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
@@ -63,8 +63,7 @@ public class LookupChangelogMergeFunctionWrapper<T>
private final ChangelogResult reusedResult = new ChangelogResult();
private final KeyValue reusedBefore = new KeyValue();
private final KeyValue reusedAfter = new KeyValue();
- private final RecordEqualiser valueEqualiser;
- private final boolean changelogRowDeduplicate;
+ @Nullable private final RecordEqualiser valueEqualiser;
private final LookupStrategy lookupStrategy;
private final @Nullable DeletionVectorsMaintainer
deletionVectorsMaintainer;
private final Comparator<KeyValue> comparator;
@@ -72,8 +71,7 @@ public class LookupChangelogMergeFunctionWrapper<T>
public LookupChangelogMergeFunctionWrapper(
MergeFunctionFactory<KeyValue> mergeFunctionFactory,
Function<InternalRow, T> lookup,
- RecordEqualiser valueEqualiser,
- boolean changelogRowDeduplicate,
+ @Nullable RecordEqualiser valueEqualiser,
LookupStrategy lookupStrategy,
@Nullable DeletionVectorsMaintainer deletionVectorsMaintainer,
@Nullable UserDefinedSeqComparator userDefinedSeqComparator) {
@@ -91,7 +89,6 @@ public class LookupChangelogMergeFunctionWrapper<T>
this.mergeFunction2 = mergeFunctionFactory.create();
this.lookup = lookup;
this.valueEqualiser = valueEqualiser;
- this.changelogRowDeduplicate = changelogRowDeduplicate;
this.lookupStrategy = lookupStrategy;
this.deletionVectorsMaintainer = deletionVectorsMaintainer;
this.comparator = createSequenceComparator(userDefinedSeqComparator);
@@ -179,7 +176,7 @@ public class LookupChangelogMergeFunctionWrapper<T>
} else {
if (!after.isAdd()) {
reusedResult.addChangelog(replaceBefore(RowKind.DELETE,
before));
- } else if (!changelogRowDeduplicate
+ } else if (valueEqualiser == null
|| !valueEqualiser.equals(before.value(), after.value())) {
reusedResult
.addChangelog(replaceBefore(RowKind.UPDATE_BEFORE,
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 328015eb8..769eec356 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
@@ -152,18 +152,15 @@ public class LookupMergeTreeCompactRewriter<T> extends
ChangelogMergeTreeRewrite
public static class LookupMergeFunctionWrapperFactory<T>
implements MergeFunctionWrapperFactory<T> {
- private final RecordEqualiser valueEqualiser;
- private final boolean changelogRowDeduplicate;
+ @Nullable private final RecordEqualiser valueEqualiser;
private final LookupStrategy lookupStrategy;
@Nullable private final UserDefinedSeqComparator
userDefinedSeqComparator;
public LookupMergeFunctionWrapperFactory(
- RecordEqualiser valueEqualiser,
- boolean changelogRowDeduplicate,
+ @Nullable RecordEqualiser valueEqualiser,
LookupStrategy lookupStrategy,
@Nullable UserDefinedSeqComparator userDefinedSeqComparator) {
this.valueEqualiser = valueEqualiser;
- this.changelogRowDeduplicate = changelogRowDeduplicate;
this.lookupStrategy = lookupStrategy;
this.userDefinedSeqComparator = userDefinedSeqComparator;
}
@@ -184,7 +181,6 @@ public class LookupMergeTreeCompactRewriter<T> extends
ChangelogMergeTreeRewrite
}
},
valueEqualiser,
- changelogRowDeduplicate,
lookupStrategy,
deletionVectorsMaintainer,
userDefinedSeqComparator);
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 2ba523e85..b62cdabae 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
@@ -99,7 +99,7 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
private final KeyValueFileWriterFactory.Builder writerFactoryBuilder;
private final Supplier<Comparator<InternalRow>> keyComparatorSupplier;
private final Supplier<FieldsComparator> udsComparatorSupplier;
- private final Supplier<RecordEqualiser> valueEqualiserSupplier;
+ private final Supplier<RecordEqualiser> logDedupEqualSupplier;
private final MergeFunctionFactory<KeyValue> mfFactory;
private final CoreOptions options;
private final FileIO fileIO;
@@ -119,7 +119,7 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
RowType valueType,
Supplier<Comparator<InternalRow>> keyComparatorSupplier,
Supplier<FieldsComparator> udsComparatorSupplier,
- Supplier<RecordEqualiser> valueEqualiserSupplier,
+ Supplier<RecordEqualiser> logDedupEqualSupplier,
MergeFunctionFactory<KeyValue> mfFactory,
FileStorePathFactory pathFactory,
Map<String, FileStorePathFactory> format2PathFactory,
@@ -165,7 +165,7 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
format2PathFactory,
options.targetFileSize(true));
this.keyComparatorSupplier = keyComparatorSupplier;
- this.valueEqualiserSupplier = valueEqualiserSupplier;
+ this.logDedupEqualSupplier = logDedupEqualSupplier;
this.mfFactory = mfFactory;
this.options = options;
}
@@ -294,8 +294,7 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
userDefinedSeqComparator,
mfFactory,
mergeSorter,
- valueEqualiserSupplier.get(),
- options.changelogRowDeduplicate());
+ logDedupEqualSupplier.get());
} else if (lookupStrategy.needLookup) {
LookupLevels.ValueProcessor<?> processor;
LookupMergeTreeCompactRewriter.MergeFunctionWrapperFactory<?>
wrapperFactory;
@@ -323,8 +322,7 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
: new KeyValueProcessor(valueType);
wrapperFactory =
new LookupMergeFunctionWrapperFactory<>(
- valueEqualiserSupplier.get(),
- options.changelogRowDeduplicate(),
+ logDedupEqualSupplier.get(),
lookupStrategy,
UserDefinedSeqComparator.create(valueType,
options));
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/ValueEqualiserSupplier.java
b/paimon-core/src/main/java/org/apache/paimon/utils/ValueEqualiserSupplier.java
index 315ac519c..5238b32dc 100644
---
a/paimon-core/src/main/java/org/apache/paimon/utils/ValueEqualiserSupplier.java
+++
b/paimon-core/src/main/java/org/apache/paimon/utils/ValueEqualiserSupplier.java
@@ -58,15 +58,14 @@ public class ValueEqualiserSupplier implements
SerializableSupplier<RecordEquali
public static ValueEqualiserSupplier fromIgnoreFields(
RowType rowType, @Nullable List<String> ignoreFields) {
- int[] projection = getProjectionWithIgnoreFields(rowType,
ignoreFields);
+ int[] projection = null;
+ if (ignoreFields != null) {
+ List<String> fieldNames = rowType.getFieldNames();
+ projection =
+ IntStream.range(0, rowType.getFieldCount())
+ .filter(idx ->
!ignoreFields.contains(fieldNames.get(idx)))
+ .toArray();
+ }
return new ValueEqualiserSupplier(rowType, projection);
}
-
- private static int[] getProjectionWithIgnoreFields(RowType rowType,
List<String> ignoreFields) {
- List<String> fieldNames = rowType.getFieldNames();
- IntStream projectionStream = IntStream.range(0,
rowType.getFieldCount());
- return projectionStream
- .filter(idx -> !ignoreFields.contains(fieldNames.get(idx)))
- .toArray();
- }
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/ChangelogMergeTreeRewriterTest.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/ChangelogMergeTreeRewriterTest.java
index 8bfcadebb..de9df4a39 100644
---
a/paimon-core/src/test/java/org/apache/paimon/mergetree/ChangelogMergeTreeRewriterTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/ChangelogMergeTreeRewriterTest.java
@@ -306,8 +306,7 @@ public class ChangelogMergeTreeRewriterTest {
@Override
protected MergeFunctionWrapper<ChangelogResult> createMergeWrapper(int
outputLevel) {
- return new FullChangelogMergeFunctionWrapper(
- mfFactory.create(), MAX_LEVEL, null, false);
+ return new FullChangelogMergeFunctionWrapper(mfFactory.create(),
MAX_LEVEL, null);
}
@Override
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 950e99d8f..85a94b1b0 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
@@ -55,7 +55,9 @@ public abstract class
FullChangelogMergeFunctionWrapperTestBase {
public void beforeEach() {
wrapper =
new FullChangelogMergeFunctionWrapper(
- createMergeFunction(), MAX_LEVEL, EQUALISER,
changelogRowDeduplicate());
+ createMergeFunction(),
+ MAX_LEVEL,
+ changelogRowDeduplicate() ? EQUALISER : null);
}
private static final List<List<KeyValue>> INPUT_KVS =
@@ -233,7 +235,7 @@ public abstract class
FullChangelogMergeFunctionWrapperTestBase {
ValueEqualiserSupplier.fromIgnoreFields(valueType,
ignoreFields);
FullChangelogMergeFunctionWrapper function =
new FullChangelogMergeFunctionWrapper(
- createMergeFunction(), MAX_LEVEL,
logDedupEqualSupplier.get(), true);
+ createMergeFunction(), MAX_LEVEL,
logDedupEqualSupplier.get());
// With level-0 'insert' record, with max level same record.
Notice that the specified
// ignored
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 576f31802..c8344c44d 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
@@ -74,8 +74,7 @@ public class LookupChangelogMergeFunctionWrapperTest {
RowType.of(DataTypes.INT()),
RowType.of(DataTypes.INT())),
highLevel::get,
- EQUALISER,
- changelogRowDeduplicate,
+ changelogRowDeduplicate ? EQUALISER : null,
LookupStrategy.from(false, true, false, false),
null,
null);
@@ -241,7 +240,6 @@ public class LookupChangelogMergeFunctionWrapperTest {
valueType),
highLevel::get,
logDedupEqualSupplier.get(),
- true,
LookupStrategy.from(false, true, false, false),
null,
userDefinedSeqComparator);
@@ -300,8 +298,7 @@ public class LookupChangelogMergeFunctionWrapperTest {
RowType.of(DataTypes.INT()),
RowType.of(DataTypes.INT())),
key -> null,
- EQUALISER,
- changelogRowDeduplicate,
+ changelogRowDeduplicate ? EQUALISER : null,
LookupStrategy.from(false, true, false, false),
null,
null);
@@ -389,8 +386,7 @@ public class LookupChangelogMergeFunctionWrapperTest {
RowType.of(DataTypes.INT()),
RowType.of(DataTypes.INT())),
highLevel::get,
- EQUALISER,
- false,
+ null,
LookupStrategy.from(false, true, false, false),
null,
UserDefinedSeqComparator.create(
