Support fetch_size API for PPL

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,58 @@ 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 greater than ``0``. In absence of ``fetch_size`` or a value of ``0``, it will use the system default behavior (no limit). The effective upper bound is governed by the ``plugins.query.size_limit`` cluster setting (defaults to ``index.max_result_window``, which is 10000).
+
++--------------------+-------------------------------------+------------------------------------+
+| 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)               |
++--------------------+-------------------------------------+------------------------------------+
+
+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);
+  }
 }

integ-test/src/test/java/org/opensearch/sql/ppl/FetchSizeIT.java

@@ -0,0 +1,198 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.ppl;
+
+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_BANK;
+import static org.opensearch.sql.plugin.rest.RestPPLQueryAction.QUERY_API_ENDPOINT;
+
+import java.io.IOException;
+import java.util.Locale;
+import org.json.JSONArray;
+import org.json.JSONObject;
+import org.junit.Assert;
+import org.junit.jupiter.api.Test;
+import org.opensearch.client.Request;
+import org.opensearch.client.RequestOptions;
+import org.opensearch.client.Response;
+
+/** Integration tests for PPL fetch_size parameter. */
+public class FetchSizeIT extends PPLIntegTestCase {
+
+  @Override
+  public void init() throws Exception {
+    super.init();
+    loadIndex(Index.ACCOUNT);
+    loadIndex(Index.BANK);
+  }
+
+  @Test
+  public void testFetchSizeLimitsResults() throws IOException {
+    // accounts index has 1000 documents, request only 5
+    JSONObject result = executeQueryWithFetchSize("source=" + TEST_INDEX_ACCOUNT, 5);
+    JSONArray dataRows = result.getJSONArray("datarows");
+    assertEquals(5, dataRows.length());
+  }
+
+  @Test
+  public void testFetchSizeWithFields() throws IOException {
+    JSONObject result =
+        executeQueryWithFetchSize(
+            String.format("source=%s | fields firstname, age", TEST_INDEX_ACCOUNT), 3);
+    JSONArray dataRows = result.getJSONArray("datarows");
+    assertEquals(3, dataRows.length());
+  }
+
+  @Test
+  public void testFetchSizeWithFilter() throws IOException {
+    // Filter + fetch_size
+    JSONObject result =
+        executeQueryWithFetchSize(
+            String.format("source=%s | where age > 30", TEST_INDEX_ACCOUNT), 10);
+    JSONArray dataRows = result.getJSONArray("datarows");
+    assertEquals(10, dataRows.length());
+  }
+
+  @Test
+  public void testFetchSizeWithSort() throws IOException {
+    // Sort + fetch_size - should get the first N results after sorting
+    JSONObject result =
+        executeQueryWithFetchSize(
+            String.format("source=%s | sort age | fields firstname, age", TEST_INDEX_ACCOUNT), 5);
+    JSONArray dataRows = result.getJSONArray("datarows");
+    assertEquals(5, dataRows.length());
+  }
+
+  @Test
+  public void testFetchSizeWithEval() throws IOException {
+    // Eval command + fetch_size - ensures fetch_size works with post-processing commands
+    JSONObject result =
+        executeQueryWithFetchSize(
+            String.format(
+                "source=%s | eval age_plus_10 = age + 10 | fields firstname, age, age_plus_10",
+                TEST_INDEX_ACCOUNT),
+            7);
+    JSONArray dataRows = result.getJSONArray("datarows");
+    assertEquals(7, dataRows.length());
+  }
+
+  @Test
+  public void testFetchSizeWithDedup() throws IOException {
+    // Dedup command + fetch_size - dedup may return fewer than fetch_size if not enough unique
+    // values
+    JSONObject result =
+        executeQueryWithFetchSize(
+            String.format("source=%s | dedup gender | fields gender", TEST_INDEX_ACCOUNT), 100);
+    JSONArray dataRows = result.getJSONArray("datarows");
+    // There are only 2 genders (M, F) in the dataset, so we should get at most 2
+    assertTrue(dataRows.length() <= 2);
+  }
+
+  @Test
+  public void testFetchSizeWithRename() throws IOException {
+    JSONObject result =
+        executeQueryWithFetchSize(
+            String.format(
+                "source=%s | rename firstname as first_name | fields first_name, age",
+                TEST_INDEX_ACCOUNT),
+            4);
+    JSONArray dataRows = result.getJSONArray("datarows");
+    assertEquals(4, dataRows.length());
+  }
+
+  @Test
+  public void testFetchSizeZeroReturnsAllResults() throws IOException {
+    // fetch_size=0 should be treated as "no limit" (use system default)
+    JSONObject result = executeQueryWithFetchSize(String.format("source=%s", TEST_INDEX_BANK), 0);
+    JSONArray dataRows = result.getJSONArray("datarows");
+    // Bank index has 7 documents
+    assertEquals(7, dataRows.length());
+  }
+
+  @Test
+  public void testFetchSizeLargerThanDataset() throws IOException {
+    // When fetch_size is larger than the dataset, return all available results
+    JSONObject result =
+        executeQueryWithFetchSize(String.format("source=%s", TEST_INDEX_BANK), 1000);
+    JSONArray dataRows = result.getJSONArray("datarows");
+    // Bank index has 7 documents, so we should get 7, not 1000
+    assertEquals(7, dataRows.length());
+  }
+
+  @Test
+  public void testFetchSizeOne() throws IOException {
+    JSONObject result =
+        executeQueryWithFetchSize(
+            String.format("source=%s | fields firstname", TEST_INDEX_ACCOUNT), 1);
+    JSONArray dataRows = result.getJSONArray("datarows");
+    assertEquals(1, dataRows.length());
+  }
+
+  @Test
+  public void testFetchSizeWithStats() throws IOException {
+    // Stats aggregation - fetch_size should still apply to aggregation results
+    JSONObject result =
+        executeQueryWithFetchSize(
+            String.format("source=%s | stats count() by gender", TEST_INDEX_ACCOUNT), 100);
+    JSONArray dataRows = result.getJSONArray("datarows");
+    // Stats by gender should return 2 rows (M and F)
+    assertEquals(2, dataRows.length());
+  }
+
+  @Test
+  public void testFetchSizeWithHead() throws IOException {
+    // Both head command and fetch_size - the smaller limit should win
+    // head 3 limits to 3, fetch_size 10 would allow 10, so we get 3
+    JSONObject result =
+        executeQueryWithFetchSize(
+            String.format("source=%s | head 3 | fields firstname", TEST_INDEX_ACCOUNT), 10);
+    JSONArray dataRows = result.getJSONArray("datarows");
+    assertEquals(3, dataRows.length());
+  }
+
+  @Test
+  public void testFetchSizeSmallerThanHead() throws IOException {
+    // fetch_size smaller than head - fetch_size should further limit
+    // head 100 would return 100, but fetch_size 5 limits to 5
+    JSONObject result =
+        executeQueryWithFetchSize(
+            String.format("source=%s | head 100 | fields firstname", TEST_INDEX_ACCOUNT), 5);
+    JSONArray dataRows = result.getJSONArray("datarows");
+    assertEquals(5, dataRows.length());
+  }
+
+  @Test
+  public void testWithoutFetchSizeReturnsDefaultBehavior() throws IOException {
+    // Without fetch_size, should return results up to system default
+    JSONObject result = executeQuery(String.format("source=%s", TEST_INDEX_BANK));
+    JSONArray dataRows = result.getJSONArray("datarows");
+    assertEquals(7, dataRows.length());
+  }
+
+  /**
+   * Execute a PPL query with fetch_size parameter.
+   *
+   * @param query the PPL query string
+   * @param fetchSize the maximum number of results to return
+   * @return the JSON response
+   */
+  protected JSONObject executeQueryWithFetchSize(String query, int fetchSize) throws IOException {
+    Request request = new Request("POST", QUERY_API_ENDPOINT);
+    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 jsonify(getResponseBody(response, true));
+  }
+}

integ-test/src/test/resources/expectedOutput/calcite/explain_fetch_size_push.yaml

@@ -0,0 +1,8 @@
+calcite:
+  logical: |
+    LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])
+      LogicalSort(fetch=[5])
+        LogicalProject(age=[$8])
+          CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])
+  physical: |
+    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[age], LIMIT->5, LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":5,"timeout":"1m","_source":{"includes":["age"],"excludes":[]}}, requestedTotalSize=5, pageSize=null, startFrom=0)])

integ-test/src/test/resources/expectedOutput/calcite/explain_fetch_size_smaller_than_head_push.yaml

@@ -0,0 +1,9 @@
+calcite:
+  logical: |
+    LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])
+      LogicalSort(fetch=[5])
+        LogicalProject(age=[$8])
+          LogicalSort(fetch=[100])
+            CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])
+  physical: |
+    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[age], LIMIT->100, LIMIT->5, LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":5,"timeout":"1m","_source":{"includes":["age"],"excludes":[]}}, requestedTotalSize=5, pageSize=null, startFrom=0)])