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 2023/06/20 11:59:42 UTC
[incubator-paimon] branch master updated: [refactor] Refactor flink actions (#1408)
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 90170043c [refactor] Refactor flink actions (#1408)
90170043c is described below
commit 90170043c9aabb19290458c179cf3c01bba86ab2
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Tue Jun 20 19:59:36 2023 +0800
[refactor] Refactor flink actions (#1408)
---
.../org/apache/paimon/flink/action/Action.java | 94 ++++++++++------------
.../apache/paimon/flink/action/CompactAction.java | 8 --
.../apache/paimon/flink/action/DeleteAction.java | 9 +--
.../paimon/flink/action/DropPartitionAction.java | 20 +----
.../paimon/flink/action/MergeIntoAction.java | 77 +++++-------------
.../paimon/flink/action/RollbackToAction.java | 9 +--
.../action/cdc/kafka/KafkaSyncDatabaseAction.java | 11 ++-
.../action/cdc/kafka/KafkaSyncTableAction.java | 8 +-
.../action/cdc/mysql/MySqlSyncDatabaseAction.java | 9 ++-
.../action/cdc/mysql/MySqlSyncTableAction.java | 8 +-
10 files changed, 90 insertions(+), 163 deletions(-)
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/Action.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/Action.java
index 8526f6834..5caf190a9 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/Action.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/Action.java
@@ -19,15 +19,15 @@
package org.apache.paimon.flink.action;
import org.apache.paimon.catalog.CatalogUtils;
+import org.apache.paimon.flink.action.cdc.kafka.KafkaSyncDatabaseAction;
import org.apache.paimon.flink.action.cdc.kafka.KafkaSyncTableAction;
import org.apache.paimon.flink.action.cdc.mysql.MySqlSyncDatabaseAction;
import org.apache.paimon.flink.action.cdc.mysql.MySqlSyncTableAction;
+import org.apache.paimon.utils.Preconditions;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.MultipleParameterTool;
-import javax.annotation.Nullable;
-
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -42,7 +42,6 @@ public interface Action {
/** The execution method of the action. */
void run() throws Exception;
- @Nullable
static Tuple3<String, String, String> getTablePath(MultipleParameterTool params) {
String warehouse = params.get("warehouse");
String database = params.get("database");
@@ -53,10 +52,8 @@ public interface Action {
int count = 0;
if (warehouse != null || database != null || table != null) {
if (warehouse == null || database == null || table == null) {
- System.err.println(
- "Warehouse, database and table must be specified all at once.\n"
- + "Run <action> --help for help.");
- return null;
+ throw new IllegalArgumentException(
+ "Warehouse, database and table must be specified all at once to specify a table.");
}
tablePath = Tuple3.of(warehouse, database, table);
count++;
@@ -71,47 +68,59 @@ public interface Action {
}
if (count != 1) {
- System.err.println(
- "Please specify either \"warehouse, database and table\" or \"path\".\n"
- + "Run <action> --help for help.");
- return null;
+ throw new IllegalArgumentException(
+ "Please specify either \"warehouse, database and table\" or \"path\".");
}
return tablePath;
}
- @Nullable
static List<Map<String, String>> getPartitions(MultipleParameterTool params) {
List<Map<String, String>> partitions = new ArrayList<>();
for (String partition : params.getMultiParameter("partition")) {
- Map<String, String> kvs = parseKeyValues(partition);
- if (kvs == null) {
- return null;
- }
+ Map<String, String> kvs = parseCommaSeparatedKeyValues(partition);
partitions.add(kvs);
}
return partitions;
}
- static Map<String, String> parseKeyValues(String keyValues) {
+ static Map<String, String> parseCommaSeparatedKeyValues(String keyValues) {
Map<String, String> kvs = new HashMap<>();
for (String kvString : keyValues.split(",")) {
- String[] kv = kvString.split("=");
- if (kv.length != 2) {
- System.err.print(
- "Invalid key-value pair \""
- + kvString
- + "\".\n"
- + "Run <action> --help for help.");
- return null;
- }
- kvs.put(kv[0].trim(), kv[1].trim());
+ parseKeyValueString(kvs, kvString);
}
-
return kvs;
}
+ static Map<String, String> optionalConfigMap(MultipleParameterTool params, String key) {
+ if (!params.has(key)) {
+ return Collections.emptyMap();
+ }
+
+ Map<String, String> config = new HashMap<>();
+ for (String kvString : params.getMultiParameter(key)) {
+ parseKeyValueString(config, kvString);
+ }
+ return config;
+ }
+
+ static void parseKeyValueString(Map<String, String> map, String kvString) {
+ String[] kv = kvString.split("=");
+ if (kv.length != 2) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Invalid key-value string '%s'. Please use format 'key=value'",
+ kvString));
+ }
+ map.put(kv[0].trim(), kv[1].trim());
+ }
+
+ static void checkRequiredArgument(MultipleParameterTool params, String key) {
+ Preconditions.checkArgument(
+ params.has(key), "Argument '%s' is required. Run '<action> --help' for help.", key);
+ }
+
/** Factory to create {@link Action}. */
class Factory {
@@ -126,6 +135,7 @@ public interface Action {
private static final String MYSQL_SYNC_DATABASE = "mysql-sync-database";
private static final String KAFKA_SYNC_TABLE = "kafka-sync-table";
+ private static final String KAFKA_SYNC_DATABASE = "kafka-sync-database";
public static Optional<Action> create(String[] args) {
String action = args[0].toLowerCase();
@@ -148,10 +158,11 @@ public interface Action {
return MySqlSyncDatabaseAction.create(actionArgs);
case KAFKA_SYNC_TABLE:
return KafkaSyncTableAction.create(actionArgs);
+ case KAFKA_SYNC_DATABASE:
+ return KafkaSyncDatabaseAction.create(actionArgs);
default:
- System.err.println("Unknown action \"" + action + "\"");
printHelp();
- return Optional.empty();
+ throw new UnsupportedOperationException("Unknown action \"" + action + "\".");
}
}
@@ -164,30 +175,13 @@ public interface Action {
System.out.println(" " + DROP_PARTITION);
System.out.println(" " + DELETE);
System.out.println(" " + MERGE_INTO);
+ System.out.println(" " + ROLLBACK_TO);
System.out.println(" " + MYSQL_SYNC_TABLE);
System.out.println(" " + MYSQL_SYNC_DATABASE);
+ System.out.println(" " + KAFKA_SYNC_TABLE);
+ System.out.println(" " + KAFKA_SYNC_DATABASE);
System.out.println("For detailed options of each action, run <action> --help");
}
}
-
- static Map<String, String> optionalConfigMap(MultipleParameterTool params, String key) {
- if (!params.has(key)) {
- return Collections.emptyMap();
- }
-
- Map<String, String> map = new HashMap<>();
- for (String param : params.getMultiParameter(key)) {
- String[] kv = param.split("=");
- if (kv.length == 2) {
- map.put(kv[0], kv[1]);
- continue;
- }
-
- throw new IllegalArgumentException(
- String.format(
- "Invalid argument '%s %s'. Please use format 'key=value'", key, param));
- }
- return map;
- }
}
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 f2064bc29..ce36dc297 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
@@ -140,10 +140,6 @@ public class CompactAction extends TableActionBase {
Tuple3<String, String, String> tablePath = getTablePath(params);
- if (tablePath == null) {
- return Optional.empty();
- }
-
Map<String, String> catalogConfig = optionalConfigMap(params, "catalog-conf");
CompactAction action =
@@ -151,10 +147,6 @@ public class CompactAction extends TableActionBase {
if (params.has("partition")) {
List<Map<String, String>> partitions = getPartitions(params);
- if (partitions == null) {
- return Optional.empty();
- }
-
action.withPartitions(partitions);
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteAction.java
index 18f2c290d..02587c60e 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteAction.java
@@ -68,13 +68,10 @@ public class DeleteAction extends TableActionBase {
Tuple3<String, String, String> tablePath = getTablePath(params);
- if (tablePath == null) {
- return Optional.empty();
- }
-
String filter = params.get("where");
if (filter == null) {
- return Optional.empty();
+ throw new IllegalArgumentException(
+ "Please specify deletion filter. If you want to delete all records, please use overwrite (see doc).");
}
Map<String, String> catalogConfig = optionalConfigMap(params, "catalog-conf");
@@ -97,7 +94,7 @@ public class DeleteAction extends TableActionBase {
System.out.println();
System.out.println(
- "The '--where <filter_spec>' part is equal to the 'WHERE' clause in SQL DELETE statement. If you want delete all records, please use overwrite (see doc).");
+ "The '--where <filter_spec>' part is equal to the 'WHERE' clause in SQL DELETE statement. If you want to delete all records, please use overwrite (see doc).");
System.out.println();
System.out.println("Examples:");
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DropPartitionAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DropPartitionAction.java
index 37c94dcde..a41a9191e 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DropPartitionAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DropPartitionAction.java
@@ -33,6 +33,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.UUID;
+import static org.apache.paimon.flink.action.Action.checkRequiredArgument;
import static org.apache.paimon.flink.action.Action.getPartitions;
import static org.apache.paimon.flink.action.Action.getTablePath;
import static org.apache.paimon.flink.action.Action.optionalConfigMap;
@@ -77,25 +78,8 @@ public class DropPartitionAction extends TableActionBase {
Tuple3<String, String, String> tablePath = getTablePath(params);
- if (tablePath == null) {
- return Optional.empty();
- }
-
- if (!params.has("partition")) {
- LOG.info(
- "Action drop-partition must specify partitions needed to be dropped.\n"
- + "Run drop-partition --help for help.");
- System.err.println(
- "Action drop-partition must specify partitions needed to be dropped.\n"
- + "Run drop-partition --help for help.");
-
- return Optional.empty();
- }
-
+ checkRequiredArgument(params, "partition");
List<Map<String, String>> partitions = getPartitions(params);
- if (partitions == null) {
- return Optional.empty();
- }
Map<String, String> catalogConfig = optionalConfigMap(params, "catalog-conf");
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MergeIntoAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MergeIntoAction.java
index a254ff71b..cd1179444 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MergeIntoAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MergeIntoAction.java
@@ -47,9 +47,10 @@ import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import static org.apache.paimon.flink.action.Action.checkRequiredArgument;
import static org.apache.paimon.flink.action.Action.getTablePath;
import static org.apache.paimon.flink.action.Action.optionalConfigMap;
-import static org.apache.paimon.flink.action.Action.parseKeyValues;
+import static org.apache.paimon.flink.action.Action.parseCommaSeparatedKeyValues;
/**
* Flink action for 'MERGE INTO', which references the syntax as follows (we use 'upsert' semantics
@@ -233,9 +234,6 @@ public class MergeIntoAction extends TableActionBase {
}
Tuple3<String, String, String> tablePath = getTablePath(params);
- if (tablePath == null) {
- return Optional.empty();
- }
Map<String, String> catalogConfig = optionalConfigMap(params, "catalog-conf");
@@ -251,14 +249,10 @@ public class MergeIntoAction extends TableActionBase {
action.withSourceSqls(sourceSqls.toArray(new String[0]));
}
- if (argumentAbsent(params, "source-table")) {
- return Optional.empty();
- }
+ checkRequiredArgument(params, "source-table");
action.withSourceTable(params.get("source-table"));
- if (argumentAbsent(params, "on")) {
- return Optional.empty();
- }
+ checkRequiredArgument(params, "on");
action.withMergeCondition(params.get("on"));
List<String> actions =
@@ -266,16 +260,12 @@ public class MergeIntoAction extends TableActionBase {
.map(String::trim)
.collect(Collectors.toList());
if (actions.contains("matched-upsert")) {
- if (argumentAbsent(params, "matched-upsert-set")) {
- return Optional.empty();
- }
+ checkRequiredArgument(params, "matched-upsert-set");
action.withMatchedUpsert(
params.get("matched-upsert-condition"), params.get("matched-upsert-set"));
}
if (actions.contains("not-matched-by-source-upsert")) {
- if (argumentAbsent(params, "not-matched-by-source-upsert-set")) {
- return Optional.empty();
- }
+ checkRequiredArgument(params, "not-matched-by-source-upsert-set");
action.withNotMatchedBySourceUpsert(
params.get("not-matched-by-source-upsert-condition"),
params.get("not-matched-by-source-upsert-set"));
@@ -288,66 +278,47 @@ public class MergeIntoAction extends TableActionBase {
params.get("not-matched-by-source-delete-condition"));
}
if (actions.contains("not-matched-insert")) {
- if (argumentAbsent(params, "not-matched-insert-values")) {
- return Optional.empty();
- }
+ checkRequiredArgument(params, "not-matched-insert-values");
action.withNotMatchedInsert(
params.get("not-matched-insert-condition"),
params.get("not-matched-insert-values"));
}
- if (!validate(action)) {
- return Optional.empty();
- }
+ validate(action);
return Optional.of(action);
}
- private static boolean argumentAbsent(MultipleParameterTool params, String key) {
- if (!params.has(key)) {
- System.err.println(key + " is absent.\nRun <action> --help for help.");
- return true;
- }
-
- return false;
- }
-
- private static boolean validate(MergeIntoAction action) {
+ private static void validate(MergeIntoAction action) {
if (!action.matchedUpsert
&& !action.notMatchedUpsert
&& !action.matchedDelete
&& !action.notMatchedDelete
&& !action.insert) {
- System.err.println(
- "Must specify at least one merge action.\nRun <action> --help for help.");
- return false;
+ throw new IllegalArgumentException(
+ "Must specify at least one merge action. Run 'merge-into --help' for help.");
}
if ((action.matchedUpsert && action.matchedDelete)
&& (action.matchedUpsertCondition == null
|| action.matchedDeleteCondition == null)) {
- System.err.println(
- "If both matched-upsert and matched-delete actions are present, their conditions must both be present too.\n"
- + "Run <action> --help for help.");
- return false;
+ throw new IllegalArgumentException(
+ "If both matched-upsert and matched-delete actions are present, their conditions must both be present too.");
}
if ((action.notMatchedUpsert && action.notMatchedDelete)
&& (action.notMatchedBySourceUpsertCondition == null
|| action.notMatchedBySourceDeleteCondition == null)) {
- System.err.println(
- "If both not-matched-by-source-upsert and not-matched-by--source-delete actions are present, their conditions must both be present too.\n"
- + "Run <action> --help for help.");
- return false;
+ throw new IllegalArgumentException(
+ "If both not-matched-by-source-upsert and not-matched-by--source-delete actions are present, "
+ + "their conditions must both be present too.\n");
}
if (action.notMatchedBySourceUpsertSet != null
&& action.notMatchedBySourceUpsertSet.equals("*")) {
- System.err.println("The '*' cannot be used in not-matched-by-source-upsert-set");
- return false;
+ throw new IllegalArgumentException(
+ "The '*' cannot be used in not-matched-by-source-upsert-set");
}
-
- return true;
}
private static void printHelp() {
@@ -499,11 +470,7 @@ public class MergeIntoAction extends TableActionBase {
// validate upsert changes
// no need to check primary keys changes because merge condition must contain all pks
// of the target table
- Map<String, String> changes = parseKeyValues(matchedUpsertSet);
- if (changes == null) {
- throw new IllegalArgumentException(
- "matched-upsert-set is invalid.\nRun <action> --help for help.");
- }
+ Map<String, String> changes = parseCommaSeparatedKeyValues(matchedUpsertSet);
for (String targetField : changes.keySet()) {
if (!targetFieldNames.contains(targetField)) {
throw new RuntimeException(
@@ -544,11 +511,7 @@ public class MergeIntoAction extends TableActionBase {
}
// validate upsert change
- Map<String, String> changes = parseKeyValues(notMatchedBySourceUpsertSet);
- if (changes == null) {
- throw new IllegalArgumentException(
- "matched-upsert-set is invalid.\nRun <action> --help for help.");
- }
+ Map<String, String> changes = parseCommaSeparatedKeyValues(notMatchedBySourceUpsertSet);
for (String targetField : changes.keySet()) {
if (!targetFieldNames.contains(targetField)) {
throw new RuntimeException(
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RollbackToAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RollbackToAction.java
index cd3b66ed5..09e2ac41f 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RollbackToAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RollbackToAction.java
@@ -28,6 +28,7 @@ import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.Optional;
+import static org.apache.paimon.flink.action.Action.checkRequiredArgument;
import static org.apache.paimon.flink.action.Action.getTablePath;
import static org.apache.paimon.flink.action.Action.optionalConfigMap;
@@ -60,14 +61,8 @@ public class RollbackToAction extends TableActionBase {
Tuple3<String, String, String> tablePath = getTablePath(params);
- if (tablePath == null) {
- throw new IllegalArgumentException("Please specific table.");
- }
-
+ checkRequiredArgument(params, "snapshot");
String snapshot = params.get("snapshot");
- if (snapshot == null) {
- throw new IllegalArgumentException("Please specific snapshot.");
- }
Map<String, String> catalogConfig = optionalConfigMap(params, "catalog-conf");
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
index a4e48b12e..676c9d344 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
@@ -54,6 +54,7 @@ import java.util.function.Supplier;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
+import static org.apache.paimon.flink.action.Action.checkRequiredArgument;
import static org.apache.paimon.flink.action.Action.optionalConfigMap;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -338,9 +339,17 @@ public class KafkaSyncDatabaseAction extends ActionBase {
return Optional.empty();
}
+ checkRequiredArgument(params, "warehouse");
+ checkRequiredArgument(params, "database");
+ checkRequiredArgument(params, "kafka-conf");
+
+ int schemaInitMaxRead = 1000;
+ if (params.has("schema-init-max-read")) {
+ schemaInitMaxRead = Integer.parseInt(params.get("schema-init-max-read"));
+ }
+
String warehouse = params.get("warehouse");
String database = params.get("database");
- int schemaInitMaxRead = Integer.parseInt(params.get("schema-init-max-read"));
boolean ignoreIncompatible = Boolean.parseBoolean(params.get("ignore-incompatible"));
String tablePrefix = params.get("table-prefix");
String tableSuffix = params.get("table-suffix");
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction.java
index ec16e294e..cbcef632c 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction.java
@@ -46,6 +46,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
+import static org.apache.paimon.flink.action.Action.checkRequiredArgument;
import static org.apache.paimon.flink.action.Action.optionalConfigMap;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -207,9 +208,6 @@ public class KafkaSyncTableAction extends ActionBase {
}
Tuple3<String, String, String> tablePath = Action.getTablePath(params);
- if (tablePath == null) {
- return Optional.empty();
- }
List<String> partitionKeys = Collections.emptyList();
if (params.has("partition-keys")) {
@@ -229,9 +227,7 @@ public class KafkaSyncTableAction extends ActionBase {
computedColumnArgs = new ArrayList<>(params.getMultiParameter("computed-column"));
}
- if (!params.has("kafka-conf")) {
- return Optional.empty();
- }
+ checkRequiredArgument(params, "kafka-conf");
Map<String, String> kafkaConfig = optionalConfigMap(params, "kafka-conf");
Map<String, String> catalogConfig = optionalConfigMap(params, "catalog-conf");
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
index b057070cc..449969678 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
@@ -54,6 +54,7 @@ import java.util.Optional;
import java.util.function.Supplier;
import java.util.regex.Pattern;
+import static org.apache.paimon.flink.action.Action.checkRequiredArgument;
import static org.apache.paimon.flink.action.Action.optionalConfigMap;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -323,6 +324,10 @@ public class MySqlSyncDatabaseAction extends ActionBase {
return Optional.empty();
}
+ checkRequiredArgument(params, "warehouse");
+ checkRequiredArgument(params, "database");
+ checkRequiredArgument(params, "mysql-conf");
+
String warehouse = params.get("warehouse");
String database = params.get("database");
boolean ignoreIncompatible = Boolean.parseBoolean(params.get("ignore-incompatible"));
@@ -331,10 +336,6 @@ public class MySqlSyncDatabaseAction extends ActionBase {
String includingTables = params.get("including-tables");
String excludingTables = params.get("excluding-tables");
- if (!params.has("mysql-conf")) {
- return Optional.empty();
- }
-
Map<String, String> mySqlConfig = optionalConfigMap(params, "mysql-conf");
Map<String, String> catalogConfig = optionalConfigMap(params, "catalog-conf");
Map<String, String> tableConfig = optionalConfigMap(params, "table-conf");
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
index 17466a947..5b83ba0f7 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
@@ -51,6 +51,7 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
+import static org.apache.paimon.flink.action.Action.checkRequiredArgument;
import static org.apache.paimon.flink.action.Action.optionalConfigMap;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -267,9 +268,6 @@ public class MySqlSyncTableAction extends ActionBase {
}
Tuple3<String, String, String> tablePath = Action.getTablePath(params);
- if (tablePath == null) {
- return Optional.empty();
- }
List<String> partitionKeys = Collections.emptyList();
if (params.has("partition-keys")) {
@@ -290,9 +288,7 @@ public class MySqlSyncTableAction extends ActionBase {
computedColumnArgs = new ArrayList<>(params.getMultiParameter("computed-column"));
}
- if (!params.has("mysql-conf")) {
- return Optional.empty();
- }
+ checkRequiredArgument(params, "mysql-conf");
Map<String, String> mySqlConfig = optionalConfigMap(params, "mysql-conf");
Map<String, String> catalogConfig = optionalConfigMap(params, "catalog-conf");