This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit eea8c4abf97c78f6acc570ba78f4548724fbfb41 Author: sanshi <[email protected]> AuthorDate: Thu Sep 17 13:10:15 2026 +0800 [cdc] Support per-table configuration for MySQL database CDC (#9871) --- docs/docs/cdc-ingestion/action-configuration.mdx | 14 +++-- docs/docs/cdc-ingestion/mysql-cdc.mdx | 21 ++++++++ .../flink/action/cdc/CdcActionCommonUtils.java | 38 +++++++++++++ .../flink/action/cdc/SyncDatabaseActionBase.java | 1 + .../action/cdc/SynchronizationActionBase.java | 23 +++++++- .../cdc/SynchronizationActionFactoryBase.java | 8 +++ .../action/cdc/mysql/MySqlSyncDatabaseAction.java | 8 ++- .../cdc/mysql/MySqlSyncDatabaseActionFactory.java | 3 +- .../flink/sink/cdc/NewTableSchemaBuilder.java | 14 ++++- .../flink/action/cdc/CdcActionCommonUtilsTest.java | 62 ++++++++++++++++++++++ .../flink/action/cdc/CdcActionITCaseBase.java | 7 +++ .../cdc/mysql/MySqlSyncDatabaseActionITCase.java | 39 ++++++++++++++ 12 files changed, 230 insertions(+), 8 deletions(-) diff --git a/docs/docs/cdc-ingestion/action-configuration.mdx b/docs/docs/cdc-ingestion/action-configuration.mdx index d32c97bae1..7632220937 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`. | | `--<source>_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=<job-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 ad661d9f0d..5a3c794f94 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 23265ae644..e7228eb202 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 org.slf4j.LoggerFactory; 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 java.util.stream.Collectors; 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<String, Map<String, String>> parseTableConfigByTable( + Collection<String> values) { + Map<String, Map<String, String>> result = new HashMap<>(); + for (String value : values) { + int colon = value.indexOf(":"); + checkArgument( + colon > 0 && colon < value.length() - 1, + "Invalid table configuration %s. Expected <source-table>:<key>=<value>.", + value); + String table = value.substring(0, colon); + Map<String, String> 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<String, String> 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<DataField> 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 d7393485ad..7300d5d79a 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 @@ public abstract class SyncDatabaseActionBase extends SynchronizationActionBase { 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 fd27f6faad..23265d0740 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<String, String> tableConfig = new HashMap<>(); + protected Map<String, Map<String, String>> 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 abstract class SynchronizationActionBase extends ActionBase { return this; } + public SynchronizationActionBase withTableConfigByTable( + Map<String, Map<String, String>> tableConfigByTable) { + this.tableConfigByTable = tableConfigByTable; + return this; + } + + protected Map<String, String> tableConfig(String sourceTable) { + Map<String, String> config = new HashMap<>(tableConfig); + Map<String, String> 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 @@ public abstract class SynchronizationActionBase extends ActionBase { EventParser.Factory<RichCdcMultiplexRecord> parserFactory); protected FileStoreTable alterTableOptions(Identifier identifier, FileStoreTable table) { + return alterTableOptions(identifier, table, tableConfig); + } + + protected FileStoreTable alterTableOptions( + Identifier identifier, FileStoreTable table, Map<String, String> options) { // doesn't support altering bucket here - Map<String, String> dynamicOptions = new HashMap<>(tableConfig); + Map<String, String> 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 0656f97c62..3e86805d54 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 abstract class SynchronizationActionFactoryBase<T extends Synchronization 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 14238fd818..0769925434 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 @@ public class MySqlSyncDatabaseAction extends SyncDatabaseActionBase { partitionKeys, primaryKeys, Collections.emptyList(), - tableConfig, + tableConfig(tableInfo.identifiers().get(0).getObjectName()), tableInfo.schema(), metadataConverters, caseSensitive, @@ -161,7 +161,11 @@ public class MySqlSyncDatabaseAction extends SyncDatabaseActionBase { Supplier<String> 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 93e25b004c..afb9e9ae90 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 class MySqlSyncDatabaseActionFactory + "[--type_mapping <option1,option2...>] \\\n" + "[--mysql_conf <mysql_cdc_source_conf> [--mysql_conf <mysql_cdc_source_conf> ...]] \\\n" + "[--catalog_conf <paimon_catalog_conf> [--catalog_conf <paimon_catalog_conf> ...]] \\\n" - + "[--table_conf <paimon_table_sink_conf> [--table_conf <paimon_table_sink_conf> ...]]"); + + "[--table_conf <paimon_table_sink_conf> [--table_conf <paimon_table_sink_conf> ...]] \\\n" + + "[--table_conf_by_table <source_table>:<key>=<value> [--table_conf_by_table <source_table>:<key>=<value> ...]]"); 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 1435cfd5da..7f235ce448 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 @@ import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.buildPaimo public class NewTableSchemaBuilder implements Serializable { private final Map<String, String> tableConfig; + private final Map<String, Map<String, String>> tableConfigByTable; private final boolean caseSensitive; private final List<String> partitionKeys; private final List<String> primaryKeys; @@ -44,6 +45,7 @@ public class NewTableSchemaBuilder implements Serializable { public NewTableSchemaBuilder( Map<String, String> tableConfig, + Map<String, Map<String, String>> tableConfigByTable, boolean caseSensitive, List<String> partitionKeys, List<String> primaryKeys, @@ -52,6 +54,7 @@ public class NewTableSchemaBuilder implements Serializable { Map<String, List<String>> partitionKeyMultiple, CdcMetadataConverter[] metadataConverters) { this.tableConfig = tableConfig; + this.tableConfigByTable = tableConfigByTable; this.caseSensitive = caseSensitive; this.metadataConverters = metadataConverters; this.partitionKeys = partitionKeys; @@ -78,7 +81,7 @@ public class NewTableSchemaBuilder implements Serializable { specifiedPartitionKeys, primaryKeys, Collections.emptyList(), - tableConfig, + tableConfigFor(record.tableName()), sourceSchema, metadataConverters, caseSensitive, @@ -86,4 +89,13 @@ public class NewTableSchemaBuilder implements Serializable { requirePrimaryKeys, syncPKeysFromSourceSchema)); } + + private Map<String, String> tableConfigFor(String sourceTable) { + Map<String, String> config = new java.util.HashMap<>(tableConfig); + Map<String, String> 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 0000000000..a8a44c45b3 --- /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<String, Map<String, String>> result = + CdcActionCommonUtils.parseTableConfigByTable( + Arrays.asList("orders:bucket=8", "orders:merge-engine=deduplicate")); + + Map<String, String> 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 1eebc2ec43..79fec23a27 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 @@ public class CdcActionITCaseBase extends ActionITCaseBase { private Map<String, String> catalogConfig = Collections.emptyMap(); private Map<String, String> tableConfig = Collections.emptyMap(); + private final List<String> tableConfigByTable = new ArrayList<>(); @Nullable private Boolean ignoreIncompatible; @Nullable private Boolean mergeShards; @Nullable private String tablePrefix; @@ -500,6 +501,11 @@ public class CdcActionITCaseBase extends ActionITCaseBase { return this; } + public SyncDatabaseActionBuilder<T> withTableConfigByTable(String... configs) { + this.tableConfigByTable.addAll(Arrays.asList(configs)); + return this; + } + public SyncDatabaseActionBuilder<T> ignoreIncompatible(boolean ignoreIncompatible) { this.ignoreIncompatible = ignoreIncompatible; return this; @@ -582,6 +588,7 @@ public class CdcActionITCaseBase extends ActionITCaseBase { 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 e0b2344cf7..209d84e906 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,45 @@ public class MySqlSyncDatabaseActionITCase extends MySqlActionITCaseBase { start(); } + @Test + @Timeout(60) + public void testTableConfigByTable() throws Exception { + Map<String, String> 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<String, String> 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 {
