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/26 09:16:03 UTC
(paimon) branch master updated: [flink] Make completely asynchronous in stream compaction job (#4049)
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 8a661eb14 [flink] Make completely asynchronous in stream compaction job (#4049)
8a661eb14 is described below
commit 8a661eb148a5d75ac3cc2206c73ab963df794094
Author: LsomeYeah <94...@users.noreply.github.com>
AuthorDate: Mon Aug 26 17:15:57 2024 +0800
[flink] Make completely asynchronous in stream compaction job (#4049)
---
.../org/apache/paimon/flink/action/CompactAction.java | 12 ++++++++++++
.../paimon/flink/action/CompactDatabaseAction.java | 12 ++++++++++++
.../paimon/flink/sink/CombinedTableCompactorSink.java | 9 +++++++--
.../flink/sink/MultiTablesStoreCompactOperator.java | 18 +++++++++++++-----
.../paimon/flink/utils/MultiTablesCompactorUtil.java | 14 --------------
5 files changed, 44 insertions(+), 21 deletions(-)
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
index 3cb3513dd..19c209709 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
@@ -124,6 +124,18 @@ public class CompactAction extends TableActionBase {
private void buildForTraditionalCompaction(
StreamExecutionEnvironment env, FileStoreTable table, boolean isStreaming)
throws Exception {
+ if (isStreaming) {
+ // for completely asynchronous compaction
+ HashMap<String, String> dynamicOptions =
+ new HashMap<String, String>() {
+ {
+ put(CoreOptions.NUM_SORTED_RUNS_STOP_TRIGGER.key(), "2147483647");
+ put(CoreOptions.SORT_SPILL_THRESHOLD.key(), "10");
+ put(CoreOptions.LOOKUP_WAIT.key(), "false");
+ }
+ };
+ table = table.copy(dynamicOptions);
+ }
CompactorSourceBuilder sourceBuilder =
new CompactorSourceBuilder(identifier.getFullName(), table);
CompactorSinkBuilder sinkBuilder = new CompactorSinkBuilder(table);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
index 7c5854e2d..ef6772e36 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
@@ -235,6 +235,18 @@ public class CompactDatabaseAction extends ActionBase {
String fullName,
FileStoreTable table,
boolean isStreaming) {
+ if (isStreaming) {
+ // for completely asynchronous compaction
+ HashMap<String, String> dynamicOptions =
+ new HashMap<String, String>() {
+ {
+ put(CoreOptions.NUM_SORTED_RUNS_STOP_TRIGGER.key(), "2147483647");
+ put(CoreOptions.SORT_SPILL_THRESHOLD.key(), "10");
+ put(CoreOptions.LOOKUP_WAIT.key(), "false");
+ }
+ };
+ table = table.copy(dynamicOptions);
+ }
CompactorSourceBuilder sourceBuilder =
new CompactorSourceBuilder(fullName, table)
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CombinedTableCompactorSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CombinedTableCompactorSink.java
index 3a8bf1a64..87a28091f 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CombinedTableCompactorSink.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CombinedTableCompactorSink.java
@@ -158,7 +158,7 @@ public class CombinedTableCompactorSink implements Serializable {
false,
options.get(SINK_COMMITTER_OPERATOR_CHAINING),
commitUser,
- createCommitterFactory(),
+ createCommitterFactory(isStreaming),
createCommittableStateManager(),
options.get(END_INPUT_WATERMARK)))
.setParallelism(written.getParallelism());
@@ -179,9 +179,14 @@ public class CombinedTableCompactorSink implements Serializable {
}
protected Committer.Factory<MultiTableCommittable, WrappedManifestCommittable>
- createCommitterFactory() {
+ createCommitterFactory(boolean isStreaming) {
Map<String, String> dynamicOptions = options.toMap();
dynamicOptions.put(CoreOptions.WRITE_ONLY.key(), "false");
+ if (isStreaming) {
+ dynamicOptions.put(CoreOptions.NUM_SORTED_RUNS_STOP_TRIGGER.key(), "2147483647");
+ dynamicOptions.put(CoreOptions.SORT_SPILL_THRESHOLD.key(), "10");
+ dynamicOptions.put(CoreOptions.LOOKUP_WAIT.key(), "false");
+ }
return context -> new StoreMultiCommitter(catalogLoader, context, true, dynamicOptions);
}
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 ccbccaa5f..513b694fb 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
@@ -36,7 +36,6 @@ import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.RowData;
import java.io.IOException;
-import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
@@ -215,10 +214,19 @@ public class MultiTablesStoreCompactOperator
while (true) {
try {
table = (FileStoreTable) catalog.getTable(tableId);
- table =
- table.copy(
- Collections.singletonMap(
- CoreOptions.WRITE_ONLY.key(), "false"));
+ HashMap<String, String> dynamicOptions =
+ new HashMap<String, String>() {
+ {
+ put(CoreOptions.WRITE_ONLY.key(), "false");
+ }
+ };
+ if (isStreaming) {
+ dynamicOptions.put(
+ CoreOptions.NUM_SORTED_RUNS_STOP_TRIGGER.key(), "2147483647");
+ dynamicOptions.put(CoreOptions.SORT_SPILL_THRESHOLD.key(), "10");
+ dynamicOptions.put(CoreOptions.LOOKUP_WAIT.key(), "false");
+ }
+ table = table.copy(dynamicOptions);
tables.put(tableId, table);
break;
} catch (Catalog.TableNotExistException e) {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/MultiTablesCompactorUtil.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/MultiTablesCompactorUtil.java
index 27264eb80..fc648b468 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/MultiTablesCompactorUtil.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/MultiTablesCompactorUtil.java
@@ -62,20 +62,6 @@ public class MultiTablesCompactorUtil {
}
}
- public static Map<String, String> partitionCompactOptions() {
-
- return new HashMap<String, String>() {
- {
- put(CoreOptions.SCAN_TIMESTAMP_MILLIS.key(), null);
- put(CoreOptions.SCAN_TIMESTAMP.key(), null);
- put(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(), null);
- put(CoreOptions.SCAN_SNAPSHOT_ID.key(), null);
- put(CoreOptions.SCAN_MODE.key(), CoreOptions.StartupMode.LATEST_FULL.toString());
- put(CoreOptions.WRITE_ONLY.key(), "false");
- }
- };
- }
-
public static boolean shouldCompactTable(
Identifier tableIdentifier, Pattern includingPattern, Pattern excludingPattern) {
String paimonFullTableName = tableIdentifier.getFullName();