Fork ES|QL execution after preMapper #132981
Conversation
idegtiarenko
added
>enhancement
Team:Analytics
|
Pinging @elastic/es-analytical-engine (Team:Analytics) |
|
Hi @idegtiarenko, I've created a changelog YAML for you. |
| this.inferenceService = services.inferenceService(); | ||
| this.preMapper = new PreMapper(services); | ||
| this.remoteClusterService = services.transportService().getRemoteClusterService(); | ||
| this.planExecutor = services.transportService().getThreadPool().executor(ThreadPool.Names.SEARCH); |
There was a problem hiding this comment.
Is this technically planExecutorExecutor? 🤔
dnhatn
left a comment
dnhatn
left a comment
There was a problem hiding this comment.
I left a question, but this change was long overdue. Thank you!
| SubscribableListener.<LogicalPlan>newForked(l -> preOptimizedPlan(analyzedPlan, l)) | ||
| .<LogicalPlan>andThen((l, p) -> preMapper.preMapper(optimizedPlan(p), l)) | ||
| .andThenApply(p -> optimizedPlan(p)) | ||
| .<LogicalPlan>andThen(planExecutor, null, (l, p) -> preMapper.preMapper(p, l)) |
There was a problem hiding this comment.
Should we specify the thread pool in the next step instead because preMapper.preMapper() can switch the executing thread during rewrite.
There was a problem hiding this comment.
That is correct, pre mapper could fork us to number of different threads.
andThen customizes the listener supplied to the given step to fork with executor after its completion.
This is also confirmed by the updated assertions (in executeOptimizedPlan right after this is completed)
There was a problem hiding this comment.
This is also a place that allow to narrow down the list of thread everywhere else in this change.
There was a problem hiding this comment.
java.lang.AssertionError: elasticsearch[test-cluster-0][transport_worker][T#8] not in [search, search_coordination] nor a test thread
at org.elasticsearch.threadpool.ThreadPool.assertCurrentThreadPool(ThreadPool.java:1116) ~[elasticsearch-9.2.0-SNAPSHOT.jar:?]
at org.elasticsearch.xpack.esql.session.EsqlSession.executeOptimizedPlan(EsqlSession.java:204) ~[?:?]
could still happen with
./gradlew ":x-pack:plugin:esql:qa:server:multi-node:javaRestTest" --tests "org.elasticsearch.xpack.esql.qa.multi_node.EsqlSpecIT" -Dtests.method="test {csv-spec:match-function.TestMatchWithSemanticTextWithEvalsAndOtherFunctionsAndStats}"
Looks like .<LogicalPlan>andThen((l, p) -> preMapper.preMapper(p, new ThreadedActionListener<>(planExecutor, l))) might be necessary instead of conditional forking logic in andThen
| this.planExecutor = planExecutor; | ||
| this.clusterService = clusterService; | ||
| this.requestExecutor = threadPool.executor(ThreadPool.Names.SEARCH); | ||
| this.requestExecutor = threadPool.executor(ThreadPool.Names.SEARCH_COORDINATION); |
There was a problem hiding this comment.
Field caps is executed on SEARCH_COORDINATION.
This sound like query metadata processing should go to SEARCH_COORDINATION and actual searching/aggregation to SEARCH/SYSTEM_READ to me.
idegtiarenko
changed the title
|
Replaced by #133313 |
3 participants
This change forks execution back to the
searchthread pool after execution preMapper.It narrows down a lot list of threads where remaining planning, optimization and request sending to data nodes can happen and eliminates a possibility of running on ml or netty/transport thread.
Related to: #131799