Support merge create table event in source operator

Author: aiwenmo

Diff for: flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/CreateTableEvent.java

@@ -79,7 +79,7 @@ public SchemaChangeEventType getType() {
     }
 
     @Override
-    public SchemaChangeEvent copy(TableId newTableId) {
+    public CreateTableEvent copy(TableId newTableId) {
         return new CreateTableEvent(newTableId, schema);
     }
 }

Diff for: flink-cdc-common/src/main/java/org/apache/flink/cdc/common/factories/Factory.java

@@ -20,7 +20,9 @@
 import org.apache.flink.cdc.common.annotation.PublicEvolving;
 import org.apache.flink.cdc.common.configuration.ConfigOption;
 import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.route.RouteRule;
 
+import java.util.List;
 import java.util.Set;
 
 /**
@@ -72,6 +74,9 @@ interface Context {
         /** Returns the configuration of current pipeline. */
         Configuration getPipelineConfiguration();
 
+        /** Returns the configuration of current routers. */
+        List<RouteRule> getRouteRules();
+
         /**
          * Returns the class loader of the current session.
          *

Diff for: flink-cdc-common/src/main/java/org/apache/flink/cdc/common/factories/FactoryHelper.java

@@ -20,9 +20,11 @@
 import org.apache.flink.cdc.common.annotation.PublicEvolving;
 import org.apache.flink.cdc.common.configuration.ConfigOption;
 import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.route.RouteRule;
 import org.apache.flink.cdc.common.utils.Preconditions;
 import org.apache.flink.table.api.ValidationException;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
@@ -151,13 +153,26 @@ public static class DefaultContext implements Factory.Context {
         private final Configuration factoryConfiguration;
         private final ClassLoader classLoader;
         private final Configuration pipelineConfiguration;
+        private final List<RouteRule> routeRules;
 
         public DefaultContext(
                 Configuration factoryConfiguration,
                 Configuration pipelineConfiguration,
                 ClassLoader classLoader) {
             this.factoryConfiguration = factoryConfiguration;
             this.pipelineConfiguration = pipelineConfiguration;
+            this.routeRules = new ArrayList<>();
+            this.classLoader = classLoader;
+        }
+
+        public DefaultContext(
+                Configuration factoryConfiguration,
+                Configuration pipelineConfiguration,
+                List<RouteRule> routeRules,
+                ClassLoader classLoader) {
+            this.factoryConfiguration = factoryConfiguration;
+            this.pipelineConfiguration = pipelineConfiguration;
+            this.routeRules = routeRules;
             this.classLoader = classLoader;
         }
 
@@ -171,6 +186,11 @@ public Configuration getPipelineConfiguration() {
             return pipelineConfiguration;
         }
 
+        @Override
+        public List<RouteRule> getRouteRules() {
+            return routeRules;
+        }
+
         @Override
         public ClassLoader getClassLoader() {
             return classLoader;

Diff for: flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java

@@ -136,7 +136,8 @@ private void translate(StreamExecutionEnvironment env, PipelineDef pipelineDef)
         OperatorIDGenerator schemaOperatorIDGenerator =
                 new OperatorIDGenerator(schemaOperatorTranslator.getSchemaOperatorUid());
         DataSource dataSource =
-                sourceTranslator.createDataSource(pipelineDef.getSource(), pipelineDefConfig, env);
+                sourceTranslator.createDataSource(
+                        pipelineDef.getSource(), pipelineDef.getRoute(), pipelineDefConfig, env);
         DataSink dataSink =
                 sinkTranslator.createDataSink(pipelineDef.getSink(), pipelineDefConfig, env);
 

Diff for: flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSourceTranslator.java

@@ -23,10 +23,12 @@
 import org.apache.flink.cdc.common.event.Event;
 import org.apache.flink.cdc.common.factories.DataSourceFactory;
 import org.apache.flink.cdc.common.factories.FactoryHelper;
+import org.apache.flink.cdc.common.route.RouteRule;
 import org.apache.flink.cdc.common.source.DataSource;
 import org.apache.flink.cdc.common.source.EventSourceProvider;
 import org.apache.flink.cdc.common.source.FlinkSourceFunctionProvider;
 import org.apache.flink.cdc.common.source.FlinkSourceProvider;
+import org.apache.flink.cdc.composer.definition.RouteDef;
 import org.apache.flink.cdc.composer.definition.SourceDef;
 import org.apache.flink.cdc.composer.flink.FlinkEnvironmentUtils;
 import org.apache.flink.cdc.composer.utils.FactoryDiscoveryUtils;
@@ -35,6 +37,9 @@
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
+import java.util.List;
+import java.util.stream.Collectors;
+
 /** Translator used to build {@link DataSource} which will generate a {@link DataStream}. */
 @Internal
 public class DataSourceTranslator {
@@ -76,14 +81,37 @@ public DataStreamSource<Event> translate(
     }
 
     public DataSource createDataSource(
-            SourceDef sourceDef, Configuration pipelineConfig, StreamExecutionEnvironment env) {
+            SourceDef sourceDef,
+            List<RouteDef> routes,
+            Configuration pipelineConfig,
+            StreamExecutionEnvironment env) {
         // Search the data source factory
         DataSourceFactory sourceFactory =
                 FactoryDiscoveryUtils.getFactoryByIdentifier(
                         sourceDef.getType(), DataSourceFactory.class);
         // Add source JAR to environment
         FactoryDiscoveryUtils.getJarPathByIdentifier(sourceFactory)
                 .ifPresent(jar -> FlinkEnvironmentUtils.addJar(env, jar));
+
+        if (routes != null && !routes.isEmpty()) {
+            List<RouteRule> routeRules =
+                    routes.stream()
+                            .map(
+                                    routeDef ->
+                                            new RouteRule(
+                                                    routeDef.getSourceTable(),
+                                                    routeDef.getSinkTable(),
+                                                    routeDef.getReplaceSymbol().isPresent()
+                                                            ? routeDef.getReplaceSymbol().get()
+                                                            : null))
+                            .collect(Collectors.toList());
+            return sourceFactory.createDataSource(
+                    new FactoryHelper.DefaultContext(
+                            sourceDef.getConfig(),
+                            pipelineConfig,
+                            routeRules,
+                            Thread.currentThread().getContextClassLoader()));
+        }
         return sourceFactory.createDataSource(
                 new FactoryHelper.DefaultContext(
                         sourceDef.getConfig(),