Skip to content

Commit d26d772

Browse files
nik9000nik9000
authored
Continue realizing sorting by aggregations (backport of #52298) (#52667)
This drops more of the `instanceof`s from `AggregationPath`. There are still a couple in `AggregationPath`. And I ended up moving two into `BucketsAggregator`, but I think this is still an improvement!
1 parent a0aa808 commit d26d772

File tree

10 files changed

+140
-95
lines changed

10 files changed

+140
-95
lines changed

‎server/src/main/java/org/elasticsearch/search/aggregations/Aggregator.java‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,9 +28,11 @@
2828
import org.elasticsearch.common.xcontent.DeprecationHandler;
2929
import org.elasticsearch.common.xcontent.XContentParser;
3030
import org.elasticsearch.search.aggregations.bucket.BucketsAggregator;
31+
import org.elasticsearch.search.aggregations.support.AggregationPath;
3132
import org.elasticsearch.search.internal.SearchContext;
3233

3334
import java.io.IOException;
35+
import java.util.Iterator;
3436

3537
/**
3638
* An Aggregator.
@@ -91,6 +93,53 @@ public static boolean descendsFromBucketAggregator(Aggregator parent) {
9193
*/
9294
public abstract Aggregator subAggregator(String name);
9395

96+
/**
97+
* Resolve the next step of the sort path as though this aggregation
98+
* supported sorting. This is usually the "first step" when resolving
99+
* a sort path because most aggs that support sorting their buckets
100+
* aren't valid in the middle of a sort path.
101+
* <p>
102+
* For example, the {@code terms} aggs supports sorting its buckets, but
103+
* that sort path itself can't contain a different {@code terms}
104+
* aggregation.
105+
*/
106+
public final Aggregator resolveSortPathOnValidAgg(AggregationPath.PathElement next, Iterator<AggregationPath.PathElement> path) {
107+
Aggregator n = subAggregator(next.name);
108+
if (n == null) {
109+
throw new IllegalArgumentException("The provided aggregation [" + next + "] either does not exist, or is "
110+
+ "a pipeline aggregation and cannot be used to sort the buckets.");
111+
}
112+
if (false == path.hasNext()) {
113+
return n;
114+
}
115+
if (next.key != null) {
116+
throw new IllegalArgumentException("Key only allowed on last aggregation path element but got [" + next + "]");
117+
}
118+
return n.resolveSortPath(path.next(), path);
119+
}
120+
121+
/**
122+
* Resolve a sort path to the target.
123+
* <p>
124+
* The default implementation throws an exception but we override it on aggregations that support sorting.
125+
*/
126+
public Aggregator resolveSortPath(AggregationPath.PathElement next, Iterator<AggregationPath.PathElement> path) {
127+
throw new IllegalArgumentException("Buckets can only be sorted on a sub-aggregator path " +
128+
"that is built out of zero or more single-bucket aggregations within the path and a final " +
129+
"single-bucket or a metrics aggregation at the path end. [" + name() + "] is not single-bucket.");
130+
}
131+
132+
/**
133+
* Validates the "key" portion of a sort on this aggregation.
134+
* <p>
135+
* The default implementation throws an exception but we override it on aggregations that support sorting.
136+
*/
137+
public void validateSortPathKey(String key) {
138+
throw new IllegalArgumentException("Buckets can only be sorted on a sub-aggregator path " +
139+
"that is built out of zero or more single-bucket aggregations within the path and a final " +
140+
"single-bucket or a metrics aggregation at the path end.");
141+
}
142+
94143
/**
95144
* Build an aggregation for data that has been collected into {@code bucket}.
96145
*/

‎server/src/main/java/org/elasticsearch/search/aggregations/bucket/BucketsAggregator.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,10 +28,12 @@
2828
import org.elasticsearch.search.aggregations.InternalAggregations;
2929
import org.elasticsearch.search.aggregations.LeafBucketCollector;
3030
import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
31+
import org.elasticsearch.search.aggregations.support.AggregationPath;
3132
import org.elasticsearch.search.internal.SearchContext;
3233

3334
import java.io.IOException;
3435
import java.util.Arrays;
36+
import java.util.Iterator;
3537
import java.util.List;
3638
import java.util.Map;
3739
import java.util.function.IntConsumer;
@@ -163,4 +165,24 @@ public final void close() {
163165
}
164166
}
165167

