From 211233457750857c1193016dc8d95009c521d6a0 Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Wed, 16 Sep 2026 00:05:42 +0800 Subject: [PATCH 1/2] [cdc] Support per-table configuration for MySQL database CDC --- .../cdc-ingestion/action-configuration.mdx | 14 ++++- docs/docs/cdc-ingestion/mysql-cdc.mdx | 21 +++++++ .../action/cdc/CdcActionCommonUtils.java | 38 ++++++++++++ .../action/cdc/SyncDatabaseActionBase.java | 1 + .../action/cdc/SynchronizationActionBase.java | 23 ++++++- .../cdc/SynchronizationActionFactoryBase.java | 8 +++ .../cdc/mysql/MySqlSyncDatabaseAction.java | 8 ++- .../mysql/MySqlSyncDatabaseActionFactory.java | 3 +- .../flink/sink/cdc/NewTableSchemaBuilder.java | 14 ++++- .../action/cdc/CdcActionCommonUtilsTest.java | 62 +++++++++++++++++++ .../flink/action/cdc/CdcActionITCaseBase.java | 7 +++ .../mysql/MySqlSyncDatabaseActionITCase.java | 37 +++++++++++ 12 files changed, 228 insertions(+), 8 deletions(-) create mode 100644 paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtilsTest.java diff --git a/docs/docs/cdc-ingestion/action-configuration.mdx b/docs/docs/cdc-ingestion/action-configuration.mdx index d32c97bae1c7..763222093719 100644 --- a/docs/docs/cdc-ingestion/action-configuration.mdx +++ b/docs/docs/cdc-ingestion/action-configuration.mdx @@ -60,7 +60,8 @@ and omit the brackets when submitting a job. | `--warehouse`, `--database`, `--table` | Locate the Paimon target; database actions do not take `--table`. | | `--_conf key=value` | Configure the source connection, source names, and event format. Repeat for each property. | | `--catalog_conf key=value` | Configure the Paimon catalog, for example `metastore=hive` and `uri=thrift://hive-metastore:9083`. | -| `--table_conf key=value` | Set target table properties and supported sink settings. Repeat for each property. | +| `--table_conf key=value` | Set default target table properties and supported job settings. Repeat for each property. | +| `--table_conf_by_table source_table:key=value` | MySQL database action only: override a table property for one source table. Repeat for each property. | | `--partition_keys`, `--primary_keys` | Set comma-separated keys where the action supports them. | | `--type_mapping` | Select [type mapping rules](./schema-evolution#mapping-options). | | `--computed_column` | Define a derived field where supported by the action. | @@ -79,10 +80,17 @@ Set `-Dpipeline.name=` to name the synchronization job. ### Table Configuration -Use `--table_conf` for table properties and supported job settings such as `sink.parallelism`. +Use `--table_conf` for global table properties and supported job settings such as `sink.parallelism`. For a new table, the action uses these properties when creating it. For an existing table, it alters mutable properties; it does not change immutable options such as `merge-engine`, or the -bucket number. See [Configurations](../maintenance/configurations) for table and catalog options. +bucket number. + +`mysql_sync_database` also supports repeated `--table_conf_by_table source_table:key=value` +arguments. The source table name is the MySQL table name, and the per-table value overrides the +matching key from `--table_conf`; tables without an override inherit the global configuration. +Only table properties can be configured this way. Sink and job options such as `sink.parallelism`, +writer resources, and committer resources must remain in the global `--table_conf`. +See [Configurations](../maintenance/configurations) for table and catalog options. ## Computed Functions diff --git a/docs/docs/cdc-ingestion/mysql-cdc.mdx b/docs/docs/cdc-ingestion/mysql-cdc.mdx index ad661d9f0d0f..5a3c794f94f9 100644 --- a/docs/docs/cdc-ingestion/mysql-cdc.mdx +++ b/docs/docs/cdc-ingestion/mysql-cdc.mdx @@ -149,6 +149,27 @@ merged into the same target. Adding a previously excluded table with historical data is different from capturing a table created after the job starts. Use the savepoint workflow below when expanding the selected set. +### Per-table table configuration + +Use repeated `--table_conf_by_table` arguments when different MySQL source tables need different +Paimon table properties: + +```bash +--table_conf bucket=4 \ +--table_conf changelog-producer=input \ +--table_conf_by_table orders:bucket=8 \ +--table_conf_by_table users:bucket=2 +``` + +The global `--table_conf` is the default. A matching per-table option overrides the same key; +`users` above uses `bucket=2`, while an unconfigured table uses `bucket=4`. Repeat the option for +each property. The source name is the MySQL table name, not the generated Paimon table name. + +This option is supported for both `divided` and `combined` database synchronization. It applies +only to Paimon table properties. Sink and job options such as `sink.parallelism`, writer resources, +and committer resources are shared runtime settings and must be configured globally with +`--table_conf`; they cannot be overridden per table. + ### Example 1: synchronize entire database ```bash diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java index 23265ae6447f..e7228eb20275 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java @@ -33,6 +33,8 @@ import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; @@ -41,6 +43,7 @@ import static org.apache.paimon.flink.action.MultiTablesSinkMode.COMBINED; import static org.apache.paimon.flink.action.MultiTablesSinkMode.DIVIDED; +import static org.apache.paimon.utils.ParameterUtils.parseKeyValueString; import static org.apache.paimon.utils.Preconditions.checkArgument; import static org.apache.paimon.utils.Preconditions.checkState; import static org.apache.paimon.utils.StringUtils.toLowerCaseIfNeed; @@ -60,6 +63,7 @@ public class CdcActionCommonUtils { public static final String TABLE_PREFIX_DB = "table_prefix_db"; public static final String TABLE_SUFFIX_DB = "table_suffix_db"; public static final String TABLE_MAPPING = "table_mapping"; + public static final String TABLE_CONF_BY_TABLE = "table_conf_by_table"; public static final String INCLUDING_TABLES = "including_tables"; public static final String EXCLUDING_TABLES = "excluding_tables"; public static final String INCLUDING_DBS = "including_dbs"; @@ -75,6 +79,40 @@ public class CdcActionCommonUtils { public static final String SYNC_PKEYS_FROM_SOURCE_SCHEMA = "sync_primary_keys_from_source_schema"; + public static Map> parseTableConfigByTable( + Collection values) { + Map> result = new HashMap<>(); + for (String value : values) { + int colon = value.indexOf(":"); + checkArgument( + colon > 0 && colon < value.length() - 1, + "Invalid table configuration %s. Expected :=.", + value); + String table = value.substring(0, colon); + Map parsed = new HashMap<>(); + parseKeyValueString(parsed, value.substring(colon + 1)); + String key = parsed.keySet().stream().findFirst().orElse(""); + checkArgument(!key.isEmpty(), "Table configuration key must not be empty."); + checkArgument( + !key.startsWith("sink."), + "Configuration %s cannot be configured per table; use table_conf instead.", + key); + checkArgument(!parsed.containsKey(""), "Table configuration key must not be empty."); + checkArgument( + parsed.size() == 1, + "Invalid table configuration %s. Expected exactly one key=value pair.", + value); + Map options = result.computeIfAbsent(table, ignored -> new HashMap<>()); + checkArgument( + !options.containsKey(key), + "Duplicate table configuration for source table %s and key %s.", + table, + key); + options.put(key, parsed.get(key)); + } + return result; + } + public static void assertSchemaCompatible( TableSchema paimonSchema, List sourceTableFields) { if (!schemaCompatible(paimonSchema, sourceTableFields)) { diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java index d7393485ad73..7300d5d79a46 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java @@ -196,6 +196,7 @@ protected EventParser.Factory buildEventParserFactory() NewTableSchemaBuilder schemaBuilder = new NewTableSchemaBuilder( tableConfig, + tableConfigByTable, caseSensitive, partitionKeys, primaryKeys, diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionBase.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionBase.java index fd27f6faadc8..23265d074095 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionBase.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionBase.java @@ -67,6 +67,7 @@ public abstract class SynchronizationActionBase extends ActionBase { protected final boolean caseSensitive; protected Map tableConfig = new HashMap<>(); + protected Map> tableConfigByTable = new HashMap<>(); protected TypeMapping typeMapping = TypeMapping.defaultMapping(); // this is to specify if we should use primary keys from source // in paimon schema if pkeys are not specified in action command @@ -93,6 +94,21 @@ public SynchronizationActionBase withTableConfig(Map tableConfig return this; } + public SynchronizationActionBase withTableConfigByTable( + Map> tableConfigByTable) { + this.tableConfigByTable = tableConfigByTable; + return this; + } + + protected Map tableConfig(String sourceTable) { + Map config = new HashMap<>(tableConfig); + Map override = tableConfigByTable.get(sourceTable); + if (override != null) { + config.putAll(override); + } + return config; + } + public SynchronizationActionBase withTypeMapping(TypeMapping typeMapping) { this.typeMapping = typeMapping; return this; @@ -198,8 +214,13 @@ protected abstract void buildSink( EventParser.Factory parserFactory); protected FileStoreTable alterTableOptions(Identifier identifier, FileStoreTable table) { + return alterTableOptions(identifier, table, tableConfig); + } + + protected FileStoreTable alterTableOptions( + Identifier identifier, FileStoreTable table, Map options) { // doesn't support altering bucket here - Map dynamicOptions = new HashMap<>(tableConfig); + Map dynamicOptions = new HashMap<>(options); dynamicOptions.remove(CoreOptions.BUCKET.key()); // remove immutable options and options with equal values diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionFactoryBase.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionFactoryBase.java index 0656f97c6216..3e86805d54fa 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionFactoryBase.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionFactoryBase.java @@ -50,6 +50,14 @@ public Optional create(MultipleParameterToolAdapter params) { T action = createAction(); action.withTableConfig(optionalConfigMap(params, TABLE_CONF)); + if (params.has(CdcActionCommonUtils.TABLE_CONF_BY_TABLE)) { + checkArgument( + "mysql_sync_database".equals(identifier()), + "table_conf_by_table is only supported by mysql_sync_database."); + action.withTableConfigByTable( + CdcActionCommonUtils.parseTableConfigByTable( + params.getMultiParameter(CdcActionCommonUtils.TABLE_CONF_BY_TABLE))); + } withParams(params, action); return Optional.of(action); diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java index 14238fd81881..0769925434ce 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java @@ -149,7 +149,7 @@ protected void beforeBuildingSourceSink() throws Exception { partitionKeys, primaryKeys, Collections.emptyList(), - tableConfig, + tableConfig(tableInfo.identifiers().get(0).getObjectName()), tableInfo.schema(), metadataConverters, caseSensitive, @@ -161,7 +161,11 @@ protected void beforeBuildingSourceSink() throws Exception { Supplier errMsg = incompatibleMessage(table.schema(), tableInfo, identifier); if (shouldMonitorTable(table.schema(), fromMySql, errMsg)) { - table = alterTableOptions(identifier, table); + table = + alterTableOptions( + identifier, + table, + tableConfig(tableInfo.identifiers().get(0).getObjectName())); tables.add(table); monitoredTables.addAll(tableInfo.identifiers()); } else { diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionFactory.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionFactory.java index 93e25b004c1e..afb9e9ae9073 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionFactory.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionFactory.java @@ -90,7 +90,8 @@ public void printHelp() { + "[--type_mapping ] \\\n" + "[--mysql_conf [--mysql_conf ...]] \\\n" + "[--catalog_conf [--catalog_conf ...]] \\\n" - + "[--table_conf [--table_conf ...]]"); + + "[--table_conf [--table_conf ...]] \\\n" + + "[--table_conf_by_table := [--table_conf_by_table := ...]]"); System.out.println(); System.out.println( diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/NewTableSchemaBuilder.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/NewTableSchemaBuilder.java index 1435cfd5da2b..7f235ce44874 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/NewTableSchemaBuilder.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/NewTableSchemaBuilder.java @@ -34,6 +34,7 @@ public class NewTableSchemaBuilder implements Serializable { private final Map tableConfig; + private final Map> tableConfigByTable; private final boolean caseSensitive; private final List partitionKeys; private final List primaryKeys; @@ -44,6 +45,7 @@ public class NewTableSchemaBuilder implements Serializable { public NewTableSchemaBuilder( Map tableConfig, + Map> tableConfigByTable, boolean caseSensitive, List partitionKeys, List primaryKeys, @@ -52,6 +54,7 @@ public NewTableSchemaBuilder( Map> partitionKeyMultiple, CdcMetadataConverter[] metadataConverters) { this.tableConfig = tableConfig; + this.tableConfigByTable = tableConfigByTable; this.caseSensitive = caseSensitive; this.metadataConverters = metadataConverters; this.partitionKeys = partitionKeys; @@ -78,7 +81,7 @@ public Optional build(RichCdcMultiplexRecord record) { specifiedPartitionKeys, primaryKeys, Collections.emptyList(), - tableConfig, + tableConfigFor(record.tableName()), sourceSchema, metadataConverters, caseSensitive, @@ -86,4 +89,13 @@ public Optional build(RichCdcMultiplexRecord record) { requirePrimaryKeys, syncPKeysFromSourceSchema)); } + + private Map tableConfigFor(String sourceTable) { + Map config = new java.util.HashMap<>(tableConfig); + Map override = tableConfigByTable.get(sourceTable); + if (override != null) { + config.putAll(override); + } + return config; + } } diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtilsTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtilsTest.java new file mode 100644 index 000000000000..a8a44c45b3fe --- /dev/null +++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtilsTest.java @@ -0,0 +1,62 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.flink.action.cdc; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class CdcActionCommonUtilsTest { + + @Test + void testParseTableConfigByTable() { + Map> result = + CdcActionCommonUtils.parseTableConfigByTable( + Arrays.asList("orders:bucket=8", "orders:merge-engine=deduplicate")); + + Map expected = new HashMap<>(); + expected.put("bucket", "8"); + expected.put("merge-engine", "deduplicate"); + assertThat(result).containsEntry("orders", expected); + } + + @Test + void testRejectDuplicateTableConfig() { + assertThatThrownBy( + () -> + CdcActionCommonUtils.parseTableConfigByTable( + Arrays.asList("orders:bucket=8", "orders:bucket=4"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Duplicate table configuration"); + } + + void testRejectSinkConfiguration() { + assertThatThrownBy( + () -> + CdcActionCommonUtils.parseTableConfigByTable( + Arrays.asList("orders:sink.parallelism=1"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("cannot be configured per table"); + } +} diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionITCaseBase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionITCaseBase.java index 1eebc2ec439c..79fec23a270e 100644 --- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionITCaseBase.java +++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionITCaseBase.java @@ -471,6 +471,7 @@ protected abstract class SyncDatabaseActionBuilder catalogConfig = Collections.emptyMap(); private Map tableConfig = Collections.emptyMap(); + private final List tableConfigByTable = new ArrayList<>(); @Nullable private Boolean ignoreIncompatible; @Nullable private Boolean mergeShards; @Nullable private String tablePrefix; @@ -500,6 +501,11 @@ public SyncDatabaseActionBuilder withTableConfig(Map tableCon return this; } + public SyncDatabaseActionBuilder withTableConfigByTable(String... configs) { + this.tableConfigByTable.addAll(Arrays.asList(configs)); + return this; + } + public SyncDatabaseActionBuilder ignoreIncompatible(boolean ignoreIncompatible) { this.ignoreIncompatible = ignoreIncompatible; return this; @@ -582,6 +588,7 @@ public T build() { args.addAll(mapToArgs(getConfKey(clazz), sourceConfig)); args.addAll(mapToArgs("--catalog-conf", catalogConfig)); args.addAll(mapToArgs("--table-conf", tableConfig)); + args.addAll(listToMultiArgs("--table-conf-by-table", tableConfigByTable)); args.addAll(nullableToArgs("--ignore-incompatible", ignoreIncompatible)); args.addAll(nullableToArgs("--merge-shards", mergeShards)); diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java index e0b2344cf702..35e301a5d576 100644 --- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java +++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java @@ -69,6 +69,43 @@ public static void startContainers() { start(); } + @Test + @Timeout(60) + public void testTableConfigByTable() throws Exception { + Map mySqlConfig = getBasicMySqlConfig(); + mySqlConfig.put("database-name", "paimon_sync_database"); + MySqlSyncDatabaseAction action = + syncDatabaseActionBuilder(mySqlConfig) + .includingTables("t1|t2") + .withTableConfig(getBasicTableConfig()) + .withTableConfigByTable("t1:bucket=2", "t2:bucket=4") + .build(); + runActionWithDefaultEnv(action); + assertThat(getFileStoreTable("t1").options()).containsEntry("bucket", "2"); + assertThat(getFileStoreTable("t2").options()).containsEntry("bucket", "4"); + } + + @Test + @Timeout(60) + public void testTableConfigByTableInCombinedMode() throws Exception { + Map mySqlConfig = getBasicMySqlConfig(); + mySqlConfig.put("database-name", "paimon_sync_database"); + try (Statement statement = getStatement()) { + statement.execute("USE paimon_sync_database"); + statement.executeUpdate("CREATE TABLE config_default (k INT, v1 VARCHAR(10), PRIMARY KEY (k))"); + } + MySqlSyncDatabaseAction action = syncDatabaseActionBuilder(mySqlConfig) + .withMode(COMBINED.configString()) + .includingTables("t1|t2|config_default") + .withTableConfig(Collections.singletonMap("bucket", "3")) + .withTableConfigByTable("t1:bucket=2", "t2:bucket=4") + .build(); + runActionWithDefaultEnv(action); + assertThat(getFileStoreTable("t1").options()).containsEntry("bucket", "2"); + assertThat(getFileStoreTable("t2").options()).containsEntry("bucket", "4"); + assertThat(getFileStoreTable("config_default").options()).containsEntry("bucket", "3"); + } + @Test @Timeout(60) public void testSchemaEvolution() throws Exception { From 0090c0ce8311c86d7c1e648eb6583a273776a39f Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Wed, 16 Sep 2026 09:59:16 +0800 Subject: [PATCH 2/2] fix code format --- .../cdc/mysql/MySqlSyncDatabaseActionITCase.java | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java index 35e301a5d576..209d84e9062c 100644 --- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java +++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java @@ -92,14 +92,16 @@ public void testTableConfigByTableInCombinedMode() throws Exception { mySqlConfig.put("database-name", "paimon_sync_database"); try (Statement statement = getStatement()) { statement.execute("USE paimon_sync_database"); - statement.executeUpdate("CREATE TABLE config_default (k INT, v1 VARCHAR(10), PRIMARY KEY (k))"); + statement.executeUpdate( + "CREATE TABLE config_default (k INT, v1 VARCHAR(10), PRIMARY KEY (k))"); } - MySqlSyncDatabaseAction action = syncDatabaseActionBuilder(mySqlConfig) - .withMode(COMBINED.configString()) - .includingTables("t1|t2|config_default") - .withTableConfig(Collections.singletonMap("bucket", "3")) - .withTableConfigByTable("t1:bucket=2", "t2:bucket=4") - .build(); + MySqlSyncDatabaseAction action = + syncDatabaseActionBuilder(mySqlConfig) + .withMode(COMBINED.configString()) + .includingTables("t1|t2|config_default") + .withTableConfig(Collections.singletonMap("bucket", "3")) + .withTableConfigByTable("t1:bucket=2", "t2:bucket=4") + .build(); runActionWithDefaultEnv(action); assertThat(getFileStoreTable("t1").options()).containsEntry("bucket", "2"); assertThat(getFileStoreTable("t2").options()).containsEntry("bucket", "4");