java.lang.ClassCastException when combining LOOKUP JOIN and remote ENRICH

[smalyshev]

Elasticsearch Version

main/9.1

Installed Plugins

No response

Java Version

bundled

OS Version

Problem Description

The LOOKUP JOIN query fails when the lookup operator is combined with ENRICH.

Steps to Reproduce

The query is:

FROM test191 | EVAL language_code = "1" | LOOKUP JOIN test-lookup191 ON color.keyword | ENRICH _remote:lang_r 

Where test191 is the test index from 191_lookup_join_text.yml and test-lookup191 is the lookup index from the same test and lang_r is enrich policy with language_code and the key.

This produces the following exception on main:

2025-06-12T16:03:09,790][WARN ][o.e.x.e.a.EsqlResponseListener] [node-1] ESQL request failed with status [INTERNAL_SERVER_ERROR]: java.lang.ClassCastException: class org.elasticsearch.xpack.esql.plan.physical.ProjectExec cannot be cast to class org.elasticsearch.xpack.esql.plan.physical.EsQueryExec (org.elasticsearch.xpack.esql.plan.physical.ProjectExec and org.elasticsearch.xpack.esql.plan.physical.EsQueryExec are in unnamed module of loader java.net.URLClassLoader @357c9bd9)
	at org.elasticsearch.xpack.esql.planner.LocalExecutionPlanner.planLookupJoin(LocalExecutionPlanner.java:719)
	at org.elasticsearch.xpack.esql.planner.LocalExecutionPlanner.plan(LocalExecutionPlanner.java:294)
	at org.elasticsearch.xpack.esql.planner.LocalExecutionPlanner.planLimit(LocalExecutionPlanner.java:838)
	at org.elasticsearch.xpack.esql.planner.LocalExecutionPlanner.plan(LocalExecutionPlanner.java:259)
	at org.elasticsearch.xpack.esql.planner.LocalExecutionPlanner.planOutput(LocalExecutionPlanner.java:387)
	at org.elasticsearch.xpack.esql.planner.LocalExecutionPlanner.plan(LocalExecutionPlanner.java:298)
	at org.elasticsearch.xpack.esql.planner.LocalExecutionPlanner.plan(LocalExecutionPlanner.java:217)
	at org.elasticsearch.xpack.esql.plugin.ComputeService.runCompute(ComputeService.java:582)
	at org.elasticsearch.xpack.esql.plugin.ComputeService.executePlan(ComputeService.java:406)
	at org.elasticsearch.xpack.esql.plugin.ComputeService.execute(ComputeService.java:198)
	at org.elasticsearch.xpack.esql.plugin.TransportEsqlQueryAction.lambda$innerExecute$3(TransportEsqlQueryAction.java:236)
	at org.elasticsearch.xpack.esql.session.EsqlSession.executeSubPlans(EsqlSession.java:248)
	at org.elasticsearch.xpack.esql.session.EsqlSession.executeOptimizedPlan(EsqlSession.java:213)
	at org.elasticsearch.xpack.esql.session.EsqlSession$1.lambda$onResponse$0(EsqlSession.java:190)
	at org.elasticsearch.server@9.1.0-SNAPSHOT/org.elasticsearch.action.ActionListenerImplementations$ResponseWrappingActionListener.onResponse(ActionListenerImplementations.java:261)
	at org.elasticsearch.xpack.esql.planner.premapper.PreMapper.lambda$preMapper$0(PreMapper.java:33)
	at org.elasticsearch.server@9.1.0-SNAPSHOT/org.elasticsearch.action.ActionListenerImplementations$ResponseWrappingActionListener.onResponse(ActionListenerImplementations.java:261)
	at org.elasticsearch.xpack.esql.expression.function.fulltext.QueryBuilderResolver.resolveQueryBuilders(QueryBuilderResolver.java:50)
	at org.elasticsearch.xpack.esql.planner.premapper.PreMapper.queryRewrite(PreMapper.java:38)
	at org.elasticsearch.xpack.esql.planner.premapper.PreMapper.preMapper(PreMapper.java:31)
	at org.elasticsearch.xpack.esql.session.EsqlSession$1.onResponse(EsqlSession.java:187)
	at org.elasticsearch.xpack.esql.session.EsqlSession$1.onResponse(EsqlSession.java:184)
	at org.elasticsearch.xpack.esql.session.EsqlSession.analyzeAndMaybeRetry(EsqlSession.java:549)

The log with the plans can be seen here: https://gist.github.com/smalyshev/22dea5574e516ef499d37556d6a61d6f

