[cdc] Support per-table configuration for MySQL database CDC

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
 

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

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<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)) {

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java

@@ -196,6 +196,7 @@ protected EventParser.Factory<RichCdcMultiplexRecord> buildEventParserFactory()
         NewTableSchemaBuilder schemaBuilder =
                 new NewTableSchemaBuilder(
                         tableConfig,
+                        tableConfigByTable,
                         caseSensitive,
                         partitionKeys,
                         primaryKeys,

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 SynchronizationActionBase withTableConfig(Map<String, String> tableConfig
         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 @@ protected abstract void buildSink(
             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

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionFactoryBase.java

@@ -50,6 +50,14 @@ public Optional<Action> 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);

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<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 {

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 <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(

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<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 NewTableSchemaBuilder(
             Map<String, List<String>> partitionKeyMultiple,
             CdcMetadataConverter[] metadataConverters) {
         this.tableConfig = tableConfig;
+        this.tableConfigByTable = tableConfigByTable;
         this.caseSensitive = caseSensitive;
         this.metadataConverters = metadataConverters;
         this.partitionKeys = partitionKeys;
@@ -78,12 +81,21 @@ public Optional<Schema> build(RichCdcMultiplexRecord record) {
                         specifiedPartitionKeys,
                         primaryKeys,
                         Collections.emptyList(),
-                        tableConfig,
+                        tableConfigFor(record.tableName()),
                         sourceSchema,
                         metadataConverters,
                         caseSensitive,
                         false,
                         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;
+    }
 }

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");
+    }
+}

paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionITCaseBase.java

@@ -471,6 +471,7 @@ protected abstract class SyncDatabaseActionBuilder<T extends SynchronizationActi
 
         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 SyncDatabaseActionBuilder<T> withTableConfig(Map<String, String> tableCon
             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 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));