[ES|QL] Enable CCS tests for subqueries (elastic#137776)

* enable CCS tests for subqueries

Author: fang-xing-esql

docs/changelog/137776.yaml

@@ -0,0 +1,5 @@
+pr: 137776
+summary: Enable CCS tests for subqueries
+area: ES|QL
+type: enhancement
+issues: []

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/MultiClusterSpecIT.java

@@ -168,12 +168,6 @@ protected void shouldSkipTest(String testName) throws IOException {
         assumeFalse("LOOKUP JOIN after SORT not yet supported in CCS", testName.contains("OnTheCoordinator"));
 
         assumeFalse("FORK not yet supported with CCS", testCase.requiredCapabilities.contains(FORK_V9.capabilityName()));
-
-        // And convertToRemoteIndices does not generate correct queries with subqueries in the FROM command yet
-        assumeFalse(
-            "Subqueries in FROM command not yet supported in CCS",
-            testCase.requiredCapabilities.contains(SUBQUERY_IN_FROM_COMMAND.capabilityName())
-        );
     }
 
     private TestFeatureService remoteFeaturesService() throws IOException {
@@ -298,6 +292,9 @@ static CsvSpecReader.CsvTestCase convertToRemoteIndices(CsvSpecReader.CsvTestCas
         if (dataLocation == null) {
             dataLocation = randomFrom(DataLocation.values());
         }
+        if (testCase.requiredCapabilities.contains(SUBQUERY_IN_FROM_COMMAND.capabilityName())) {
+            return convertSubqueryToRemoteIndices(testCase);
+        }
         String query = testCase.query;
         // If true, we're using *:index, otherwise we're using *:index,index
         boolean onlyRemotes = canUseRemoteIndicesOnly() && randomBoolean();
@@ -388,4 +385,13 @@ protected boolean supportsTDigestField() {
             throw new RuntimeException(e);
         }
     }
+
+    /**
+     * Convert index patterns and subqueries in FROM commands to use remote indices for a given test case.
+     */
+    private static CsvSpecReader.CsvTestCase convertSubqueryToRemoteIndices(CsvSpecReader.CsvTestCase testCase) {
+        String query = testCase.query;
+        testCase.query = EsqlTestUtils.convertSubqueryToRemoteIndices(query);
+        return testCase;
+    }
 }

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java

@@ -60,6 +60,8 @@
 import org.elasticsearch.index.mapper.blockloader.BlockLoaderFunctionConfig;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.logging.LogManager;
+import org.elasticsearch.logging.Logger;
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.search.aggregations.bucket.geogrid.GeoTileUtils;
 import org.elasticsearch.search.crossproject.CrossProjectModeDecider;
@@ -225,6 +227,8 @@ public final class EsqlTestUtils {
     public static final Literal FIVE = new Literal(Source.EMPTY, 5, DataType.INTEGER);
     public static final Literal SIX = new Literal(Source.EMPTY, 6, DataType.INTEGER);
 
+    private static final Logger LOGGER = LogManager.getLogger(EsqlTestUtils.class);
+
     public static Equals equalsOf(Expression left, Expression right) {
         return new Equals(EMPTY, left, right, null);
     }
@@ -1375,4 +1379,100 @@ private static String unquote(String index, int numOfQuotes) {
         return index.substring(numOfQuotes, index.length() - numOfQuotes);
     }
 
+    /**
+     * Convert index patterns and subqueries in FROM commands to use remote indices.
+     */
+    public static String convertSubqueryToRemoteIndices(String testQuery) {
+        String query = testQuery;
+        // find the main from command, ignoring pipes inside subqueries
+        List<String> mainFromCommandAndTheRest = splitIgnoringParentheses(query, "|");
+        String mainFrom = mainFromCommandAndTheRest.get(0).strip();
+        List<String> theRest = mainFromCommandAndTheRest.size() > 1
+            ? mainFromCommandAndTheRest.subList(1, mainFromCommandAndTheRest.size())
+            : List.of();
+        // check for metadata in the main from command
+        List<String> mainFromCommandWithMetadata = splitIgnoringParentheses(mainFrom, "metadata");
+        mainFrom = mainFromCommandWithMetadata.get(0).strip();
+        // if there is metadata, we need to add it back later
+        String metadata = mainFromCommandWithMetadata.size() > 1 ? " metadata " + mainFromCommandWithMetadata.get(1) : "";
+        // the main from command could be a comma separated list of index patterns, and subqueries
+        List<String> indexPatternsAndSubqueries = splitIgnoringParentheses(mainFrom, ",");
+        List<String> transformed = new ArrayList<>();
+        for (String indexPatternOrSubquery : indexPatternsAndSubqueries) {
+            // remove the from keyword if it's there
+            indexPatternOrSubquery = indexPatternOrSubquery.strip();
+            if (indexPatternOrSubquery.toLowerCase(Locale.ROOT).startsWith("from ")) {
+                indexPatternOrSubquery = indexPatternOrSubquery.strip().substring(5);
+            }
+            // substitute the index patterns or subquery with remote index patterns
+            if (isSubquery(indexPatternOrSubquery)) {
+                // it's a subquery, we need to process it recursively
+                String subquery = indexPatternOrSubquery.strip().substring(1, indexPatternOrSubquery.length() - 1);
+                String transformedSubquery = convertSubqueryToRemoteIndices(subquery);
+                transformed.add("(" + transformedSubquery + ")");
+            } else {
+                // It's an index pattern, we need to convert it to remote index pattern.
+                String remoteIndex = unquoteAndRequoteAsRemote(indexPatternOrSubquery, false);
+                transformed.add(remoteIndex);
+            }
+        }
+        // rebuild from command from transformed index patterns and subqueries
+        String transformedFrom = "FROM " + String.join(", ", transformed) + metadata;
+        // rebuild the whole query
+        mainFromCommandAndTheRest.set(0, transformedFrom);
+        testQuery = String.join(" | ", mainFromCommandAndTheRest);
+
+        LOGGER.trace("Transform query: \nFROM: {}\nTO:   {}", query, testQuery);
+        return testQuery;
+    }
+
+    /**
+     * Checks if the given string is a subquery (enclosed in parentheses).
+     */
+    private static boolean isSubquery(String indexPatternOrSubquery) {
+        String trimmed = indexPatternOrSubquery.strip();
+        return trimmed.startsWith("(") && trimmed.endsWith(")");
+    }
+
+    /**
+     * Splits the input string by the given delimiter, ignoring delimiters inside parentheses.
+     */
+    public static List<String> splitIgnoringParentheses(String input, String delimiter) {
+        List<String> results = new ArrayList<>();
+        if (input == null || input.isEmpty()) return results;
+
+        int depth = 0; // parentheses nesting
+        int lastSplit = 0;
+        int delimiterLength = delimiter.length();
+
+        for (int i = 0; i <= input.length() - delimiterLength; i++) {
+            char c = input.charAt(i);
+
+            if (c == '(') {
+                depth++;
+            } else if (c == ')') {
+                if (depth > 0) depth--;
+            }
+
+            // check delimiter only outside parentheses
+            if (depth == 0) {
+                boolean match;
+                if (delimiter.length() == 1) {
+                    match = c == delimiter.charAt(0);
+                } else {
+                    match = input.regionMatches(true, i, delimiter, 0, delimiterLength);
+                }
+
+                if (match) {
+                    results.add(input.substring(lastSplit, i).trim());
+                    lastSplit = i + delimiterLength;
+                    i += delimiterLength - 1; // skip the delimiter
+                }
+            }
+        }
+        // add remaining part
+        results.add(input.substring(lastSplit).trim());
+
+        return results;
+    }
 }