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 9c53136e4 [core] Introduce UpgradeStrategy to simplify logical (#2781)
9c53136e4 is described below
commit 9c53136e4d0ac5d87dacc0d38b9ed2718969dbc9
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jan 24 15:19:31 2024 +0800
[core] Introduce UpgradeStrategy to simplify logical (#2781)
---
.../compact/ChangelogMergeTreeRewriter.java | 36 ++++++++++++----------
.../compact/FirstRowMergeTreeCompactRewriter.java | 18 +++++++++--
.../FullChangelogMergeTreeCompactRewriter.java | 7 +++--
.../compact/LookupMergeTreeCompactRewriter.java | 29 ++++++++++++++---
4 files changed, 66 insertions(+), 24 deletions(-)
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 4626d8fe0..7be5fc741 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
@@ -18,7 +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;
import org.apache.paimon.compact.CompactResult;
@@ -44,13 +44,13 @@ import java.util.stream.Collectors;
public abstract class ChangelogMergeTreeRewriter extends
MergeTreeCompactRewriter {
protected final int maxLevel;
- protected final CoreOptions.MergeEngine mergeEngine;
+ protected final MergeEngine mergeEngine;
protected final RecordEqualiser valueEqualiser;
protected final boolean changelogRowDeduplicate;
public ChangelogMergeTreeRewriter(
int maxLevel,
- CoreOptions.MergeEngine mergeEngine,
+ MergeEngine mergeEngine,
KeyValueFileReaderFactory readerFactory,
KeyValueFileWriterFactory writerFactory,
Comparator<InternalRow> keyComparator,
@@ -68,7 +68,7 @@ public abstract class ChangelogMergeTreeRewriter extends
MergeTreeCompactRewrite
protected abstract boolean rewriteChangelog(
int outputLevel, boolean dropDelete, List<List<SortedRun>>
sections);
- protected abstract boolean upgradeChangelog(int outputLevel, DataFileMeta
file);
+ protected abstract UpgradeStrategy upgradeChangelog(int outputLevel,
DataFileMeta file);
protected abstract MergeFunctionWrapper<ChangelogResult>
createMergeWrapper(int outputLevel);
@@ -164,26 +164,30 @@ public abstract class ChangelogMergeTreeRewriter extends
MergeTreeCompactRewrite
@Override
public CompactResult upgrade(int outputLevel, DataFileMeta file) throws
Exception {
- if (upgradeChangelog(outputLevel, file)) {
+ UpgradeStrategy strategy = upgradeChangelog(outputLevel, file);
+ if (strategy.changelog) {
return rewriteChangelogCompaction(
outputLevel,
Collections.singletonList(
Collections.singletonList(SortedRun.fromSingle(file))),
- !rewriteCompactFileForUpgrade(outputLevel));
+ strategy.rewrite);
} else {
return super.upgrade(outputLevel, file);
}
}
- /**
- * For upgrade, there are two scenarios can skip rewrite compact file.
- *
- * <ul>
- * <li>mergeEngine == DEDUPLICATE, no need to consider previous records
- * <li>outputLevel == maxLevel, no previous records
- * </ul>
- */
- private boolean rewriteCompactFileForUpgrade(int outputLevel) {
- return mergeEngine == CoreOptions.MergeEngine.DEDUPLICATE ||
outputLevel == maxLevel;
+ /** Strategy for upgrade. */
+ protected enum UpgradeStrategy {
+ NO_CHANGELOG(false, false),
+ CHANGELOG_NO_REWRITE(true, false),
+ CHANGELOG_WITH_REWRITE(true, true);
+
+ private final boolean changelog;
+ private final boolean rewrite;
+
+ UpgradeStrategy(boolean changelog, boolean rewrite) {
+ this.changelog = changelog;
+ this.rewrite = rewrite;
+ }
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FirstRowMergeTreeCompactRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FirstRowMergeTreeCompactRewriter.java
index 7336ade38..24f872da1 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FirstRowMergeTreeCompactRewriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FirstRowMergeTreeCompactRewriter.java
@@ -36,6 +36,9 @@ import java.io.UncheckedIOException;
import java.util.Comparator;
import java.util.List;
+import static
org.apache.paimon.mergetree.compact.ChangelogMergeTreeRewriter.UpgradeStrategy.CHANGELOG_NO_REWRITE;
+import static
org.apache.paimon.mergetree.compact.ChangelogMergeTreeRewriter.UpgradeStrategy.CHANGELOG_WITH_REWRITE;
+import static
org.apache.paimon.mergetree.compact.ChangelogMergeTreeRewriter.UpgradeStrategy.NO_CHANGELOG;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/**
@@ -77,8 +80,19 @@ public class FirstRowMergeTreeCompactRewriter extends
ChangelogMergeTreeRewriter
}
@Override
- protected boolean upgradeChangelog(int outputLevel, DataFileMeta file) {
- return file.level() == 0;
+ protected UpgradeStrategy upgradeChangelog(int outputLevel, DataFileMeta
file) {
+ if (file.level() != 0) {
+ return NO_CHANGELOG;
+ }
+
+ if (outputLevel == maxLevel) {
+ return CHANGELOG_NO_REWRITE;
+ }
+
+ // FIRST_ROW must rewrite file, because some records that are already
at higher level may be
+ // skipped
+ // See LookupMergeFunction, it just returns newly records.
+ return CHANGELOG_WITH_REWRITE;
}
@Override
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 a0f387b24..b8d64fa86 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
@@ -33,6 +33,9 @@ import java.io.IOException;
import java.util.Comparator;
import java.util.List;
+import static
org.apache.paimon.mergetree.compact.ChangelogMergeTreeRewriter.UpgradeStrategy.CHANGELOG_NO_REWRITE;
+import static
org.apache.paimon.mergetree.compact.ChangelogMergeTreeRewriter.UpgradeStrategy.NO_CHANGELOG;
+
/** A {@link MergeTreeCompactRewriter} which produces changelog files for each
full compaction. */
public class FullChangelogMergeTreeCompactRewriter extends
ChangelogMergeTreeRewriter {
@@ -71,8 +74,8 @@ public class FullChangelogMergeTreeCompactRewriter extends
ChangelogMergeTreeRew
}
@Override
- protected boolean upgradeChangelog(int outputLevel, DataFileMeta file) {
- return outputLevel == maxLevel;
+ protected UpgradeStrategy upgradeChangelog(int outputLevel, DataFileMeta
file) {
+ return outputLevel == maxLevel ? CHANGELOG_NO_REWRITE : NO_CHANGELOG;
}
@Override
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 68df5593d..c78ded5ff 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,7 +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;
import org.apache.paimon.data.InternalRow;
@@ -34,6 +34,10 @@ import java.io.UncheckedIOException;
import java.util.Comparator;
import java.util.List;
+import static
org.apache.paimon.mergetree.compact.ChangelogMergeTreeRewriter.UpgradeStrategy.CHANGELOG_NO_REWRITE;
+import static
org.apache.paimon.mergetree.compact.ChangelogMergeTreeRewriter.UpgradeStrategy.CHANGELOG_WITH_REWRITE;
+import static
org.apache.paimon.mergetree.compact.ChangelogMergeTreeRewriter.UpgradeStrategy.NO_CHANGELOG;
+
/**
* A {@link MergeTreeCompactRewriter} which produces changelog files by lookup
for the compaction
* involving level 0 files.
@@ -44,7 +48,7 @@ public class LookupMergeTreeCompactRewriter extends
ChangelogMergeTreeRewriter {
public LookupMergeTreeCompactRewriter(
int maxLevel,
- CoreOptions.MergeEngine mergeEngine,
+ MergeEngine mergeEngine,
LookupLevels lookupLevels,
KeyValueFileReaderFactory readerFactory,
KeyValueFileWriterFactory writerFactory,
@@ -73,8 +77,25 @@ public class LookupMergeTreeCompactRewriter extends
ChangelogMergeTreeRewriter {
}
@Override
- protected boolean upgradeChangelog(int outputLevel, DataFileMeta file) {
- return file.level() == 0;
+ protected UpgradeStrategy upgradeChangelog(int outputLevel, DataFileMeta
file) {
+ if (file.level() != 0) {
+ return NO_CHANGELOG;
+ }
+
+ if (outputLevel == maxLevel) {
+ return CHANGELOG_NO_REWRITE;
+ }
+
+ // DEDUPLICATE retains the latest records as the final result, so
merging has no impact on
+ // it at all
+ if (mergeEngine == MergeEngine.DEDUPLICATE) {
+ return CHANGELOG_NO_REWRITE;
+ }
+
+ // other merge engines must rewrite file, because some records that
are already at higher
+ // level may be merged
+ // See LookupMergeFunction, it just returns newly records.
+ return CHANGELOG_WITH_REWRITE;
}
@Override