The basic problem IMHO is that there is a problem in processing LookupJoin when it is on the remote side (inside FragmentExec). I have had problems with "Duplicate name ids are not allowed" in Layout.java but when trying to construct the reduced example of this I've also discovered this issue. By default a lot of mapping operations leave LookupJoin to execute on coordinator side of the computation - due to this check:

if (left instanceof FragmentExec fragment) {
                return new FragmentExec(bp);
}

which often fails when FragmentExec is not the immediate child of LookupJoin (same issue we dealt with when working on remote ENRICH) but if I manage to force it on the FragmentExec side, other problems start.

The test above is constructed to test duplicate fields (both test191 and test-lookup191 have color and description fields) but it looks like current code fails even before getting to the remote side.

Logs (if relevant)

https://gist.github.com/smalyshev/22dea5574e516ef499d37556d6a61d6f

[elasticsearchmachine]

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

[alex-spies]

Reproduced this locally. The issue is our mapping to physical plans:

[2025-06-17T19:01:35,864][DEBUG][o.e.x.e.s.EsqlSession    ] [runTask-0] Optimized logicalPlan plan:
Enrich[REMOTE,[6d 79 2d 70 6f 6c 69 63 79][KEYWORD],field{r}#43,{"match":{"indices":[],"match_field":"field","enrich_fie
lds":["otherfield"]}},{=.enrich-my-policy-1750179616163},[otherfield{r}#53]]
\_Limit[1000[INTEGER],true]
  \_Join[LEFT,[field{r}#43],[field{r}#43],[field{f}#48]]
    |_EsqlProject[[field.keyword{f}#47 AS field#43]]
    | \_Limit[1000[INTEGER],false]
    |   \_EsRelation[test][field{f}#46, field.keyword{f}#47]
    \_EsRelation[lu][LOOKUP][field{f}#48, otherfield.keyword{f}#50]

[2025-06-17T19:01:35,864][DEBUG][o.e.x.e.s.EsqlSession    ] [runTask-0] Physical plan:
LimitExec[1000[INTEGER],null]
\_LookupJoinExec[[field{r}#43],[field{f}#48],[otherfield.keyword{f}#50]]
  |_LimitExec[1000[INTEGER],null]
  | \_ExchangeExec[[],false]
  |   \_FragmentExec[filter=null, estimatedRowSize=0, reducer=[], fragment=[<>
Enrich[REMOTE,[6d 79 2d 70 6f 6c 69 63 79][KEYWORD],field{r}#43,{"match":{"indices":[],"match_field":"field","enrich_fie
lds":["otherfield"]}},{=.enrich-my-policy-1750179616163},[otherfield{r}#53]]
\_Limit[1000[INTEGER],true]
  \_Join[LEFT,[field{r}#43],[field{r}#43],[field{f}#48]]
    |_EsqlProject[[field.keyword{f}#47 AS field#43]]
    | \_Limit[1000[INTEGER],false]
    |   \_EsRelation[test][field{f}#46, field.keyword{f}#47]
    \_EsRelation[lu][LOOKUP][field{f}#48, otherfield.keyword{f}#50]<>]]

Note how the lookup join is just duplicated.

[smalyshev]

Is this because the join does not have the same "fragmentexec" hack that the enrich does?
With the hack, I am getting:

[2025-06-17T11:50:24,901][DEBUG][o.e.x.e.s.EsqlSession    ] [node-1] Physical plan:
LimitExec[1000[INTEGER],null]
\_LimitExec[1000[INTEGER],null]
  \_ExchangeExec[[],false]
    \_FragmentExec[filter=null, estimatedRowSize=0, reducer=[], fragment=[<>
Enrich[REMOTE,[6c 61 6e 67 5f 72][KEYWORD],language_code{r}#2,{"match":{"indices":[],"match_field":"language_code","enri
ch_fields":["language_name"]}},{=.enrich-lang_r-1728002375328},[language_name{r}#17]]
\_Limit[1000[INTEGER],true]
  \_Join[LEFT,[color.keyword{f}#7],[color.keyword{f}#7],[color.keyword{f}#11]]
    |_Eval[[[31][KEYWORD] AS language_code#2]]
    | \_Limit[1000[INTEGER],false]
    |   \_EsRelation[test191][color{f}#6, color.keyword{f}#7, description{f}#8, d..]
    \_EsRelation[test-lookup191][LOOKUP][color{f}#10, color.keyword{f}#11, description{f}#12, ..]<>]]

Not sure what's up with limit being there twice, that's a bit weird, but otherwise doesn't look wrong?

[smalyshev]

The hack is basically this: https://github.com/elastic/elasticsearch/pull/129013/files#diff-d5b4b1e3d2bc9f20cf5bd4d49d935997ea6f63ec37a30efd6ca7291bf3be01d4R203 (caution: extremely WIP code, but gives the general idea)

[alex-spies]

It's the opposite: the planning of remote enrich is hacky in itself. We planned to clean that up: #115897

For the time being, I think it's necessary to to just disallow remote enrich after lookup joins, and consider this a current limitation of remote enriches.

[alex-spies]

What's happening: in the mapper, we decide that if there's any fragment upstream from the remote enrich, that fragment should contain the remote enrich. So far so good.

But we only update the fragment inside the already mapped physical plan; we keep all physical plan nodes downstream from the fragment! This has to incur duplications.

It looks roughly like this:

Plan to be mapped:

Enrich[REMOTE,my-policy[KEYWORD],field{f}#52,{"match":{"indices":[],"match_field":"field","enrich_fields":["otherfield"]
}},{=.enrich-my-policy-1750949818391},[otherfield{r}#58 AS $$otherfield$temp_name$59#60]]
\_Limit[1000[INTEGER],true]
  \_Join[LEFT,[field{f}#52],[field{f}#52],[field{f}#53]]
    |_Limit[1000[INTEGER],false]
     ...

physical mapping of the child plan:

LimitExec[1000[INTEGER],null]
\_LookupJoinExec[[field{f}#52],[field{f}#53],[otherfield{f}#54, otherfield.keyword{f}#55]]
  |_LimitExec[1000[INTEGER],null]
  | \_ExchangeExec[[],false]
  |   \_FragmentExec[filter=null, estimatedRowSize=0, reducer=[], fragment=[<>
Limit[1000[INTEGER],false]
  ...

Now we transform the physical mapping of the child plan to place the enrich into the fragment - duplicating the lookup join!

LimitExec[1000[INTEGER],null]
\_LookupJoinExec[[field{f}#52],[field{f}#53],[otherfield{f}#54, otherfield.keyword{f}#55]]
  |_LimitExec[1000[INTEGER],null]
  | \_ExchangeExec[[],false]
  |   \_FragmentExec[filter=null, estimatedRowSize=0, reducer=[], fragment=[<>
Enrich[REMOTE,my-policy[KEYWORD],field{f}#52,{"match":{"indices":[],"match_field":"field","enrich_fields":["otherfield"]
}},{=.enrich-my-policy-1750949818391},[otherfield{r}#58 AS $$otherfield$temp_name$59#60]]
\_Limit[1000[INTEGER],true]
  \_Join[LEFT,[field{f}#52],[field{f}#52],[field{f}#53]]
    |_Limit[1000[INTEGER],false]
     ...

Not to mention that we do this with every fragment we encounter - and in case of lookup joins, there's typically 2 fragments!

[smalyshev]

It looks like this part:

LimitExec[1000[INTEGER],null]
\_LookupJoinExec[[field{f}#52],[field{f}#53],[otherfield{f}#54, otherfield.keyword{f}#55]]
  |_LimitExec[1000[INTEGER],null]
  | \_ExchangeExec[[],false]
  |   \_FragmentExec[filter=null, estimatedRowSize=0, reducer=[], fragment=[<>
Limit[1000[INTEGER],false]

is already not right - why isn't _LookupJoinExec part of the fragment? This means that LOOKUP JOIN happens on the coordinator side, as I understand - but there's no reason for it to do so, it only seems to happen because its mapping is not smart enough to detect there is a _FragmentExec around.

[alex-spies]

is already not right - why isn't _LookupJoinExec part of the fragment?

The limit upstream from the lookup join caused the fragment to break at this place, because we first need to reduce the number of rows to 1000, which requires work to continue on the coordinator. As far as I can see, this behavior is correct - performing the lookup join on the data node would mean it would run over 1000 * n rows, where n is the number of data nodes, rather than just over 1000 rows.

[smalyshev]

performing the lookup join on the data node would mean it would run over 1000 * n rows, where n is the number of data nodes, rather than just over 1000 rows.

But that's how limit always works, doesn't it? Any limit that applies on the remote side would have to be run there, and then run again on the coordinator side. And every operation that runs on remote will be run on 1000 * n rows, and then limited to 1000 rows when it comes back to coordinator. Join changes nothing in this case.