Update CalciteExplainIT, and fixes

Signed-off-by: Kai Huang <ahkcs@amazon.com>

# Conflicts:
#	integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java

Author: ahkcs

core/src/main/java/org/opensearch/sql/calcite/plan/LogicalPaginationLimit.java

@@ -35,6 +35,13 @@ public enum SystemLimitType {
     JOIN_SUBSEARCH_MAXOUT,
     /** Max output to return from a subsearch. */
     SUBSEARCH_MAXOUT,
+    /**
+     * Pagination limit type for API pagination support.
+     *
+     * <p>This type is used to apply LIMIT/OFFSET for paginating query results. Unlike
+     * QUERY_SIZE_LIMIT, pagination is applied as post-processing after the user's query executes.
+     */
+    PAGINATION,
   }
 
   @Getter private final SystemLimitType type;

core/src/main/java/org/opensearch/sql/calcite/plan/LogicalSystemLimit.java

@@ -5,8 +5,6 @@
 
 package org.opensearch.sql.executor;
 
-import java.security.AccessController;
-import java.security.PrivilegedAction;
 import java.util.List;
 import java.util.Optional;
 import javax.annotation.Nullable;
@@ -39,7 +37,6 @@
 import org.opensearch.sql.calcite.CalciteRelNodeVisitor;
 import org.opensearch.sql.calcite.OpenSearchSchema;
 import org.opensearch.sql.calcite.SysLimit;
-import org.opensearch.sql.calcite.plan.LogicalPaginationLimit;
 import org.opensearch.sql.calcite.plan.LogicalSystemLimit;
 import org.opensearch.sql.calcite.plan.LogicalSystemLimit.SystemLimitType;
 import org.opensearch.sql.common.response.ResponseListener;
@@ -118,6 +115,36 @@ public void explain(
     }
   }
 