168+
@Override
169+
public Aggregator resolveSortPath(AggregationPath.PathElement next, Iterator<AggregationPath.PathElement> path) {
170+
if (this instanceof SingleBucketAggregator) {
171+
return resolveSortPathOnValidAgg(next, path);
172+
}
173+
return super.resolveSortPath(next, path);
174+
}
175+
176+
@Override
177+
public void validateSortPathKey(String key) {
178+
if (false == this instanceof SingleBucketAggregator) {
179+
super.validateSortPathKey(key);
180+
return;
181+
}
182+
if (key != null && false == "doc_count".equals(key)) {
183+
throw new IllegalArgumentException("Ordering on a single-bucket aggregation can only be done on its doc_count. " +
184+
"Either drop the key (a la \"" + name() + "\") or change it to \"doc_count\" (a la \"" + name() +
185+
".doc_count\")");
186+
}
187+
}
166188
}

‎server/src/main/java/org/elasticsearch/search/aggregations/bucket/DeferringBucketCollector.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,9 +25,11 @@
2525
import org.elasticsearch.search.aggregations.BucketCollector;
2626
import org.elasticsearch.search.aggregations.InternalAggregation;
2727
import org.elasticsearch.search.aggregations.LeafBucketCollector;
28+
import org.elasticsearch.search.aggregations.support.AggregationPath.PathElement;
2829
import org.elasticsearch.search.internal.SearchContext;
2930

3031
import java.io.IOException;
32+
import java.util.Iterator;
3133

