honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Cut bucket_script and bucket_selector to registerPipelineAggregation 

and remove their PROTOTYPEs.

Relates to #17085
This commit is contained in:
Nik Everett 2016-04-15 15:49:07 -04:00
parent cdb36a2b0c
commit b8003de409
5 changed files with 262 additions and 334 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
core/src/main/java/org/elasticsearch/search/SearchModule.java
View File

@ -208,10 +208,10 @@ import org.elasticsearch.search.aggregations.pipeline.bucketmetrics.stats.extend
import org.elasticsearch.search.aggregations.pipeline.bucketmetrics.stats.extended.ExtendedStatsBucketPipelineAggregatorBuilder;
import org.elasticsearch.search.aggregations.pipeline.bucketmetrics.sum.SumBucketParser;
import org.elasticsearch.search.aggregations.pipeline.bucketmetrics.sum.SumBucketPipelineAggregator;
import org.elasticsearch.search.aggregations.pipeline.bucketscript.BucketScriptParser;
import org.elasticsearch.search.aggregations.pipeline.bucketscript.BucketScriptPipelineAggregator;
import org.elasticsearch.search.aggregations.pipeline.bucketselector.BucketSelectorParser;
import org.elasticsearch.search.aggregations.pipeline.bucketscript.BucketScriptPipelineAggregatorBuilder;
import org.elasticsearch.search.aggregations.pipeline.bucketselector.BucketSelectorPipelineAggregator;
import org.elasticsearch.search.aggregations.pipeline.bucketselector.BucketSelectorPipelineAggregatorBuilder;
import org.elasticsearch.search.aggregations.pipeline.cumulativesum.CumulativeSumPipelineAggregator;
import org.elasticsearch.search.aggregations.pipeline.cumulativesum.CumulativeSumPipelineAggregatorBuilder;
import org.elasticsearch.search.aggregations.pipeline.derivative.DerivativePipelineAggregator;
@ -514,8 +514,10 @@ public class SearchModule extends AbstractModule {
registerPipelineParser(new MovAvgParser(movAvgModelParserMapper));
registerPipelineAggregation(CumulativeSumPipelineAggregatorBuilder::new, CumulativeSumPipelineAggregatorBuilder::parse,
CumulativeSumPipelineAggregatorBuilder.AGGREGATION_NAME_FIELD);
registerPipelineParser(new BucketScriptParser());
registerPipelineParser(new BucketSelectorParser());
registerPipelineAggregation(BucketScriptPipelineAggregatorBuilder::new, BucketScriptPipelineAggregatorBuilder::parse,
BucketScriptPipelineAggregatorBuilder.AGGREGATION_NAME_FIELD);
registerPipelineAggregation(BucketSelectorPipelineAggregatorBuilder::new, BucketSelectorPipelineAggregatorBuilder::parse,
BucketSelectorPipelineAggregatorBuilder.AGGREGATION_NAME_FIELD);
registerPipelineAggregation(SerialDiffPipelineAggregatorBuilder::new, SerialDiffPipelineAggregatorBuilder::parse,
SerialDiffPipelineAggregatorBuilder.AGGREGATION_NAME_FIELD);

134
core/src/main/java/org/elasticsearch/search/aggregations/pipeline/bucketscript/BucketScriptParser.java
View File

@ -1,134 +0,0 @@
/*
* Licensed to Elasticsearch under one or more contributor
* license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright
* ownership. Elasticsearch licenses this file to you under
* the Apache License, Version 2.0 (the "License"); you may
* not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.elasticsearch.search.aggregations.pipeline.bucketscript;
import org.elasticsearch.common.ParseField;
import org.elasticsearch.common.ParsingException;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.index.query.QueryParseContext;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.Script.ScriptField;
import org.elasticsearch.search.aggregations.pipeline.BucketHelpers.GapPolicy;
import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class BucketScriptParser implements PipelineAggregator.Parser {
public static final ParseField FORMAT = new ParseField("format");
public static final ParseField GAP_POLICY = new ParseField("gap_policy");
public static final ParseField PARAMS_FIELD = new ParseField("params");
@Override
public String type() {
return BucketScriptPipelineAggregator.TYPE.name();
}
@Override
public BucketScriptPipelineAggregatorBuilder parse(String reducerName, QueryParseContext context) throws IOException {
XContentParser parser = context.parser();
XContentParser.Token token;
Script script = null;
String currentFieldName = null;
Map<String, String> bucketsPathsMap = null;
String format = null;
GapPolicy gapPolicy = null;
while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
if (token == XContentParser.Token.FIELD_NAME) {
currentFieldName = parser.currentName();
} else if (token == XContentParser.Token.VALUE_STRING) {
if (context.getParseFieldMatcher().match(currentFieldName, FORMAT)) {
format = parser.text();
} else if (context.getParseFieldMatcher().match(currentFieldName, BUCKETS_PATH)) {
bucketsPathsMap = new HashMap<>();
bucketsPathsMap.put("_value", parser.text());
} else if (context.getParseFieldMatcher().match(currentFieldName, GAP_POLICY)) {
gapPolicy = GapPolicy.parse(context, parser.text(), parser.getTokenLocation());
} else if (context.getParseFieldMatcher().match(currentFieldName, ScriptField.SCRIPT)) {
script = Script.parse(parser, context.getParseFieldMatcher());
} else {
throw new ParsingException(parser.getTokenLocation(),
"Unknown key for a " + token + " in [" + reducerName + "]: [" + currentFieldName + "].");
}
} else if (token == XContentParser.Token.START_ARRAY) {
if (context.getParseFieldMatcher().match(currentFieldName, BUCKETS_PATH)) {
List<String> paths = new ArrayList<>();
while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
String path = parser.text();
paths.add(path);
}
bucketsPathsMap = new HashMap<>();
for (int i = 0; i < paths.size(); i++) {
bucketsPathsMap.put("_value" + i, paths.get(i));
}
} else {
throw new ParsingException(parser.getTokenLocation(),
"Unknown key for a " + token + " in [" + reducerName + "]: [" + currentFieldName + "].");
}
} else if (token == XContentParser.Token.START_OBJECT) {
if (context.getParseFieldMatcher().match(currentFieldName, ScriptField.SCRIPT)) {
script = Script.parse(parser, context.getParseFieldMatcher());
} else if (context.getParseFieldMatcher().match(currentFieldName, BUCKETS_PATH)) {
Map<String, Object> map = parser.map();
bucketsPathsMap = new HashMap<>();
for (Map.Entry<String, Object> entry : map.entrySet()) {
bucketsPathsMap.put(entry.getKey(), String.valueOf(entry.getValue()));
}
} else {
throw new ParsingException(parser.getTokenLocation(),
"Unknown key for a " + token + " in [" + reducerName + "]: [" + currentFieldName + "].");
}
} else {
throw new ParsingException(parser.getTokenLocation(), "Unexpected token " + token + " in [" + reducerName + "].");
}
}
if (bucketsPathsMap == null) {
throw new ParsingException(parser.getTokenLocation(), "Missing required field [" + BUCKETS_PATH.getPreferredName()
+ "] for series_arithmetic aggregation [" + reducerName + "]");
}
if (script == null) {
throw new ParsingException(parser.getTokenLocation(), "Missing required field [" + ScriptField.SCRIPT.getPreferredName()
+ "] for series_arithmetic aggregation [" + reducerName + "]");
}
BucketScriptPipelineAggregatorBuilder factory =
new BucketScriptPipelineAggregatorBuilder(reducerName, bucketsPathsMap, script);
if (format != null) {
factory.format(format);
}
if (gapPolicy != null) {
factory.gapPolicy(gapPolicy);
}
return factory;
}
@Override
public BucketScriptPipelineAggregatorBuilder getFactoryPrototype() {
return BucketScriptPipelineAggregatorBuilder.PROTOTYPE;
}
}

168
core/src/main/java/org/elasticsearch/search/aggregations/pipeline/bucketscript/BucketScriptPipelineAggregatorBuilder.java
View File

@ -19,28 +19,36 @@
package org.elasticsearch.search.aggregations.pipeline.bucketscript;
import org.elasticsearch.common.ParseField;
import org.elasticsearch.common.ParsingException;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.index.query.QueryParseContext;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.Script.ScriptField;
import org.elasticsearch.search.DocValueFormat;
import org.elasticsearch.search.aggregations.pipeline.BucketHelpers.GapPolicy;
import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
import org.elasticsearch.search.aggregations.pipeline.PipelineAggregatorBuilder;
import org.elasticsearch.search.DocValueFormat;
import java.io.IOException;
import java.util.Collections;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.TreeMap;
public class BucketScriptPipelineAggregatorBuilder extends PipelineAggregatorBuilder<BucketScriptPipelineAggregatorBuilder> {
import static org.elasticsearch.search.aggregations.pipeline.PipelineAggregator.Parser.BUCKETS_PATH;
import static org.elasticsearch.search.aggregations.pipeline.PipelineAggregator.Parser.FORMAT;
import static org.elasticsearch.search.aggregations.pipeline.PipelineAggregator.Parser.GAP_POLICY;
static final BucketScriptPipelineAggregatorBuilder PROTOTYPE = new BucketScriptPipelineAggregatorBuilder("", Collections.emptyMap(),
new Script(""));
public class BucketScriptPipelineAggregatorBuilder extends PipelineAggregatorBuilder<BucketScriptPipelineAggregatorBuilder> {
public static final String NAME = BucketScriptPipelineAggregator.TYPE.name();
public static final ParseField AGGREGATION_NAME_FIELD = new ParseField(NAME);
private final Script script;
private final Map<String, String> bucketsPathsMap;
@ -58,6 +66,38 @@ public class BucketScriptPipelineAggregatorBuilder extends PipelineAggregatorBui
this(name, convertToBucketsPathMap(bucketsPaths), script);
}
/**
* Read from a stream.
*/
public BucketScriptPipelineAggregatorBuilder(StreamInput in) throws IOException {
super(in, BucketScriptPipelineAggregator.TYPE.name());
int mapSize = in.readVInt();
bucketsPathsMap = new HashMap<String, String>(mapSize);
for (int i = 0; i < mapSize; i++) {
bucketsPathsMap.put(in.readString(), in.readString());
}
script = Script.readScript(in);
format = in.readOptionalString();
gapPolicy = GapPolicy.readFrom(in);
}
@Override
protected void doWriteTo(StreamOutput out) throws IOException {
out.writeVInt(bucketsPathsMap.size());
for (Entry<String, String> e : bucketsPathsMap.entrySet()) {
out.writeString(e.getKey());
out.writeString(e.getValue());
}
script.writeTo(out);
out.writeOptionalString(format);
gapPolicy.writeTo(out);
}
@Override
protected boolean usesNewStyleSerialization() {
return true;
}
private static Map<String, String> convertToBucketsPathMap(String[] bucketsPaths) {
Map<String, String> bucketsPathsMap = new HashMap<>();
for (int i = 0; i < bucketsPaths.length; i++) {
@ -117,46 +157,101 @@ public class BucketScriptPipelineAggregatorBuilder extends PipelineAggregatorBui
@Override
protected XContentBuilder internalXContent(XContentBuilder builder, Params params) throws IOException {
builder.field(BucketScriptParser.BUCKETS_PATH.getPreferredName(), bucketsPathsMap);
builder.field(BUCKETS_PATH.getPreferredName(), bucketsPathsMap);
builder.field(ScriptField.SCRIPT.getPreferredName(), script);
if (format != null) {
builder.field(BucketScriptParser.FORMAT.getPreferredName(), format);
builder.field(FORMAT.getPreferredName(), format);
}
builder.field(BucketScriptParser.GAP_POLICY.getPreferredName(), gapPolicy.getName());
builder.field(GAP_POLICY.getPreferredName(), gapPolicy.getName());
return builder;
}
public static BucketScriptPipelineAggregatorBuilder parse(String reducerName, QueryParseContext context) throws IOException {
XContentParser parser = context.parser();
XContentParser.Token token;
Script script = null;
String currentFieldName = null;
Map<String, String> bucketsPathsMap = null;
String format = null;
GapPolicy gapPolicy = null;
while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
if (token == XContentParser.Token.FIELD_NAME) {
currentFieldName = parser.currentName();
} else if (token == XContentParser.Token.VALUE_STRING) {
if (context.getParseFieldMatcher().match(currentFieldName, FORMAT)) {
format = parser.text();
} else if (context.getParseFieldMatcher().match(currentFieldName, BUCKETS_PATH)) {
bucketsPathsMap = new HashMap<>();
bucketsPathsMap.put("_value", parser.text());
} else if (context.getParseFieldMatcher().match(currentFieldName, GAP_POLICY)) {
gapPolicy = GapPolicy.parse(context, parser.text(), parser.getTokenLocation());
} else if (context.getParseFieldMatcher().match(currentFieldName, ScriptField.SCRIPT)) {
script = Script.parse(parser, context.getParseFieldMatcher());
} else {
throw new ParsingException(parser.getTokenLocation(),
"Unknown key for a " + token + " in [" + reducerName + "]: [" + currentFieldName + "].");
}
} else if (token == XContentParser.Token.START_ARRAY) {
if (context.getParseFieldMatcher().match(currentFieldName, BUCKETS_PATH)) {
List<String> paths = new ArrayList<>();
while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
String path = parser.text();
paths.add(path);
}
bucketsPathsMap = new HashMap<>();
for (int i = 0; i < paths.size(); i++) {
bucketsPathsMap.put("_value" + i, paths.get(i));
}
} else {
throw new ParsingException(parser.getTokenLocation(),
"Unknown key for a " + token + " in [" + reducerName + "]: [" + currentFieldName + "].");
}
} else if (token == XContentParser.Token.START_OBJECT) {
if (context.getParseFieldMatcher().match(currentFieldName, ScriptField.SCRIPT)) {
script = Script.parse(parser, context.getParseFieldMatcher());
} else if (context.getParseFieldMatcher().match(currentFieldName, BUCKETS_PATH)) {
Map<String, Object> map = parser.map();
bucketsPathsMap = new HashMap<>();
for (Map.Entry<String, Object> entry : map.entrySet()) {
bucketsPathsMap.put(entry.getKey(), String.valueOf(entry.getValue()));
}
} else {
throw new ParsingException(parser.getTokenLocation(),
"Unknown key for a " + token + " in [" + reducerName + "]: [" + currentFieldName + "].");
}
} else {
throw new ParsingException(parser.getTokenLocation(), "Unexpected token " + token + " in [" + reducerName + "].");
}
}
if (bucketsPathsMap == null) {
throw new ParsingException(parser.getTokenLocation(), "Missing required field [" + BUCKETS_PATH.getPreferredName()
+ "] for series_arithmetic aggregation [" + reducerName + "]");
}
if (script == null) {
throw new ParsingException(parser.getTokenLocation(), "Missing required field [" + ScriptField.SCRIPT.getPreferredName()
+ "] for series_arithmetic aggregation [" + reducerName + "]");
}
BucketScriptPipelineAggregatorBuilder factory =
new BucketScriptPipelineAggregatorBuilder(reducerName, bucketsPathsMap, script);
if (format != null) {
factory.format(format);
}
if (gapPolicy != null) {
factory.gapPolicy(gapPolicy);
}
return factory;
}
@Override
protected boolean overrideBucketsPath() {
return true;
}
@Override
protected BucketScriptPipelineAggregatorBuilder doReadFrom(String name, String[] bucketsPaths, StreamInput in) throws IOException {
Map<String, String> bucketsPathsMap = new HashMap<String, String>();
int mapSize = in.readVInt();
for (int i = 0; i < mapSize; i++) {
bucketsPathsMap.put(in.readString(), in.readString());
}
Script script = Script.readScript(in);
BucketScriptPipelineAggregatorBuilder factory = new BucketScriptPipelineAggregatorBuilder(name, bucketsPathsMap, script);
factory.format = in.readOptionalString();
factory.gapPolicy = GapPolicy.readFrom(in);
return factory;
}
@Override
protected void doWriteTo(StreamOutput out) throws IOException {
out.writeVInt(bucketsPathsMap.size());
for (Entry<String, String> e : bucketsPathsMap.entrySet()) {
out.writeString(e.getKey());
out.writeString(e.getValue());
}
script.writeTo(out);
out.writeOptionalString(format);
gapPolicy.writeTo(out);
}
@Override
protected int doHashCode() {
return Objects.hash(bucketsPathsMap, script, format, gapPolicy);
@ -168,4 +263,9 @@ public class BucketScriptPipelineAggregatorBuilder extends PipelineAggregatorBui
return Objects.equals(bucketsPathsMap, other.bucketsPathsMap) && Objects.equals(script, other.script)
&& Objects.equals(format, other.format) && Objects.equals(gapPolicy, other.gapPolicy);
}
@Override
public String getWriteableName() {
return NAME;
}
}

