Support fetch_size API for PPL

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/ast/statement/Query.java

@@ -7,7 +7,6 @@
 
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
-import lombok.RequiredArgsConstructor;
 import lombok.Setter;
 import lombok.ToString;
 import org.opensearch.sql.ast.AbstractNodeVisitor;
@@ -19,13 +18,18 @@
 @Setter
 @ToString
 @EqualsAndHashCode(callSuper = false)
-@RequiredArgsConstructor
 public class Query extends Statement {
 
   protected final UnresolvedPlan plan;
   protected final int fetchSize;
   private final QueryType queryType;
 
+  public Query(UnresolvedPlan plan, int fetchSize, QueryType queryType) {
+    this.plan = plan;
+    this.fetchSize = fetchSize;
+    this.queryType = queryType;
+  }
+
   @Override
   public <R, C> R accept(AbstractNodeVisitor<R, C> visitor, C context) {
     return visitor.visitQuery(this, context);

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

@@ -100,9 +100,9 @@ public void executeWithCalcite(
                 QueryProfiling.activate(QueryContext.isProfileEnabled());
             ProfileMetric analyzeMetric = profileContext.getOrCreateMetric(MetricName.ANALYZE);
             long analyzeStart = System.nanoTime();
+            SysLimit sysLimit = SysLimit.fromSettings(settings);
             CalcitePlanContext context =
-                CalcitePlanContext.create(
-                    buildFrameworkConfig(), SysLimit.fromSettings(settings), queryType);
+                CalcitePlanContext.create(buildFrameworkConfig(), sysLimit, queryType);
             RelNode relNode = analyze(plan, context);
             RelNode calcitePlan = convertToCalcitePlan(relNode, context);
             analyzeMetric.set(System.nanoTime() - analyzeStart);
@@ -236,16 +236,18 @@ public void executePlan(
           .getSplit()
           .ifPresentOrElse(
               split -> executionEngine.execute(plan(plan), new ExecutionContext(split), listener),
-              () ->
-                  executionEngine.execute(
-                      plan(plan),
-                      ExecutionContext.querySizeLimit(
-                          // For pagination, querySizeLimit shouldn't take effect.
-                          // See {@link PaginationWindowIT::testQuerySizeLimitDoesNotEffectPageSize}
-                          plan instanceof LogicalPaginate
-                              ? null
-                              : SysLimit.fromSettings(settings).querySizeLimit()),
-                      listener));
+              () -> {
+                Integer effectiveLimit;
+                if (plan instanceof LogicalPaginate) {
+                  // For pagination, querySizeLimit shouldn't take effect.
+                  // See {@link PaginationWindowIT::testQuerySizeLimitDoesNotEffectPageSize}
+                  effectiveLimit = null;
+                } else {
+                  effectiveLimit = SysLimit.fromSettings(settings).querySizeLimit();
+                }
+                executionEngine.execute(
+                    plan(plan), ExecutionContext.querySizeLimit(effectiveLimit), listener);
+              });
     } catch (Exception e) {
       listener.onFailure(e);
     }

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

@@ -29,7 +29,7 @@ public class QueryPlan extends AbstractPlan {
 
   protected final Optional<Integer> pageSize;
 
-  /** Constructor. */
+  /** Constructor without page size. */
   public QueryPlan(
       QueryId queryId,
       QueryType queryType,
@@ -43,7 +43,7 @@ public QueryPlan(
     this.pageSize = Optional.empty();
   }
 
-  /** Constructor with page size. */
+  /** Constructor with page size (for pagination). */
   public QueryPlan(
       QueryId queryId,
       QueryType queryType,

docs/user/interfaces/endpoint.rst

@@ -208,13 +208,13 @@ Explain::
       }
     }
 
-Cursor
-======
+Cursor (SQL)
+============
 
 Description
 -----------
 
-To get paginated response for a query, user needs to provide `fetch_size` parameter as part of normal query. The value of `fetch_size` should be greater than `0`. In absence of `fetch_size` or a value of `0`, it will fallback to non-paginated response. This feature is only available over `jdbc` format for now.
+To get paginated response for a SQL query, user needs to provide `fetch_size` parameter as part of normal query. The value of `fetch_size` should be greater than `0`. In absence of `fetch_size` or a value of `0`, it will fallback to non-paginated response. This feature is only available over `jdbc` format for now.
 
 Example
 -------
@@ -266,3 +266,60 @@ Result set::
       "size": 5,
       "status": 200
     }
+
+Fetch Size (PPL)
+================
+
+Description
+-----------
+
+PPL also supports the ``fetch_size`` parameter, but with different semantics from SQL. In PPL, ``fetch_size`` limits the number of rows returned in a single, complete response. **PPL does not support cursor-based pagination** — no cursor is returned and there is no way to fetch additional pages. The value of ``fetch_size`` should be between ``1`` and ``10000``. In absence of ``fetch_size`` or a value of ``0``, it will use the system default behavior (no limit).
+
++--------------------+-------------------------------------+------------------------------------+
+| Aspect             | SQL ``fetch_size``                  | PPL ``fetch_size``                 |
++====================+=====================================+====================================+
+| Purpose            | Cursor-based pagination             | Response size limiting             |
++--------------------+-------------------------------------+------------------------------------+
+| Returns cursor?    | Yes                                 | No                                 |
++--------------------+-------------------------------------+------------------------------------+
+| Can fetch more?    | Yes (with cursor)                   | No (single response)               |
++--------------------+-------------------------------------+------------------------------------+
+| Maximum value      | No hard limit                       | 10,000                             |
++--------------------+-------------------------------------+------------------------------------+
+
+Example
+-------
+
+PPL query::
+
+	>> curl -H 'Content-Type: application/json' -X POST localhost:9200/_plugins/_ppl -d '{
+	  "fetch_size" : 5,
+	  "query" : "source = accounts | fields firstname, lastname | where age > 20"
+	}'
+
+Result set::
+
+    {
+      "schema": [
+        {
+          "name": "firstname",
+          "type": "text"
+        },
+        {
+          "name": "lastname",
+          "type": "text"
+        }
+      ],
+      "total": 5,
+      "datarows": [
+        ["Cherry", "Carey"],
+        ["Lindsey", "Hawkins"],
+        ["Sargent", "Powers"],
+        ["Campos", "Olsen"],
+        ["Savannah", "Kirby"]
+      ],
+      "size": 5,
+      "status": 200
+    }
+
+Note that unlike the SQL response above, there is no ``cursor`` field in the PPL response. The response is complete and final.

docs/user/ppl/limitations/limitations.md

@@ -62,7 +62,7 @@ For the following functionalities, the query will be forwarded to the V2 query e
     * ML  
     * Kmeans  
 * `show datasources` and command  
-* Commands with `fetch_size` parameter  
+* SQL commands with `fetch_size` parameter (cursor-based pagination). Note: PPL's `fetch_size` (response size limiting, no cursor) is supported in Calcite Engine.
 
 
 ## Malformed Field Names in Object Fields

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

@@ -5,6 +5,7 @@
 
 package org.opensearch.sql.calcite.remote;
 
+import static org.opensearch.sql.legacy.TestUtils.getResponseBody;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_ACCOUNT;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_ALIAS;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BANK;
@@ -25,8 +26,12 @@
 
 import java.io.IOException;
 import java.util.Locale;
+import org.junit.Assert;
 import org.junit.Ignore;
 import org.junit.Test;
+import org.opensearch.client.Request;
+import org.opensearch.client.RequestOptions;
+import org.opensearch.client.Response;
 import org.opensearch.sql.ast.statement.ExplainMode;
 import org.opensearch.sql.common.setting.Settings;
 import org.opensearch.sql.common.setting.Settings.Key;
@@ -2497,4 +2502,68 @@ public void testExplainMvCombine() throws IOException {
     String expected = loadExpectedPlan("explain_mvcombine.yaml");
     assertYamlEqualsIgnoreId(expected, actual);
   }
+
+  // ==================== fetch_size explain tests ====================
+
+  @Test
+  public void testExplainFetchSizePushDown() throws IOException {
+    // fetch_size=5 injects Head(5, 0) on top of the plan
+    // Logical plan: LogicalSort(fetch=[5]) wraps the Project
+    String expected = loadExpectedPlan("explain_fetch_size_push.yaml");
+    assertYamlEqualsIgnoreId(
+        expected,
+        explainQueryWithFetchSizeYaml(
+            String.format("source=%s | fields age", TEST_INDEX_ACCOUNT), 5));
+  }
+
+  @Test
+  public void testExplainFetchSizeWithSmallerHead() throws IOException {
+    // fetch_size=10 with user's | head 3
+    // Two LogicalSort nodes: inner fetch=[3] from user head, outer fetch=[10] from fetch_size
+    // Effective limit = min(3, 10) = 3
+    String expected = loadExpectedPlan("explain_fetch_size_with_head_push.yaml");
+    assertYamlEqualsIgnoreId(
+        expected,
+        explainQueryWithFetchSizeYaml(
+            String.format("source=%s | head 3 | fields age", TEST_INDEX_ACCOUNT), 10));
+  }
+
+  @Test
+  public void testExplainFetchSizeSmallerThanHead() throws IOException {
+    // fetch_size=5 with user's | head 100
+    // Two LogicalSort nodes: inner fetch=[100] from user head, outer fetch=[5] from fetch_size
+    // Effective limit = min(100, 5) = 5
+    String expected = loadExpectedPlan("explain_fetch_size_smaller_than_head_push.yaml");
+    assertYamlEqualsIgnoreId(
+        expected,
+        explainQueryWithFetchSizeYaml(
+            String.format("source=%s | head 100 | fields age", TEST_INDEX_ACCOUNT), 5));
+  }
+
+  /**
+   * Send an explain request with fetch_size in the JSON body and return YAML output.
+   *
+   * @param query the PPL query string
+   * @param fetchSize the fetch_size parameter value
+   * @return the explain output as YAML string
+   */
+  private String explainQueryWithFetchSizeYaml(String query, int fetchSize) throws IOException {
+    Request request =
+        new Request(
+            "POST",
+            String.format(
+                "/_plugins/_ppl/_explain?format=%s&mode=%s", Format.YAML, ExplainMode.STANDARD));
+    String jsonBody =
+        String.format(
+            Locale.ROOT, "{\n  \"query\": \"%s\",\n  \"fetch_size\": %d\n}", query, fetchSize);
+    request.setJsonEntity(jsonBody);
+
+    RequestOptions.Builder restOptionsBuilder = RequestOptions.DEFAULT.toBuilder();
+    restOptionsBuilder.addHeader("Content-Type", "application/json");
+    request.setOptions(restOptionsBuilder);
+
+    Response response = client().performRequest(request);
+    Assert.assertEquals(200, response.getStatusLine().getStatusCode());
+    return getResponseBody(response, true);
+  }
 }