fix compile error

Signed-off-by: Lantao Jin <ltjin@amazon.com>

Author: LantaoJin

core/src/main/java/org/opensearch/sql/planner/physical/CursorCloseOperator.java

@@ -64,8 +64,8 @@ public void close() {
   }
 
   private void forceCloseChildren(PhysicalPlan node) {
-    if (node instanceof TableScanOperator scan) {
-      scan.forceClose();
+    if (node instanceof TableScanOperator) {
+      ((TableScanOperator) node).forceClose();
     } else {
       node.getChild().forEach(this::forceCloseChildren);
     }

opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchClient.java

@@ -76,6 +76,13 @@ public interface OpenSearchClient {
    */
   Map<String, String> meta();
 
+  /**
+   * Force to clean up resources related to the search request.
+   *
+   * @param request search request
+   */
+  void forceCleanup(OpenSearchRequest request);
+
   /**
    * Clean up resources related to the search request, for example scroll context.
    *

opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchNodeClient.java

@@ -166,6 +166,27 @@ public Map<String, String> meta() {
         client.settings().get("plugins.sql.pagination.api", "true"));
   }
 
+  @Override
+  public void forceCleanup(OpenSearchRequest request) {
+    if (request instanceof OpenSearchScrollRequest) {
+      request.forceClean(
+          scrollId -> {
+            try {
+              client.prepareClearScroll().addScrollId(scrollId).get();
+            } catch (Exception e) {
+              throw new IllegalStateException(
+                  "Failed to clean up resources for search request " + request, e);
+            }
+          });
+    } else {
+      request.forceClean(
+          pitId -> {
+            DeletePitRequest deletePitRequest = new DeletePitRequest(pitId);
+            deletePit(deletePitRequest);
+          });
+    }
+  }
+
   @Override
   public void cleanup(OpenSearchRequest request) {
     if (request instanceof OpenSearchScrollRequest) {

opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchRestClient.java

@@ -175,6 +175,29 @@ public Map<String, String> meta() {
     }
   }
 
+  @Override
+  public void forceCleanup(OpenSearchRequest request) {
+    if (request instanceof OpenSearchScrollRequest) {
+      request.forceClean(
+          scrollId -> {
+            try {
+              ClearScrollRequest clearRequest = new ClearScrollRequest();
+              clearRequest.addScrollId(scrollId);
+              client.clearScroll(clearRequest, RequestOptions.DEFAULT);
+            } catch (IOException e) {
+              throw new IllegalStateException(
+                  "Failed to clean up resources for search request " + request, e);
+            }
+          });
+    } else {
+      request.forceClean(
+          pitId -> {
+            DeletePitRequest deletePitRequest = new DeletePitRequest(pitId);
+            deletePit(deletePitRequest);
+          });
+    }
+  }
+
   @Override
   public void cleanup(OpenSearchRequest request) {
     if (request instanceof OpenSearchScrollRequest) {

opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchQueryRequest.java

@@ -221,6 +221,19 @@ public void clean(Consumer<String> cleanAction) {
     }
   }
 
+  @Override
+  public void forceClean(Consumer<String> cleanAction) {
+    try {
+      if (this.pitId != null) {
+        cleanAction.accept(this.pitId);
+        searchDone = true;
+      }
+    } finally {
+      this.pitId = null;
+      this.searchAfter = null;
+    }
+  }
+
   @Override
   public boolean hasAnotherBatch() {
     if (this.pitId != null) {

opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchRequest.java

@@ -43,6 +43,9 @@ OpenSearchResponse search(
    */
   void clean(Consumer<String> cleanAction);
 
+  /** Force clean the request. */
+  void forceClean(Consumer<String> cleanAction);
+
   /**
    * Get the OpenSearchExprValueFactory.
    *

opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchScrollRequest.java

@@ -123,6 +123,12 @@ public void clean(Consumer<String> cleanAction) {
     }
   }
 
+  @Override
+  public void forceClean(Consumer<String> cleanAction) {
+    throw new UnsupportedOperationException(
+        "Force clean is unsupported in OpenSearchScrollRequest");
+  }
+
   /**
    * Is scroll started which means pages after first is being requested.
    *