[Feature][Connector] add get source method to all source connector (#3846)

* add transform doc

* add transform v2 document

* remove transform v1 from document

* improve document

* fix dead link

* fix dead link

* fix dead link

* update supported connnector num

* Update docs/en/transform-v2/replace.md

Co-authored-by: Zongwen Li <zongwen.li.tech@gmail.com>

* fix ci

* fix ci error

* add Parallelism and SchemaProjection inteface to Source Connector

* update schemaprojection to columnprojection

* fix code style

* tmp

* all connector add getSourceClass method

* fix ci error

* fix ci error

Co-authored-by: Zongwen Li <zongwen.li.tech@gmail.com>

Author: EricJoy2048

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/factory/FactoryUtil.java

@@ -22,6 +22,7 @@
 import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.source.SourceCommonOptions;
 import org.apache.seatunnel.api.source.SourceSplit;
+import org.apache.seatunnel.api.source.SupportParallelism;
 import org.apache.seatunnel.api.table.catalog.Catalog;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.connector.TableSource;
@@ -163,15 +164,19 @@ public static List<Factory> discoverFactories(ClassLoader classLoader) {
      * This method is called by SeaTunnel Web to get the full option rule of a source.
      * @return
      */
-    public static OptionRule sourceFullOptionRule(@NonNull Factory factory) {
+    public static OptionRule sourceFullOptionRule(@NonNull TableSourceFactory factory) {
         OptionRule sourceOptionRule = factory.optionRule();
         if (sourceOptionRule == null) {
             throw new FactoryException("sourceOptionRule can not be null");
         }
 
-        OptionRule sourceCommonOptionRule =
-            OptionRule.builder().optional(SourceCommonOptions.PARALLELISM).build();
-        sourceOptionRule.getOptionalOptions().addAll(sourceCommonOptionRule.getOptionalOptions());
+        Class<? extends SeaTunnelSource> sourceClass = factory.getSourceClass();
+        if (sourceClass.isAssignableFrom(SupportParallelism.class)) {
+            OptionRule sourceCommonOptionRule =
+                OptionRule.builder().optional(SourceCommonOptions.PARALLELISM).build();
+            sourceOptionRule.getOptionalOptions().addAll(sourceCommonOptionRule.getOptionalOptions());
+        }
+
         return sourceOptionRule;
     }
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/factory/TableSourceFactory.java

@@ -17,6 +17,7 @@
 
 package org.apache.seatunnel.api.table.factory;
 
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.source.SourceSplit;
 import org.apache.seatunnel.api.table.connector.TableSource;
 
@@ -36,4 +37,6 @@ default <T, SplitT extends SourceSplit, StateT extends Serializable> TableSource
         TableFactoryContext context) {
         throw new UnsupportedOperationException("unsupported now");
     }
+
+    Class<? extends SeaTunnelSource> getSourceClass();
 }

seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceFactory.java

@@ -24,6 +24,7 @@
 import static org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.AmazonDynamoDBConfig.URL;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -42,4 +43,9 @@ public OptionRule optionRule() {
         return OptionRule.builder()
             .required(URL, REGION, ACCESS_KEY_ID, SECRET_ACCESS_KEY, TABLE, SeaTunnelSchema.SCHEMA).build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return AmazonDynamoDBSource.class;
+    }
 }

seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.cdc.base.option.JdbcSourceOptions;
@@ -51,4 +52,9 @@ public OptionRule optionRule() {
                 JdbcSourceOptions.CONNECTION_POOL_SIZE)
             .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return MySqlIncrementalSource.class;
+    }
 }

seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSourceFactory.java

@@ -24,6 +24,7 @@
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.USERNAME;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 
@@ -40,4 +41,9 @@ public String factoryIdentifier() {
     public OptionRule optionRule() {
         return OptionRule.builder().required(HOST, DATABASE, SQL, USERNAME, PASSWORD).build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return ClickhouseSource.class;
+    }
 }

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceFactory.java

@@ -26,6 +26,7 @@
 import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.SourceConfig.SOURCE;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -44,4 +45,9 @@ public OptionRule optionRule() {
         return OptionRule.builder().required(HOSTS, INDEX).optional(USERNAME, PASSWORD, SCROLL_TIME, SCROLL_SIZE)
             .exclusive(SOURCE, SeaTunnelSchema.SCHEMA).build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return ElasticsearchSource.class;
+    }
 }

seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceFactory.java

@@ -26,6 +26,7 @@
 import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeOption.STRING_LENGTH;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -44,4 +45,9 @@ public OptionRule optionRule() {
         return OptionRule.builder().required(SeaTunnelSchema.SCHEMA).optional(ROW_NUM, SPLIT_NUM, SPLIT_READ_INTERVAL, MAP_SIZE,
             ARRAY_SIZE, BYTES_LENGTH, STRING_LENGTH).build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return FakeSource.class;
+    }
 }

seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.file.ftp.source;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -55,4 +56,9 @@ public OptionRule optionRule() {
             .optional(BaseSourceConfig.TIME_FORMAT)
             .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return FtpFileSource.class;
+    }
 }

seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.file.hdfs.source;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -52,4 +53,9 @@ public OptionRule optionRule() {
             .optional(BaseSourceConfig.TIME_FORMAT)
             .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return HdfsFileSource.class;
+    }
 }

seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.file.local.source;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -51,4 +52,9 @@ public OptionRule optionRule() {
             .optional(BaseSourceConfig.TIME_FORMAT)
             .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return LocalFileSource.class;
+    }
 }