+  /**
+   * Explain with pagination support.
+   *
+   * @param plan the unresolved plan
+   * @param queryType the query type
+   * @param listener the response listener
+   * @param format the explain format
+   * @param pageSize page size for pagination
+   * @param offset pagination offset (0-based)
+   */
+  public void explain(
+      UnresolvedPlan plan,
+      QueryType queryType,
+      ResponseListener<ExecutionEngine.ExplainResponse> listener,
+      Explain.ExplainFormat format,
+      int pageSize,
+      int offset) {
+    if (shouldUseCalcite(queryType)) {
+      explainWithCalcite(plan, queryType, listener, format, pageSize, offset);
+    } else {
+      // For legacy path (SQL), wrap with Paginate when pagination is enabled
+      if (pageSize > 0) {
+        explainWithLegacy(
+            new Paginate(pageSize, plan), queryType, listener, format, Optional.empty());
+      } else {
+        explainWithLegacy(plan, queryType, listener, format, Optional.empty());
+      }
+    }
+  }
+
   public void executeWithCalcite(
       UnresolvedPlan plan,
       QueryType queryType,
@@ -143,26 +170,21 @@ public void executeWithCalcite(
     CalcitePlanContext.run(
         () -> {
           try {
-            AccessController.doPrivileged(
-                (PrivilegedAction<Void>)
-                    () -> {
-                      CalcitePlanContext context =
-                          CalcitePlanContext.create(
-                              buildFrameworkConfig(), SysLimit.fromSettings(settings), queryType);
+            CalcitePlanContext context =
+                CalcitePlanContext.create(
+                    buildFrameworkConfig(), SysLimit.fromSettings(settings), queryType);
 
-                      // Set pagination parameters if enabled
-                      if (pageSize > 0) {
-                        context.setPaginationSize(pageSize);
-                        context.setPaginationOffset(offset);
-                      }
+            // Set pagination parameters if enabled
+            if (pageSize > 0) {
+              context.setPaginationSize(pageSize);
+              context.setPaginationOffset(offset);
+            }
 
-                      RelNode relNode = analyze(plan, context);
-                      relNode = mergeAdjacentFilters(relNode);
-                      RelNode optimized = optimize(relNode, context);
-                      RelNode calcitePlan = convertToCalcitePlan(optimized);
-                      executionEngine.execute(calcitePlan, context, listener);
-                      return null;
-                    });
+            RelNode relNode = analyze(plan, context);
+            relNode = mergeAdjacentFilters(relNode);
+            RelNode optimized = optimize(relNode, context);
+            RelNode calcitePlan = convertToCalcitePlan(optimized);
+            executionEngine.execute(calcitePlan, context, listener);
           } catch (Throwable t) {
             if (isCalciteFallbackAllowed(t) && !(t instanceof NonFallbackCalciteException)) {
               log.warn("Fallback to V2 query engine since got exception", t);
@@ -191,12 +213,39 @@ public void explainWithCalcite(
       QueryType queryType,
       ResponseListener<ExecutionEngine.ExplainResponse> listener,
       Explain.ExplainFormat format) {
+    explainWithCalcite(plan, queryType, listener, format, 0, 0);
+  }
+
+  /**
+   * Explain with Calcite engine and pagination support.
+   *
+   * @param plan the unresolved plan
+   * @param queryType the query type
+   * @param listener the response listener
+   * @param format the explain format
+   * @param pageSize page size (0 = no pagination)
+   * @param offset pagination offset (0-based)
+   */
+  public void explainWithCalcite(
+      UnresolvedPlan plan,
+      QueryType queryType,
+      ResponseListener<ExecutionEngine.ExplainResponse> listener,
+      Explain.ExplainFormat format,
+      int pageSize,
+      int offset) {
     CalcitePlanContext.run(
         () -> {
           try {
             CalcitePlanContext context =
                 CalcitePlanContext.create(
                     buildFrameworkConfig(), SysLimit.fromSettings(settings), queryType);
+
+            // Set pagination parameters if enabled
+            if (pageSize > 0) {
+              context.setPaginationSize(pageSize);
+              context.setPaginationOffset(offset);
+            }
+
             context.run(
                 () -> {
                   RelNode relNode = analyze(plan, context);
@@ -344,9 +393,8 @@ public PhysicalPlan plan(LogicalPlan plan) {
   public RelNode optimize(RelNode plan, CalcitePlanContext context) {
     if (context.isPaginationEnabled()) {
       // Apply pagination LIMIT/OFFSET on top of user's complete query result
-      // Using LogicalPaginationLimit to preserve input's collation and ensure
-      // pagination is applied as post-processing, not merged with user's head/sort
-      return LogicalPaginationLimit.create(
+      return LogicalSystemLimit.create(
+          SystemLimitType.PAGINATION,
           plan,
           context.relBuilder.literal(context.getPaginationOffset()),
           context.relBuilder.literal(context.getPaginationSize()));

core/src/main/java/org/opensearch/sql/executor/QueryService.java

@@ -6,7 +6,6 @@
 package org.opensearch.sql.executor.execution;
 
 import java.util.Optional;
-import org.apache.commons.lang3.NotImplementedException;
 import org.opensearch.sql.ast.statement.Explain;
 import org.opensearch.sql.ast.tree.UnresolvedPlan;
 import org.opensearch.sql.common.response.ResponseListener;
@@ -88,9 +87,8 @@ public void execute() {
   public void explain(
       ResponseListener<ExecutionEngine.ExplainResponse> listener, Explain.ExplainFormat format) {
     if (pageSize.isPresent()) {
-      listener.onFailure(
-          new NotImplementedException(
-              "`explain` feature for paginated requests is not implemented yet."));
+      queryService.explain(
+          plan, getQueryType(), listener, format, pageSize.get(), paginationOffset);
     } else {
       queryService.explain(plan, getQueryType(), listener, format);
     }

core/src/main/java/org/opensearch/sql/executor/execution/QueryPlan.java

@@ -2130,4 +2130,73 @@ public void testRexStandardizationForScript() throws IOException {
                 TEST_INDEX_BANK),
             true));
   }
+
+  // Pagination explain tests
+
+  /**
+   * Test explain for head X from Y syntax. This verifies the explain plan shows the correct
+   * LIMIT/OFFSET structure.
+   */
+  @Test
+  public void testExplainHeadFromOffset() throws IOException {
+    enabledOnlyWhenPushdownIsEnabled();
+    String query =
+        String.format(
+            "source=%s | fields firstname, age | sort age | head 3 from 2", TEST_INDEX_BANK);
+    var result = explainQueryYaml(query);
+    String expected = loadExpectedPlan("explain_head_from_offset.yaml");
+    assertYamlEqualsIgnoreId(expected, result);
+  }
+
+  /**
+   * Test explain for head X from Y with filter. This verifies the explain plan shows the correct
+   * LIMIT/OFFSET structure with filter applied first.
+   */
+  @Test
+  public void testExplainHeadFromOffsetWithFilter() throws IOException {
+    enabledOnlyWhenPushdownIsEnabled();
+    String query =
+        String.format(
+            "source=%s | where age > 30 | fields firstname, age | sort age | head 2 from 3",
+            TEST_INDEX_BANK);
+    var result = explainQueryYaml(query);
+    String expected = loadExpectedPlan("explain_head_from_offset_with_filter.yaml");
+    assertYamlEqualsIgnoreId(expected, result);
+  }
+
+  /**
+   * Test explain for pagination API. This verifies that using pageSize and offset parameters
+   * produces a plan with LogicalSystemLimit(type=PAGINATION) applied on top.
+   */
+  @Test
+  public void testExplainPaginationApi() throws IOException {
+    enabledOnlyWhenPushdownIsEnabled();
+    String query = String.format("source=%s | fields firstname, age | sort age", TEST_INDEX_BANK);
+    var result = explainQueryYamlWithPagination(query, 3, 2);
+    String expected = loadExpectedPlan("explain_pagination_api.yaml");
+    assertYamlEqualsIgnoreId(expected, result);
+  }
+
+  /**
+   * Test using both pagination API and head X from Y together. The query uses head to limit results
+   * first, then pagination API applies additional LIMIT/OFFSET on top of that result.
+   *
+   * <p>Example: Query "head 5 from 1" returns rows 2-6 (5 rows starting from offset 1). Then
+   * pagination with pageSize=2, offset=1 returns rows 2-3 of that result (i.e., original rows 3-4).
+   */
+  @Test
+  public void testExplainPaginationApiWithHeadFrom() throws IOException {
+    enabledOnlyWhenPushdownIsEnabled();
+    // Query with head 5 from 1, then apply pagination pageSize=2, offset=1
+    String query =
+        String.format(
+            "source=%s | fields firstname, age | sort age | head 5 from 1", TEST_INDEX_BANK);
+    var result = explainQueryYamlWithPagination(query, 2, 1);
+
+    // The plan should show both:
+    // 1. LogicalSystemLimit with type=[PAGINATION] (from pagination API)
+    // 2. LogicalSort with offset and fetch (from head 5 from 1)
+    String expected = loadExpectedPlan("explain_pagination_api_with_head_from.yaml");
+    assertYamlEqualsIgnoreId(expected, result);
+  }
 }