ES|QL: Support STATS after FORK

[ioanatia]

related: #121950

It was reported that using STATS immediately after FORK results in an error.

Example:

FROM employees | FORK (EVAL x = 1) ( EVAL y = 1) | stats max(salary)

ends up with

"stack_trace": "org.elasticsearch.ElasticsearchException$1: Cannot invoke \"org.elasticsearch.xpack.esql.planner.Layout$ChannelAndType.channel()\" because the return value of \"org.elasticsearch.xpack.esql.planner.Layout.get(org.elasticsearch.xpack.esql.core.expression.NameId)\" is null

The problem has to do with how we set the aggregatorMode in AbstractPhysicalOperationProviders.javagroupingPhysicalOperation. We just assume that if the AggregateExec follows a ExchangeSourceExec, we should set the aggregatorMode to INTERMEDIARY because we are on the data node reduce step.
With the introduction of FORK and #126389 this is no longer always true, so we explicitly check if the ExchangeSourceExec is indeed between aggregations.

[carlosdelest]

LGTM 👍

[elasticsearchmachine]

Hi @ioanatia, I've created a changelog YAML for you.

[elasticsearchmachine]

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

[elasticsearchmachine]

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

[ioanatia]

I am putting back the non-issue label, because we don't want a release note for this particular PR.
FORK is still in snapshot and release notes are used in user facing docs.
I don't think we need to document that we fixed a bug for a feature that was not released yet.

[fang-xing-esql]

This change looks fine to me, but I'd like to get another pair of eyes from @alex-spies to double check if the aggregatorMode is set properly, as this was introduced by Alex originally.

[ioanatia]

the reason why we added this check is because a query like:

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

will now be split into multiple sub plans using ExchangeSourceExec and ExchangeSinkExec:

and so we need to explicitly check with what type of ExchangeSourceExec we are dealing with.