ES|QL: Support LOOKUP JOIN with FORK by ioanatia · Pull Request #128839 · elastic/elasticsearch

ioanatia · 2025-06-03T15:26:04Z

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 · 2025-06-04T07:49:23Z

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

ChrisHegarty

LGTM

tteofili

LGTM

Fix ProjectAwayColumns for FORK

5442608

ioanatia added >non-issue :Analytics/ES|QL AKA ESQL v9.1.0 Team:Search - Relevance The Search organization Search Relevance team labels Jun 3, 2025

ioanatia changed the title ~~Support LOOKUP JOIN with FORK~~ ES|QL: Support LOOKUP JOIN with FORK Jun 3, 2025

Merge branch 'main' into fork_lookup_join

c23f9b6

ioanatia mentioned this pull request Jun 3, 2025

Initial FORK with restrictions #121950

Closed

23 tasks

ioanatia requested review from ChrisHegarty, carlosdelest, fang-xing-esql and tteofili June 4, 2025 07:48

ioanatia marked this pull request as ready for review June 4, 2025 07:48

elasticsearchmachine added Team:Analytics Meta label for analytical engine team (ESQL/Aggs/Geo) and removed Team:Search - Relevance The Search organization Search Relevance team labels Jun 4, 2025

ChrisHegarty approved these changes Jun 4, 2025

View reviewed changes

tteofili approved these changes Jun 4, 2025

View reviewed changes

carlosdelest approved these changes Jun 4, 2025

View reviewed changes

Merge branch 'main' into fork_lookup_join

ce67cb6

ioanatia merged commit ca904ce into elastic:main Jun 4, 2025
18 checks passed

ioanatia deleted the fork_lookup_join branch June 4, 2025 15:50

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ES|QL: Support LOOKUP JOIN with FORK#128839

ES|QL: Support LOOKUP JOIN with FORK#128839
ioanatia merged 3 commits intoelastic:mainfrom
ioanatia:fork_lookup_join

ioanatia commented Jun 3, 2025 •

edited

Loading

Uh oh!

elasticsearchmachine commented Jun 4, 2025

Uh oh!

ChrisHegarty left a comment

Uh oh!

tteofili left a comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

	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();

	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);
	});

	// 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);
	}

Conversation

ioanatia commented Jun 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

elasticsearchmachine commented Jun 4, 2025

Uh oh!

ChrisHegarty left a comment

Choose a reason for hiding this comment

Uh oh!

tteofili left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

ioanatia commented Jun 3, 2025 •

edited

Loading