[Feature][Elastic search] Support multi-table source feature

docs/en/connector-v2/source/Elasticsearch.md

@@ -19,24 +19,25 @@ support version >= 2.x and <= 8.x.
 
 ## Options
 
-|          name           |  type   | required |   default value   |
-|-------------------------|---------|----------|-------------------|
-| hosts                   | array   | yes      | -                 |
-| username                | string  | no       | -                 |
-| password                | string  | no       | -                 |
-| index                   | string  | yes      | -                 |
-| source                  | array   | no       | -                 |
-| query                   | json    | no       | {"match_all": {}} |
-| scroll_time             | string  | no       | 1m                |
-| scroll_size             | int     | no       | 100               |
-| tls_verify_certificate  | boolean | no       | true              |
-| tls_verify_hostnames    | boolean | no       | true              |
-| array_column            | map     | no       |                   |
-| tls_keystore_path       | string  | no       | -                 |
-| tls_keystore_password   | string  | no       | -                 |
-| tls_truststore_path     | string  | no       | -                 |
-| tls_truststore_password | string  | no       | -                 |
-| common-options          |         | no       | -                 |
+| name                    | type    | required | default value                        |
+| ----------------------- | ------- | -------- | ------------------------------------ |
+| hosts                   | array   | yes      | -                                    |
+| username                | string  | no       | -                                    |
+| password                | string  | no       | -                                    |
+| index                   | string  | yes      | -                                    |
+| source                  | array   | no       | -                                    |
+| query                   | json    | no       | {"match_all": {}}                    |
+| scroll_time             | string  | no       | 1m                                   |
+| scroll_size             | int     | no       | 100                                  |
+| index_list              | array   | no       | used to define a multiple table task |
+| tls_verify_certificate  | boolean | no       | true                                 |
+| tls_verify_hostnames    | boolean | no       | true                                 |
+| array_column            | map     | no       |                                      |
+| tls_keystore_path       | string  | no       | -                                    |
+| tls_keystore_password   | string  | no       | -                                    |
+| tls_truststore_path     | string  | no       | -                                    |
+| tls_truststore_password | string  | no       | -                                    |
+| common-options          |         | no       | -                                    |
 
 ### hosts [array]
 
@@ -78,6 +79,10 @@ Amount of time Elasticsearch will keep the search context alive for scroll reque
 
 Maximum number of hits to be returned with each Elasticsearch scroll request.
 
+### index_list [array]
+
+The `index_list` is used to define multi-index synchronization tasks. It is an array that contains the parameters required for single-table synchronization, such as `query`, `source/schema`, `scroll_size`, and `scroll_time`. It is recommended that `index_list` and `query` should not be configured at the same level simultaneously. Please refer to the upcoming multi-table synchronization example for more details.
+
 ### tls_verify_certificate [boolean]
 
 Enable certificates validation for HTTPS endpoints
@@ -147,6 +152,90 @@ Elasticsearch {
 }
 ```
 
+Multi-table synchronization
+
+> This example demonstrates how to read data from read_index1 and read_index2 and write it into the multi_source_write_test_index index. In read_index1, I used source to specify the fields to be read and to indicate which fields are array fields. In read_index2, I used schema to define the fields to be read (not recommended).
+
+```hocon
+source {
+  Elasticsearch {
+    hosts = ["https://elasticsearch:9200"]
+    username = "elastic"
+    password = "elasticsearch"
+    tls_verify_certificate = false
+    tls_verify_hostname = false
+    index_list = [
+       {
+           index = "read_index1"
+           query = {"range": {"c_int": {"gte": 10, "lte": 20}}}
+           source = [
+           c_map,
+           c_array,
+           c_string,
+           c_boolean,
+           c_tinyint,
+           c_smallint,
+           c_bigint,
+           c_float,
+           c_double,
+           c_decimal,
+           c_bytes,
+           c_int,
+           c_date,
+           c_timestamp]
+           array_column = {
+           c_array = "array<tinyint>"
+           }
+       }
+       {
+           index = "read_index2"
+           query = {"match_all": {}}
+           schema = {
+             fields {
+               c_map = "map<string, tinyint>"
+               c_array = "array<tinyint>"
+               c_string = string
+               c_boolean = boolean
+               c_tinyint = tinyint
+               c_smallint = smallint
+               c_bigint = bigint
+               c_float = float
+               c_double = double
+               c_decimal = "decimal(2, 1)"
+               c_bytes = bytes
+               c_int = int
+               c_date = date
+               c_timestamp = timestamp
+             }
+           }
+       }
+
+    ]
+
+  }
+}
+
+transform {
+}
+
+sink {
+  Elasticsearch {
+    hosts = ["https://elasticsearch:9200"]
+    username = "elastic"
+    password = "elasticsearch"
+    tls_verify_certificate = false
+    tls_verify_hostname = false
+
+    index = "multi_source_write_test_index"
+    index_type = "st"
+    "schema_save_mode"="CREATE_SCHEMA_WHEN_NOT_EXIST"
+    "data_save_mode"="APPEND_DATA"
+  }
+}
+```
+
+
+
 SSL (Disable certificates validation)
 
 ```hocon
