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/03/13 15:04:35 UTC
(incubator-paimon) branch master updated: [flink] Introduce FlinkConnectorOptions.prepareCommitWaitCompaction to add document (#3002)
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 497403d10 [flink] Introduce FlinkConnectorOptions.prepareCommitWaitCompaction to add document (#3002)
497403d10 is described below
commit 497403d10d462c1ce9b493d8e0600a9f0e265c99
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Wed Mar 13 23:04:30 2024 +0800
[flink] Introduce FlinkConnectorOptions.prepareCommitWaitCompaction to add document (#3002)
---
.../compact/LookupMergeTreeCompactRewriter.java | 3 ++-
.../apache/paimon/flink/FlinkConnectorOptions.java | 21 +++++++++++++++++++--
.../org/apache/paimon/flink/sink/FlinkSink.java | 10 ++--------
.../flink/sink/MultiTablesStoreCompactOperator.java | 9 ++-------
4 files changed, 25 insertions(+), 18 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 23d2cdb26..02c3e9c6d 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
@@ -101,8 +101,9 @@ public class LookupMergeTreeCompactRewriter<T> extends ChangelogMergeTreeRewrite
return NO_CHANGELOG;
}
- // In deletionVector mode, since drop delete is required, rewrite is always required.
+ // TODO In deletionVector mode, since drop delete is required, rewrite is always required.
// TODO wait https://github.com/apache/incubator-paimon/pull/2962
+ // TODO but should be careful to not be deleted by DeletionVectorsMaintainer!
if (dvMaintainer != null) {
return CHANGELOG_WITH_REWRITE;
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
index d98709f05..94371cb60 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
@@ -19,11 +19,13 @@
package org.apache.paimon.flink;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.CoreOptions.ChangelogProducer;
import org.apache.paimon.CoreOptions.StreamingReadMode;
import org.apache.paimon.annotation.Documentation.ExcludeFromDocumentation;
import org.apache.paimon.options.ConfigOption;
import org.apache.paimon.options.ConfigOptions;
import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.options.Options;
import org.apache.paimon.options.description.DescribedEnum;
import org.apache.paimon.options.description.Description;
import org.apache.paimon.options.description.InlineElement;
@@ -34,6 +36,7 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
+import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED;
import static org.apache.paimon.CoreOptions.STREAMING_READ_MODE;
import static org.apache.paimon.options.ConfigOptions.key;
import static org.apache.paimon.options.description.TextElement.text;
@@ -140,7 +143,7 @@ public class FlinkConnectorOptions {
"When "
+ CoreOptions.CHANGELOG_PRODUCER.key()
+ " is set to "
- + CoreOptions.ChangelogProducer.FULL_COMPACTION.name()
+ + ChangelogProducer.FULL_COMPACTION.name()
+ ", full compaction will be constantly triggered after this interval.");
public static final ConfigOption<Boolean> CHANGELOG_PRODUCER_LOOKUP_WAIT =
@@ -151,7 +154,7 @@ public class FlinkConnectorOptions {
"When "
+ CoreOptions.CHANGELOG_PRODUCER.key()
+ " is set to "
- + CoreOptions.ChangelogProducer.LOOKUP.name()
+ + ChangelogProducer.LOOKUP.name()
+ ", commit will wait for changelog generation by lookup.");
public static final ConfigOption<WatermarkEmitStrategy> SCAN_WATERMARK_EMIT_STRATEGY =
@@ -344,6 +347,20 @@ public class FlinkConnectorOptions {
return list;
}
+ public static boolean prepareCommitWaitCompaction(Options options) {
+ if (options.get(DELETION_VECTORS_ENABLED)) {
+ // DeletionVector (DV) is maintained in the compaction thread, but it needs to be
+ // read into a file during prepareCommit (write thread) to commit it.
+ // We must set waitCompaction to true so that there are no multiple threads
+ // operating DV simultaneously.
+ return true;
+ }
+
+ ChangelogProducer changelogProducer = options.get(CoreOptions.CHANGELOG_PRODUCER);
+ return changelogProducer == ChangelogProducer.LOOKUP
+ && options.get(CHANGELOG_PRODUCER_LOOKUP_WAIT);
+ }
+
/** The mode of lookup cache. */
public enum LookupCacheMode {
/** Auto mode, try to use partial mode. */
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
index 40cd90e3b..545bd7f07 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
@@ -54,15 +54,14 @@ import java.util.Set;
import java.util.UUID;
import static org.apache.flink.configuration.ClusterOptions.ENABLE_FINE_GRAINED_RESOURCE_MANAGEMENT;
-import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED;
import static org.apache.paimon.CoreOptions.FULL_COMPACTION_DELTA_COMMITS;
import static org.apache.paimon.flink.FlinkConnectorOptions.CHANGELOG_PRODUCER_FULL_COMPACTION_TRIGGER_INTERVAL;
-import static org.apache.paimon.flink.FlinkConnectorOptions.CHANGELOG_PRODUCER_LOOKUP_WAIT;
import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_AUTO_TAG_FOR_SAVEPOINT;
import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_COMMITTER_CPU;
import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_COMMITTER_MEMORY;
import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_MANAGED_WRITER_BUFFER_MEMORY;
import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_USE_MANAGED_MEMORY;
+import static org.apache.paimon.flink.FlinkConnectorOptions.prepareCommitWaitCompaction;
import static org.apache.paimon.flink.utils.ManagedMemoryUtils.declareManagedMemory;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -102,12 +101,7 @@ public abstract class FlinkSink<T> implements Serializable {
} else {
Options options = table.coreOptions().toConfiguration();
ChangelogProducer changelogProducer = table.coreOptions().changelogProducer();
- // todo: deletion vectors support lookup wait
- waitCompaction =
- (changelogProducer == ChangelogProducer.LOOKUP
- && options.get(CHANGELOG_PRODUCER_LOOKUP_WAIT))
- || options.get(DELETION_VECTORS_ENABLED);
-
+ waitCompaction = prepareCommitWaitCompaction(options);
int deltaCommits = -1;
if (options.contains(FULL_COMPACTION_DELTA_COMMITS)) {
deltaCommits = options.get(FULL_COMPACTION_DELTA_COMMITS);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesStoreCompactOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesStoreCompactOperator.java
index dbb220bfd..766fb762a 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesStoreCompactOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesStoreCompactOperator.java
@@ -43,10 +43,9 @@ import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
-import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED;
import static org.apache.paimon.CoreOptions.FULL_COMPACTION_DELTA_COMMITS;
import static org.apache.paimon.flink.FlinkConnectorOptions.CHANGELOG_PRODUCER_FULL_COMPACTION_TRIGGER_INTERVAL;
-import static org.apache.paimon.flink.FlinkConnectorOptions.CHANGELOG_PRODUCER_LOOKUP_WAIT;
+import static org.apache.paimon.flink.FlinkConnectorOptions.prepareCommitWaitCompaction;
import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
/**
@@ -241,11 +240,7 @@ public class MultiTablesStoreCompactOperator
Options options = fileStoreTable.coreOptions().toConfiguration();
CoreOptions.ChangelogProducer changelogProducer =
fileStoreTable.coreOptions().changelogProducer();
- // todo: deletion vectors support lookup wait
- waitCompaction =
- (changelogProducer == CoreOptions.ChangelogProducer.LOOKUP
- && options.get(CHANGELOG_PRODUCER_LOOKUP_WAIT))
- || options.get(DELETION_VECTORS_ENABLED);
+ waitCompaction = prepareCommitWaitCompaction(options);
int deltaCommits = -1;
if (options.contains(FULL_COMPACTION_DELTA_COMMITS)) {
deltaCommits = options.get(FULL_COMPACTION_DELTA_COMMITS);