3234
/**
3335
* A {@link BucketCollector} that records collected doc IDs and buckets and
@@ -120,6 +122,15 @@ public void postCollection() throws IOException {
120122
"Deferred collectors cannot be collected directly. They must be collected through the recording wrapper.");
121123
}
122124

125+
@Override
126+
public Aggregator resolveSortPath(PathElement next, Iterator<PathElement> path) {
127+
return in.resolveSortPath(next, path);
128+
}
129+
130+
@Override
131+
public void validateSortPathKey(String key) {
132+
in.validateSortPathKey(key);
133+
}
123134
}
124135

125136
}

‎server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/TermsAggregator.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
4444
import org.elasticsearch.search.aggregations.support.AggregationPath;
4545
import org.elasticsearch.search.internal.SearchContext;
46+
import org.elasticsearch.search.profile.aggregation.ProfilingAggregator;
4647

4748
import java.io.IOException;
4849
import java.util.Comparator;
@@ -249,7 +250,12 @@ private boolean subAggsNeedScore() {
249250
*/
250251
public Comparator<Bucket> bucketComparator(AggregationPath path, boolean asc) {
251252

252-
final Aggregator aggregator = path.resolveAggregator(this);
253+
Aggregator agg = path.resolveAggregator(this);
254+
// TODO Move this method into Aggregator or AggregationPath.
255+
if (agg instanceof ProfilingAggregator) {
256+
agg = ProfilingAggregator.unwrap(agg);
257+
}
258+
final Aggregator aggregator = agg;
253259
final String key = path.lastPathElement().key;
254260

255261
if (aggregator instanceof SingleBucketAggregator) {

‎server/src/main/java/org/elasticsearch/search/aggregations/metrics/InternalNumericMetricsAggregation.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -107,7 +107,7 @@ public Object getProperty(List<String> path) {
107107
@Override
108108
public final double sortValue(String key) {
109109
if (key == null) {
110-
throw new IllegalArgumentException("Missing value key in [" + key+ "] which refers to a multi-value metric aggregation");
110+
throw new IllegalArgumentException("Missing value key in [" + key + "] which refers to a multi-value metric aggregation");
111111
}
112112
return value(key);
113113
}

‎server/src/main/java/org/elasticsearch/search/aggregations/metrics/NumericMetricsAggregator.java‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,14 @@ protected SingleValue(String name, SearchContext context, Aggregator parent, Lis
4141
}
4242

4343
public abstract double metric(long owningBucketOrd);
44+
45+
@Override
46+
public void validateSortPathKey(String key) {
47+
if (key != null && false == "value".equals(key)) {
48+
throw new IllegalArgumentException("Ordering on a single-value metrics aggregation can only be done on its value. " +
49+
"Either drop the key (a la \"" + name() + "\") or change it to \"value\" (a la \"" + name() + ".value\")");
50+
}
51+
}
4452
}
4553

4654
public abstract static class MultiValue extends NumericMetricsAggregator {
@@ -53,5 +61,16 @@ protected MultiValue(String name, SearchContext context, Aggregator parent, List
5361
public abstract boolean hasMetric(String name);
5462

5563
public abstract double metric(String name, long owningBucketOrd);
64+
65+
@Override
66+
public void validateSortPathKey(String key) {
67+
if (key == null) {
68+
throw new IllegalArgumentException("When ordering on a multi-value metrics aggregation a metric name must be specified.");
69+
}
70+
if (false == hasMetric(key)) {
71+
throw new IllegalArgumentException(
72+
"Unknown metric name [" + key + "] on multi-value metrics aggregation [" + name() + "]");
73+
}
74+
}
5675
}
5776
}

‎server/src/main/java/org/elasticsearch/search/aggregations/support/AggregationPath.java‎

Lines changed: 15 additions & 90 deletions
Original file line numberDiff line numberDiff line change
@@ -182,7 +182,7 @@ public List<String> getPathElementsAsStringList() {
182182
return stringPathElements;
183183
}
184184

185-
public AggregationPath subPath(int offset, int length) {
185+
private AggregationPath subPath(int offset, int length) {
186186
List<PathElement> subTokens = new ArrayList<>(pathElements.subList(offset, offset + length));
187187
return new AggregationPath(subTokens);
188188
}
@@ -196,38 +196,29 @@ public double resolveValue(InternalAggregations aggregations) {
196196
assert path.hasNext();
197197
return aggregations.sortValue(path.next(), path);
198198
} catch (IllegalArgumentException e) {
199-
throw new IllegalArgumentException("Invalid order path [" + this + "]. " + e.getMessage(), e);
199+
throw new IllegalArgumentException("Invalid aggregation order path [" + this + "]. " + e.getMessage(), e);
200200
}
201201
}
202202

203203
/**
204-
* Resolves the aggregator pointed by this path using the given root as a point of reference.
205-
*
206-
* @param root The point of reference of this path
207-
* @return The aggregator pointed by this path starting from the given aggregator as a point of reference
204+
* Resolves the {@linkplain Aggregator} pointed to by this path against
205+
* the given root {@linkplain Aggregator}.
208206
*/
209207
public Aggregator resolveAggregator(Aggregator root) {
210-
Aggregator aggregator = root;
211-
for (int i = 0; i < pathElements.size(); i++) {
212-
AggregationPath.PathElement token = pathElements.get(i);
213-
aggregator = ProfilingAggregator.unwrap(aggregator.subAggregator(token.name));
214-
assert (aggregator instanceof SingleBucketAggregator && i <= pathElements.size() - 1)
215-
|| (aggregator instanceof NumericMetricsAggregator && i == pathElements.size() - 1) :
216-
"this should be picked up before aggregation execution - on validate";
217-
}
218-
return aggregator;
208+
Iterator<PathElement> path = pathElements.iterator();
209+
assert path.hasNext();
210+
return root.resolveSortPathOnValidAgg(path.next(), path);
219211
}
220212

221213
/**
222-
* Resolves the topmost aggregator pointed by this path using the given root as a point of reference.
223-
*
224-
* @param root The point of reference of this path
225-
* @return The first child aggregator of the root pointed by this path
214+
* Resolves the {@linkplain Aggregator} pointed to by the first element
215+
* of this path against the given root {@linkplain Aggregator}.
226216
*/
227217
public Aggregator resolveTopmostAggregator(Aggregator root) {
228218
AggregationPath.PathElement token = pathElements.get(0);
219+
// TODO both unwrap and subAggregator are only used here!
229220
Aggregator aggregator = ProfilingAggregator.unwrap(root.subAggregator(token.name));
230-
assert (aggregator instanceof SingleBucketAggregator )
221+
assert (aggregator instanceof SingleBucketAggregator)
231222
|| (aggregator instanceof NumericMetricsAggregator) : "this should be picked up before aggregation execution - on validate";
232223
return aggregator;
233224
}
@@ -239,76 +230,10 @@ public Aggregator resolveTopmostAggregator(Aggregator root) {
239230
* @throws AggregationExecutionException on validation error
240231
*/
241232
public void validate(Aggregator root) throws AggregationExecutionException {
242-
Aggregator aggregator = root;
243-
for (int i = 0; i < pathElements.size(); i++) {
244-
String name = pathElements.get(i).name;
245-
aggregator = ProfilingAggregator.unwrap(aggregator.subAggregator(name));
246-
if (aggregator == null) {
247-
throw new AggregationExecutionException("Invalid aggregator order path [" + this + "]. The " +
248-
"provided aggregation [" + name + "] either does not exist, or is a pipeline aggregation " +
249-
"and cannot be used to sort the buckets.");
250-
}
251-
252-
if (i < pathElements.size() - 1) {
253-
254-
// we're in the middle of the path, so the aggregator can only be a single-bucket aggregator
255-
256-
if (!(aggregator instanceof SingleBucketAggregator)) {
257-
throw new AggregationExecutionException("Invalid aggregation order path [" + this +
258-
"]. Buckets can only be sorted on a sub-aggregator path " +
259-
"that is built out of zero or more single-bucket aggregations within the path and a final " +
260-
"single-bucket or a metrics aggregation at the path end. Sub-path [" +
261-
subPath(0, i + 1) + "] points to non single-bucket aggregation");
262-
}
263-
264-
if (pathElements.get(i).key != null) {
265-
throw new AggregationExecutionException("Invalid aggregation order path [" + this +
266-
"]. Buckets can only be sorted on a sub-aggregator path " +
267-
"that is built out of zero or more single-bucket aggregations within the path and a " +
268-
"final single-bucket or a metrics aggregation at the path end. Sub-path [" +
269-
subPath(0, i + 1) + "] points to non single-bucket aggregation");
270-
}
271-
}
272-
}
273-
boolean singleBucket = aggregator instanceof SingleBucketAggregator;
274-
if (!singleBucket && !(aggregator instanceof NumericMetricsAggregator)) {
275-
throw new AggregationExecutionException("Invalid aggregation order path [" + this +
276-
"]. Buckets can only be sorted on a sub-aggregator path " +
277-
"that is built out of zero or more single-bucket aggregations within the path and a final " +
278-
"single-bucket or a metrics aggregation at the path end.");
279-
}
280-
281-
AggregationPath.PathElement lastToken = lastPathElement();
282-
283-
if (singleBucket) {
284-
if (lastToken.key != null && !"doc_count".equals(lastToken.key)) {
285-
throw new AggregationExecutionException("Invalid aggregation order path [" + this +
286-
"]. Ordering on a single-bucket aggregation can only be done on its doc_count. " +
287-
"Either drop the key (a la \"" + lastToken.name + "\") or change it to \"doc_count\" (a la \"" + lastToken.name +
288-
".doc_count\")");
289-
}
290-
return; // perfectly valid to sort on single-bucket aggregation (will be sored on its doc_count)
291-
}
292-
293-
if (aggregator instanceof NumericMetricsAggregator.SingleValue) {
294-
if (lastToken.key != null && !"value".equals(lastToken.key)) {
295-
throw new AggregationExecutionException("Invalid aggregation order path [" + this +
296-
"]. Ordering on a single-value metrics aggregation can only be done on its value. " +
297-
"Either drop the key (a la \"" + lastToken.name + "\") or change it to \"value\" (a la \"" + lastToken.name +
298-
".value\")");
299-
}
300-
return; // perfectly valid to sort on single metric aggregation (will be sorted on its associated value)
301-
}
302-
303-
// the aggregator must be of a multi-value metrics type
304-
if (lastToken.key == null) {
305-
throw new AggregationExecutionException("Invalid aggregation order path [" + this +
306-
"]. When ordering on a multi-value metrics aggregation a metric name must be specified");
307-
}
308-
309-
if (!((NumericMetricsAggregator.MultiValue) aggregator).hasMetric(lastToken.key)) {
310-
throw new AggregationExecutionException("Invalid aggregation order path [" + this +
311-
"]. Unknown metric name [" + lastToken.key + "] on multi-value metrics aggregation [" + lastToken.name + "]");
233+
try {
234+
resolveAggregator(root).validateSortPathKey(lastPathElement().key);
235+
} catch (IllegalArgumentException e) {
236+
throw new AggregationExecutionException("Invalid aggregation order path [" + this + "]. " + e.getMessage(), e);
312237
}
313238
}
314239

‎server/src/main/java/org/elasticsearch/search/profile/aggregation/ProfilingAggregator.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,12 @@
2424
import org.elasticsearch.search.aggregations.Aggregator;
2525
import org.elasticsearch.search.aggregations.InternalAggregation;
2626
import org.elasticsearch.search.aggregations.LeafBucketCollector;
27+
import org.elasticsearch.search.aggregations.support.AggregationPath.PathElement;
2728
import org.elasticsearch.search.internal.SearchContext;
2829
import org.elasticsearch.search.profile.Timer;
2930

3031
import java.io.IOException;
32+
import java.util.Iterator;
3133

3234
public class ProfilingAggregator extends Aggregator {
3335

@@ -70,6 +72,16 @@ public Aggregator subAggregator(String name) {
7072
return delegate.subAggregator(name);
7173
}
7274

75+
@Override
76+
public Aggregator resolveSortPath(PathElement next, Iterator<PathElement> path) {
77+
return delegate.resolveSortPath(next, path);
78+
}
79+
80+
@Override
81+
public void validateSortPathKey(String key) {
82+
delegate.validateSortPathKey(key);
83+
}
84+
7385
@Override
7486
public InternalAggregation buildAggregation(long bucket) throws IOException {
7587
Timer timer = profileBreakdown.getTimer(AggregationTimingType.BUILD_AGGREGATION);

‎server/src/test/java/org/elasticsearch/search/aggregations/bucket/terms/TermsAggregatorTests.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1237,7 +1237,7 @@ public void testOrderByPipelineAggregation() throws Exception {
12371237

12381238
AggregationExecutionException e = expectThrows(AggregationExecutionException.class,
12391239
() -> createAggregator(termsAgg, indexSearcher, fieldType));
1240-
assertEquals("Invalid aggregator order path [script]. The provided aggregation [script] " +
1240+
assertEquals("Invalid aggregation order path [script]. The provided aggregation [script] " +
12411241
"either does not exist, or is a pipeline aggregation and cannot be used to sort the buckets.",
12421242
e.getMessage());
12431243
}

‎server/src/test/java/org/elasticsearch/search/profile/aggregation/AggregationProfilerIT.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,19 +28,20 @@
2828
import org.elasticsearch.search.profile.ProfileResult;
2929
import org.elasticsearch.search.profile.ProfileShardResult;
3030
import org.elasticsearch.test.ESIntegTestCase;
31+
3132
import java.util.ArrayList;
3233
import java.util.List;
3334
import java.util.Map;
3435
import java.util.stream.Collectors;
3536

3637
import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
37-
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
38-
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertSearchResponse;
3938
import static org.elasticsearch.search.aggregations.AggregationBuilders.avg;
4039
import static org.elasticsearch.search.aggregations.AggregationBuilders.diversifiedSampler;
4140
import static org.elasticsearch.search.aggregations.AggregationBuilders.histogram;
4241
import static org.elasticsearch.search.aggregations.AggregationBuilders.max;
4342
import static org.elasticsearch.search.aggregations.AggregationBuilders.terms;
43+
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
44+
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertSearchResponse;
4445
import static org.hamcrest.Matchers.equalTo;
4546
import static org.hamcrest.Matchers.greaterThan;
4647
import static org.hamcrest.Matchers.notNullValue;

0 commit comments

Comments
 (0)