[Feature][connector-elasticsearch] elasticsearch source support PIT

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

@@ -41,6 +41,9 @@ support version >= 2.x and <= 8.x.
 | tls_keystore_password   | string  | no       | -                                                              |
 | tls_truststore_path     | string  | no       | -                                                              |
 | tls_truststore_password | string  | no       | -                                                              |
+| use_pit                 | boolean | no       | false                                                          |
+| pit_keep_alive          | long    | no       | 60000 (1 minute)                                               |
+| pit_batch_size          | int     | no       | 100                                                            |
 | common-options          |         | no       | -                                                              |
 
 
@@ -113,6 +116,15 @@ The path to PEM or JKS trust store. This file must be readable by the operating
 
 The key password for the trust store specified
 
+### use_pit [boolean]
+Whether to use Point-in-Time (PIT) API instead of scroll API
+
+### pit_keep_alive [long]
+The amount of time (in milliseconds) for which the PIT should be keep alive
+
+### pit_batch_size  [long]
+Maximum number of hits to be returned with each PIT search request
+
 ### common options
 
 Source plugin common parameters, please refer to [Source Common Options](../source-common-options.md) for details
@@ -266,6 +278,25 @@ source {
 }
 ```
 
+Demo7:  PIT
+```hocon
+source {
+  Elasticsearch {
+    hosts = ["https://elasticsearch:9200"]
+    username = "elastic"
+    password = "elasticsearch"
+    tls_verify_certificate = false
+    tls_verify_hostname = false
+
+    index = "st_index"
+    query = {"range": {"c_int": {"gte": 10, "lte": 20}}}
+
+    # Enable PIT API
+    use_pit = true
+    pit_keep_alive = 60000  # 1 minute in milliseconds
+    pit_batch_size = 100
+```
+
 ## Changelog
 
 <ChangeLog />

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

@@ -39,7 +39,10 @@ import ChangeLog from '../changelog/connector-elasticsearch.md';
 | tls_keystore_password   | string  | no       | -                                   |
 | tls_truststore_path     | string  | no       | -                                   |
 | tls_truststore_password | string  | no       | -                                   |
-| common-options          |         | no       | -                                   |
+| use_pit                 | boolean | no       | false                                |
+| pit_keep_alive          | long    | no       | 60000 (1 minute)                    |
+| pit_batch_size          | int     | no       | 100                                 |
+| common-options          |         | no       | -                                                              |
 
 ### hosts [array]
 
@@ -115,6 +118,16 @@ PEM 或 JKS 信任库的路径。该文件必须对运行 SeaTunnel 的操作系
 
 指定信任库的密钥密码。
 
+
+### use_pit [boolean]
+是否使用时间点 (PIT) API 代替滚动 API
+
+### pit_keep_alive [long]
+PIT 应保持活动的时间量（以毫秒为单位）
+
+### pit_batch_size  [long]
+每次 PIT 搜索请求返回的最大数量
+
 ### common options
 
 Source 插件常用参数，具体请参考 [Source 常用选项](../source-common-options.md)
@@ -267,6 +280,25 @@ source {
 }
 ```
 
+Demo7:  PIT方式滚动查询
+```hocon
+source {
+  Elasticsearch {
+    hosts = ["https://elasticsearch:9200"]
+    username = "elastic"
+    password = "elasticsearch"
+    tls_verify_certificate = false
+    tls_verify_hostname = false
+
+    index = "st_index"
+    query = {"range": {"c_int": {"gte": 10, "lte": 20}}}
+
+    # Enable PIT API
+    use_pit = true
+    pit_keep_alive = 60000  # 1 minute in milliseconds
+    pit_batch_size = 100
+```
+
 ## 变更日志
 
 <ChangeLog />

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClient.java

@@ -30,6 +30,7 @@
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.BulkResponse;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.ElasticsearchClusterInfo;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.source.IndexDocsCount;
+import org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.source.PointInTimeResult;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.source.ScrollResult;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.exception.ElasticsearchConnectorErrorCode;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.exception.ElasticsearchConnectorException;
@@ -61,6 +62,7 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
@@ -876,4 +878,208 @@ public void addField(String index, BasicTypeDefine<EsType> fieldTypeDefine) {
                     ex);
         }
     }
