[Feature][Transform] Support merge shading-tables to one table (#8360)

Author: hailin0

docs/en/transform-v2/table-merge.md

@@ -0,0 +1,68 @@
+# TableRename
+
+> TableMerge transform plugin
+
+## Description
+
+TableMerge transform plugin for merge sharding-tables.
+
+## Options
+
+|   name   | type   | required | default value | Description               |
+|:--------:|--------|----------|---------------|---------------------------|
+| database | string | no       |               | Specify new database name |
+|  schema  | string | no       |               | Specify new schema name   |
+|  table   | string | yes      |               | Specify new table name    |
+
+## Examples
+
+### Merge sharding-tables
+
+`
+```hocon
+env {
+    parallelism = 1
+    job.mode = "BATCH"
+}
+
+source {
+    MySQL-CDC {
+        plugin_output = "customers_mysql_cdc"
+        
+        username = "root"
+        password = "123456"
+        table-names = ["source.user_1", "source.user_2", "source.shop"]
+        base-url = "jdbc:mysql://localhost:3306/source"
+    }
+}
+
+transform {
+  TableMerge {
+    plugin_input = "customers_mysql_cdc"
+    plugin_output = "trans_result"
+    
+    table_match_regex = "source.user_.*"
+    database = "user_db"
+    table = "user_all"
+  }
+}
+
+sink {
+  Jdbc {
+    plugin_input = "trans_result"
+    
+    driver="com.mysql.cj.jdbc.Driver"
+    url="jdbc:mysql://localhost:3306/sink"
+    user="myuser"
+    password="mypwd"
+    
+    generate_sink_sql = true
+    database = "${database_name}"
+    table = "${table_name}"
+    primary_keys = ["${primary_key}"]
+    
+    schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
+    data_save_mode = "APPEND_DATA"
+  }
+}
+```

docs/zh/transform-v2/table-merge.md

@@ -0,0 +1,67 @@
+# TableRename
+
+> TableMerge transform plugin
+
+## Description
+
+表合并插件，用于分库分表合并为一个表。
+
+## Options
+
+|   name   | type   | required | default value | Description      |
+|:--------:|--------|----------|---------------|------------------|
+| database | string | no       |               | 指定新的 database 名称 |
+|  schema  | string | no       |               | 指定新的 schema 名称   |
+|  table   | string | yes      |               | 指定新的 table 名称    |
+
+## Examples
+
+### 合并分库分表为一个表
+
+```hocon
+env {
+    parallelism = 1
+    job.mode = "BATCH"
+}
+
+source {
+    MySQL-CDC {
+        plugin_output = "customers_mysql_cdc"
+        
+        username = "root"
+        password = "123456"
+        table-names = ["source.user_1", "source.user_2", "source.shop"]
+        base-url = "jdbc:mysql://localhost:3306/source"
+    }
+}
+
+transform {
+  TableMerge {
+    plugin_input = "customers_mysql_cdc"
+    plugin_output = "trans_result"
+    
+    table_match_regex = "source.user_.*"
+    database = "user_db"
+    table = "user_all"
+  }
+}
+
+sink {
+  Jdbc {
+    plugin_input = "trans_result"
+    
+    driver="com.mysql.cj.jdbc.Driver"
+    url="jdbc:mysql://localhost:3306/sink"
+    user="myuser"
+    password="mypwd"
+    
+    generate_sink_sql = true
+    database = "${database_name}"
+    table = "${table_name}"
+    primary_keys = ["${primary_key}"]
+    
+    schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
+    data_save_mode = "APPEND_DATA"
+  }
+}
+```

plugin-mapping.properties

@@ -157,3 +157,4 @@ seatunnel.transform.RowKindExtractor = seatunnel-transforms-v2
 seatunnel.transform.Metadata = seatunnel-transforms-v2
 seatunnel.transform.FieldRename = seatunnel-transforms-v2
 seatunnel.transform.TableRename = seatunnel-transforms-v2
+seatunnel.transform.TableMerge = seatunnel-transforms-v2

seatunnel-dist/src/test/java/org/apache/seatunnel/api/connector/TransformSpecificationCheckTest.java

@@ -43,7 +43,7 @@ public void testAllTransformUseFactory() {
                 FactoryUtil.discoverFactories(
                         Thread.currentThread().getContextClassLoader(),
                         TableTransformFactory.class);
-        Assertions.assertEquals(15, factories.size());
+        Assertions.assertEquals(16, factories.size());
     }
 
     @Test

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/java/org/apache/seatunnel/e2e/transform/TestTableMergeIT.java

@@ -0,0 +1,36 @@
+/*
+ * 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.seatunnel.e2e.transform;
+
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.TestTemplate;
+import org.testcontainers.containers.Container;
+
+import java.io.IOException;
+
+public class TestTableMergeIT extends TestSuiteBase {
+
+    @TestTemplate
+    public void testMergeMultiTable(TestContainer container)
+            throws IOException, InterruptedException {
+        Container.ExecResult execResult = container.executeJob("/table_merge_multi_table.conf");
+        Assertions.assertEquals(0, execResult.getExitCode());
+    }
+}

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/table_merge_multi_table.conf

@@ -0,0 +1,165 @@
+#
+# 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.
+#
+
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+}
+
+source {
+  FakeSource {
+    plugin_output = "source1"
+
+    tables_configs = [
+      {
+        row.num = 3
+        schema = {
+          table = "test.user_1"
+          columns = [
+            {
+              name = "id"
+              type = "bigint"
+            },
+            {
+              name = "name"
+              type = "string"
+            }
+          ]
+        }
+      },
+      {
+        row.num = 3
+        schema = {
+          table = "test.user_2"
+          columns = [
+            {
+              name = "id"
+              type = "bigint"
+            },
+            {
+              name = "name"
+              type = "string"
+            }
+          ]
+        }
+      },
+      {
+        row.num = 5
+        schema = {
+          table = "test.xyz"
+          columns = [
+            {
+              name = "id"
+              type = "bigint"
+            },
+            {
+              name = "age"
+              type = "int"
+            }
+          ]
+        }
+      }
+    ]
+  }
+}
+transform {
+  TableMerge {
+    plugin_input = "source1"
+    plugin_output = "transform1"
+
+    table_match_regex = "test.user_.*"
+    database = "sink"
+    table = "user_all"
+  }
+}
+sink {
+  Assert {
+    plugin_input = "transform1"
+
+    rules =
+      {
+        tables_configs = [
+          {
+            table_path = "sink.user_all"
+            row_rules = [
+              {
+                rule_type = MAX_ROW
+                rule_value = 6
+              },
+              {
+                rule_type = MIN_ROW
+                rule_value = 6
+              }
+            ],
+            field_rules = [
+                {
+                  field_name = id
+                  field_type = bigint
+                  field_value = [
+                    {
+                      rule_type = NOT_NULL
+                    }
+                  ]
+                },
+                {
+                  field_name = name
+                  field_type = string
+                  field_value = [
+                    {
+                      rule_type = NOT_NULL
+                    }
+                  ]
+                }
+            ]
+          },
+          {
+            table_path = "test.xyz"
+            row_rules = [
+            {
+              rule_type = MAX_ROW
+              rule_value = 5
+            },
+            {
+              rule_type = MIN_ROW
+              rule_value = 5
+            }
+            ],
+            field_rules = [
+              {
+                field_name = id
+                field_type = bigint
+                field_value = [
+                  {
+                    rule_type = NOT_NULL
+                  }
+                ]
+              },
+              {
+                field_name = age
+                field_type = int
+                field_value = [
+                  {
+                    rule_type = NOT_NULL
+                  }
+                ]
+              }
+            ]
+        }
+        ]
+      }
+  }
+}