ES|QL: Support LOOKUP JOIN with FORK

[ioanatia]

related: #121950

This removes a TODO for fixing ProjectAwayColumns for FORK, which also fixes support for using LOOKUP JOIN and FORK.

It was reported that when we using LOOKUP JOIN in combination with FORK, this results in an error:

Example:

FROM employees
| EVAL language_code = languages
| LOOKUP JOIN languages_lookup ON language_code
| FORK (EVAL x = 1) ( EVAL y = 1)
| keep l*

results in org.elasticsearch.xpack.esql.plan.physical.ProjectExec cannot be cast to class org.elasticsearch.xpack.esql.plan.physical.EsQueryExec

This is raised here in LocalExecutionPlanner#L702:

elasticsearch/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/LocalExecutionPlanner.java

Lines 694 to 702 in d3d2d9b

	private PhysicalOperation planLookupJoin(LookupJoinExec join, LocalExecutionPlannerContext context) {
	PhysicalOperation source = plan(join.left(), context);
	Layout.Builder layoutBuilder = source.layout.builder();
	for (Attribute f : join.addedFields()) {
	layoutBuilder.append(f);
	}
	Layout layout = layoutBuilder.build();
	EsQueryExec localSourceExec = (EsQueryExec) join.lookup();

Normally, the right child of LookUpJoinExec is a FragmentExec that wraps a EsRelation.
Then this FragmentExec gets mapped to a EsQueryExec.

However, when using FORK, when we split the plan into sub plans, we add a Project node inside each FragmentExec which causes issues when using LOOKUP JOIN in combination with FORK.

Notice the TODO:

elasticsearch/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java

Lines 83 to 92 in d3d2d9b

	public static Tuple<List<PhysicalPlan>, PhysicalPlan> breakPlanIntoSubPlansAndMainPlan(PhysicalPlan plan) {
	var subplans = new Holder<List<PhysicalPlan>>();
	PhysicalPlan mainPlan = plan.transformUp(MergeExec.class, me -> {
	subplans.set(me.children().stream().map(child -> {
	// TODO: we are adding a Project plan to force InsertFieldExtraction - we should remove this transformation
	child = child.transformUp(FragmentExec.class, f -> {
	var logicalFragment = f.fragment();
	logicalFragment = new Project(logicalFragment.source(), logicalFragment, logicalFragment.output());
	return new FragmentExec(logicalFragment);
	});

We have done this because we needed to fix ProjectAwayColumns, where we simply skip MergeExec (the physical plan for FORK):

elasticsearch/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/physical/ProjectAwayColumns.java

Lines 47 to 51 in d3d2d9b

	// This will require updating should we choose to have non-unary execution plans in the future.
	return plan.transformDown(currentPlanNode -> {
	if (currentPlanNode instanceof MergeExec) {
	keepTraversing.set(FALSE);
	}

The fix was therefore to remove the hack from PlannerUtils.breakPlanIntoSubPlansAndMainPlan and to make sure that ProjectAwayColumns works correctly when FORK is used.

[elasticsearchmachine]

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

[ChrisHegarty]

LGTM

[tteofili]

LGTM