diff --git a/x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/MultiClusterSpecIT.java b/x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/MultiClusterSpecIT.java index 6e48b13b28d65..be106e23342ea 100644 --- a/x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/MultiClusterSpecIT.java +++ b/x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/MultiClusterSpecIT.java @@ -18,6 +18,8 @@ import org.elasticsearch.client.RestClient; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.core.IOUtils; +import org.elasticsearch.logging.LogManager; +import org.elasticsearch.logging.Logger; import org.elasticsearch.test.TestClustersThreadFilter; import org.elasticsearch.test.cluster.ElasticsearchCluster; import org.elasticsearch.test.rest.TestFeatureService; @@ -36,6 +38,7 @@ import java.io.ByteArrayInputStream; import java.io.IOException; import java.net.URL; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Locale; @@ -81,6 +84,8 @@ public class MultiClusterSpecIT extends EsqlSpecTestCase { private static RestClient remoteClusterClient; private static DataLocation dataLocation = null; + private static final Logger LOGGER = LogManager.getLogger(MultiClusterSpecIT.class); + @ParametersFactory(argumentFormatting = "csv-spec:%2$s.%3$s") public static List readScriptSpec() throws Exception { List urls = classpathResources("/*.csv-spec"); @@ -96,7 +101,16 @@ public MultiClusterSpecIT( CsvTestCase testCase, String instructions ) { - super(fileName, groupName, testName, lineNumber, convertToRemoteIndices(testCase), instructions); + super( + fileName, + groupName, + testName, + lineNumber, + testCase.requiredCapabilities.contains(SUBQUERY_IN_FROM_COMMAND.capabilityName()) + ? convertSubqueryToRemoteIndices(testCase) + : convertToRemoteIndices(testCase), + instructions + ); } // TODO: think how to handle this better @@ -168,12 +182,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 { @@ -399,6 +407,120 @@ 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 for a given test case. + */ + private static CsvSpecReader.CsvTestCase convertSubqueryToRemoteIndices(CsvSpecReader.CsvTestCase testCase) { + if (dataLocation == null) { + dataLocation = randomFrom(DataLocation.values()); + } + String query = testCase.query; + testCase.query = convertSubqueryToRemoteIndices(query); + return testCase; + } + + /** + * Convert index patterns and subqueries in FROM commands to use remote indices. + */ + private static String convertSubqueryToRemoteIndices(String testQuery) { + String query = testQuery; + // find the main from command, ignoring pipes inside subqueries + List mainFromCommandAndTheRest = splitIgnoringParentheses(query, "|"); + String mainFrom = mainFromCommandAndTheRest.get(0).strip(); + List theRest = mainFromCommandAndTheRest.size() > 1 + ? mainFromCommandAndTheRest.subList(1, mainFromCommandAndTheRest.size()) + : List.of(); + // check for metadata in the main from command + List 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 indexPatternsAndSubqueries = splitIgnoringParentheses(mainFrom, ","); + List 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.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. + // indexPatternOrSubquery could be a comma separated list of indices, we need to process each index separately + List indexPatterns = splitIgnoringParentheses(indexPatternOrSubquery, ","); + String remoteIndex = indexPatterns.stream() + .map(String::strip) + // Data is loaded on one of the clusters not both, remote only may not guarantee data is found on the remote cluster. + .map(index -> unquoteAndRequoteAsRemote(index, false)) + .collect(Collectors.joining(",")); + transformed.add(remoteIndex); + } + } + // rebuild from command from transformed index patterns and subqueries + String transformedFrom = "from " + String.join(", ", transformed) + metadata; + // rebuild the whole query + testQuery = transformedFrom + (theRest.isEmpty() ? "" : " | " + String.join(" | ", theRest)); + + 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 splitIgnoringParentheses(String input, String delimiter) { + List 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; + } + @Override protected boolean enableRoundingDoubleValuesOnAsserting() { return true; diff --git a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/subquery.csv-spec b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/subquery.csv-spec index 930ca874488b6..8ad31d82b7220 100644 --- a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/subquery.csv-spec +++ b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/subquery.csv-spec @@ -3,7 +3,6 @@ // subqueryInFrom -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees, (FROM sample_data) @@ -26,7 +25,6 @@ null | null | 172.21.3.15 ; subqueryInFromWithoutMainIndexPattern -required_capability: fork_v9 required_capability: subquery_in_from_command FROM (FROM employees) @@ -42,7 +40,6 @@ emp_no:integer | languages:integer ; subqueryInFromWithoutMainIndexPatternWithMultipleSubqueries -required_capability: fork_v9 required_capability: subquery_in_from_command FROM (FROM employees), (FROM sample_data) @@ -65,7 +62,6 @@ null | null | 172.21.3.15 ; subqueryInFromWithIdenticalIndexPatternsInMainAndSubquery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees, (FROM employees) @@ -84,11 +80,11 @@ emp_no:integer | languages:integer ; subqueryInFromWithEvalInSubquery -required_capability: fork_v9 required_capability: subquery_in_from_command -FROM employees, (FROM sample_data | EVAL x = client_ip::keyword ) metadata _index +FROM employees, (FROM sample_data metadata _index | EVAL x = client_ip::keyword ) metadata _index | WHERE ( emp_no >= 10091 AND emp_no < 10094) OR emp_no IS NULL +| EVAL _index = MV_LAST(SPLIT(_index, ":")) | SORT emp_no, client_ip | KEEP _index, emp_no, languages, client_ip, x ; @@ -97,23 +93,23 @@ _index:keyword | emp_no:integer | languages:integer | client_ip:ip | x:keyword employees | 10091 | 3 | null | null employees | 10092 | 1 | null | null employees | 10093 | 3 | null | null -null | null | null | 172.21.0.5 | 172.21.0.5 -null | null | null | 172.21.2.113 | 172.21.2.113 -null | null | null | 172.21.2.162 | 172.21.2.162 -null | null | null | 172.21.3.15 | 172.21.3.15 -null | null | null | 172.21.3.15 | 172.21.3.15 -null | null | null | 172.21.3.15 | 172.21.3.15 -null | null | null | 172.21.3.15 | 172.21.3.15 +sample_data | null | null | 172.21.0.5 | 172.21.0.5 +sample_data | null | null | 172.21.2.113 | 172.21.2.113 +sample_data | null | null | 172.21.2.162 | 172.21.2.162 +sample_data | null | null | 172.21.3.15 | 172.21.3.15 +sample_data | null | null | 172.21.3.15 | 172.21.3.15 +sample_data | null | null | 172.21.3.15 | 172.21.3.15 +sample_data | null | null | 172.21.3.15 | 172.21.3.15 ; subqueryInFromWithWhereInSubquery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees, (FROM sample_data metadata _index | WHERE client_ip == "172.21.3.15" ) metadata _index | WHERE ( emp_no >= 10091 AND emp_no < 10094) OR emp_no IS NULL +| EVAL _index = MV_LAST(SPLIT(_index, ":")) | SORT emp_no | KEEP _index, emp_no, languages, client_ip ; @@ -129,133 +125,121 @@ sample_data | null | null | 172.21.3.15 ; subqueryInFromWithStatsInSubquery -required_capability: fork_v9 required_capability: subquery_in_from_command -FROM employees, (FROM sample_data metadata _index - | STATS cnt = count(*) by client_ip ) - metadata _index +FROM employees, (FROM sample_data metadata _index + | STATS cnt = count(*) by _index, client_ip ) + metadata _index | WHERE ( emp_no >= 10091 AND emp_no < 10094) OR emp_no IS NULL -| SORT emp_no, client_ip -| KEEP _index, emp_no, languages, cnt, client_ip +| EVAL _index = MV_LAST(SPLIT(_index, ":")) +| SORT _index, emp_no, client_ip +| KEEP _index, emp_no, languages, cnt, client_ip ; _index:keyword | emp_no:integer | languages:integer | cnt:long | client_ip:ip -employees | 10091 | 3 | null | null -employees | 10092 | 1 | null | null -employees | 10093 | 3 | null | null -null | null | null | 1 | 172.21.0.5 -null | null | null | 1 | 172.21.2.113 -null | null | null | 1 | 172.21.2.162 -null | null | null | 4 | 172.21.3.15 +employees | 10091 | 3 | null | null +employees | 10092 | 1 | null | null +employees | 10093 | 3 | null | null +sample_data | null | null | 1 | 172.21.0.5 +sample_data | null | null | 1 | 172.21.2.113 +sample_data | null | null | 1 | 172.21.2.162 +sample_data | null | null | 4 | 172.21.3.15 ; subqueryInFromWithStatsInSubqueryConjunctiveFilterInMainQuery -required_capability: fork_v9 required_capability: subquery_in_from_command -FROM employees, (FROM sample_data - | STATS cnt = count(*) by client_ip ) - , (FROM sample_data_str - | STATS cnt = count(*) by client_ip ) - metadata _index -| EVAL client_ip = client_ip::ip +FROM employees, (FROM sample_data metadata _index + | STATS cnt = count(*) by _index, client_ip ) + , (FROM sample_data_str metadata _index + | STATS cnt = count(*) by _index, client_ip ) + metadata _index +| EVAL client_ip = client_ip::ip, _index = MV_LAST(SPLIT(_index, ":")) | WHERE client_ip == "172.21.3.15" AND cnt >0 -| SORT emp_no, client_ip -| KEEP _index, emp_no, languages, cnt, client_ip +| SORT _index, emp_no, client_ip +| KEEP _index, emp_no, languages, cnt, client_ip ; _index:keyword | emp_no:integer | languages:integer | cnt:long | client_ip:ip -null | null | null | 4 | 172.21.3.15 -null | null | null | 4 | 172.21.3.15 +sample_data |null | null | 4 | 172.21.3.15 +sample_data_str |null | null | 4 | 172.21.3.15 ; subqueryInFromWithStatsInSubqueryDisjunctiveFilterInMainQuery -required_capability: fork_v9 required_capability: subquery_in_from_command -FROM employees, (FROM sample_data - | STATS cnt = count(*) by client_ip ) - , (FROM sample_data_str - | STATS cnt = count(*) by client_ip ) +FROM employees, (FROM sample_data metadata _index + | STATS cnt = count(*) by _index, client_ip ) + , (FROM sample_data_str metadata _index + | STATS cnt = count(*) by _index, client_ip ) metadata _index -| EVAL client_ip = client_ip::ip +| EVAL client_ip = client_ip::ip, _index = MV_LAST(SPLIT(_index, ":")) | WHERE ( emp_no >= 10091 AND emp_no < 10094) OR client_ip == "172.21.3.15" -| SORT emp_no, client_ip -| KEEP _index, emp_no, languages, cnt, client_ip +| SORT _index, emp_no, client_ip +| KEEP _index, emp_no, languages, cnt, client_ip ; _index:keyword | emp_no:integer | languages:integer | cnt:long | client_ip:ip -employees | 10091 | 3 | null | null -employees | 10092 | 1 | null | null -employees | 10093 | 3 | null | null -null | null | null | 4 | 172.21.3.15 -null | null | null | 4 | 172.21.3.15 +employees | 10091 | 3 | null | null +employees |10092 | 1 | null | null +employees |10093 | 3 | null | null +sample_data | null | null | 4 | 172.21.3.15 +sample_data_str | null | null | 4 | 172.21.3.15 ; subqueryInFromWithStatsInMainQuery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM sample_data, sample_data_str, - (FROM sample_data_ts_nanos metadata _index + (FROM sample_data_ts_nanos | WHERE client_ip == "172.21.3.15") , - (FROM sample_data_ts_long metadata _index + (FROM sample_data_ts_long | EVAL @timestamp = @timestamp::date_nanos | WHERE client_ip == "172.21.0.5") - metadata _index | EVAL client_ip = client_ip::ip -| STATS cnt = count(*) BY _index, client_ip -| SORT _index, client_ip +| STATS cnt = count(*) BY client_ip +| SORT client_ip ; -cnt:long | _index:keyword | client_ip:ip -1 | sample_data | 172.21.0.5 -1 | sample_data | 172.21.2.113 -1 | sample_data | 172.21.2.162 -4 | sample_data | 172.21.3.15 -1 | sample_data_str | 172.21.0.5 -1 | sample_data_str | 172.21.2.113 -1 | sample_data_str | 172.21.2.162 -4 | sample_data_str | 172.21.3.15 -1 | sample_data_ts_long | 172.21.0.5 -4 | sample_data_ts_nanos | 172.21.3.15 +cnt:long | client_ip:ip +3 | 172.21.0.5 +2 | 172.21.2.113 +2 | 172.21.2.162 +12 | 172.21.3.15 ; subqueryInFromWithLookupJoinInSubquery -required_capability: fork_v9 required_capability: subquery_in_from_command -FROM employees, (FROM sample_data metadata _index +FROM employees, (FROM sample_data | EVAL client_ip = client_ip::keyword | LOOKUP JOIN clientips_lookup ON client_ip ) - metadata _index | WHERE ( emp_no >= 10091 AND emp_no < 10094) OR emp_no IS NULL | SORT emp_no, client_ip -| KEEP _index, emp_no, languages, client_ip, env +| KEEP emp_no, languages, client_ip, env ; -_index:keyword | emp_no:integer | languages:integer | client_ip:keyword | env:keyword -employees | 10091 | 3 | null | null -employees | 10092 | 1 | null | null -employees | 10093 | 3 | null | null -sample_data | null | null | 172.21.0.5 | Development -sample_data | null | null | 172.21.2.113 | QA -sample_data | null | null | 172.21.2.162 | QA -sample_data | null | null | 172.21.3.15 | Production -sample_data | null | null | 172.21.3.15 | Production -sample_data | null | null | 172.21.3.15 | Production -sample_data | null | null | 172.21.3.15 | Production +emp_no:integer | languages:integer | client_ip:keyword | env:keyword +10091 | 3 | null | null +10092 | 1 | null | null +10093 | 3 | null | null +null | null | 172.21.0.5 | Development +null | null | 172.21.2.113 | QA +null | null | 172.21.2.162 | QA +null | null | 172.21.3.15 | Production +null | null | 172.21.3.15 | Production +null | null | 172.21.3.15 | Production +null | null | 172.21.3.15 | Production ; subqueryInFromWithLookupJoinInMainQuery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees, (FROM sample_data metadata _index | EVAL client_ip = client_ip::keyword ) metadata _index | WHERE ( emp_no >= 10091 AND emp_no < 10094) OR emp_no IS NULL +| EVAL _index = MV_LAST(SPLIT(_index, ":")) | LOOKUP JOIN clientips_lookup ON client_ip | SORT emp_no, client_ip | KEEP _index, emp_no, languages, client_ip, env @@ -275,79 +259,71 @@ sample_data | null | null | 172.21.3.15 | Produc ; subqueryInFromWithEnrichInSubquery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees, (FROM employees_incompatible | ENRICH languages_policy on languages with language_name ) - metadata _index | EVAL emp_no = emp_no::long | WHERE emp_no >= 10091 AND emp_no < 10094 -| SORT _index, emp_no -| KEEP _index, emp_no, languages, language_name +| SORT emp_no, language_name +| KEEP emp_no, languages, language_name ; -_index:keyword | emp_no:long | languages:integer | language_name:keyword -employees | 10091 | 3 | null -employees | 10092 | 1 | null -employees | 10093 | 3 | null -null | 10091 | 3 | Spanish -null | 10092 | 1 | English -null | 10093 | 3 | Spanish +emp_no:long | languages:integer | language_name:keyword +10091 | 3 | Spanish +10091 | 3 | null +10092 | 1 | English +10092 | 1 | null +10093 | 3 | Spanish +10093 | 3 | null ; subqueryInFromWithEnrichInMainQuery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees, (FROM employees_incompatible | KEEP emp_no, languages) - metadata _index | EVAL emp_no = emp_no::long | WHERE emp_no >= 10091 AND emp_no < 10094 | ENRICH languages_policy on languages with language_name -| SORT _index, emp_no -| KEEP _index, emp_no, languages, language_name +| SORT emp_no +| KEEP emp_no, languages, language_name ; -_index:keyword | emp_no:long | languages:integer | language_name:keyword -employees | 10091 | 3 | Spanish -employees | 10092 | 1 | English -employees | 10093 | 3 | Spanish -null | 10091 | 3 | Spanish -null | 10092 | 1 | English -null | 10093 | 3 | Spanish +emp_no:long | languages:integer | language_name:keyword +10091 | 3 | Spanish +10091 | 3 | Spanish +10092 | 1 | English +10092 | 1 | English +10093 | 3 | Spanish +10093 | 3 | Spanish ; subqueryInFromWithSortInSubquery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees, (FROM sample_data | STATS cnt = count(*) by client_ip | SORT cnt DESC | LIMIT 1 ) - metadata _index | WHERE ( emp_no >= 10091 AND emp_no < 10094) OR emp_no IS NULL | SORT emp_no, client_ip -| KEEP _index, emp_no, languages, cnt, client_ip +| KEEP emp_no, languages, cnt, client_ip ; -_index:keyword | emp_no:integer | languages:integer | cnt:long | client_ip:ip -employees | 10091 | 3 | null | null -employees | 10092 | 1 | null | null -employees | 10093 | 3 | null | null -null | null | null | 4 | 172.21.3.15 +emp_no:integer | languages:integer | cnt:long | client_ip:ip +10091 | 3 | null | null +10092 | 1 | null | null +10093 | 3 | null | null +null | null | 4 | 172.21.3.15 ; subqueryInFromWithGrokInSubquery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees_incompatible, (FROM employees | GROK concat(first_name, " ", last_name) "%{WORD:a} %{WORD:b}" | KEEP emp_no, a, b ) - metadata _index | EVAL emp_no = emp_no::long | WHERE emp_no >= 10091 AND emp_no < 10094 | SORT emp_no, a, b @@ -364,13 +340,11 @@ emp_no:long | a:keyword | b:keyword ; subqueryInFromWithGrokInMainQuery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees_incompatible, (FROM employees | WHERE emp_no > 10091 AND emp_no < 10094 | KEEP emp_no, first_name, last_name ) - metadata _index | EVAL emp_no = emp_no::long, first_name = first_name::keyword, last_name = last_name::keyword | WHERE emp_no >= 10091 AND emp_no < 10094 | GROK concat(first_name, " ", last_name) "%{WORD:a} %{WORD:b}" @@ -387,7 +361,6 @@ emp_no:long | a:keyword | b:keyword ; subqueryInFromWithDissectInSubquery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees, (FROM employees_incompatible @@ -410,13 +383,11 @@ emp_no:long | a:keyword | b:keyword ; subqueryInFromWithDissectInMainQuery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees, (FROM employees_incompatible | WHERE emp_no > 10091 AND emp_no < 10094 | KEEP emp_no, first_name, last_name ) - metadata _index | EVAL emp_no = emp_no::long, first_name = first_name::keyword, last_name = last_name::keyword | WHERE emp_no >= 10091 AND emp_no < 10094 | EVAL name = concat(first_name, "1 ", last_name) @@ -434,7 +405,6 @@ emp_no:long | a:keyword | b:keyword ; subqueryInFromWithMvExpandInSubquery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees_incompatible, (FROM employees @@ -462,7 +432,6 @@ emp_no:long | first_name:keyword | last_name:keyword | job_positions:keyword ; subqueryInFromWithMvExpandInMainQuery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees_incompatible, (FROM employees @@ -495,7 +464,6 @@ emp_no:long | first_name:keyword | last_name:keyword | job_positions:keyword ; subqueryInFromWithInlineStatsInSubquery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees, (FROM employees_incompatible @@ -517,17 +485,16 @@ emp_no:long | first_name:keyword | last_name:keyword | cnt:long | gender:keyword ; subqueryInFromWithInlineStatsInMainQuery-Ignore -required_capability: fork_v9 required_capability: subquery_in_from_command FROM sample_data, sample_data_str, - (FROM sample_data_ts_nanos + (FROM sample_data_ts_nanos metadata _index | WHERE client_ip == "172.21.3.15") , - (FROM sample_data_ts_long + (FROM sample_data_ts_long metadata _index | EVAL @timestamp = @timestamp::date_nanos | WHERE client_ip == "172.21.0.5") metadata _index -| EVAL client_ip = client_ip::ip +| EVAL client_ip = client_ip::ip, _index = MV_LAST(SPLIT(_index, ":")) | INLINE STATS cnt = count(*) BY _index, client_ip | SORT _index, @timestamp ; @@ -547,15 +514,14 @@ FROM sample_data, sample_data_str, 2023-10-23T13:52:55.015Z | 8268153 | Connection error | 4 | sample_data_str | 172.21.3.15 2023-10-23T13:53:55.832Z | 5033755 | Connection error | 4 | sample_data_str | 172.21.3.15 2023-10-23T13:55:01.543Z | 1756467 | Connected to 10.1.0.1 | 4 | sample_data_str | 172.21.3.15 -1970-01-01T00:28:18.068014937Z | 1232382 | Disconnected | 1 | null | 172.21.0.5 -2023-10-23T13:51:54.732123456Z | 725448 | Connection error | 4 | null | 172.21.3.15 -2023-10-23T13:52:55.015123456Z | 8268153 | Connection error | 4 | null | 172.21.3.15 -2023-10-23T13:53:55.832123456Z | 5033755 | Connection error | 4 | null | 172.21.3.15 -2023-10-23T13:55:01.543123456Z | 1756467 | Connected to 10.1.0.1 | 4 | null | 172.21.3.15 +1970-01-01T00:28:18.068014937Z | 1232382 | Disconnected | 1 | sample_data_ts_long | 172.21.0.5 +2023-10-23T13:51:54.732123456Z | 725448 | Connection error | 4 | sample_data_ts_nanos | 172.21.3.15 +2023-10-23T13:52:55.015123456Z | 8268153 | Connection error | 4 | sample_data_ts_nanos | 172.21.3.15 +2023-10-23T13:53:55.832123456Z | 5033755 | Connection error | 4 | sample_data_ts_nanos | 172.21.3.15 +2023-10-23T13:55:01.543123456Z | 1756467 | Connected to 10.1.0.1 | 4 | sample_data_ts_nanos | 172.21.3.15 ; subqueryInFromWithRenameInSubquery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees_incompatible, (FROM employees @@ -577,19 +543,19 @@ emp_no:long | first_name:text | last_name:text | a:keyword | b:keyword ; subqueryInFromWithRenameInMainQuery -required_capability: fork_v9 required_capability: subquery_in_from_command FROM sample_data, sample_data_str, - (FROM sample_data_ts_nanos + (FROM sample_data_ts_nanos metadata _index | WHERE client_ip == "172.21.3.15") , - (FROM sample_data_ts_long + (FROM sample_data_ts_long metadata _index | EVAL @timestamp = @timestamp::date_nanos | WHERE client_ip == "172.21.0.5") metadata _index -| EVAL client_ip = client_ip::ip +| EVAL client_ip = client_ip::ip, _index = MV_LAST(SPLIT(_index, ":")) | DROP event_duration, message | RENAME client_ip AS clientip +| KEEP @timestamp, _index, clientip | SORT _index, @timestamp ; @@ -608,15 +574,14 @@ FROM sample_data, sample_data_str, 2023-10-23T13:52:55.015Z | sample_data_str | 172.21.3.15 2023-10-23T13:53:55.832Z | sample_data_str | 172.21.3.15 2023-10-23T13:55:01.543Z | sample_data_str | 172.21.3.15 -1970-01-01T00:28:18.068014937Z | null | 172.21.0.5 -2023-10-23T13:51:54.732123456Z | null | 172.21.3.15 -2023-10-23T13:52:55.015123456Z | null | 172.21.3.15 -2023-10-23T13:53:55.832123456Z | null | 172.21.3.15 -2023-10-23T13:55:01.543123456Z | null | 172.21.3.15 +1970-01-01T00:28:18.068014937Z | sample_data_ts_long | 172.21.0.5 +2023-10-23T13:51:54.732123456Z | sample_data_ts_nanos | 172.21.3.15 +2023-10-23T13:52:55.015123456Z | sample_data_ts_nanos | 172.21.3.15 +2023-10-23T13:53:55.832123456Z | sample_data_ts_nanos | 172.21.3.15 +2023-10-23T13:55:01.543123456Z | sample_data_ts_nanos | 172.21.3.15 ; subqueryInFromWithSampleInSubquery -required_capability: fork_v9 required_capability: sample_v3 required_capability: subquery_in_from_command @@ -640,7 +605,6 @@ null | null | 172.21.3.15 ; subqueryInFromWithSampleInMainQuery -required_capability: fork_v9 required_capability: sample_v3 required_capability: subquery_in_from_command @@ -665,7 +629,6 @@ null | null | 172.21.3.15 ; subqueryInFromWithChangePointInSubquery -required_capability: fork_v9 required_capability: change_point required_capability: subquery_in_from_command @@ -686,7 +649,6 @@ null | 10023 | 1000000 | spike | 0.0 ; subqueryInFromWithChangePointInMainQuery -required_capability: fork_v9 required_capability: change_point required_capability: subquery_in_from_command @@ -706,7 +668,6 @@ emp_no:long | salary:long | type:keyword | pvalue:double ; subqueryInFromWithCompletionInSubquery -required_capability: fork_v9 required_capability: completion required_capability: match_operator_colon required_capability: subquery_in_from_command @@ -729,51 +690,50 @@ null | null | 2023-10-23T1 ; subqueryInFromWithCompletionInMainQuery -required_capability: fork_v9 required_capability: completion required_capability: subquery_in_from_command -FROM books, (FROM sample_data +FROM books, (FROM sample_data metadata _index | WHERE client_ip == "172.21.0.5" - | KEEP @timestamp, client_ip) + | KEEP @timestamp, client_ip, _index) + metadata _index | SORT book_no | LIMIT 2 | COMPLETION title WITH { "inference_id" : "test_completion" } -| KEEP title, completion, @timestamp, client_ip +| KEEP _index, title, completion, @timestamp, client_ip ; ignoreOrder:true -title:text | completion:keyword | @timestamp:datetime | client_ip:ip -Realms of Tolkien: Images of Middle-earth | REALMS OF TOLKIEN: IMAGES OF MIDDLE-EARTH | null | null -The brothers Karamazov | THE BROTHERS KARAMAZOV | null | null +_index: keyword | title:text | completion:keyword | @timestamp:datetime | client_ip:ip +books | Realms of Tolkien: Images of Middle-earth | REALMS OF TOLKIEN: IMAGES OF MIDDLE-EARTH | null | null +books | The brothers Karamazov | THE BROTHERS KARAMAZOV | null | null ; subqueryInFromWithRerankInSubquery -required_capability: fork_v9 required_capability: rerank required_capability: match_operator_colon required_capability: subquery_in_from_command -FROM sample_data, (FROM books METADATA _score +FROM sample_data, (FROM books METADATA _score, _index | WHERE title:"war and peace" AND author:"Tolstoy" | SORT _score DESC, book_no ASC | LIMIT 2 | RERANK "war and peace" ON title WITH { "inference_id" : "test_reranker" } | EVAL _score=ROUND(_score, 2) - | KEEP book_no, title, author, _score) + | KEEP book_no, title, author, _score, _index) + metadata _index | WHERE client_ip == "172.21.0.5" or client_ip IS NULL -| KEEP book_no, title, author, _score, @timestamp +| KEEP _index, book_no, title, author, _score, @timestamp ; ignoreOrder:true -book_no:keyword | title:text | author:text | _score:double | @timestamp:datetime -4536 | War and Peace (Signet Classics) | [John Hockenberry, Leo Tolstoy, Pat Conroy] | 0.03 | null -5327 | War and Peace | Leo Tolstoy | 0.08 | null -null | null | null | null | 2023-10-23T13:33:34.937Z +_index:keyword | book_no:keyword | title:text | author:text | _score:double | @timestamp:datetime +books | 4536 | War and Peace (Signet Classics) | [John Hockenberry, Leo Tolstoy, Pat Conroy] | 0.03 | null +books | 5327 | War and Peace | Leo Tolstoy | 0.08 | null +sample_data | null | null | null | null | 2023-10-23T13:33:34.937Z ; subqueryInFromWithRerankInMainQuery -required_capability: fork_v9 required_capability: rerank required_capability: subquery_in_from_command @@ -794,7 +754,6 @@ book_no:keyword | title:text | author:text ; subqueryInFromWithUnionTypesWithCommonTypes -required_capability: fork_v9 required_capability: subquery_in_from_command FROM employees, (FROM employees_incompatible, employees @@ -803,259 +762,248 @@ FROM employees, (FROM employees_incompatible, employees , last_name = last_name::keyword | WHERE emp_no < 10093 | KEEP emp_no, first_name, last_name) - metadata _index | WHERE emp_no >= 10091 AND emp_no < 10094 -| SORT _index, emp_no -| KEEP _index, emp_no, first_name, last_name +| SORT emp_no +| KEEP emp_no, first_name, last_name ; -_index:keyword | emp_no:integer | first_name:keyword | last_name:keyword -employees | 10091 | Amabile | Gomatam -employees | 10092 | Valdiodio | Niizuma -employees | 10093 | Sailaja | Desikan -null | 10091 | Amabile | Gomatam -null | 10091 | Amabile | Gomatam -null | 10092 | Valdiodio | Niizuma -null | 10092 | Valdiodio | Niizuma +emp_no:integer | first_name:keyword | last_name:keyword +10091 | Amabile | Gomatam +10091 | Amabile | Gomatam +10091 | Amabile | Gomatam +10092 | Valdiodio | Niizuma +10092 | Valdiodio | Niizuma +10092 | Valdiodio | Niizuma +10093 | Sailaja | Desikan ; subqueryInFromWithUnionTypesWithoutKeepingFieldsWithCommonTypes -required_capability: fork_v9 required_capability: subquery_in_from_command FROM sample_data, - (FROM sample_data_ts_nanos + (FROM sample_data_ts_nanos | WHERE client_ip == "172.21.3.15") , (FROM sample_data_ts_long | EVAL @timestamp = @timestamp::date_nanos | WHERE client_ip == "172.21.0.5") - metadata _index -| SORT _index, @timestamp +| SORT @timestamp ; -@timestamp:date_nanos | client_ip:ip | event_duration:long | message:keyword | _index:keyword -2023-10-23T12:15:03.360Z | 172.21.2.162 | 3450233 | Connected to 10.1.0.3 | sample_data -2023-10-23T12:27:28.948Z | 172.21.2.113 | 2764889 | Connected to 10.1.0.2 | sample_data -2023-10-23T13:33:34.937Z | 172.21.0.5 | 1232382 | Disconnected | sample_data -2023-10-23T13:51:54.732Z | 172.21.3.15 | 725448 | Connection error | sample_data -2023-10-23T13:52:55.015Z | 172.21.3.15 | 8268153 | Connection error | sample_data -2023-10-23T13:53:55.832Z | 172.21.3.15 | 5033755 | Connection error | sample_data -2023-10-23T13:55:01.543Z | 172.21.3.15 | 1756467 | Connected to 10.1.0.1 | sample_data -1970-01-01T00:28:18.068014937Z | 172.21.0.5 | 1232382 | Disconnected | null -2023-10-23T13:51:54.732123456Z | 172.21.3.15 | 725448 | Connection error | null -2023-10-23T13:52:55.015123456Z | 172.21.3.15 | 8268153 | Connection error | null -2023-10-23T13:53:55.832123456Z | 172.21.3.15 | 5033755 | Connection error | null -2023-10-23T13:55:01.543123456Z | 172.21.3.15 | 1756467 | Connected to 10.1.0.1 | null +@timestamp:date_nanos | client_ip:ip | event_duration:long | message:keyword +1970-01-01T00:28:18.068014937Z | 172.21.0.5 | 1232382 | Disconnected +2023-10-23T12:15:03.360Z | 172.21.2.162 | 3450233 | Connected to 10.1.0.3 +2023-10-23T12:27:28.948Z | 172.21.2.113 | 2764889 | Connected to 10.1.0.2 +2023-10-23T13:33:34.937Z | 172.21.0.5 | 1232382 | Disconnected +2023-10-23T13:51:54.732Z | 172.21.3.15 | 725448 | Connection error +2023-10-23T13:51:54.732123456Z | 172.21.3.15 | 725448 | Connection error +2023-10-23T13:52:55.015Z | 172.21.3.15 | 8268153 | Connection error +2023-10-23T13:52:55.015123456Z | 172.21.3.15 | 8268153 | Connection error +2023-10-23T13:53:55.832Z | 172.21.3.15 | 5033755 | Connection error +2023-10-23T13:53:55.832123456Z | 172.21.3.15 | 5033755 | Connection error +2023-10-23T13:55:01.543Z | 172.21.3.15 | 1756467 | Connected to 10.1.0.1 +2023-10-23T13:55:01.543123456Z | 172.21.3.15 | 1756467 | Connected to 10.1.0.1 ; subqueryInFromWithUnionTypesWithoutCommonTypesExplicitCasting -required_capability: fork_v9 required_capability: subquery_in_from_command FROM sample_data, sample_data_str, - (FROM sample_data_ts_nanos + (FROM sample_data_ts_nanos metadata _index | WHERE client_ip == "172.21.3.15") , - (FROM sample_data_ts_long + (FROM sample_data_ts_long metadata _index | EVAL @timestamp = @timestamp::date_nanos | WHERE client_ip == "172.21.0.5") metadata _index -| EVAL client_ip = client_ip::ip +| EVAL _index = MV_LAST(SPLIT(_index, ":")), client_ip = client_ip::ip | SORT _index, @timestamp ; -@timestamp:date_nanos | event_duration:long | message:keyword | _index:keyword | client_ip:ip -2023-10-23T12:15:03.360Z | 3450233 | Connected to 10.1.0.3 | sample_data | 172.21.2.162 -2023-10-23T12:27:28.948Z | 2764889 | Connected to 10.1.0.2 | sample_data | 172.21.2.113 -2023-10-23T13:33:34.937Z | 1232382 | Disconnected | sample_data | 172.21.0.5 -2023-10-23T13:51:54.732Z | 725448 | Connection error | sample_data | 172.21.3.15 -2023-10-23T13:52:55.015Z | 8268153 | Connection error | sample_data | 172.21.3.15 -2023-10-23T13:53:55.832Z | 5033755 | Connection error | sample_data | 172.21.3.15 -2023-10-23T13:55:01.543Z | 1756467 | Connected to 10.1.0.1 | sample_data | 172.21.3.15 -2023-10-23T12:15:03.360Z | 3450233 | Connected to 10.1.0.3 | sample_data_str | 172.21.2.162 -2023-10-23T12:27:28.948Z | 2764889 | Connected to 10.1.0.2 | sample_data_str | 172.21.2.113 -2023-10-23T13:33:34.937Z | 1232382 | Disconnected | sample_data_str | 172.21.0.5 -2023-10-23T13:51:54.732Z | 725448 | Connection error | sample_data_str | 172.21.3.15 -2023-10-23T13:52:55.015Z | 8268153 | Connection error | sample_data_str | 172.21.3.15 -2023-10-23T13:53:55.832Z | 5033755 | Connection error | sample_data_str | 172.21.3.15 -2023-10-23T13:55:01.543Z | 1756467 | Connected to 10.1.0.1 | sample_data_str | 172.21.3.15 -1970-01-01T00:28:18.068014937Z | 1232382 | Disconnected | null | 172.21.0.5 -2023-10-23T13:51:54.732123456Z | 725448 | Connection error | null | 172.21.3.15 -2023-10-23T13:52:55.015123456Z | 8268153 | Connection error | null | 172.21.3.15 -2023-10-23T13:53:55.832123456Z | 5033755 | Connection error | null | 172.21.3.15 -2023-10-23T13:55:01.543123456Z | 1756467 | Connected to 10.1.0.1 | null | 172.21.3.15 +@timestamp:date_nanos | event_duration:long | message:keyword | _index:keyword | client_ip:ip +2023-10-23T12:15:03.360Z | 3450233 | Connected to 10.1.0.3 | sample_data | 172.21.2.162 +2023-10-23T12:27:28.948Z | 2764889 | Connected to 10.1.0.2 | sample_data | 172.21.2.113 +2023-10-23T13:33:34.937Z | 1232382 | Disconnected | sample_data | 172.21.0.5 +2023-10-23T13:51:54.732Z | 725448 | Connection error | sample_data | 172.21.3.15 +2023-10-23T13:52:55.015Z | 8268153 | Connection error | sample_data | 172.21.3.15 +2023-10-23T13:53:55.832Z | 5033755 | Connection error | sample_data | 172.21.3.15 +2023-10-23T13:55:01.543Z | 1756467 | Connected to 10.1.0.1 | sample_data | 172.21.3.15 +2023-10-23T12:15:03.360Z | 3450233 | Connected to 10.1.0.3 | sample_data_str | 172.21.2.162 +2023-10-23T12:27:28.948Z | 2764889 | Connected to 10.1.0.2 | sample_data_str | 172.21.2.113 +2023-10-23T13:33:34.937Z | 1232382 | Disconnected | sample_data_str | 172.21.0.5 +2023-10-23T13:51:54.732Z | 725448 | Connection error | sample_data_str | 172.21.3.15 +2023-10-23T13:52:55.015Z | 8268153 | Connection error | sample_data_str | 172.21.3.15 +2023-10-23T13:53:55.832Z | 5033755 | Connection error | sample_data_str | 172.21.3.15 +2023-10-23T13:55:01.543Z | 1756467 | Connected to 10.1.0.1 | sample_data_str | 172.21.3.15 +1970-01-01T00:28:18.068014937Z | 1232382 | Disconnected | sample_data_ts_long | 172.21.0.5 +2023-10-23T13:51:54.732123456Z | 725448 | Connection error | sample_data_ts_nanos | 172.21.3.15 +2023-10-23T13:52:55.015123456Z | 8268153 | Connection error | sample_data_ts_nanos | 172.21.3.15 +2023-10-23T13:53:55.832123456Z | 5033755 | Connection error | sample_data_ts_nanos | 172.21.3.15 +2023-10-23T13:55:01.543123456Z | 1756467 | Connected to 10.1.0.1 | sample_data_ts_nanos | 172.21.3.15 ; subqueryInFromWithUnionTypesWithoutCommonTypesMultipleExplicitCasting -required_capability: fork_v9 required_capability: subquery_in_from_command FROM sample_data, sample_data_str, - (FROM sample_data_ts_nanos metadata _index + (FROM sample_data_ts_nanos | WHERE client_ip == "172.21.3.15") , - (FROM sample_data_ts_long metadata _index + (FROM sample_data_ts_long | EVAL @timestamp = @timestamp::date_nanos | WHERE client_ip == "172.21.0.5") - metadata _index | EVAL client_ip = client_ip::ip, client_ip_str = client_ip::keyword | WHERE client_ip IN ("172.21.0.5", "172.21.3.15", "172.21.2.162") -| SORT _index, @timestamp -; - -@timestamp:date_nanos | event_duration:long | message:keyword | _index:keyword | client_ip:ip | client_ip_str:keyword -2023-10-23T12:15:03.360Z | 3450233 | Connected to 10.1.0.3 | sample_data | 172.21.2.162 | 172.21.2.162 -2023-10-23T13:33:34.937Z | 1232382 | Disconnected | sample_data | 172.21.0.5 | 172.21.0.5 -2023-10-23T13:51:54.732Z | 725448 | Connection error | sample_data | 172.21.3.15 | 172.21.3.15 -2023-10-23T13:52:55.015Z | 8268153 | Connection error | sample_data | 172.21.3.15 | 172.21.3.15 -2023-10-23T13:53:55.832Z | 5033755 | Connection error | sample_data | 172.21.3.15 | 172.21.3.15 -2023-10-23T13:55:01.543Z | 1756467 | Connected to 10.1.0.1 | sample_data | 172.21.3.15 | 172.21.3.15 -2023-10-23T12:15:03.360Z | 3450233 | Connected to 10.1.0.3 | sample_data_str | 172.21.2.162 | 172.21.2.162 -2023-10-23T13:33:34.937Z | 1232382 | Disconnected | sample_data_str | 172.21.0.5 | 172.21.0.5 -2023-10-23T13:51:54.732Z | 725448 | Connection error | sample_data_str | 172.21.3.15 | 172.21.3.15 -2023-10-23T13:52:55.015Z | 8268153 | Connection error | sample_data_str | 172.21.3.15 | 172.21.3.15 -2023-10-23T13:53:55.832Z | 5033755 | Connection error | sample_data_str | 172.21.3.15 | 172.21.3.15 -2023-10-23T13:55:01.543Z | 1756467 | Connected to 10.1.0.1 | sample_data_str | 172.21.3.15 | 172.21.3.15 -1970-01-01T00:28:18.068014937Z | 1232382 | Disconnected | sample_data_ts_long | 172.21.0.5 | 172.21.0.5 -2023-10-23T13:51:54.732123456Z | 725448 | Connection error | sample_data_ts_nanos | 172.21.3.15 | 172.21.3.15 -2023-10-23T13:52:55.015123456Z | 8268153 | Connection error | sample_data_ts_nanos | 172.21.3.15 | 172.21.3.15 -2023-10-23T13:53:55.832123456Z | 5033755 | Connection error | sample_data_ts_nanos | 172.21.3.15 | 172.21.3.15 -2023-10-23T13:55:01.543123456Z | 1756467 | Connected to 10.1.0.1 | sample_data_ts_nanos | 172.21.3.15 | 172.21.3.15 +| SORT @timestamp +; + +@timestamp:date_nanos | event_duration:long | message:keyword | client_ip:ip | client_ip_str:keyword +1970-01-01T00:28:18.068014937Z | 1232382 | Disconnected | 172.21.0.5 | 172.21.0.5 +2023-10-23T12:15:03.360Z | 3450233 | Connected to 10.1.0.3 | 172.21.2.162 | 172.21.2.162 +2023-10-23T12:15:03.360Z | 3450233 | Connected to 10.1.0.3 | 172.21.2.162 | 172.21.2.162 +2023-10-23T13:33:34.937Z | 1232382 | Disconnected | 172.21.0.5 | 172.21.0.5 +2023-10-23T13:33:34.937Z | 1232382 | Disconnected | 172.21.0.5 | 172.21.0.5 +2023-10-23T13:51:54.732Z | 725448 | Connection error | 172.21.3.15 | 172.21.3.15 +2023-10-23T13:51:54.732Z | 725448 | Connection error | 172.21.3.15 | 172.21.3.15 +2023-10-23T13:51:54.732123456Z | 725448 | Connection error | 172.21.3.15 | 172.21.3.15 +2023-10-23T13:52:55.015Z | 8268153 | Connection error | 172.21.3.15 | 172.21.3.15 +2023-10-23T13:52:55.015Z | 8268153 | Connection error | 172.21.3.15 | 172.21.3.15 +2023-10-23T13:52:55.015123456Z | 8268153 | Connection error | 172.21.3.15 | 172.21.3.15 +2023-10-23T13:53:55.832Z | 5033755 | Connection error | 172.21.3.15 | 172.21.3.15 +2023-10-23T13:53:55.832Z | 5033755 | Connection error | 172.21.3.15 | 172.21.3.15 +2023-10-23T13:53:55.832123456Z | 5033755 | Connection error | 172.21.3.15 | 172.21.3.15 +2023-10-23T13:55:01.543Z | 1756467 | Connected to 10.1.0.1 | 172.21.3.15 | 172.21.3.15 +2023-10-23T13:55:01.543Z | 1756467 | Connected to 10.1.0.1 | 172.21.3.15 | 172.21.3.15 +2023-10-23T13:55:01.543123456Z | 1756467 | Connected to 10.1.0.1 | 172.21.3.15 | 172.21.3.15 ; subqueryInFromWithUnionTypesWithoutCommonTypesWithoutExplicitCasting -required_capability: fork_v9 required_capability: subquery_in_from_command FROM sample_data, sample_data_str, - (FROM sample_data_ts_nanos + (FROM sample_data_ts_nanos | WHERE client_ip == "172.21.3.15") , (FROM sample_data_ts_long | EVAL @timestamp = @timestamp::date_nanos | WHERE client_ip == "172.21.0.5") - metadata _index -| SORT _index, @timestamp -; - -@timestamp:date_nanos | event_duration:long | message:keyword | _index:keyword | client_ip:keyword -2023-10-23T12:15:03.360Z | 3450233 | Connected to 10.1.0.3 | sample_data | null -2023-10-23T12:27:28.948Z | 2764889 | Connected to 10.1.0.2 | sample_data | null -2023-10-23T13:33:34.937Z | 1232382 | Disconnected | sample_data | null -2023-10-23T13:51:54.732Z | 725448 | Connection error | sample_data | null -2023-10-23T13:52:55.015Z | 8268153 | Connection error | sample_data | null -2023-10-23T13:53:55.832Z | 5033755 | Connection error | sample_data | null -2023-10-23T13:55:01.543Z | 1756467 | Connected to 10.1.0.1 | sample_data | null -2023-10-23T12:15:03.360Z | 3450233 | Connected to 10.1.0.3 | sample_data_str | null -2023-10-23T12:27:28.948Z | 2764889 | Connected to 10.1.0.2 | sample_data_str | null -2023-10-23T13:33:34.937Z | 1232382 | Disconnected | sample_data_str | null -2023-10-23T13:51:54.732Z | 725448 | Connection error | sample_data_str | null -2023-10-23T13:52:55.015Z | 8268153 | Connection error | sample_data_str | null -2023-10-23T13:53:55.832Z | 5033755 | Connection error | sample_data_str | null -2023-10-23T13:55:01.543Z | 1756467 | Connected to 10.1.0.1 | sample_data_str | null -1970-01-01T00:28:18.068014937Z | 1232382 | Disconnected | null | null -2023-10-23T13:51:54.732123456Z | 725448 | Connection error | null | null -2023-10-23T13:52:55.015123456Z | 8268153 | Connection error | null | null -2023-10-23T13:53:55.832123456Z | 5033755 | Connection error | null | null -2023-10-23T13:55:01.543123456Z | 1756467 | Connected to 10.1.0.1 | null | null +| KEEP @timestamp, event_duration, message, client_ip +| SORT @timestamp +; + +@timestamp:date_nanos | event_duration:long | message:keyword | client_ip:keyword +1970-01-01T00:28:18.068014937Z | 1232382 | Disconnected | null +2023-10-23T12:15:03.360Z | 3450233 | Connected to 10.1.0.3 | null +2023-10-23T12:15:03.360Z | 3450233 | Connected to 10.1.0.3 | null +2023-10-23T12:27:28.948Z | 2764889 | Connected to 10.1.0.2 | null +2023-10-23T12:27:28.948Z | 2764889 | Connected to 10.1.0.2 | null +2023-10-23T13:33:34.937Z | 1232382 | Disconnected | null +2023-10-23T13:33:34.937Z | 1232382 | Disconnected | null +2023-10-23T13:51:54.732Z | 725448 | Connection error | null +2023-10-23T13:51:54.732Z | 725448 | Connection error | null +2023-10-23T13:51:54.732123456Z | 725448 | Connection error | null +2023-10-23T13:52:55.015Z | 8268153 | Connection error | null +2023-10-23T13:52:55.015Z | 8268153 | Connection error | null +2023-10-23T13:52:55.015123456Z | 8268153 | Connection error | null +2023-10-23T13:53:55.832Z | 5033755 | Connection error | null +2023-10-23T13:53:55.832Z | 5033755 | Connection error | null +2023-10-23T13:53:55.832123456Z | 5033755 | Connection error | null +2023-10-23T13:55:01.543Z | 1756467 | Connected to 10.1.0.1 | null +2023-10-23T13:55:01.543Z | 1756467 | Connected to 10.1.0.1 | null +2023-10-23T13:55:01.543123456Z | 1756467 | Connected to 10.1.0.1 | null ; subqueryInFromWithTimeSeriesDataTypesInSubquery -required_capability: fork_v9 required_capability: subquery_in_from_command -FROM sample_data, (FROM k8s metadata _index) metadata _index +FROM sample_data, (FROM k8s) | WHERE @timestamp < "2024-05-10T00:01:00.000Z" -| KEEP _index, @timestamp, client.ip, event_duration, cluster, network.total_bytes_in, network.eth0.tx -| SORT _index, @timestamp, client.ip, cluster +| KEEP @timestamp, client.ip, event_duration, cluster, network.total_bytes_in, network.eth0.tx +| SORT @timestamp, client.ip, cluster ; -_index:keyword | @timestamp:datetime | client.ip:ip | event_duration:long | cluster:keyword | network.total_bytes_in:long | network.eth0.tx:integer -k8s | 2024-05-10T00:00:29.000Z | 10.10.20.34 | null | staging | 953 | 81 -k8s | 2024-05-10T00:00:33.000Z | 10.10.20.34 | null | staging | 1111 | 48 -k8s | 2024-05-10T00:00:51.000Z | 10.10.20.30 | null | prod | 278 | 58 -k8s | 2024-05-10T00:00:57.000Z | 10.10.20.30 | null | prod | 955 | 131 -sample_data | 2023-10-23T12:15:03.360Z | null | 3450233 | null | null | null -sample_data | 2023-10-23T12:27:28.948Z | null | 2764889 | null | null | null -sample_data | 2023-10-23T13:33:34.937Z | null | 1232382 | null | null | null -sample_data | 2023-10-23T13:51:54.732Z | null | 725448 | null | null | null -sample_data | 2023-10-23T13:52:55.015Z | null | 8268153 | null | null | null -sample_data | 2023-10-23T13:53:55.832Z | null | 5033755 | null | null | null -sample_data | 2023-10-23T13:55:01.543Z | null | 1756467 | null | null | null +@timestamp:datetime | client.ip:ip | event_duration:long | cluster:keyword | network.total_bytes_in:long | network.eth0.tx:integer +2023-10-23T12:15:03.360Z | null | 3450233 | null | null | null +2023-10-23T12:27:28.948Z | null | 2764889 | null | null | null +2023-10-23T13:33:34.937Z | null | 1232382 | null | null | null +2023-10-23T13:51:54.732Z | null | 725448 | null | null | null +2023-10-23T13:52:55.015Z | null | 8268153 | null | null | null +2023-10-23T13:53:55.832Z | null | 5033755 | null | null | null +2023-10-23T13:55:01.543Z | null | 1756467 | null | null | null +2024-05-10T00:00:29.000Z | 10.10.20.34 | null | staging | 953 | 81 +2024-05-10T00:00:33.000Z | 10.10.20.34 | null | staging | 1111 | 48 +2024-05-10T00:00:51.000Z | 10.10.20.30 | null | prod | 278 | 58 +2024-05-10T00:00:57.000Z | 10.10.20.30 | null | prod | 955 | 131 ; subqueryInFromWithTimeSeriesDataTypesInMainQuery -required_capability: fork_v9 required_capability: subquery_in_from_command -FROM k8s-downsampled, (FROM sample_data metadata _index) metadata _index +FROM k8s-downsampled, (FROM sample_data) | WHERE @timestamp <= "2024-05-09T23:30:00.000Z" -| KEEP _index, @timestamp, client.ip, event_duration, cluster, network.total_bytes_in, network.eth0.tx -| SORT _index, @timestamp, client.ip, cluster, network.total_bytes_in -; - -_index:keyword | @timestamp:datetime | client.ip:ip | event_duration:long | cluster:keyword | network.total_bytes_in:long | network.eth0.tx:aggregate_metric_double -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.30 | null | prod | 285 | {"min":565.0,"max":829.0,"sum":7290.0,"value_count":10} -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.30 | null | qa | 1143 | {"min":605.0,"max":605.0,"sum":605.0,"value_count":1} -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.30 | null | staging | 930 | {"min":341.0,"max":592.0,"sum":1956.0,"value_count":5} -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.31 | null | prod | 1038 | {"min":20.0,"max":190.0,"sum":370.0,"value_count":10} -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.31 | null | qa | 363 | {"min":346.0,"max":356.0,"sum":1765.0,"value_count":5} -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.31 | null | qa | 1032 | {"min":304.0,"max":1148.0,"sum":8590.0,"value_count":10} -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.33 | null | prod | 210 | {"min":201.0,"max":582.0,"sum":1794.0,"value_count":6} -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.34 | null | staging | 821 | {"min":263.0,"max":740.0,"sum":5390.0,"value_count":10} -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.34 | null | staging | 838 | {"min":442.0,"max":1011.0,"sum":3850.0,"value_count":7} -sample_data | 2023-10-23T12:15:03.360Z | null | 3450233 | null | null | null -sample_data | 2023-10-23T12:27:28.948Z | null | 2764889 | null | null | null -sample_data | 2023-10-23T13:33:34.937Z | null | 1232382 | null | null | null -sample_data | 2023-10-23T13:51:54.732Z | null | 725448 | null | null | null -sample_data | 2023-10-23T13:52:55.015Z | null | 8268153 | null | null | null -sample_data | 2023-10-23T13:53:55.832Z | null | 5033755 | null | null | null -sample_data | 2023-10-23T13:55:01.543Z | null | 1756467 | null | null | null +| KEEP @timestamp, client.ip, event_duration, cluster, network.total_bytes_in, network.eth0.tx +| SORT @timestamp, client.ip, cluster, network.total_bytes_in +; + +@timestamp:datetime | client.ip:ip | event_duration:long | cluster:keyword | network.total_bytes_in:long | network.eth0.tx:aggregate_metric_double +2023-10-23T12:15:03.360Z | null | 3450233 | null | null | null +2023-10-23T12:27:28.948Z | null | 2764889 | null | null | null +2023-10-23T13:33:34.937Z | null | 1232382 | null | null | null +2023-10-23T13:51:54.732Z | null | 725448 | null | null | null +2023-10-23T13:52:55.015Z | null | 8268153 | null | null | null +2023-10-23T13:53:55.832Z | null | 5033755 | null | null | null +2023-10-23T13:55:01.543Z | null | 1756467 | null | null | null +2024-05-09T23:30:00.000Z | 10.10.20.30 | null | prod | 285 | {"min":565.0,"max":829.0,"sum":7290.0,"value_count":10} +2024-05-09T23:30:00.000Z | 10.10.20.30 | null | qa | 1143 | {"min":605.0,"max":605.0,"sum":605.0,"value_count":1} +2024-05-09T23:30:00.000Z | 10.10.20.30 | null | staging | 930 | {"min":341.0,"max":592.0,"sum":1956.0,"value_count":5} +2024-05-09T23:30:00.000Z | 10.10.20.31 | null | prod | 1038 | {"min":20.0,"max":190.0,"sum":370.0,"value_count":10} +2024-05-09T23:30:00.000Z | 10.10.20.31 | null | qa | 363 | {"min":346.0,"max":356.0,"sum":1765.0,"value_count":5} +2024-05-09T23:30:00.000Z | 10.10.20.31 | null | qa | 1032 | {"min":304.0,"max":1148.0,"sum":8590.0,"value_count":10} +2024-05-09T23:30:00.000Z | 10.10.20.33 | null | prod | 210 | {"min":201.0,"max":582.0,"sum":1794.0,"value_count":6} +2024-05-09T23:30:00.000Z | 10.10.20.34 | null | staging | 821 | {"min":263.0,"max":740.0,"sum":5390.0,"value_count":10} +2024-05-09T23:30:00.000Z | 10.10.20.34 | null | staging | 838 | {"min":442.0,"max":1011.0,"sum":3850.0,"value_count":7} ; subqueryInFromWithTimeSeriesDataTypesInMainQueryAndSubquery -required_capability: fork_v9 required_capability: subquery_in_from_command -FROM k8s, (FROM k8s-downsampled metadata _index | WHERE @timestamp <= "2024-05-09T23:30:00.000Z") metadata _index +FROM k8s, (FROM k8s-downsampled | WHERE @timestamp <= "2024-05-09T23:30:00.000Z") | WHERE @timestamp <= "2024-05-10T00:01:00.000Z" -| KEEP _index, @timestamp, client.ip, cluster, network.total_bytes_in, network.eth0.tx -| SORT _index, @timestamp, client.ip, cluster, network.total_bytes_in -; - -_index:keyword | @timestamp:datetime | client.ip:ip | cluster:keyword | network.total_bytes_in:long | network.eth0.tx:keyword -k8s | 2024-05-10T00:00:29.000Z | 10.10.20.34 | staging | 953 | null -k8s | 2024-05-10T00:00:33.000Z | 10.10.20.34 | staging | 1111 | null -k8s | 2024-05-10T00:00:51.000Z | 10.10.20.30 | prod | 278 | null -k8s | 2024-05-10T00:00:57.000Z | 10.10.20.30 | prod | 955 | null -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.30 | prod | 285 | null -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.30 | qa | 1143 | null -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.30 | staging | 930 | null -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.31 | prod | 1038 | null -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.31 | qa | 363 | null -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.31 | qa | 1032 | null -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.33 | prod | 210 | null -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.34 | staging | 821 | null -k8s-downsampled | 2024-05-09T23:30:00.000Z | 10.10.20.34 | staging | 838 | null +| KEEP @timestamp, client.ip, cluster, network.total_bytes_in, network.eth0.tx +| SORT @timestamp, client.ip, cluster, network.total_bytes_in +; + +@timestamp:datetime | client.ip:ip | cluster:keyword | network.total_bytes_in:long | network.eth0.tx:keyword +2024-05-09T23:30:00.000Z | 10.10.20.30 | prod | 285 | null +2024-05-09T23:30:00.000Z | 10.10.20.30 | qa | 1143 | null +2024-05-09T23:30:00.000Z | 10.10.20.30 | staging | 930 | null +2024-05-09T23:30:00.000Z | 10.10.20.31 | prod | 1038 | null +2024-05-09T23:30:00.000Z | 10.10.20.31 | qa | 363 | null +2024-05-09T23:30:00.000Z | 10.10.20.31 | qa | 1032 | null +2024-05-09T23:30:00.000Z | 10.10.20.33 | prod | 210 | null +2024-05-09T23:30:00.000Z | 10.10.20.34 | staging | 821 | null +2024-05-09T23:30:00.000Z | 10.10.20.34 | staging | 838 | null +2024-05-10T00:00:29.000Z | 10.10.20.34 | staging | 953 | null +2024-05-10T00:00:33.000Z | 10.10.20.34 | staging | 1111 | null +2024-05-10T00:00:51.000Z | 10.10.20.30 | prod | 278 | null +2024-05-10T00:00:57.000Z | 10.10.20.30 | prod | 955 | null ; subqueryInFromWithFullTextFunctionInMainQueryAndSubquery -required_capability: fork_v9 required_capability: subquery_in_from_command required_capability: match_function -FROM books, (FROM books | WHERE author:"Faulkner" and ratings > 4) metadata _index +FROM books, (FROM books | WHERE author:"Faulkner" and ratings > 4) | WHERE match(title, "Faulkner") OR qstr("publisher:Oxford") | EVAL ratings = ROUND(ratings, 2) -| KEEP _index, book_no, title, author, publisher, ratings -| SORT _index, book_no -; - -_index:keyword | book_no:keyword | title:text | author:text | publisher:text | ratings:double -books | 2713 | Collected Stories of William Faulkner | William Faulkner | Vintage | 4.53 -books | 2776 | The Devil and Other Stories (Oxford World's Classics) | Leo Tolstoy | OUP Oxford | 5.0 -books | 2883 | A Summer of Faulkner: As I Lay Dying/The Sound and the Fury/Light in August (Oprah's Book Club) | William Faulkner | Vintage Books | 3.89 -books | 2924 | A Gentle Creature and Other Stories: White Nights, A Gentle Creature, and The Dream of a Ridiculous Man (The World's Classics) | [Alan Myers, Fyodor Dostoevsky, W. J. Leatherbarrow] | Oxford Paperbacks | 4.0 -books | 5119 | William Faulkner | William Faulkner | Vintage | 4.0 -books | 5948 | That We Are Gentle Creatures | Fyodor Dostoevsky | OUP Oxford | 4.33 -books | 8534 | Crime and Punishment (Oxford World's Classics) | Fyodor Dostoevsky | Oxford University Press | 4.38 -books | 9801 | The Karamazov Brothers (Oxford World's Classics) | Fyodor Dostoevsky | Oxford University Press | 4.4 -null | 2713 | Collected Stories of William Faulkner | William Faulkner | Vintage | 4.53 +| KEEP book_no, title, author, publisher, ratings +| SORT book_no +; + +book_no:keyword | title:text | author:text | publisher:text | ratings:double +2713 | Collected Stories of William Faulkner | William Faulkner | Vintage | 4.53 +2713 | Collected Stories of William Faulkner | William Faulkner | Vintage | 4.53 +2776 | The Devil and Other Stories (Oxford World's Classics) | Leo Tolstoy | OUP Oxford | 5.0 +2883 | A Summer of Faulkner: As I Lay Dying/The Sound and the Fury/Light in August (Oprah's Book Club) | William Faulkner | Vintage Books | 3.89 +2924 | A Gentle Creature and Other Stories: White Nights, A Gentle Creature, and The Dream of a Ridiculous Man (The World's Classics) | [Alan Myers, Fyodor Dostoevsky, W. J. Leatherbarrow] | Oxford Paperbacks | 4.0 +5119 | William Faulkner | William Faulkner | Vintage | 4.0 +5948 | That We Are Gentle Creatures | Fyodor Dostoevsky | OUP Oxford | 4.33 +8534 | Crime and Punishment (Oxford World's Classics) | Fyodor Dostoevsky | Oxford University Press | 4.38 +9801 | The Karamazov Brothers (Oxford World's Classics) | Fyodor Dostoevsky | Oxford University Press | 4.4 ; diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterSubqueryIT.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterSubqueryIT.java new file mode 100644 index 0000000000000..e25c6e529c6fd --- /dev/null +++ b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterSubqueryIT.java @@ -0,0 +1,510 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the Elastic License + * 2.0; you may not use this file except in compliance with the Elastic License + * 2.0. + */ + +package org.elasticsearch.xpack.esql.action; + +import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest; +import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequestBuilder; +import org.elasticsearch.client.internal.Client; +import org.elasticsearch.index.query.RangeQueryBuilder; +import org.elasticsearch.xpack.esql.VerificationException; +import org.junit.Before; + +import java.io.IOException; +import java.time.Duration; +import java.time.ZoneOffset; +import java.time.ZonedDateTime; +import java.time.format.DateTimeFormatter; +import java.util.List; +import java.util.Locale; + +import static org.elasticsearch.core.TimeValue.timeValueSeconds; +import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked; +import static org.elasticsearch.xpack.esql.EsqlTestUtils.getValuesList; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.empty; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; +import static org.hamcrest.Matchers.hasItems; +import static org.hamcrest.Matchers.hasSize; + +// @TestLogging(value = "org.elasticsearch.xpack.esql.session:DEBUG", reason = "to better understand planning") +public class CrossClusterSubqueryIT extends AbstractCrossClusterTestCase { + + @Before + public void checkSubqueryInFromCommandSupport() { + assumeTrue("Requires subquery in FROM command support", EsqlCapabilities.Cap.SUBQUERY_IN_FROM_COMMAND.isEnabled()); + } + + public void testSubquery() throws IOException { + setupClusters(3); + + try (EsqlQueryResponse resp = runQuery(""" + FROM (FROM logs-* metadata _index),(FROM *:logs-* metadata _index) + | SORT _index, id + """, randomBoolean())) { + var columns = resp.columns().stream().map(ColumnInfoImpl::name).toList(); + assertThat(columns, hasItems("id", "tag", "v", "const", "_index")); + int constIndex = columns.indexOf("const"); + int vIndex = columns.indexOf("v"); + int indexIndex = columns.indexOf("_index"); + + List> values = getValuesList(resp); + assertThat(values, hasSize(30)); + for (int i = 0; i < values.size(); i++) { + var row = values.get(i); + assertThat(row, hasSize(5)); + assertNull(row.get(constIndex)); + String indexName = (String) row.get(indexIndex); + if (i < 10) { + assertEquals("cluster-a:logs-2", indexName); + } else if (i < 20) { + assertEquals("logs-1", indexName); + } else { + assertEquals("remote-b:logs-2", indexName); + } + assertThat((Long) row.get(vIndex), greaterThanOrEqualTo(0L)); + } + + EsqlExecutionInfo executionInfo = resp.getExecutionInfo(); + assertCCSExecutionInfoDetails(executionInfo); + } + } + + public void testSubqueryWithAliases() throws IOException { + setupClusters(3); + setupAlias(LOCAL_CLUSTER, "logs-1", "logs-a"); + setupAlias(REMOTE_CLUSTER_1, "logs-2", "logs-a"); + setupAlias(REMOTE_CLUSTER_2, "logs-2", "logs-a"); + + try (EsqlQueryResponse resp = runQuery(""" + FROM logs-a,(FROM *:logs-a metadata _index) metadata _index + | STATS c = count(*) by _index + | SORT _index + """, randomBoolean())) { + var columns = resp.columns().stream().map(ColumnInfoImpl::name).toList(); + assertThat(columns, hasItems("c", "_index")); + + List> values = getValuesList(resp); + assertThat(values, hasSize(3)); + List> expected = List.of( + List.of(10L, "cluster-a:logs-2"), + List.of(10L, "logs-1"), + List.of(10L, "remote-b:logs-2") + ); + assertEquals(expected, values); + + EsqlExecutionInfo executionInfo = resp.getExecutionInfo(); + assertCCSExecutionInfoDetails(executionInfo); + } + } + + public void testSubqueryWithDateMath() throws IOException { + setupClusters(3); + + ZonedDateTime nowUtc = ZonedDateTime.now(ZoneOffset.UTC); + ZonedDateTime nextMidnight = nowUtc.plusDays(1).withHour(0).withMinute(0).withSecond(0).withNano(0); + // If we're too close to midnight, we could create index with one day and query with another, and it'd fail. + assumeTrue("Skip if too close to midnight", Duration.between(nowUtc, nextMidnight).toMinutes() >= 5); + DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy.MM.dd", Locale.ROOT); + String indexName = "idx_" + nowUtc.format(formatter); + + populateIndex(LOCAL_CLUSTER, indexName, randomIntBetween(1, 5), 5); + populateIndex(REMOTE_CLUSTER_1, indexName, randomIntBetween(1, 5), 5); + populateIndex(REMOTE_CLUSTER_2, indexName, randomIntBetween(1, 5), 5); + + try (EsqlQueryResponse resp = runQuery(""" + FROM + (FROM | STATS c = count(*) | EVAL cluster = "local"), + (FROM *: | STATS c = count(*) | EVAL cluster = "remote") + | SORT c DESC + """, randomBoolean())) { + var columns = resp.columns().stream().map(ColumnInfoImpl::name).toList(); + assertThat(columns, hasItems("c", "cluster")); + + List> values = getValuesList(resp); + assertThat(values, hasSize(2)); + List> expected = List.of(List.of(10L, "remote"), List.of(5L, "local")); + assertEquals(expected, values); + + EsqlExecutionInfo executionInfo = resp.getExecutionInfo(); + assertCCSExecutionInfoDetails(executionInfo); + } + } + + public void testSubqueryWithMissingRemoteIndex() throws IOException { + setupClusters(3); + populateIndex(LOCAL_CLUSTER, "local_idx", randomIntBetween(1, 5), 5); + populateIndex(REMOTE_CLUSTER_2, "remote_idx", randomIntBetween(1, 5), 5); + + try { + for (boolean skipUnavailableOnRemote : List.of(true, false)) { + setSkipUnavailable(REMOTE_CLUSTER_1, skipUnavailableOnRemote); + try (EsqlQueryResponse resp = runQuery(""" + FROM local*,(FROM *:remote* metadata _index) metadata _index + | STATS c = count(*) by _index + | SORT _index + """, randomBoolean())) { + var columns = resp.columns().stream().map(ColumnInfoImpl::name).toList(); + assertThat(columns, hasItems("c", "_index")); + + List> values = getValuesList(resp); + assertThat(values, hasSize(2)); + List> expected = List.of(List.of(5L, "local_idx"), List.of(5L, "remote-b:remote_idx")); + assertEquals(expected, values); + + EsqlExecutionInfo executionInfo = resp.getExecutionInfo(); + + var localCluster = executionInfo.getCluster(LOCAL_CLUSTER); + assertThat(localCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL)); + var remoteCluster = executionInfo.getCluster(REMOTE_CLUSTER_2); + assertThat(localCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL)); + remoteCluster = executionInfo.getCluster(REMOTE_CLUSTER_1); + // This is successful, given the index does not exist on remote-1 but exists on remote-2, is this as expected? + assertThat(remoteCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL)); + assertThat(remoteCluster.getFailures(), empty()); + } + + // The subquery does not have any index matching the index pattern, + // the query should fail regardless skipUnavailable=true or false, + // index resolution in Analyzer will fail on the branch anyway, it does not prune the branch with missing indices + // TODO think about how to prune branches with no matching indices + var ex = expectThrows(VerificationException.class, () -> runQuery(""" + FROM (FROM c*:remote* metadata _index),(FROM r*:remote* metadata _index) metadata _index + | STATS c = count(*) by _index + | SORT _index + """, randomBoolean())); + assertThat(ex.getMessage(), containsString("Unknown index [c*:remote*]")); + } + } finally { + clearSkipUnavailable(3); + } + } + + public void testSubqueryWithMissingLocalIndex() throws IOException { + setupClusters(3); + populateIndex(REMOTE_CLUSTER_1, "remote_idx", randomIntBetween(1, 5), 5); + populateIndex(REMOTE_CLUSTER_2, "remote_idx", randomIntBetween(1, 5), 5); + + // no local index exists, the query should fail regardless skipUnavailable=true or false, + // index resolution in Analyzer will fail on the branch anyway, it does not prune the branch with missing indices + var ex = expectThrows(VerificationException.class, () -> runQuery(""" + FROM local*,(FROM *:remote* metadata _index) metadata _index + | STATS c = count(*) by _index + | SORT _index + """, randomBoolean())); + assertThat(ex.getMessage(), containsString("Unknown index [local*]")); + + ex = expectThrows(VerificationException.class, () -> runQuery(""" + FROM (FROM local* metadata _index),(FROM *:remote* metadata _index) metadata _index + | STATS c = count(*) by _index + | SORT _index + """, randomBoolean())); + assertThat(ex.getMessage(), containsString("Unknown index [local*]")); + + try (EsqlQueryResponse resp = runQuery(""" + FROM *,(FROM *:* metadata _index) metadata _index + | STATS c = count(*) by _index + | SORT _index + """, randomBoolean())) { + var columns = resp.columns().stream().map(ColumnInfoImpl::name).toList(); + assertThat(columns, hasItems("c", "_index")); + + List> values = getValuesList(resp); + assertThat(values, hasSize(5)); + List> expected = List.of( + List.of(10L, "cluster-a:logs-2"), + List.of(5L, "cluster-a:remote_idx"), + List.of(10L, "logs-1"), + List.of(10L, "remote-b:logs-2"), + List.of(5L, "remote-b:remote_idx") + ); + assertEquals(expected, values); + + EsqlExecutionInfo executionInfo = resp.getExecutionInfo(); + + var localCluster = executionInfo.getCluster(LOCAL_CLUSTER); + assertThat(localCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL)); + var remoteCluster = executionInfo.getCluster(REMOTE_CLUSTER_2); + assertThat(localCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL)); + remoteCluster = executionInfo.getCluster(REMOTE_CLUSTER_1); + // This is successful, given the index does not exist on remote-1 but exists on remote-2, is this as expected? + assertThat(remoteCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL)); + assertThat(remoteCluster.getFailures(), empty()); + } + } + + public void testSubqueryWithFilter() throws IOException { + setupClusters(3); + + try (EsqlQueryResponse resp = runQuery(""" + FROM + (FROM logs-* metadata _index | where v > 5), + (FROM *:logs-* metadata _index | where v >1) + | WHERE v < 7 + | DROP const, id + | SORT _index, v + """, randomBoolean())) { + var columns = resp.columns().stream().map(ColumnInfoImpl::name).toList(); + assertThat(columns, hasItems("tag", "v", "_index")); + + List> values = getValuesList(resp); + List> expected = List.of( + List.of("remote", 4L, "cluster-a:logs-2"), + List.of("local", 6L, "logs-1"), + List.of("remote", 4L, "remote-b:logs-2") + ); + assertTrue(values.equals(expected)); + + EsqlExecutionInfo executionInfo = resp.getExecutionInfo(); + assertCCSExecutionInfoDetails(executionInfo); + } + } + + public void testSubqueryWithFullTextFunctionInFilter() throws IOException { + setupClusters(3); + + try (EsqlQueryResponse resp = runQuery(""" + FROM + logs-*, + (FROM *:logs-* metadata _index | where v < 4 ) + metadata _index + | WHERE tag:"remote" + | DROP const, id + | SORT _index, v + """, randomBoolean())) { + var columns = resp.columns().stream().map(ColumnInfoImpl::name).toList(); + assertThat(columns, hasItems("tag", "v", "_index")); + + List> values = getValuesList(resp); + List> expected = List.of( + List.of("remote", 0L, "cluster-a:logs-2"), + List.of("remote", 1L, "cluster-a:logs-2"), + List.of("remote", 0L, "remote-b:logs-2"), + List.of("remote", 1L, "remote-b:logs-2") + ); + assertTrue(values.equals(expected)); + + EsqlExecutionInfo executionInfo = resp.getExecutionInfo(); + assertCCSExecutionInfoDetails(executionInfo); + } + } + + public void testSubqueryWithStats() throws IOException { + setupClusters(3); + + try (EsqlQueryResponse resp = runQuery(""" + FROM + logs-*, + (FROM *:logs-* | where v < 4 ) + | WHERE v < 5 + | STATS sum = sum(v) BY tag + | SORT tag + """, randomBoolean())) { + var columns = resp.columns().stream().map(ColumnInfoImpl::name).toList(); + assertThat(columns, hasItems("tag", "sum")); + + List> values = getValuesList(resp); + List> expected = List.of(List.of(10L, "local"), List.of(2L, "remote")); + assertTrue(values.equals(expected)); + + EsqlExecutionInfo executionInfo = resp.getExecutionInfo(); + assertCCSExecutionInfoDetails(executionInfo); + } + } + + public void testSubqueryWithLookupJoin() throws IOException { + setupClusters(3); + populateLookupIndex(LOCAL_CLUSTER, "values_lookup", 10); + populateLookupIndex(REMOTE_CLUSTER_1, "values_lookup", 10); + populateLookupIndex(REMOTE_CLUSTER_2, "values_lookup", 10); + + // lookup join inside subquery is supported + try (EsqlQueryResponse resp = runQuery(""" + FROM + (FROM logs-* metadata _index | where v > 5 | LOOKUP JOIN values_lookup on v == lookup_key), + (FROM *:logs-* metadata _index | where v >1 | LOOKUP JOIN values_lookup on v == lookup_key) + | WHERE v < 7 + | KEEP tag, v, _index, lookup_tag + | SORT _index, v + """, randomBoolean())) { + var columns = resp.columns().stream().map(ColumnInfoImpl::name).toList(); + assertThat(columns, hasItems("tag", "v", "_index", "lookup_tag")); + + List> values = getValuesList(resp); + List> expected = List.of( + List.of("remote", 4L, "cluster-a:logs-2", "cluster-a"), + List.of("local", 6L, "logs-1", "local"), + List.of("remote", 4L, "remote-b:logs-2", "remote-b") + ); + assertTrue(values.equals(expected)); + + EsqlExecutionInfo executionInfo = resp.getExecutionInfo(); + assertCCSExecutionInfoDetails(executionInfo); + } + + // lookup join in main query after subqueries is not supported yet, because there is remote index pattern and limit + // TODO remove the limit added for subqueries + VerificationException ex = expectThrows(VerificationException.class, () -> runQuery(""" + FROM logs-*,(FROM c*:logs-*), (FROM r*:logs-*) + | LOOKUP JOIN values_lookup on v == lookup_key + """, randomBoolean())); + assertThat( + ex.getMessage(), + containsString("LOOKUP JOIN with remote indices can't be executed after [FROM logs-*,(FROM c*:logs-*), (FROM r*:logs-*)]") + ); + } + + public void testSubqueryWithInlineStatsInSubquery() throws IOException { + setupClusters(3); + + // inline stats inside subquery is supported + try (EsqlQueryResponse resp = runQuery(""" + FROM + logs-*, + (FROM *:logs-* | INLINE STATS sum = sum(v) WHERE v < 10 ) + | WHERE v < 4 + | DROP const, id + | SORT tag, v + """, randomBoolean())) { + var columns = resp.columns().stream().map(ColumnInfoImpl::name).toList(); + assertThat(columns, hasItems("tag", "sum", "v")); + int tagIndex = columns.indexOf("tag"); + int vIndex = columns.indexOf("v"); + int sumIndex = columns.indexOf("sum"); + List> values = getValuesList(resp); + for (int i = 0; i < values.size(); i++) { + var row = values.get(i); + if (i < 4) { + assertNull(row.get(sumIndex)); + assertEquals("local", row.get(tagIndex)); + assertEquals((long) i, row.get(vIndex)); + } else { + assertEquals(28L, row.get(sumIndex)); + assertEquals("remote", row.get(tagIndex)); + assertEquals(i > 5 ? 1L : 0L, row.get(vIndex)); + } + } + + EsqlExecutionInfo executionInfo = resp.getExecutionInfo(); + assertCCSExecutionInfoDetails(executionInfo); + } + + // inline stats in main query after subqueries is not supported yet, because there is limit + // TODO remove the limit added for subqueries + VerificationException ex = expectThrows(VerificationException.class, () -> runQuery(""" + FROM logs-*,(FROM c*:logs-*), (FROM r*:logs-*) + | INLINE STATS sum = sum(v) + """, randomBoolean())); + String errorMessage = ex.getMessage(); + assertThat( + errorMessage, + containsString( + "INLINE STATS after subquery is not supported, " + + "as INLINE STATS cannot be used after an explicit or implicit LIMIT command" + ) + ); + assertThat( + errorMessage, + containsString( + "INLINE STATS cannot be used after an explicit or implicit LIMIT command, " + + "but was [INLINE STATS sum = sum(v)] after [FROM logs-*,(FROM c*:logs-*), (FROM r*:logs-*)]" + ) + ); + } + + public void testSubqueryWithFilterInRequest() throws IOException { + setupClusters(3); + + EsqlQueryRequest request = randomBoolean() ? EsqlQueryRequest.asyncEsqlQueryRequest() : EsqlQueryRequest.syncEsqlQueryRequest(); + request.query(""" + FROM + (FROM logs-* metadata _index | where v > 5), + (FROM *:logs-* metadata _index | where v >1) + | DROP const, id + | SORT _index, v + """); + request.pragmas(AbstractEsqlIntegTestCase.randomPragmas()); + request.profile(randomInt(5) == 2); + request.columnar(randomBoolean()); + request.includeCCSMetadata(randomBoolean()); + request.filter(new RangeQueryBuilder("v").lt(7)); + request.waitForCompletionTimeout(timeValueSeconds(30)); + + try (EsqlQueryResponse resp = runQuery(request)) { + var columns = resp.columns().stream().map(ColumnInfoImpl::name).toList(); + assertThat(columns, hasItems("tag", "v", "_index")); + + List> values = getValuesList(resp); + List> expected = List.of( + List.of("remote", 4L, "cluster-a:logs-2"), + List.of("local", 6L, "logs-1"), + List.of("remote", 4L, "remote-b:logs-2") + ); + assertTrue(values.equals(expected)); + + EsqlExecutionInfo executionInfo = resp.getExecutionInfo(); + assertCCSExecutionInfoDetails(executionInfo); + } + } + + public void testNestedSubqueries() throws IOException { + setupClusters(3); + + // nested subqueries are not supported yet + VerificationException ex = expectThrows(VerificationException.class, () -> runQuery(""" + FROM logs-*,(FROM c*:logs-*, (FROM r*:logs-*)) + """, randomBoolean())); + assertThat(ex.getMessage(), containsString("Nested subqueries are not supported")); + } + + public void testSubqueryWithFork() throws IOException { + setupClusters(3); + + // fork after subqueries is not supported yet + VerificationException ex = expectThrows(VerificationException.class, () -> runQuery(""" + FROM logs-*,(FROM c*:logs-*), (FROM r*:logs-*) + | FORK + (WHERE v > 5) + (WHERE v < 3) + """, randomBoolean())); + assertThat(ex.getMessage(), containsString("FORK after subquery is not supported")); + + // fork inside subquery is not supported yet + ex = expectThrows(VerificationException.class, () -> runQuery(""" + FROM + logs-*, + (FROM c*:logs-*), + (FROM r*:logs-* + | FORK (WHERE v > 5) (WHERE v < 3)) + """, randomBoolean())); + assertThat(ex.getMessage(), containsString("FORK inside subquery is not supported")); + } + + private static void assertCCSExecutionInfoDetails(EsqlExecutionInfo executionInfo) { + assertNotNull(executionInfo); + assertThat(executionInfo.overallTook().millis(), greaterThanOrEqualTo(0L)); + assertTrue(executionInfo.isCrossClusterSearch()); + List clusters = executionInfo.clusterAliases().stream().map(executionInfo::getCluster).toList(); + + for (EsqlExecutionInfo.Cluster cluster : clusters) { + assertThat(cluster.getTook().millis(), greaterThanOrEqualTo(0L)); + assertThat(cluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL)); + assertThat(cluster.getSkippedShards(), equalTo(0)); + assertThat(cluster.getFailedShards(), equalTo(0)); + } + } + + protected void setupAlias(String clusterAlias, String indexName, String aliasName) { + Client client = client(clusterAlias); + IndicesAliasesRequestBuilder indicesAliasesRequestBuilder = client.admin() + .indices() + .prepareAliases(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT) + .addAliasAction(IndicesAliasesRequest.AliasActions.add().index(indexName).alias(aliasName)); + assertAcked(client.admin().indices().aliases(indicesAliasesRequestBuilder.request())); + } +} diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterSubqueryUnavailableRemotesIT.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterSubqueryUnavailableRemotesIT.java new file mode 100644 index 0000000000000..dca7128acaad1 --- /dev/null +++ b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterSubqueryUnavailableRemotesIT.java @@ -0,0 +1,246 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the Elastic License + * 2.0; you may not use this file except in compliance with the Elastic License + * 2.0. + */ + +package org.elasticsearch.xpack.esql.action; + +import org.elasticsearch.ElasticsearchException; +import org.elasticsearch.ExceptionsHelper; +import org.elasticsearch.test.transport.MockTransportService; +import org.elasticsearch.transport.TransportChannel; +import org.elasticsearch.transport.TransportResponse; +import org.elasticsearch.transport.TransportService; +import org.junit.Before; + +import java.io.IOException; +import java.util.List; + +import static org.elasticsearch.xpack.esql.EsqlTestUtils.getValuesList; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.empty; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.hasItems; +import static org.hamcrest.Matchers.hasSize; +import static org.hamcrest.Matchers.not; + +//@TestLogging(value = "org.elasticsearch.xpack.esql.session:DEBUG", reason = "to better understand error handling") +public class CrossClusterSubqueryUnavailableRemotesIT extends AbstractCrossClusterTestCase { + + @Before + public void checkSubqueryInFromCommandSupport() { + assumeTrue("Requires subquery in FROM command support", EsqlCapabilities.Cap.SUBQUERY_IN_FROM_COMMAND.isEnabled()); + } + + @Override + protected boolean reuseClusters() { + return false; + } + + /* + * Test that when skip_unavailable is true, and a remote cluster fails to respond, + * the query does not error out and data is returned from the other clusters. + */ + public void testSubqueryFailToReceiveClusterResponseWithSkipUnavailableTrue() throws IOException { + setupClusters(3); + setSkipUnavailable(REMOTE_CLUSTER_1, true); + setSkipUnavailable(REMOTE_CLUSTER_2, true); + Exception simulatedFailure = mockTransportServiceToRecevieSimulatedFailure(); + + try { + try (EsqlQueryResponse resp = runQuery(""" + FROM logs-*,(FROM *:logs-*) + | KEEP v, tag + """, randomBoolean())) { + var columns = resp.columns().stream().map(ColumnInfoImpl::name).toList(); + assertThat(columns, hasItems("v", "tag")); + + List> values = getValuesList(resp); + assertThat(values, hasSize(20)); + EsqlExecutionInfo executionInfo = resp.getExecutionInfo(); + + var localCluster = executionInfo.getCluster(LOCAL_CLUSTER); + assertThat(localCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL)); + var remoteCluster = executionInfo.getCluster(REMOTE_CLUSTER_1); + assertThat(remoteCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SKIPPED)); + var remoteCluster2 = executionInfo.getCluster(REMOTE_CLUSTER_2); + assertThat(remoteCluster2.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL)); + assertThat(remoteCluster.getFailures(), not(empty())); + var failure = remoteCluster.getFailures().get(0); + assertThat(failure.reason(), containsString(simulatedFailure.getMessage())); + } + + // No remote index can be found in a subquery, the query should fail regardless skipUnavailable=true or false, + // index resolution in Analyzer will fail on the branch anyway, it does not prune the branch with missing indices. + // TODO think about how to prune branches with no matching indices + Exception ex = expectThrows(Exception.class, () -> runQuery(""" + FROM logs-*,(FROM c*:logs-*), (FROM r*:logs-*) + | KEEP v, tag + """, randomBoolean())); + String message = ex.getCause() == null ? ex.getMessage() : ex.getCause().getMessage(); + assertThat(message, containsString(simulatedFailure.getMessage())); + } finally { + for (TransportService transportService : cluster(REMOTE_CLUSTER_1).getInstances(TransportService.class)) { + MockTransportService ts = asInstanceOf(MockTransportService.class, transportService); + ts.clearAllRules(); + } + } + } + + /* + * Test that when skip_unavailable is false, and a remote cluster fails to respond, the query fails. + */ + public void testSubqueryFailToReceiveClusterResponseWithSkipUnavailableFalse() throws IOException { + setupClusters(3); + setSkipUnavailable(REMOTE_CLUSTER_1, false); + setSkipUnavailable(REMOTE_CLUSTER_2, false); + Exception simulatedFailure = mockTransportServiceToRecevieSimulatedFailure(); + + try { + // with local indices + Exception ex = expectThrows(Exception.class, () -> runQuery(""" + FROM logs-*,(FROM c*:logs-*), (FROM r*:logs-*) + | KEEP v, tag + """, randomBoolean())); + String message = ex.getCause() == null ? ex.getMessage() : ex.getCause().getMessage(); + assertThat(message, containsString(simulatedFailure.getMessage())); + + // without local indices + ex = expectThrows(Exception.class, () -> runQuery(""" + FROM (FROM c*:logs-*), (FROM r*:logs-*) + | KEEP v, tag + """, randomBoolean())); + message = ex.getCause() == null ? ex.getMessage() : ex.getCause().getMessage(); + assertThat(message, containsString(simulatedFailure.getMessage())); + } finally { + for (TransportService transportService : cluster(REMOTE_CLUSTER_1).getInstances(TransportService.class)) { + MockTransportService ts = asInstanceOf(MockTransportService.class, transportService); + ts.clearAllRules(); + } + } + } + + /* + * Test that when skip_unavailable is true, a disconnected remote cluster is skipped. + */ + public void testSubqueryWithDisconnectedRemoteClusterWithSkipUnavailableTrue() throws IOException { + setupClusters(3); + setSkipUnavailable(REMOTE_CLUSTER_1, true); + setSkipUnavailable(REMOTE_CLUSTER_2, true); + try { + // close the remote cluster so that it is unavailable + cluster(REMOTE_CLUSTER_1).close(); + + try (EsqlQueryResponse resp = runQuery(""" + FROM logs-*,(FROM *:logs-*) + | KEEP v, tag + """, randomBoolean())) { + var columns = resp.columns().stream().map(ColumnInfoImpl::name).toList(); + assertThat(columns, hasItems("v", "tag")); + List> values = getValuesList(resp); + assertThat(values, hasSize(20)); + EsqlExecutionInfo executionInfo = resp.getExecutionInfo(); + + var remoteCluster = executionInfo.getCluster(REMOTE_CLUSTER_1); + assertThat(remoteCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SKIPPED)); + assertThat(remoteCluster.getFailures(), not(empty())); + var failure = remoteCluster.getFailures().get(0); + assertThat( + failure.reason(), + containsString("Remote cluster [cluster-a] (with setting skip_unavailable=true) is not available") + ); + } + + // This is how it behaves today, if the remote cluster that is referenced by a subquery is not available, + // and there is no other cluster available in the same subquery, then no data is returned, no exception is thrown. + // This seems make sense, as Analyzer will fail this query anyway, because the indices on remate-1 cannot be found. + try (EsqlQueryResponse resp = runQuery(""" + FROM logs-*,(FROM c*:logs-*), (FROM r*:logs-*) + | KEEP v, tag + """, randomBoolean())) { + var columns = resp.columns().stream().map(ColumnInfoImpl::name).toList(); + assertThat(columns, hasItems("")); + + List> values = getValuesList(resp); + assertThat(values, hasSize(0)); + EsqlExecutionInfo executionInfo = resp.getExecutionInfo(); + + var localCluster = executionInfo.getCluster(LOCAL_CLUSTER); + assertThat(localCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL)); + var remoteCluster = executionInfo.getCluster(REMOTE_CLUSTER_1); + assertThat(remoteCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SKIPPED)); + assertThat(remoteCluster.getFailures(), not(empty())); + var failure = remoteCluster.getFailures().get(0); + assertThat(failure.reason(), containsString("unable to connect to remote cluster")); + } + } finally { + clearSkipUnavailable(3); + } + } + + /* + * Test that when skip_unavailable is false, a disconnected remote cluster causes the query to fail. + */ + public void testSubqueryWithDisconnectedRemoteClusterWithSkipUnavailableFalse() throws IOException { + setupClusters(3); + setSkipUnavailable(REMOTE_CLUSTER_1, false); + setSkipUnavailable(REMOTE_CLUSTER_2, false); + + try { + // close the remote cluster so that it is unavailable + cluster(REMOTE_CLUSTER_1).close(); + + Exception ex = expectThrows(ElasticsearchException.class, () -> runQuery(""" + FROM logs-*,(FROM *:logs-*) + | KEEP v, tag + """, randomBoolean())); + assertTrue(ExceptionsHelper.isRemoteUnavailableException(ex)); + + ex = expectThrows(ElasticsearchException.class, () -> runQuery(""" + FROM logs-*,(FROM c*:logs-*), (FROM r*:logs-*) + | KEEP v, tag + """, randomBoolean())); + assertTrue(ExceptionsHelper.isRemoteUnavailableException(ex)); + + ex = expectThrows(ElasticsearchException.class, () -> runQuery(""" + FROM (FROM c*:logs-*), (FROM r*:logs-*) + | KEEP v, tag + """, randomBoolean())); + assertTrue(ExceptionsHelper.isRemoteUnavailableException(ex)); + } finally { + clearSkipUnavailable(3); + } + } + + /** + * Mocks a bad REMOTE_CLUSTER_1. + */ + private Exception mockTransportServiceToRecevieSimulatedFailure() { + Exception simulatedFailure = randomFailure(); + for (TransportService transportService : cluster(REMOTE_CLUSTER_1).getInstances(TransportService.class)) { + MockTransportService ts = asInstanceOf(MockTransportService.class, transportService); + ts.addRequestHandlingBehavior( + EsqlResolveFieldsAction.RESOLVE_REMOTE_TYPE.name(), + (handler, request, channel, task) -> handler.messageReceived(request, new TransportChannel() { + @Override + public String getProfileName() { + return channel.getProfileName(); + } + + @Override + public void sendResponse(TransportResponse response) { + sendResponse(simulatedFailure); + } + + @Override + public void sendResponse(Exception exception) { + channel.sendResponse(exception); + } + }, task) + ); + } + return simulatedFailure; + } +}