ES|QL: Allow FORK with remote indices

[ioanatia]

related #121652

We initially skipped CCS tests (PR) after we merged streaming support for FORK because the tests were flaky. In theory, given the FORK's execution model, queries using FORK with remote indices should just work.

This was back in May and since then CCS has stabilized for the GA release and we no longer see these failures.
I am not sure whether there was a single fix that enabled us to enable querying remote indices with FORK.
There are probably more separate changes (#127328, #127328 etc) that compounded to this moment where we can support FORK with CCS.

[elasticsearchmachine]

Pinging @elastic/es-analytical-engine (Team:Analytics)

[elasticsearchmachine]

Pinging @elastic/es-search-relevance (Team:Search Relevance)

[ioanatia]

got another failure after merging main - cannot replicate locally 😞

seems to be the same type of issue as before with results not coming back:

java.lang.AssertionError: Expected more data but no more entries found after [3] |  
-- | --
  | Data mismatch: |  
  |   |  
  | Actual: |  
  | _fork:keyword \| emp_no:integer \| x:keyword \| y:keyword \| z:keyword \| a:keyword |  
  | fork1         \| 10048          \| Florian   \| 10048     \| Syrotiuk  \| Florian 10048 Syrotiuk |  
  | fork1         \| 10081          \| Zhongwei  \| 10081     \| Rosen     \| Zhongwei 10081 Rosen |  
  | fork2         \| null           \| 2         \| 10081     \| 10048     \| null |  
  |   |  
  | Expected: |  
  | _fork:keyword \| emp_no:integer \| x:keyword \| y:keyword \| z:keyword \| a:keyword |  
  | fork1         \| 10048          \| Florian   \| 10048     \| Syrotiuk  \| Florian 10048 Syrotiuk |  
  | fork1         \| 10081          \| Zhongwei  \| 10081     \| Rosen     \| Zhongwei 10081 Rosen |  
  | fork2         \| null           \| 2         \| 10081     \| 10048     \| null |  
  | fork3         \| 10048          \| Syrotiuk  \| null      \| null      \| null |  
  | fork3         \| 10081          \| Rosen     \| null      \| null      \| null |  
  | fork4         \| 10048          \| abc       \| aaa       \| null      \| null |  
  | fork4         \| 10081          \| abc       \| aaa       \| null      \| null |  
  |  

putting this back in draft until I figure it out

[idegtiarenko]

👍
Looking forward for this stabilized and merged

[ioanatia]

all the tests that were querying the languages index failed - with rows being duplicated:

Actual: |  
-- | --
language_name:keyword \| x:integer \| _fork:keyword |  
English               \| 1         \| fork1 |  
English               \| 1         \| fork1 |  
French                \| null      \| fork2 |  
French                \| null      \| fork2 |  
German                \| null      \| fork2 |  
German                \| null      \| fork2 |  
Spanish               \| null      \| fork2 |  
Spanish               \| null      \| fork2 |  
  |  
Expected: |  
language_name:keyword \| x:integer \| _fork:keyword |  
English               \| 1         \| fork1 |  
French                \| null      \| fork2 |  
German                \| null      \| fork2 |  
Spanish               \| null      \| fork2

debugging these test showed that we were querying both the local and remote index, which is why we had duplicate results.

languages was an index that was primarily used for testing lookup join for which we had some special handling such that we disallow duplicates:

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

Lines 298 to 302 in b1081c0

	if (Arrays.stream(localIndices).anyMatch(i -> LOOKUP_INDICES.contains(i.trim().toLowerCase(Locale.ROOT)))) {
	// If the query contains lookup indices, use only remotes to avoid duplication
	onlyRemotes = true;
	}
	final boolean onlyRemotesFinal = onlyRemotes;

languages is not part of LOOKUP_INDICES (but it should be?)
anyway, I thought that it might be better to rewrite these tests with an index that has no special handling - so I used employees here instead of languages.

[ioanatia]

we could have made ENABLE_FORK_FOR_REMOTE_INDICES capability available only in snapshots and use it here - such that CCS support is only available in snapshots - but I don't think we need to 🤔

[smalyshev]

Is it still true that cross-cluster FORK requires coordinator-side aggregation? I'm not sure I understand the mechanism why.

[ioanatia]

Is it still true that cross-cluster FORK requires coordinator-side aggregation? I'm not sure I understand the mechanism why.

I am not sure what you mean by coordinator-side aggregation.
A query using FORK gets split into multiple plans - a subplan for each branch and a main coordinator plan.
Each subplan that pertains to a FORK branch gets split into a data node plan and coordinator plan.

For example:

FROM test
| FORK
    ( WHERE content:"fox" ) // sub plan 1
    ( WHERE content:"dog" ) // sub plan 2
| SORT _fork
| KEEP _fork, id, content

gets translated to something like this:

It was a conscious decision to execute FORK this way and one of the benefits was that we did not need to make any special concessions for CCS support.
Anyway, this PR does not make any fundamental changes in how FORK gets planned - we simply enable CCS support here.

[ioanatia]

Is it still true that cross-cluster FORK requires coordinator-side aggregation? I'm not sure I understand the mechanism why.

Checked with @smalyshev - this is about having FORK marked as ExecutesOn.Coordinator, similar to STATS:

elasticsearch/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/Fork.java

Line 36 in ebb94bd

public class Fork extends LogicalPlan implements PostAnalysisPlanVerificationAware, TelemetryAware, ExecutesOn.Coordinator {

We agreed that for the beginning this is fine, even though it puts some restrictions in place such as we cannot execute a remote lookup join after FORK.
We will look into removing this restriction in the future - e.g. if we can push more plans that come after FORK to the subplans etc.

[leemthompo]

@ioanatia I guess can update the docs to mention that the the limitation disappears in 9.2.0

elasticsearch/docs/reference/query-languages/esql/_snippets/commands/layout/fork.md

Line 42 in 208cd45

- Using remote cluster references and `FORK` is not supported.

[dnhatn]

Looks good. Thanks @ioanatia

[ioanatia]

@ioanatia I guess can update the docs to mention that the the limitation disappears in 9.2.0

I will get that addressed too. @leemthompo