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.
Refactor design by moving filter away from the Aggregate plan node into
the aggregate function. This helps keep the filter embedded in the agg which signals to existing rules this aggs shouldn't be optimized or confused with those that do not define a filter.
- Loading branch information
Showing
14 changed files
with
277 additions
and
156 deletions.
There are no files selected for viewing
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
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
105 changes: 105 additions & 0 deletions
105
.../java/org/elasticsearch/xpack/esql/expression/function/aggregate/FilteredAggregation.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,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.expression.function.aggregate; | ||
|
|
||
| import org.elasticsearch.common.io.stream.NamedWriteableRegistry; | ||
| import org.elasticsearch.common.io.stream.StreamInput; | ||
| import org.elasticsearch.common.io.stream.StreamOutput; | ||
| import org.elasticsearch.compute.aggregation.AggregatorFunctionSupplier; | ||
| import org.elasticsearch.compute.aggregation.FilteredAggregatorFunctionSupplier; | ||
| import org.elasticsearch.xpack.esql.EsqlIllegalArgumentException; | ||
| import org.elasticsearch.xpack.esql.core.expression.Expression; | ||
| import org.elasticsearch.xpack.esql.core.tree.NodeInfo; | ||
| import org.elasticsearch.xpack.esql.core.tree.Source; | ||
| import org.elasticsearch.xpack.esql.core.type.DataType; | ||
| import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput; | ||
| import org.elasticsearch.xpack.esql.planner.ToAggregator; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.List; | ||
|
|
||
| import static java.util.Collections.singletonList; | ||
|
|
||
| /** | ||
| * Basic wrapper for aggregate functions declared with a nested filter (typically in stats). | ||
| * The optimizer uses this class to replace the base aggregations from {@code FilteredExpression} with this | ||
| * specialized class which encapsulate the underlying aggregation supplier. | ||
| */ | ||
| public class FilteredAggregation extends AggregateFunction implements ToAggregator { | ||
| public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry( | ||
| Expression.class, | ||
| "FilteredAggregation", | ||
| FilteredAggregation::new | ||
| ); | ||
|
|
||
| private final AggregateFunction delegate; | ||
| private final Expression filter; | ||
|
|
||
| public FilteredAggregation(Source source, AggregateFunction delegate, Expression filter) { | ||
| super(source, delegate, singletonList(filter)); | ||
| this.delegate = delegate; | ||
| this.filter = filter; | ||
| } | ||
|
|
||
| public FilteredAggregation(StreamInput in) throws IOException { | ||
| this( | ||
| Source.readFrom((PlanStreamInput) in), | ||
| (AggregateFunction) in.readNamedWriteable(Expression.class), | ||
| in.readNamedWriteable(Expression.class) | ||
| ); | ||
| } | ||
|
|
||
| @Override | ||
| public void writeTo(StreamOutput out) throws IOException { | ||
| Source.EMPTY.writeTo(out); | ||
| out.writeNamedWriteable(delegate); | ||
| out.writeNamedWriteable(filter); | ||
| } | ||
|
|
||
| @Override | ||
| public DataType dataType() { | ||
| return delegate.dataType(); | ||
| } | ||
|
|
||
| public AggregateFunction delegate() { | ||
| return delegate; | ||
| } | ||
|
|
||
| public Expression filter() { | ||
| return filter; | ||
| } | ||
|
|
||
| @Override | ||
| public FilteredAggregation replaceChildren(List<Expression> newChildren) { | ||
| return new FilteredAggregation(source(), (AggregateFunction) newChildren.get(0), newChildren.get(1)); | ||
| } | ||
|
|
||
| @Override | ||
| protected NodeInfo<FilteredAggregation> info() { | ||
| return NodeInfo.create(this, FilteredAggregation::new, delegate, filter); | ||
| } | ||
|
|
||
| @Override | ||
| public String getWriteableName() { | ||
| return ENTRY.name; | ||
| } | ||
|
|
||
| public FilteredAggregation with(Expression filter) { | ||
| return new FilteredAggregation(source(), delegate, filter); | ||
| } | ||
|
|
||
| @Override | ||
| public AggregatorFunctionSupplier supplier(List<Integer> inputChannels) { | ||
| if (delegate instanceof ToAggregator toAggregator) { | ||
| AggregatorFunctionSupplier aggSupplier = toAggregator.supplier(inputChannels); | ||
| return new FilteredAggregatorFunctionSupplier(aggSupplier, null); | ||
| } else { | ||
| throw new EsqlIllegalArgumentException("Cannot create aggregator for " + delegate); | ||
| } | ||
| } | ||
| } |
91 changes: 91 additions & 0 deletions
91
...n/java/org/elasticsearch/xpack/esql/expression/function/aggregate/FilteredExpression.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,91 @@ | ||
| /* | ||
| * 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.expression.function.aggregate; | ||
|
|
||
| import org.elasticsearch.common.io.stream.NamedWriteableRegistry; | ||
| import org.elasticsearch.common.io.stream.StreamInput; | ||
| import org.elasticsearch.common.io.stream.StreamOutput; | ||
| import org.elasticsearch.xpack.esql.core.expression.Expression; | ||
| import org.elasticsearch.xpack.esql.core.expression.Nullability; | ||
| import org.elasticsearch.xpack.esql.core.tree.NodeInfo; | ||
| import org.elasticsearch.xpack.esql.core.tree.Source; | ||
| import org.elasticsearch.xpack.esql.core.type.DataType; | ||
| import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.List; | ||
|
|
||
| import static java.util.Arrays.asList; | ||
|
|
||
| /** | ||
| * Basic wrapper for expressions declared with a nested filter (typically in stats). | ||
| */ | ||
| public class FilteredExpression extends Expression { | ||
| public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry( | ||
| Expression.class, | ||
| "FilteredExpression", | ||
| FilteredExpression::new | ||
| ); | ||
|
|
||
| private final Expression delegate; | ||
| private final Expression filter; | ||
|
|
||
| public FilteredExpression(Source source, Expression delegate, Expression filter) { | ||
| super(source, asList(delegate, filter)); | ||
| this.delegate = delegate; | ||
| this.filter = filter; | ||
| } | ||
|
|
||
| public FilteredExpression(StreamInput in) throws IOException { | ||
| this( | ||
| Source.readFrom((PlanStreamInput) in), | ||
| in.readNamedWriteable(Expression.class), | ||
| in.readNamedWriteable(Expression.class) | ||
| ); | ||
| } | ||
|
|
||
| @Override | ||
| public void writeTo(StreamOutput out) throws IOException { | ||
| Source.EMPTY.writeTo(out); | ||
| out.writeNamedWriteable(delegate); | ||
| out.writeNamedWriteable(filter); | ||
| } | ||
|
|
||
| @Override | ||
| public String getWriteableName() { | ||
| return ENTRY.name; | ||
| } | ||
|
|
||
| public Expression delegate() { | ||
| return delegate; | ||
| } | ||
|
|
||
| public Expression filter() { | ||
| return filter; | ||
| } | ||
|
|
||
| @Override | ||
| public DataType dataType() { | ||
| return delegate.dataType(); | ||
| } | ||
|
|
||
| @Override | ||
| public Nullability nullable() { | ||
| return delegate.nullable(); | ||
| } | ||
|
|
||
| @Override | ||
| protected NodeInfo<FilteredExpression> info() { | ||
| return NodeInfo.create(this, FilteredExpression::new, delegate, filter); | ||
| } | ||
|
|
||
| @Override | ||
| public Expression replaceChildren(List<Expression> newChildren) { | ||
| return new FilteredExpression(source(), newChildren.get(0), newChildren.get(1)); | ||
| } | ||
| } |
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
Oops, something went wrong.