[flink] make warehouse in Flink action optional

apache · Dec 18, 2024 · a2f1902 · a2f1902
1 parent f3813f0
commit a2f1902
Show file tree

Hide file tree

Showing 103 changed files with 355 additions and 612 deletions.
diff --git a/paimon-flink/paimon-flink-1.15/src/test/java/org/apache/paimon/flink/FlinkActionITCase.java b/paimon-flink/paimon-flink-1.15/src/test/java/org/apache/paimon/flink/FlinkActionITCase.java
@@ -47,7 +47,8 @@ public void testDeleteAction() throws Exception {
         batchSql("INSERT INTO T VALUES (1, 'Hi'), (2, 'Hello'), (3, 'World')");
 
         DeleteAction action =
-                new DeleteAction(path, "default", "T", "k = 1", Collections.emptyMap());
+                new DeleteAction(
+                        "default", "T", "k = 1", Collections.singletonMap("warehouse", path));
 
         BlockingIterator<Row, Row> iterator =
                 BlockingIterator.of(sEnv.executeSql("SELECT * FROM T").collect());

diff --git a/...-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/CompactDatabaseProcedure.java b/...-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/CompactDatabaseProcedure.java
@@ -123,10 +123,9 @@ public String[] call(
             String partitionIdleTime,
             String compactStrategy)
             throws Exception {
-        String warehouse = catalog.warehouse();
         Map<String, String> catalogOptions = catalog.options();
         CompactDatabaseAction action =
-                new CompactDatabaseAction(warehouse, catalogOptions)
+                new CompactDatabaseAction(catalogOptions)
                         .includingDatabases(nullable(includingDatabases))
                         .includingTables(nullable(includingTables))
                         .excludingTables(nullable(excludingTables))

diff --git a/...k/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java b/...k/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java
@@ -139,8 +139,6 @@ public String[] call(
             String partitionIdleTime,
             String compactStrategy)
             throws Exception {
-
-        String warehouse = catalog.warehouse();
         Map<String, String> catalogOptions = catalog.options();
         Map<String, String> tableConf =
                 StringUtils.isNullOrWhitespaceOnly(tableOptions)
@@ -152,7 +150,6 @@ public String[] call(
         if (orderStrategy.isEmpty() && orderByColumns.isEmpty()) {
             action =
                     new CompactAction(
-                            warehouse,
                             identifier.getDatabaseName(),
                             identifier.getObjectName(),
                             catalogOptions,
@@ -171,7 +168,6 @@ public String[] call(
                     "sort compact do not support 'partition_idle_time'.");
             action =
                     new SortCompactAction(
-                                    warehouse,
                                     identifier.getDatabaseName(),
                                     identifier.getObjectName(),
                                     catalogOptions,

diff --git a/...paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MergeIntoProcedure.java b/...paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MergeIntoProcedure.java
@@ -179,15 +179,11 @@ public String[] call(
             String notMatchedInsertCondition,
             String notMatchedInsertValues,
             String matchedDeleteCondition) {
-        String warehouse = catalog.warehouse();
         Map<String, String> catalogOptions = catalog.options();
         Identifier identifier = Identifier.fromString(targetTableId);
         MergeIntoAction action =
                 new MergeIntoAction(
-                        warehouse,
-                        identifier.getDatabaseName(),
-                        identifier.getObjectName(),
-                        catalogOptions);
+                        identifier.getDatabaseName(), identifier.getObjectName(), catalogOptions);
         action.withTargetAlias(nullable(targetAlias));
 
         if (!sourceSqls.isEmpty()) {

diff --git a/...cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java b/...cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java
@@ -54,13 +54,12 @@
 public abstract class MessageQueueSyncTableActionBase extends SyncTableActionBase {
 
     public MessageQueueSyncTableActionBase(
-            String warehouse,
             String database,
             String table,
             Map<String, String> catalogConfig,
             Map<String, String> mqConfig,
             SyncJobHandler.SourceType sourceType) {
-        super(warehouse, database, table, catalogConfig, mqConfig, sourceType);
+        super(database, table, catalogConfig, mqConfig, sourceType);
     }
 
     @Override

diff --git a/...on-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java b/...on-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
@@ -66,13 +66,11 @@ public abstract class SyncDatabaseActionBase extends SynchronizationActionBase {
     protected Map<String, List<String>> partitionKeyMultiple = new HashMap<>();
 
     public SyncDatabaseActionBase(
-            String warehouse,
             String database,
             Map<String, String> catalogConfig,
             Map<String, String> cdcSourceConfig,
             SyncJobHandler.SourceType sourceType) {
         super(
-                warehouse,
                 database,
                 catalogConfig,
                 cdcSourceConfig,

diff --git a/...k-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionFactoryBase.java b/...k-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionFactoryBase.java
@@ -42,13 +42,11 @@
 public abstract class SyncDatabaseActionFactoryBase<T extends SyncDatabaseActionBase>
         extends SynchronizationActionFactoryBase<T> {
 
-    protected String warehouse;
     protected String database;
 
     @Override
     public Optional<Action> create(MultipleParameterToolAdapter params) {
-        this.warehouse = getRequiredValue(params, WAREHOUSE);
-        this.database = getRequiredValue(params, DATABASE);
+        this.database = params.getRequired(DATABASE);
         return super.create(params);
     }
 

diff --git a/...aimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncTableActionBase.java b/...aimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncTableActionBase.java
@@ -59,14 +59,12 @@ public abstract class SyncTableActionBase extends SynchronizationActionBase {
     protected List<ComputedColumn> computedColumns = new ArrayList<>();
 
     public SyncTableActionBase(
-            String warehouse,
             String database,
             String table,
             Map<String, String> catalogConfig,
             Map<String, String> cdcSourceConfig,
             SyncJobHandler.SourceType sourceType) {
         super(
-                warehouse,
                 database,
                 catalogConfig,
                 cdcSourceConfig,

diff --git a/...link-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncTableActionFactoryBase.java b/...link-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncTableActionFactoryBase.java
@@ -22,8 +22,6 @@
 import org.apache.paimon.flink.action.ActionFactory;
 import org.apache.paimon.flink.action.MultipleParameterToolAdapter;
 
-import org.apache.flink.api.java.tuple.Tuple3;
-
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -39,11 +37,13 @@
 public abstract class SyncTableActionFactoryBase
         extends SynchronizationActionFactoryBase<SyncTableActionBase> {
 
-    protected Tuple3<String, String, String> tablePath;
+    protected String database;
+    protected String table;
 
     @Override
     public Optional<Action> create(MultipleParameterToolAdapter params) {
-        this.tablePath = getTablePath(params);
+        this.database = params.getRequired(DATABASE);
+        this.table = params.getRequired(TABLE);
         return super.create(params);
     }
 

diff --git a/...flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionBase.java b/...flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionBase.java
@@ -71,12 +71,11 @@ public abstract class SynchronizationActionBase extends ActionBase {
     protected CdcMetadataConverter[] metadataConverters = new CdcMetadataConverter[] {};
 
     public SynchronizationActionBase(
-            String warehouse,
             String database,
             Map<String, String> catalogConfig,
             Map<String, String> cdcSourceConfig,
             SyncJobHandler syncJobHandler) {
-        super(warehouse, catalogConfig);
+        super(catalogConfig);
         this.database = database;
         this.cdcSourceConfig = Configuration.fromMap(cdcSourceConfig);
         this.syncJobHandler = syncJobHandler;

diff --git a/...dc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionFactoryBase.java b/...dc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionFactoryBase.java
@@ -25,6 +25,8 @@
 import java.util.Map;
 import java.util.Optional;
 
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
 /** Base {@link ActionFactory} for table/database synchronizing job. */
 public abstract class SynchronizationActionFactoryBase<T extends SynchronizationActionBase>
         implements ActionFactory {
@@ -38,8 +40,11 @@ public abstract class SynchronizationActionFactoryBase<T extends Synchronization
 
     @Override
     public Optional<Action> create(MultipleParameterToolAdapter params) {
-        checkRequiredArgument(params, cdcConfigIdentifier());
-        this.catalogConfig = optionalConfigMap(params, CATALOG_CONF);
+        checkArgument(
+                params.has(cdcConfigIdentifier()),
+                "Argument '%s' is required. Run '<action> --help' for help.",
+                cdcConfigIdentifier());
+        this.catalogConfig = catalogConfigMap(params);
         this.cdcSourceConfig = optionalConfigMap(params, cdcConfigIdentifier());
 
         T action = createAction();

diff --git a/...k-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java b/...k-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
@@ -29,11 +29,8 @@
 public class KafkaSyncDatabaseAction extends SyncDatabaseActionBase {
 
     public KafkaSyncDatabaseAction(
-            String warehouse,
-            String database,
-            Map<String, String> catalogConfig,
-            Map<String, String> kafkaConfig) {
-        super(warehouse, database, catalogConfig, kafkaConfig, SyncJobHandler.SourceType.KAFKA);
+            String database, Map<String, String> catalogConfig, Map<String, String> kafkaConfig) {
+        super(database, catalogConfig, kafkaConfig, SyncJobHandler.SourceType.KAFKA);
     }
 
     @Override

diff --git a/...rc/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseActionFactory.java b/...rc/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseActionFactory.java
@@ -40,7 +40,7 @@ protected String cdcConfigIdentifier() {
 
     @Override
     public KafkaSyncDatabaseAction createAction() {
-        return new KafkaSyncDatabaseAction(warehouse, database, catalogConfig, cdcSourceConfig);
+        return new KafkaSyncDatabaseAction(database, catalogConfig, cdcSourceConfig);
     }
 
     @Override

diff --git a/...link-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction.java b/...link-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction.java
@@ -27,17 +27,10 @@
 public class KafkaSyncTableAction extends MessageQueueSyncTableActionBase {
 
     public KafkaSyncTableAction(
-            String warehouse,
             String database,
             String table,
             Map<String, String> catalogConfig,
             Map<String, String> kafkaConfig) {
-        super(
-                warehouse,
-                database,
-                table,
-                catalogConfig,
-                kafkaConfig,
-                SyncJobHandler.SourceType.KAFKA);
+        super(database, table, catalogConfig, kafkaConfig, SyncJobHandler.SourceType.KAFKA);
     }
 }
diff --git a/...c/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionFactory.java b/...c/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionFactory.java
@@ -40,11 +40,7 @@ public String cdcConfigIdentifier() {
     @Override
     public KafkaSyncTableAction createAction() {
         return new KafkaSyncTableAction(
-                this.tablePath.f0,
-                this.tablePath.f1,
-                this.tablePath.f2,
-                this.catalogConfig,
-                this.cdcSourceConfig);
+                this.database, this.table, this.catalogConfig, this.cdcSourceConfig);
     }
 
     @Override

diff --git a/...c/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java b/...c/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
@@ -52,11 +52,8 @@
 public class MongoDBSyncDatabaseAction extends SyncDatabaseActionBase {
 
     public MongoDBSyncDatabaseAction(
-            String warehouse,
-            String database,
-            Map<String, String> catalogConfig,
-            Map<String, String> mongodbConfig) {
-        super(warehouse, database, catalogConfig, mongodbConfig, SyncJobHandler.SourceType.MONGODB);
+            String database, Map<String, String> catalogConfig, Map<String, String> mongodbConfig) {
+        super(database, catalogConfig, mongodbConfig, SyncJobHandler.SourceType.MONGODB);
     }
 
     @Override

diff --git a/...ain/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseActionFactory.java b/...ain/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseActionFactory.java
@@ -40,7 +40,7 @@ protected String cdcConfigIdentifier() {
 
     @Override
     public MongoDBSyncDatabaseAction createAction() {
-        return new MongoDBSyncDatabaseAction(warehouse, database, catalogConfig, cdcSourceConfig);
+        return new MongoDBSyncDatabaseAction(database, catalogConfig, cdcSourceConfig);
     }
 
     @Override

diff --git a/...-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java b/...-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java
@@ -51,18 +51,11 @@
 public class MongoDBSyncTableAction extends SyncTableActionBase {
 
     public MongoDBSyncTableAction(
-            String warehouse,
             String database,
             String table,
             Map<String, String> catalogConfig,
             Map<String, String> mongodbConfig) {
-        super(
-                warehouse,
-                database,
-                table,
-                catalogConfig,
-                mongodbConfig,
-                SyncJobHandler.SourceType.MONGODB);
+        super(database, table, catalogConfig, mongodbConfig, SyncJobHandler.SourceType.MONGODB);
     }
 
     @Override

diff --git a/...c/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableActionFactory.java b/...c/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableActionFactory.java
@@ -46,11 +46,7 @@ public String cdcConfigIdentifier() {
     @Override
     public SyncTableActionBase createAction() {
         return new MongoDBSyncTableAction(
-                this.tablePath.f0,
-                this.tablePath.f1,
-                this.tablePath.f2,
-                this.catalogConfig,
-                this.cdcSourceConfig);
+                database, table, this.catalogConfig, this.cdcSourceConfig);
     }
 
     @Override

diff --git a/...k-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java b/...k-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
@@ -102,11 +102,8 @@ public class MySqlSyncDatabaseAction extends SyncDatabaseActionBase {
     private final List<Identifier> excludedTables = new ArrayList<>();
 
     public MySqlSyncDatabaseAction(
-            String warehouse,
-            String database,
-            Map<String, String> catalogConfig,
-            Map<String, String> mySqlConfig) {
-        super(warehouse, database, catalogConfig, mySqlConfig, SyncJobHandler.SourceType.MYSQL);
+            String database, Map<String, String> catalogConfig, Map<String, String> mySqlConfig) {
+        super(database, catalogConfig, mySqlConfig, SyncJobHandler.SourceType.MYSQL);
         this.mode = DIVIDED;
     }
 

diff --git a/...rc/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionFactory.java b/...rc/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionFactory.java
@@ -49,7 +49,7 @@ protected String cdcConfigIdentifier() {
 
     @Override
     public MySqlSyncDatabaseAction createAction() {
-        return new MySqlSyncDatabaseAction(warehouse, database, catalogConfig, cdcSourceConfig);
+        return new MySqlSyncDatabaseAction(database, catalogConfig, cdcSourceConfig);
     }
 
     @Override

diff --git a/...link-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java b/...link-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
@@ -76,18 +76,11 @@ public class MySqlSyncTableAction extends SyncTableActionBase {
     private JdbcSchemasInfo mySqlSchemasInfo;
 
     public MySqlSyncTableAction(
-            String warehouse,
             String database,
             String table,
             Map<String, String> catalogConfig,
             Map<String, String> mySqlConfig) {
-        super(
-                warehouse,
-                database,
-                table,
-                catalogConfig,
-                mySqlConfig,
-                SyncJobHandler.SourceType.MYSQL);
+        super(database, table, catalogConfig, mySqlConfig, SyncJobHandler.SourceType.MYSQL);
     }
 
     @Override

diff --git a/...c/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionFactory.java b/...c/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionFactory.java
@@ -39,12 +39,7 @@ public String cdcConfigIdentifier() {
 
     @Override
     public MySqlSyncTableAction createAction() {
-        return new MySqlSyncTableAction(
-                this.tablePath.f0,
-                this.tablePath.f1,
-                this.tablePath.f2,
-                this.catalogConfig,
-                this.cdcSourceConfig);
+        return new MySqlSyncTableAction(database, table, this.catalogConfig, this.cdcSourceConfig);
     }
 
     @Override

diff --git a/...dc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableAction.java b/...dc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableAction.java
@@ -77,18 +77,11 @@ public class PostgresSyncTableAction extends SyncTableActionBase {
     private JdbcSchemasInfo postgresSchemasInfo;
 
     public PostgresSyncTableAction(
-            String warehouse,
             String database,
             String table,
             Map<String, String> catalogConfig,
             Map<String, String> postgresConfig) {
-        super(
-                warehouse,
-                database,
-                table,
-                catalogConfig,
-                postgresConfig,
-                SyncJobHandler.SourceType.POSTGRES);
+        super(database, table, catalogConfig, postgresConfig, SyncJobHandler.SourceType.POSTGRES);
     }
 
     @Override

diff --git a/...main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableActionFactory.java b/...main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresSyncTableActionFactory.java
@@ -40,11 +40,7 @@ public String cdcConfigIdentifier() {
     @Override
     public PostgresSyncTableAction createAction() {
         return new PostgresSyncTableAction(
-                this.tablePath.f0,
-                this.tablePath.f1,
-                this.tablePath.f2,
-                this.catalogConfig,
-                this.cdcSourceConfig);
+                database, table, this.catalogConfig, this.cdcSourceConfig);
     }
 
     @Override

diff --git a/...cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseAction.java b/...cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseAction.java
@@ -29,11 +29,8 @@
 public class PulsarSyncDatabaseAction extends SyncDatabaseActionBase {
 
     public PulsarSyncDatabaseAction(
-            String warehouse,
-            String database,
-            Map<String, String> catalogConfig,
-            Map<String, String> pulsarConfig) {
-        super(warehouse, database, catalogConfig, pulsarConfig, SyncJobHandler.SourceType.PULSAR);
+            String database, Map<String, String> catalogConfig, Map<String, String> pulsarConfig) {
+        super(database, catalogConfig, pulsarConfig, SyncJobHandler.SourceType.PULSAR);
     }
 
     @Override