forked from elastic/elasticsearch
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
EQL: Introduce sequence internal paging (elastic#58859)
Refactor sequence matching classes in order to decouple querying from results consumption (and matching). Rename some classes to better convey their intent. Introduce internal pagination of sequence algorithm, that is getting the data in slices and, if needed, moving forward in order to find more matches until either the dataset is consumer or the number of results desired is found.
- Loading branch information
Showing
17 changed files
with
543 additions
and
345 deletions.
There are no files selected for viewing
71 changes: 71 additions & 0 deletions
71
.../eql/src/main/java/org/elasticsearch/xpack/eql/execution/assembler/BoxedQueryRequest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License; | ||
| * you may not use this file except in compliance with the Elastic License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.xpack.eql.execution.assembler; | ||
|
|
||
| import org.elasticsearch.index.query.BoolQueryBuilder; | ||
| import org.elasticsearch.index.query.RangeQueryBuilder; | ||
| import org.elasticsearch.search.builder.SearchSourceBuilder; | ||
| import org.elasticsearch.xpack.eql.execution.search.Ordinal; | ||
| import org.elasticsearch.xpack.eql.execution.search.QueryRequest; | ||
|
|
||
| import static org.elasticsearch.index.query.QueryBuilders.boolQuery; | ||
| import static org.elasticsearch.index.query.QueryBuilders.rangeQuery; | ||
|
|
||
| public class BoxedQueryRequest implements QueryRequest { | ||
|
|
||
| private final RangeQueryBuilder timestampRange; | ||
| private final RangeQueryBuilder tiebreakerRange; | ||
|
|
||
| private final SearchSourceBuilder searchSource; | ||
|
|
||
| public BoxedQueryRequest(QueryRequest original, String timestamp, String tiebreaker) { | ||
| searchSource = original.searchSource(); | ||
|
|
||
| // setup range queries and preserve their reference to simplify the update | ||
| timestampRange = rangeQuery(timestamp).timeZone("UTC").format("epoch_millis"); | ||
| BoolQueryBuilder filter = boolQuery().filter(timestampRange); | ||
| if (tiebreaker != null) { | ||
| tiebreakerRange = rangeQuery(tiebreaker); | ||
| filter.filter(tiebreakerRange); | ||
| } else { | ||
| tiebreakerRange = null; | ||
| } | ||
| // add ranges to existing query | ||
| searchSource.query(filter.must(searchSource.query())); | ||
| } | ||
|
|
||
| @Override | ||
| public SearchSourceBuilder searchSource() { | ||
| return searchSource; | ||
| } | ||
|
|
||
| @Override | ||
| public void next(Ordinal ordinal) { | ||
| // reset existing constraints | ||
| timestampRange.gte(null).lte(null); | ||
| if (tiebreakerRange != null) { | ||
| tiebreakerRange.gte(null).lte(null); | ||
| } | ||
| // and leave only search_after | ||
| searchSource.searchAfter(ordinal.toArray()); | ||
| } | ||
|
|
||
| public BoxedQueryRequest between(Ordinal begin, Ordinal end) { | ||
| timestampRange.gte(begin.timestamp()).lte(end.timestamp()); | ||
|
|
||
| if (tiebreakerRange != null) { | ||
| tiebreakerRange.gte(begin.tiebreaker()).lte(end.tiebreaker()); | ||
| } | ||
|
|
||
| return this; | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return searchSource.toString(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
74 changes: 74 additions & 0 deletions
74
x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/assembler/Matcher.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,74 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License; | ||
| * you may not use this file except in compliance with the Elastic License. | ||
| */ | ||
| package org.elasticsearch.xpack.eql.execution.assembler; | ||
|
|
||
| import org.elasticsearch.common.collect.Tuple; | ||
| import org.elasticsearch.common.unit.TimeValue; | ||
| import org.elasticsearch.search.SearchHit; | ||
| import org.elasticsearch.xpack.eql.execution.search.Limit; | ||
| import org.elasticsearch.xpack.eql.execution.search.Ordinal; | ||
| import org.elasticsearch.xpack.eql.execution.sequence.Sequence; | ||
| import org.elasticsearch.xpack.eql.execution.sequence.SequenceStateMachine; | ||
| import org.elasticsearch.xpack.eql.session.Payload; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| /** | ||
| * Executable tracking sequences at runtime. | ||
| */ | ||
| class Matcher { | ||
|
|
||
| // NB: just like in a list, this represents the total number of stages yet counting starts at 0 | ||
| private final SequenceStateMachine stateMachine; | ||
| private final int numberOfStages; | ||
|
|
||
| Matcher(int numberOfStages, TimeValue maxSpan, Limit limit) { | ||
| this.numberOfStages = numberOfStages; | ||
| this.stateMachine = new SequenceStateMachine(numberOfStages, maxSpan, limit); | ||
| } | ||
|
|
||
| /** | ||
| * Match hits for the given stage. | ||
| * Returns false if the process needs to be stopped. | ||
| */ | ||
| boolean match(int stage, Iterable<Tuple<KeyAndOrdinal, SearchHit>> hits) { | ||
| for (Tuple<KeyAndOrdinal, SearchHit> tuple : hits) { | ||
| KeyAndOrdinal ko = tuple.v1(); | ||
| SearchHit hit = tuple.v2(); | ||
|
|
||
| if (stage == 0) { | ||
| Sequence seq = new Sequence(ko.key, numberOfStages, ko.ordinal, hit); | ||
| stateMachine.trackSequence(seq); | ||
| } else { | ||
| stateMachine.match(stage, ko.key, ko.ordinal, hit); | ||
|
|
||
| // early skip in case of reaching the limit | ||
| // check the last stage to avoid calling the state machine in other stages | ||
| if (stateMachine.reachedLimit()) { | ||
| return false; | ||
| } | ||
| } | ||
| } | ||
| return true; | ||
| } | ||
|
|
||
| boolean until(Iterable<Ordinal> markers) { | ||
| // no-op so far | ||
|
|
||
| return false; | ||
| } | ||
|
|
||
|
|
||
| public boolean hasCandidates(int stage) { | ||
| return stateMachine.hasCandidates(stage); | ||
| } | ||
|
|
||
| Payload payload(long startTime) { | ||
| List<Sequence> completed = stateMachine.completeSequences(); | ||
| TimeValue tookTime = new TimeValue(System.currentTimeMillis() - startTime); | ||
| return new SequencePayload(completed, false, tookTime); | ||
| } | ||
| } |
Oops, something went wrong.