+
+    /**
+     * Creates a Point-in-Time (PIT) for the specified index.
+     *
+     * @param index The index to create a PIT for
+     * @param keepAlive The time to keep the PIT alive (in milliseconds)
+     * @return The PIT ID
+     */
+    public String createPointInTime(String index, long keepAlive) {
+        String endpoint = String.format("/%s/_pit?keep_alive=%dms", index.toLowerCase(), keepAlive);
+        Request request = new Request("POST", endpoint);
+        try {
+            Response response = restClient.performRequest(request);
+            if (response == null) {
+                throw new ElasticsearchConnectorException(
+                        ElasticsearchConnectorErrorCode.CREATE_PIT_FAILED,
+                        "POST " + endpoint + " response null");
+            }
+            if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
+                String entity = EntityUtils.toString(response.getEntity());
+                JsonNode jsonNode = JsonUtils.parseObject(entity);
+                return jsonNode.get("id").asText();
+            } else {
+                throw new ElasticsearchConnectorException(
+                        ElasticsearchConnectorErrorCode.CREATE_PIT_FAILED,
+                        String.format(
+                                "POST %s response status code=%d",
+                                endpoint, response.getStatusLine().getStatusCode()));
+            }
+        } catch (IOException ex) {
+            throw new ElasticsearchConnectorException(
+                    ElasticsearchConnectorErrorCode.CREATE_PIT_FAILED, ex);
+        }
+    }
+
+    /**
+     * Deletes a Point-in-Time (PIT).
+     *
+     * @param pitId The PIT ID to delete
+     * @return True if the PIT was successfully deleted
+     */
+    public boolean deletePointInTime(String pitId) {
+        String endpoint = "/_pit";
+        Request request = new Request("DELETE", endpoint);
+        Map<String, String> requestBody = new HashMap<>();
+        requestBody.put("id", pitId);
+        request.setJsonEntity(JsonUtils.toJsonString(requestBody));
+        try {
+            Response response = restClient.performRequest(request);
+            if (response == null) {
+                throw new ElasticsearchConnectorException(
+                        ElasticsearchConnectorErrorCode.DELETE_PIT_FAILED,
+                        "DELETE " + endpoint + " response null");
+            }
+            if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
+                String entity = EntityUtils.toString(response.getEntity());
+                JsonNode jsonNode = JsonUtils.parseObject(entity);
+                return jsonNode.get("succeeded").asBoolean();
+            } else {
+                throw new ElasticsearchConnectorException(
+                        ElasticsearchConnectorErrorCode.DELETE_PIT_FAILED,
+                        String.format(
+                                "DELETE %s response status code=%d",
+                                endpoint, response.getStatusLine().getStatusCode()));
+            }
+        } catch (IOException ex) {
+            throw new ElasticsearchConnectorException(
+                    ElasticsearchConnectorErrorCode.DELETE_PIT_FAILED, ex);
+        }
+    }
+
+    /**
+     * Searches using a Point-in-Time (PIT).
+     *
+     * @param pitId The PIT ID to use
+     * @param source The fields to include in the response
+     * @param query The query to execute
+     * @param batchSize The number of documents to return
+     * @param searchAfter The sort values to search after (for pagination)
+     * @param keepAlive The time to keep the PIT alive (in milliseconds)
+     * @return The search results
+     */
+    public PointInTimeResult searchWithPointInTime(
+            String pitId,
+            List<String> source,
+            Map<String, Object> query,
+            int batchSize,
+            Object[] searchAfter,
+            long keepAlive) {
+
+        Map<String, Object> requestBody = new HashMap<>();
+        requestBody.put("size", batchSize);
+        requestBody.put("query", query);
+        requestBody.put("_source", source);
+
+        // Add PIT information
+        Map<String, Object> pit = new HashMap<>();
+        pit.put("id", pitId);
+        pit.put("keep_alive", keepAlive + "ms");
+        requestBody.put("pit", pit);
+
+        // Add sort for search_after
+        List<Map<String, String>> sort = new ArrayList<>();
+        Map<String, String> sortField = new HashMap<>();
+        sortField.put("order", "asc");
+        sort.add(Collections.singletonMap("_shard_doc", "asc"));
+        requestBody.put("sort", sort);
+
+        // Add search_after if provided
+        if (searchAfter != null && searchAfter.length > 0) {
+            requestBody.put("search_after", searchAfter);
+        }
+
+        String endpoint = "/_search";
+        Request request = new Request("POST", endpoint);
+        request.setJsonEntity(JsonUtils.toJsonString(requestBody));
+
+        try {
+            Response response = restClient.performRequest(request);
+            if (response == null) {
+                throw new ElasticsearchConnectorException(
+                        ElasticsearchConnectorErrorCode.SEARCH_WITH_PIT_FAILED,
+                        "POST " + endpoint + " response null");
+            }
+            if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
+                String entity = EntityUtils.toString(response.getEntity());
+                return parsePointInTimeResponse(entity, pitId);
+            } else {
+                throw new ElasticsearchConnectorException(
+                        ElasticsearchConnectorErrorCode.SEARCH_WITH_PIT_FAILED,
+                        String.format(
+                                "POST %s response status code=%d",
+                                endpoint, response.getStatusLine().getStatusCode()));
+            }
+        } catch (IOException ex) {
+            throw new ElasticsearchConnectorException(
+                    ElasticsearchConnectorErrorCode.SEARCH_WITH_PIT_FAILED, ex);
+        }
+    }
+
+    /**
+     * Parses the response from a Point-in-Time search.
+     *
+     * @param responseJson The JSON response from Elasticsearch
+     * @param pitId The PIT ID used for the search
+     * @return The parsed search results
+     */
+    private PointInTimeResult parsePointInTimeResponse(String responseJson, String pitId) {
+        JsonNode rootNode = JsonUtils.parseObject(responseJson);
+        JsonNode hitsNode = rootNode.get("hits");
+        JsonNode totalNode = hitsNode.get("total");
+        long totalHits = totalNode.get("value").asLong();
+
+        List<Map<String, Object>> docs = new ArrayList<>();
+        JsonNode hitsArray = hitsNode.get("hits");
+        Object[] searchAfter = null;
+
+        for (JsonNode hit : hitsArray) {
+            Map<String, Object> doc = new HashMap<>();
+            // Add metadata fields
+            doc.put("_index", hit.get("_index").textValue());
+            doc.put("_id", hit.get("_id").textValue());
+            if (hit.has("_type")) {
+                doc.put("_type", hit.get("_type").textValue());
+            }
+
+            // Extract document source fields
+            JsonNode source = hit.get("_source");
+            for (Iterator<Map.Entry<String, JsonNode>> iterator = source.fields();
+                    iterator.hasNext(); ) {
+                Map.Entry<String, JsonNode> entry = iterator.next();
+                String fieldName = entry.getKey();
+                if (entry.getValue() instanceof TextNode) {
+                    doc.put(fieldName, entry.getValue().textValue());
+                } else {
+                    doc.put(fieldName, entry.getValue());
+                }
+            }
+            docs.add(doc);
+
+            // Get sort values from the last document for search_after
+            if (hit.has("sort")) {
+                searchAfter = new Object[hit.get("sort").size()];
+                for (int i = 0; i < searchAfter.length; i++) {
+                    JsonNode sortValue = hit.get("sort").get(i);
+                    if (sortValue.isNumber()) {
+                        searchAfter[i] = sortValue.asDouble();
+                    } else if (sortValue.isTextual()) {
+                        searchAfter[i] = sortValue.asText();
+                    } else {
+                        searchAfter[i] = sortValue.toString();
+                    }
+                }
+            }
+        }
+
+        // Get the updated PIT ID
+        String updatedPitId = rootNode.has("pit_id") ? rootNode.get("pit_id").asText() : pitId;
+
+        // Determine if there are more results
+        boolean hasMore = docs.size() > 0 && totalHits > 0 && docs.size() < totalHits;
+
+        return new PointInTimeResult(updatedPitId, docs, totalHits, searchAfter, hasMore);
+    }
 }

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

