Fix IT

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

Author: LantaoJin

docs/user/ppl/cmd/mvcombine.md

@@ -27,6 +27,26 @@ mvcombine <field>
 
 ---
 
+## Configuration
+
+The mvcombine command leverages OpenSearch's top hits aggregation pushdown, which requires increasing the `index.max_inner_result_window` setting to 10000 or larger.
+
+Change the `index.max_inner_result_window` to `10000`:
+
+```bash ppl
+curl -sS -H 'Content-Type: application/json' \
+-X PUT localhost:9200/mvcombine_data/_settings \
+-d '{"index" : {"max_inner_result_window" : "10000"}}'
+```
+
+```text
+{
+  "acknowledged": true
+}
+```
+
+---
+
 ## Example 1: Basic mvcombine
 
 Given the following input rows:

doctest/markdown_parser.py

@@ -34,7 +34,7 @@ class MarkdownDocTestParser:
 
     # Regex to match Markdown code fences with optional attributes
     CODE_FENCE_PATTERN = re.compile(
-        r'^```(\w+)([^\n]*?)\s*\n'   # ```language [attributes] (no newlines in attributes)
+        r'^```([\w][\w ]*\w|\w+)([^\n]*?)\s*\n'  # ```language [attributes] (language may contain spaces, e.g. "bash ppl")
         r'(.*?)'                     # code content (non-greedy)
         r'^```\s*$',                 # closing ```
         re.MULTILINE | re.DOTALL

doctest/test_docs.py

@@ -472,9 +472,9 @@ def create_markdown_suite(filepaths, category_name, setup_func):
         transform=transform,
     )
 
-    # Prepare globs for bash commands
+    # Prepare globs for bash commands (needed when input_langs includes "bash ppl")
     test_globs = {}
-    if "bash" in category_name:
+    if "bash" in category_name or "bash ppl" in input_langs:
         test_globs = {
             "sh": partial(
                 subprocess.run,

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

@@ -17,6 +17,7 @@
 import java.util.List;
 import org.json.JSONArray;
 import org.json.JSONObject;
+import org.junit.After;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.opensearch.client.ResponseException;
@@ -31,6 +32,12 @@ public void init() throws Exception {
     super.init();
     enableCalcite();
     loadIndex(Index.MVCOMBINE);
+    updateIndexSettings(INDEX, "{ \"index\": { \"max_inner_result_window\":" + 10000 + " } }");
+  }
+
+  @After
+  public void afterTest() throws IOException {
+    updateIndexSettings(INDEX, "{ \"index\": { \"max_inner_result_window\":" + 100 + " } }");
   }
 
   // ---------------------------

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

@@ -246,12 +246,17 @@ private void verifyQuery(JSONObject result) throws IOException {
   public void testMvCombineUnsupportedInV2() throws IOException {
     JSONObject result;
     try {
+      updateIndexSettings(
+          TEST_INDEX_BANK, "{ \"index\": { \"max_inner_result_window\":" + 10000 + " } }");
       result =
           executeQuery(
               String.format(
                   "source=%s | fields state, city, age | mvcombine age", TEST_INDEX_BANK));
     } catch (ResponseException e) {
       result = new JSONObject(TestUtils.getResponseBody(e.getResponse()));
+    } finally {
+      updateIndexSettings(
+          TEST_INDEX_BANK, "{ \"index\": { \"max_inner_result_window\":" + 100 + " } }");
     }
     verifyQuery(result);
   }

integ-test/src/test/java/org/opensearch/sql/security/CalciteCrossClusterSearchIT.java

@@ -389,20 +389,26 @@ public void testCrossClusterAppend() throws IOException {
   /** CrossClusterSearchIT Test for mvcombine. */
   @Test
   public void testCrossClusterMvcombine() throws IOException {
-
-    JSONObject result =
-        executeQuery(
-            String.format(
-                "search source=%s | where firstname='Hattie' or firstname='Nanette' "
-                    + "| fields firstname, age | mvcombine age",
-                TEST_INDEX_BANK_REMOTE));
-
-    verifyColumn(result, columnName("firstname"), columnName("age"));
-
-    verifyDataRows(
-        result,
-        rows("Hattie", new org.json.JSONArray().put(36)),
-        rows("Nanette", new org.json.JSONArray().put(28)));
+    try {
+      updateIndexSettings(
+          TEST_INDEX_BANK_REMOTE, "{ \"index\": { \"max_inner_result_window\":" + 10000 + " } }");
+      JSONObject result =
+          executeQuery(
+              String.format(
+                  "search source=%s | where firstname='Hattie' or firstname='Nanette' "
+                      + "| fields firstname, age | mvcombine age",
+                  TEST_INDEX_BANK_REMOTE));
+
+      verifyColumn(result, columnName("firstname"), columnName("age"));
+
+      verifyDataRows(
+          result,
+          rows("Hattie", new org.json.JSONArray().put(36)),
+          rows("Nanette", new org.json.JSONArray().put(28)));
+    } finally {
+      updateIndexSettings(
+          TEST_INDEX_BANK_REMOTE, "{ \"index\": { \"max_inner_result_window\":" + 100 + " } }");
+    }
   }
 
   /** CrossClusterSearchIT Test for fieldformat. */

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

@@ -129,7 +129,7 @@ public boolean updateIndexSettings(Map<String, Object> settings, String... index
       return response.isAcknowledged();
     } catch (IOException e) {
       throw new IllegalStateException(
-          String.format("Failed to update index settings %s for %s", settings, indexExpression, e));
+          String.format("Failed to update index settings %s for %s", settings, indexExpression), e);
     }
   }
 

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

@@ -625,38 +625,6 @@ yield switch (functionName) {
     };
   }
 
-  private static TopHitsAggregationBuilder getTopHitsAggregationBuilder(
-      AggregateCall aggCall,
-      List<Pair<RexNode, String>> args,
-      String aggName,
-      AggregateBuilderHelper helper,
-      Integer topHitsSize) {
-    // Disable fetchSource since TopHitsParser only parses fetchField currently.
-    TopHitsAggregationBuilder topHitsAggregationBuilder =
-        AggregationBuilders.topHits(aggName).from(0).size(topHitsSize);
-    List<String> sources = new ArrayList<>();
-    List<SearchSourceBuilder.ScriptField> scripts = new ArrayList<>();
-    args.forEach(
-        rex -> {
-          if (rex.getKey() instanceof RexInputRef) {
-            sources.add(helper.inferNamedField(rex.getKey()).getReference());
-          } else if (rex.getKey() instanceof RexCall || rex.getKey() instanceof RexLiteral) {
-            scripts.add(
-                new SearchSourceBuilder.ScriptField(
-                    rex.getValue(), helper.inferScript(rex.getKey()).getScript(), false));
-          } else {
-            throw new AggregateAnalyzerException(
-                String.format(
-                    "Unsupported push-down aggregator %s due to rex type is %s",
-                    aggCall.getAggregation(), rex.getKey().getKind()));
-          }
-        });
-    topHitsAggregationBuilder.fetchSource(
-        sources.stream().distinct().toArray(String[]::new), new String[0]);
-    topHitsAggregationBuilder.scriptFields(scripts);
-    return topHitsAggregationBuilder;
-  }
-
   private static boolean supportsMaxMinAggregation(ExprType fieldType) {
     ExprType coreType =
         (fieldType instanceof OpenSearchDataType)