@@ -196,5 +285,4 @@ source {
 
 - Add Elasticsearch Source Connector
 - [Feature] Support https protocol & compatible with opensearch ([3997](https://github.com/apache/seatunnel/pull/3997))
-- [Feature] Support DSL
-
+- [Feature] Support DSL

release-note.md

@@ -58,6 +58,7 @@
 - [Connector-v2] [Mongodb] Support to convert to double from numeric type that mongodb saved it as numeric internally (#6997)
 - [Connector-v2] [Redis] Using scan replace keys operation command,support batchWrite in single mode(#7030,#7085)
 - [Connector-V2] [Clickhouse] Add a new optional configuration `clickhouse.config` to the source connector of ClickHouse (#7143)
+- [Connector-V2] [ElasticsSource] Source support multiSource (#6730)
 
 ### Zeta(ST-Engine)
 

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/SourceConfig.java

@@ -17,15 +17,31 @@
 
 package org.apache.seatunnel.connectors.seatunnel.elasticsearch.config;
 
+import org.apache.seatunnel.shade.com.fasterxml.jackson.core.type.TypeReference;
+
 import org.apache.seatunnel.api.configuration.Option;
 import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+
+import lombok.Getter;
+import lombok.Setter;
 
+import java.io.Serializable;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-public class SourceConfig {
+@Getter
+@Setter
+public class SourceConfig implements Serializable {
+
+    public static final Option<List<Map<String, Object>>> INDEX_LIST =
+            Options.key("index_list")
+                    .type(new TypeReference<List<Map<String, Object>>>() {})
+                    .noDefaultValue()
+                    .withDescription("index_list for multiTable sync");
 
     public static final Option<String> INDEX =
             Options.key("index")
@@ -61,11 +77,30 @@ public class SourceConfig {
                     .withDescription(
                             "Maximum number of hits to be returned with each Elasticsearch scroll request");
 
-    public static final Option<Map> QUERY =
+    public static final Option<Map<String, Object>> QUERY =
             Options.key("query")
-                    .objectType(Map.class)
+                    .type(new TypeReference<Map<String, Object>>() {})
                     .defaultValue(
                             Collections.singletonMap("match_all", new HashMap<String, String>()))
                     .withDescription(
                             "Elasticsearch query language. You can control the range of data read");
+
+    private String index;
+    private List<String> source;
+    private Map<String, Object> query;
+    private String scrollTime;
+    private int scrollSize;
+
+    private CatalogTable catalogTable;
+
+    public SourceConfig clone() {
+        SourceConfig sourceConfig = new SourceConfig();
+        sourceConfig.setIndex(index);
+        sourceConfig.setSource(new ArrayList<>(source));
+        sourceConfig.setQuery(new HashMap<>(query));
+        sourceConfig.setScrollTime(scrollTime);
+        sourceConfig.setScrollSize(scrollSize);
+        sourceConfig.setCatalogTable(catalogTable);
+        return sourceConfig;
+    }
 }

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

@@ -137,7 +137,9 @@ SeaTunnelRow convert(ElasticsearchRecord rowRecord) {
                             fieldName, value, seaTunnelDataType, JsonUtils.toJsonString(rowRecord)),
                     ex);
         }
-        return new SeaTunnelRow(seaTunnelFields);
+        SeaTunnelRow seaTunnelRow = new SeaTunnelRow(seaTunnelFields);
+        seaTunnelRow.setTableId(rowRecord.getTableId());
+        return seaTunnelRow;
     }
 
     Object convertValue(SeaTunnelDataType<?> fieldType, String fieldValue)

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

@@ -30,4 +30,6 @@
 public class ElasticsearchRecord {
     private Map<String, Object> doc;
     private List<String> source;
+
+    private String tableId;
 }