@@ -40,6 +40,13 @@ public class ElasticsearchConfig implements Serializable {
     private SearchTypeEnum searchType;
     private String sqlQuery;
 
+    // PIT API related fields
+    private boolean usePit;
+    private long pitKeepAlive;
+    private int pitBatchSize;
+    private String pitId;
+    private Object[] searchAfter;
+
     private CatalogTable catalogTable;
 
     public ElasticsearchConfig clone() {
@@ -52,6 +59,14 @@ public ElasticsearchConfig clone() {
         elasticsearchConfig.setCatalogTable(catalogTable);
         elasticsearchConfig.setSearchType(searchType);
         elasticsearchConfig.setSqlQuery(sqlQuery);
+
+        // PIT API related fields
+        elasticsearchConfig.setUsePit(usePit);
+        elasticsearchConfig.setPitKeepAlive(pitKeepAlive);
+        elasticsearchConfig.setPitBatchSize(pitBatchSize);
+        elasticsearchConfig.setPitId(pitId);
+        elasticsearchConfig.setSearchAfter(searchAfter != null ? searchAfter.clone() : null);
+
         return elasticsearchConfig;
     }
 }

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

@@ -29,6 +29,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 @Getter
 @Setter
@@ -87,4 +88,25 @@ public class ElasticsearchSourceOptions extends ElasticsearchBaseOptions {
                             Collections.singletonMap("match_all", new HashMap<String, String>()))
                     .withDescription(
                             "Elasticsearch query language. You can control the range of data read");
+
+    public static final Option<Boolean> USE_PIT =
+            Options.key("use_pit")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to use Point-in-Time (PIT) API instead of scroll API. PIT API is more efficient and is the recommended approach in newer Elasticsearch versions (7.10+).");
+
+    public static final Option<Long> PIT_KEEP_ALIVE =
+            Options.key("pit_keep_alive")
+                    .longType()
+                    .defaultValue(TimeUnit.MINUTES.toMillis(1)) // 1 minute in milliseconds
+                    .withDescription(
+                            "The amount of time (in milliseconds) for which the PIT should be kept alive. Default is 1 minute.");
+
+    public static final Option<Integer> PIT_BATCH_SIZE =
+            Options.key("pit_batch_size")
+                    .intType()
+                    .defaultValue(100)
+                    .withDescription(
+                            "Maximum number of hits to be returned with each PIT search request. Similar to scroll_size but for PIT API.");
 }