Push down field extraction to time-series source #127445
There are no files selected for viewing
This file contains hidden or 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 hidden or 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 hidden or 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 hidden or 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 hidden or 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 hidden or 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 hidden or 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,105 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License | ||
| * 2.0; you may not use this file except in compliance with the Elastic License | ||
| * 2.0. | ||
| */ | ||
|
|
||
| package org.elasticsearch.xpack.esql.optimizer.rules.physical.local; | ||
|
|
||
| import org.elasticsearch.index.IndexMode; | ||
| import org.elasticsearch.xpack.esql.core.expression.Attribute; | ||
| import org.elasticsearch.xpack.esql.core.util.Holder; | ||
| import org.elasticsearch.xpack.esql.optimizer.LocalPhysicalOptimizerContext; | ||
| import org.elasticsearch.xpack.esql.optimizer.PhysicalOptimizerRules; | ||
| import org.elasticsearch.xpack.esql.plan.physical.EsQueryExec; | ||
| import org.elasticsearch.xpack.esql.plan.physical.FieldExtractExec; | ||
| import org.elasticsearch.xpack.esql.plan.physical.FilterExec; | ||
| import org.elasticsearch.xpack.esql.plan.physical.LimitExec; | ||
| import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan; | ||
| import org.elasticsearch.xpack.esql.plan.physical.TimeSeriesSourceExec; | ||
| import org.elasticsearch.xpack.esql.plan.physical.TopNExec; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.HashSet; | ||
| import java.util.List; | ||
| import java.util.Set; | ||
|
|
||
| /** | ||
| * A rule that pushes down field extractions to occur before filter/limit/topN in the time-series source plan. | ||
| */ | ||
| public class PushDownFieldExtractionToTimeSeriesSource extends PhysicalOptimizerRules.ParameterizedOptimizerRule< | ||
| PhysicalPlan, | ||
| LocalPhysicalOptimizerContext> { | ||
|
|
||
| @Override | ||
| public PhysicalPlan rule(PhysicalPlan plan, LocalPhysicalOptimizerContext context) { | ||
| if (plan.anyMatch(p -> p instanceof EsQueryExec q && q.indexMode() == IndexMode.TIME_SERIES) == false) { | ||
| return plan; | ||
| } | ||
| final List<FieldExtractExec> pushDownExtracts = new ArrayList<>(); | ||
|
Member
Author
There was a problem hiding this comment. This is the main change in planning, where field extractions are pushed down to the time-series source |
||
| final Holder<Boolean> keepDocIds = new Holder<>(Boolean.FALSE); | ||
| plan.forEachDown(p -> { | ||
| if (p instanceof FieldExtractExec) { | ||
| pushDownExtracts.add((FieldExtractExec) p); | ||
| } else if (stopPushDownExtract(p)) { | ||
| if (pushDownExtracts.isEmpty() == false) { | ||
| keepDocIds.set(Boolean.TRUE); | ||
| pushDownExtracts.clear(); | ||
| } | ||
| } | ||
| }); | ||
| final Holder<Boolean> aborted = new Holder<>(Boolean.FALSE); | ||
| return plan.transformUp(PhysicalPlan.class, p -> { | ||
| if (aborted.get()) { | ||
| return p; | ||
| } | ||
| if (p instanceof EsQueryExec q && q.indexMode() == IndexMode.TIME_SERIES) { | ||
| return addFieldExtract(context, q, keepDocIds.get(), pushDownExtracts); | ||
| } | ||
| if (stopPushDownExtract(p)) { | ||
| aborted.set(Boolean.TRUE); | ||
| return p; | ||
| } | ||
| if (p instanceof FieldExtractExec e) { | ||
| return e.child(); | ||
| } | ||
| return p; | ||
| }); | ||
| } | ||
|
|
||
| private static boolean stopPushDownExtract(PhysicalPlan p) { | ||
| return p instanceof FilterExec || p instanceof TopNExec || p instanceof LimitExec; | ||
| } | ||
|
|
||
| private TimeSeriesSourceExec addFieldExtract( | ||
| LocalPhysicalOptimizerContext context, | ||
| EsQueryExec query, | ||
| boolean keepDocAttribute, | ||
| List<FieldExtractExec> extracts | ||
| ) { | ||
| Set<Attribute> docValuesAttributes = new HashSet<>(); | ||
| Set<Attribute> boundsAttributes = new HashSet<>(); | ||
| List<Attribute> attributesToExtract = new ArrayList<>(); | ||
| for (FieldExtractExec extract : extracts) { | ||
| docValuesAttributes.addAll(extract.docValuesAttributes()); | ||
| boundsAttributes.addAll(extract.boundsAttributes()); | ||
| attributesToExtract.addAll(extract.attributesToExtract()); | ||
| } | ||
| List<Attribute> attrs = query.attrs(); | ||
| if (keepDocAttribute == false) { | ||
| attrs = attrs.stream().filter(a -> EsQueryExec.isSourceAttribute(a) == false).toList(); | ||
| } | ||
| return new TimeSeriesSourceExec( | ||
| query.source(), | ||
| attrs, | ||
| query.query(), | ||
| query.limit(), | ||
| context.configuration().pragmas().fieldExtractPreference(), | ||
| docValuesAttributes, | ||
| boundsAttributes, | ||
| attributesToExtract, | ||
| query.estimatedRowSize() | ||
| ); | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Don't we want filtering to happen before field extraction? Or combine them, at least? Maybe I misunderstood this comment..
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
For the query
TS index | WHERE host = 'a' | STATS max(rate(counter)) BY host, bucket(1minute), we should extract only thehostfield (and tsid and timestamp) in the time-series source command. Thecounterfield should be extracted later by the ValuesSourceReaderOperator because we don't know how many rows of will be filtered out.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Correct, what if we have
I'd think we want to push down the filters on
hostand@timestampto Lucene, to run before extracting the fields. If this is the case, maybe clarify in the comment that filters on the extracted fields are still pushed down.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I expanded the javadoc: 164d788