Introduce batched query execution and data-node side reduce #121885
Conversation
Shortest version I could think of. Still WIP, have to make some test adjustments and polish rough edges, but it shouldn't get longer than this.
original-brownbear
added
WIP
:Search Foundations/Search
|
trivial dependency: #121887 |
|
Another trivial dependency #121922 to avoid some duplication here and remove existing dead code. |
An easy change we can split out of elastic#121885 to make that shorter.
| @@ -404,12 +414,19 @@ public SearchService( | |||
| enableQueryPhaseParallelCollection = QUERY_PHASE_PARALLEL_COLLECTION_ENABLED.get(settings); | |||
| clusterService.getClusterSettings() | |||
| .addSettingsUpdateConsumer(QUERY_PHASE_PARALLEL_COLLECTION_ENABLED, this::setEnableQueryPhaseParallelCollection); | |||
|
|
|||
| batchQueryPhase = BATCHED_QUERY_PHASE.get(settings) | |||
| && (BATCHED_QUERY_PHASE.exists(settings) || DiscoveryNode.isStateless(settings) == false); | |||
There was a problem hiding this comment.
Talking to @jimczi about this change, he raised a point that this feature should be under a feature flag to start with. That would allow us to have benchmarks and tests running with it enabled, for both stateful and stateless. Given the branches we are targeting to merge this in and our release schedule for 8.19 and 9.1, I foresee that we'd remove the feature flag before shipping 9.1 and 8.19, hence having a feature flag for stateful does not make a huge difference, but it probably does so that we run serverless tests and benchmarks with the feature enabled, before enabling it by default in serverless envs, which we plan on doing gradually. I think that that's a valid concern, and the feature flag would not be a replacement for the escape hatch via setting which we still need for different reasons.
Also, would a feature flag also allow us to remove the isStateless conditional here (which is quite odd) ?
@jimczi I hope I represented your thoughts accurately, please correct me if that isn't the case.
There was a problem hiding this comment.
Makes sense, feature flag added :) First time I did that, but looks like it worked :D
|
Alrighty, flag is in place, serverless setting reference is removed :) Thanks Luca + Jim! I'll open a back port PR to 8.19 after some benchmarking! :) |
…astic#122188) An easy change we can split out of elastic#121885 to make that shorter.
…121885) This change moves the query phase a single roundtrip per node just like can_match or field_caps work already. A a result of executing multiple shard queries from a single request we can also partially reduce each node's query results on the data node side before responding to the coordinating node. As a result this change significantly reduces the impact of network latencies on the end-to-end query performance, reduces the amount of work done (memory and cpu) on the coordinating node and the network traffic by factors of up to the number of shards per data node! Benchmarking shows up to orders of magnitude improvements in heap and network traffic dimensions in querying across a larger number of shards.
…#126563) * Introduce batched query execution and data-node side reduce (#121885) This change moves the query phase a single roundtrip per node just like can_match or field_caps work already. A a result of executing multiple shard queries from a single request we can also partially reduce each node's query results on the data node side before responding to the coordinating node. As a result this change significantly reduces the impact of network latencies on the end-to-end query performance, reduces the amount of work done (memory and cpu) on the coordinating node and the network traffic by factors of up to the number of shards per data node! Benchmarking shows up to orders of magnitude improvements in heap and network traffic dimensions in querying across a larger number of shards. * Filter out empty top docs results before merging (#126385) `Lucene.EMPTY_TOP_DOCS` to identify empty to docs results. These were previously null results, but did not need to be send over transport as incremental reduction was performed only on the data node. Now it can happen that the coord node received a merge result with empty top docs, which has nothing interesting for merging, but that can lead to an exception because the type of the empty array does not match the type of other shards results, for instance if the query was sorted by field. To resolve this, we filter out empty top docs results before merging. Closes #126118 --------- Co-authored-by: Luca Cavanna <[email protected]>
Shortest version I could think of for this from where we are now.
This change moves the query phase a single roundtrip per node just like can_match or field_caps work already.
A a result of executing multiple shard queries from a single request we can also partially reduce each node's query results on the data node side before responding to the coordinating node.
As a result this change significantly reduces the impact of network latencies on the end-to-end query performance, reduces the amount of work done (memory and cpu) on the coordinating node and the network traffic by factors of up to the number of shards per data node!
Benchmarking shows up to orders of magnitude improvements in heap and network traffic dimensions in querying across a larger number of shards.