Use PIT + search_after for delete in serverless mode (#691) (#693)

* Use PIT + search_after for delete in serverless mode (#691)

Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

* UPDATE CHANGELOG

Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

---------

Signed-off-by: Sotaro Hikita <bering1814@gmail.com>
Signed-off-by: Harsha Vamsi Kalluri <harshavamsi096@gmail.com>
Co-authored-by: Harsha Vamsi Kalluri <harshavamsi096@gmail.com>

Author: lawofcycles

CHANGELOG.md

@@ -19,6 +19,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Fixed opensearch-spark-40 dependency license SHAs and added dependabot config for sql-40 module ([#688](https://github.com/opensearch-project/opensearch-hadoop/pull/688))
 - Fixed global refresh when using dynamic index patterns ([#686](https://github.com/opensearch-project/opensearch-hadoop/pull/686))
 - Fixed build failures when downloading Apache project dependencies (Hadoop, Hive, Spark) ([#595](https://github.com/opensearch-project/opensearch-hadoop/pull/595))
+- Fixed serverless mode SaveMode.Overwrite failing when document count exceeds scroll size ([#693](https://github.com/opensearch-project/opensearch-hadoop/pull/693))
 
 ### Security
 

mr/src/main/java/org/opensearch/hadoop/rest/RestRepository.java

@@ -438,14 +438,6 @@ public void delete() {
         // 250 results
 
         int batchSize = 500;
-        StringBuilder sb = new StringBuilder(resources.getResourceWrite().index());
-        if (resources.getResourceWrite().isTyped()) {
-            sb.append('/').append(resources.getResourceWrite().type());
-        }
-        sb.append("/_search?scroll=10m&_source=false&size=");
-        sb.append(batchSize);
-        sb.append("&sort=_doc");
-        String scanQuery = sb.toString();
         ScrollReader scrollReader = new ScrollReader(
                 ScrollReaderConfigBuilder.builder(new JdkValueReader(), settings)
                         .setReadMetadata(true)
@@ -458,8 +450,27 @@ public void delete() {
                         .setErrorHandlerLoader(new AbortOnlyHandlerLoader()) // Only abort since this is internal
         );
 
+        ScrollQuery sq;
+        if (this.settings.getServerlessMode()) {
+            // Serverless mode: use PIT + search_after instead of scroll API
+            String searchQuery = "_search?size=" + batchSize + "&_source=false&track_total_hits=true";
+            BytesArray body = new BytesArray("{\"query\":{\"match_all\":{}},\"sort\":[\"_doc\",\"_id\"]}");
+            String index = resources.getResourceWrite().index();
+            String keepAlive = settings.getPitKeepAlive();
+            sq = scanLimitSearchAfter(searchQuery, body, -1, scrollReader, index, keepAlive);
+        } else {
+            StringBuilder sb = new StringBuilder(resources.getResourceWrite().index());
+            if (resources.getResourceWrite().isTyped()) {
+                sb.append('/').append(resources.getResourceWrite().type());
+            }
+            sb.append("/_search?scroll=10m&_source=false&size=");
+            sb.append(batchSize);
+            sb.append("&sort=_doc");
+            String scanQuery = sb.toString();
+            sq = scanAll(scanQuery, null, scrollReader);
+        }
+
         // start iterating
-        ScrollQuery sq = scanAll(scanQuery, null, scrollReader);
         try {
             BytesArray entry = new BytesArray(0);