129
core/src/main/java/org/elasticsearch/search/aggregations/pipeline/bucketselector/BucketSelectorParser.java
View File

@ -1,129 +0,0 @@
/*
* Licensed to Elasticsearch under one or more contributor
* license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright
* ownership. Elasticsearch licenses this file to you under
* the Apache License, Version 2.0 (the "License"); you may
* not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.elasticsearch.search.aggregations.pipeline.bucketselector;
import org.elasticsearch.common.ParseField;
import org.elasticsearch.common.ParsingException;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.index.query.QueryParseContext;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.Script.ScriptField;
import org.elasticsearch.search.aggregations.pipeline.BucketHelpers.GapPolicy;
import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class BucketSelectorParser implements PipelineAggregator.Parser {
public static final ParseField FORMAT = new ParseField("format");
public static final ParseField GAP_POLICY = new ParseField("gap_policy");
public static final ParseField PARAMS_FIELD = new ParseField("params");
@Override
public String type() {
return BucketSelectorPipelineAggregator.TYPE.name();
}
@Override
public BucketSelectorPipelineAggregatorBuilder parse(String reducerName, QueryParseContext context) throws IOException {
XContentParser parser = context.parser();
XContentParser.Token token;
Script script = null;
String currentFieldName = null;
Map<String, String> bucketsPathsMap = null;
GapPolicy gapPolicy = null;
while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
if (token == XContentParser.Token.FIELD_NAME) {
currentFieldName = parser.currentName();
} else if (token == XContentParser.Token.VALUE_STRING) {
if (context.getParseFieldMatcher().match(currentFieldName, BUCKETS_PATH)) {
bucketsPathsMap = new HashMap<>();
bucketsPathsMap.put("_value", parser.text());
} else if (context.getParseFieldMatcher().match(currentFieldName, GAP_POLICY)) {
gapPolicy = GapPolicy.parse(context, parser.text(), parser.getTokenLocation());
} else if (context.getParseFieldMatcher().match(currentFieldName, ScriptField.SCRIPT)) {
script = Script.parse(parser, context.getParseFieldMatcher());
} else {
throw new ParsingException(parser.getTokenLocation(),
"Unknown key for a " + token + " in [" + reducerName + "]: [" + currentFieldName + "].");
}
} else if (token == XContentParser.Token.START_ARRAY) {
if (context.getParseFieldMatcher().match(currentFieldName, BUCKETS_PATH)) {
List<String> paths = new ArrayList<>();
while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
String path = parser.text();
paths.add(path);
}
bucketsPathsMap = new HashMap<>();
for (int i = 0; i < paths.size(); i++) {
bucketsPathsMap.put("_value" + i, paths.get(i));
}
} else {
throw new ParsingException(parser.getTokenLocation(),
"Unknown key for a " + token + " in [" + reducerName + "]: [" + currentFieldName + "].");
}
} else if (token == XContentParser.Token.START_OBJECT) {
if (context.getParseFieldMatcher().match(currentFieldName, ScriptField.SCRIPT)) {
script = Script.parse(parser, context.getParseFieldMatcher());
} else if (context.getParseFieldMatcher().match(currentFieldName, BUCKETS_PATH)) {
Map<String, Object> map = parser.map();
bucketsPathsMap = new HashMap<>();
for (Map.Entry<String, Object> entry : map.entrySet()) {
bucketsPathsMap.put(entry.getKey(), String.valueOf(entry.getValue()));
}
} else {
throw new ParsingException(parser.getTokenLocation(),
"Unknown key for a " + token + " in [" + reducerName + "]: [" + currentFieldName + "].");
}
} else {
throw new ParsingException(parser.getTokenLocation(), "Unexpected token " + token + " in [" + reducerName + "].");
}
}
if (bucketsPathsMap == null) {
throw new ParsingException(parser.getTokenLocation(), "Missing required field [" + BUCKETS_PATH.getPreferredName()
+ "] for bucket_selector aggregation [" + reducerName + "]");
}
if (script == null) {
throw new ParsingException(parser.getTokenLocation(), "Missing required field [" + ScriptField.SCRIPT.getPreferredName()
+ "] for bucket_selector aggregation [" + reducerName + "]");
}
BucketSelectorPipelineAggregatorBuilder factory =
new BucketSelectorPipelineAggregatorBuilder(reducerName, bucketsPathsMap, script);
if (gapPolicy != null) {
factory.gapPolicy(gapPolicy);
}
return factory;
}
@Override
public BucketSelectorPipelineAggregatorBuilder getFactoryPrototype() {
return BucketSelectorPipelineAggregatorBuilder.PROTOTYPE;
}
}

155
core/src/main/java/org/elasticsearch/search/aggregations/pipeline/bucketselector/BucketSelectorPipelineAggregatorBuilder.java
View File

@ -19,32 +19,38 @@
package org.elasticsearch.search.aggregations.pipeline.bucketselector;
import org.elasticsearch.common.ParseField;
import org.elasticsearch.common.ParsingException;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.index.query.QueryParseContext;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.Script.ScriptField;
import org.elasticsearch.search.aggregations.pipeline.BucketHelpers.GapPolicy;
import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
import org.elasticsearch.search.aggregations.pipeline.PipelineAggregatorBuilder;
import org.elasticsearch.search.aggregations.pipeline.bucketscript.BucketScriptParser;
import java.io.IOException;
import java.util.Collections;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.TreeMap;
import static org.elasticsearch.search.aggregations.pipeline.PipelineAggregator.Parser.BUCKETS_PATH;
import static org.elasticsearch.search.aggregations.pipeline.PipelineAggregator.Parser.GAP_POLICY;
public class BucketSelectorPipelineAggregatorBuilder extends PipelineAggregatorBuilder<BucketSelectorPipelineAggregatorBuilder> {
public static final String NAME = BucketSelectorPipelineAggregator.TYPE.name();
public static final ParseField AGGREGATION_NAME_FIELD = new ParseField(NAME);
static final BucketSelectorPipelineAggregatorBuilder PROTOTYPE = new BucketSelectorPipelineAggregatorBuilder("",
Collections.emptyMap(), new Script(""));
private final Map<String, String> bucketsPathsMap;
private Script script;
private GapPolicy gapPolicy = GapPolicy.SKIP;
private final Map<String, String> bucketsPathsMap;
public BucketSelectorPipelineAggregatorBuilder(String name, Map<String, String> bucketsPathsMap, Script script) {
super(name, BucketSelectorPipelineAggregator.TYPE.name(), new TreeMap<>(bucketsPathsMap).values()
@ -57,6 +63,36 @@ public class BucketSelectorPipelineAggregatorBuilder extends PipelineAggregatorB
this(name, convertToBucketsPathMap(bucketsPaths), script);
}
/**
* Read from a stream.
*/
public BucketSelectorPipelineAggregatorBuilder(StreamInput in) throws IOException {
super(in, BucketSelectorPipelineAggregator.TYPE.name());
int mapSize = in.readVInt();
bucketsPathsMap = new HashMap<String, String>(mapSize);
for (int i = 0; i < mapSize; i++) {
bucketsPathsMap.put(in.readString(), in.readString());
}
script = Script.readScript(in);
gapPolicy = GapPolicy.readFrom(in);
}
@Override
protected void doWriteTo(StreamOutput out) throws IOException {
out.writeVInt(bucketsPathsMap.size());
for (Entry<String, String> e : bucketsPathsMap.entrySet()) {
out.writeString(e.getKey());
out.writeString(e.getValue());
}
script.writeTo(out);
gapPolicy.writeTo(out);
}
@Override
protected boolean usesNewStyleSerialization() {
return true;
}
private static Map<String, String> convertToBucketsPathMap(String[] bucketsPaths) {
Map<String, String> bucketsPathsMap = new HashMap<>();
for (int i = 0; i < bucketsPaths.length; i++) {
@ -90,42 +126,91 @@ public class BucketSelectorPipelineAggregatorBuilder extends PipelineAggregatorB
@Override
protected XContentBuilder internalXContent(XContentBuilder builder, Params params) throws IOException {
builder.field(BucketScriptParser.BUCKETS_PATH.getPreferredName(), bucketsPathsMap);
builder.field(BUCKETS_PATH.getPreferredName(), bucketsPathsMap);
builder.field(ScriptField.SCRIPT.getPreferredName(), script);
builder.field(BucketScriptParser.GAP_POLICY.getPreferredName(), gapPolicy.getName());
builder.field(GAP_POLICY.getPreferredName(), gapPolicy.getName());
return builder;
}
public static BucketSelectorPipelineAggregatorBuilder parse(String reducerName, QueryParseContext context) throws IOException {
XContentParser parser = context.parser();
XContentParser.Token token;
Script script = null;
String currentFieldName = null;
Map<String, String> bucketsPathsMap = null;
GapPolicy gapPolicy = null;
while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
if (token == XContentParser.Token.FIELD_NAME) {
currentFieldName = parser.currentName();
} else if (token == XContentParser.Token.VALUE_STRING) {
if (context.getParseFieldMatcher().match(currentFieldName, BUCKETS_PATH)) {
bucketsPathsMap = new HashMap<>();
bucketsPathsMap.put("_value", parser.text());
} else if (context.getParseFieldMatcher().match(currentFieldName, GAP_POLICY)) {
gapPolicy = GapPolicy.parse(context, parser.text(), parser.getTokenLocation());
} else if (context.getParseFieldMatcher().match(currentFieldName, ScriptField.SCRIPT)) {
script = Script.parse(parser, context.getParseFieldMatcher());
} else {
throw new ParsingException(parser.getTokenLocation(),
"Unknown key for a " + token + " in [" + reducerName + "]: [" + currentFieldName + "].");
}
} else if (token == XContentParser.Token.START_ARRAY) {
if (context.getParseFieldMatcher().match(currentFieldName, BUCKETS_PATH)) {
List<String> paths = new ArrayList<>();
while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
String path = parser.text();
paths.add(path);
}
bucketsPathsMap = new HashMap<>();
for (int i = 0; i < paths.size(); i++) {
bucketsPathsMap.put("_value" + i, paths.get(i));
}
} else {
throw new ParsingException(parser.getTokenLocation(),
"Unknown key for a " + token + " in [" + reducerName + "]: [" + currentFieldName + "].");
}
} else if (token == XContentParser.Token.START_OBJECT) {
if (context.getParseFieldMatcher().match(currentFieldName, ScriptField.SCRIPT)) {
script = Script.parse(parser, context.getParseFieldMatcher());
} else if (context.getParseFieldMatcher().match(currentFieldName, BUCKETS_PATH)) {
Map<String, Object> map = parser.map();
bucketsPathsMap = new HashMap<>();
for (Map.Entry<String, Object> entry : map.entrySet()) {
bucketsPathsMap.put(entry.getKey(), String.valueOf(entry.getValue()));
}
} else {
throw new ParsingException(parser.getTokenLocation(),
"Unknown key for a " + token + " in [" + reducerName + "]: [" + currentFieldName + "].");
}
} else {
throw new ParsingException(parser.getTokenLocation(), "Unexpected token " + token + " in [" + reducerName + "].");
}
}
if (bucketsPathsMap == null) {
throw new ParsingException(parser.getTokenLocation(), "Missing required field [" + BUCKETS_PATH.getPreferredName()
+ "] for bucket_selector aggregation [" + reducerName + "]");
}
if (script == null) {
throw new ParsingException(parser.getTokenLocation(), "Missing required field [" + ScriptField.SCRIPT.getPreferredName()
+ "] for bucket_selector aggregation [" + reducerName + "]");
}
BucketSelectorPipelineAggregatorBuilder factory =
new BucketSelectorPipelineAggregatorBuilder(reducerName, bucketsPathsMap, script);
if (gapPolicy != null) {
factory.gapPolicy(gapPolicy);
}
return factory;
}
@Override
protected boolean overrideBucketsPath() {
return true;
}
@Override
protected BucketSelectorPipelineAggregatorBuilder doReadFrom(String name, String[] bucketsPaths, StreamInput in)
throws IOException {
Map<String, String> bucketsPathsMap = new HashMap<String, String>();
int mapSize = in.readVInt();
for (int i = 0; i < mapSize; i++) {
bucketsPathsMap.put(in.readString(), in.readString());
}
Script script = Script.readScript(in);
BucketSelectorPipelineAggregatorBuilder factory = new BucketSelectorPipelineAggregatorBuilder(name, bucketsPathsMap, script);
factory.gapPolicy = GapPolicy.readFrom(in);
return factory;
}
@Override
protected void doWriteTo(StreamOutput out) throws IOException {
out.writeVInt(bucketsPathsMap.size());
for (Entry<String, String> e : bucketsPathsMap.entrySet()) {
out.writeString(e.getKey());
out.writeString(e.getValue());
}
script.writeTo(out);
gapPolicy.writeTo(out);
}
@Override
protected int doHashCode() {
return Objects.hash(bucketsPathsMap, script, gapPolicy);
@ -138,4 +223,8 @@ public class BucketSelectorPipelineAggregatorBuilder extends PipelineAggregatorB
&& Objects.equals(gapPolicy, other.gapPolicy);
}
@Override
public String getWriteableName() {
return NAME;
}
}