diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/dataframe/DataFrameField.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/dataframe/DataFrameField.java index d2956e6559b0e..9749cd915b54e 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/dataframe/DataFrameField.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/dataframe/DataFrameField.java @@ -14,9 +14,12 @@ public final class DataFrameField { // common parse fields + public static final ParseField AGGREGATIONS = new ParseField("aggregations"); + public static final ParseField AGGS = new ParseField("aggs"); public static final ParseField ID = new ParseField("id"); public static final ParseField TRANSFORMS = new ParseField("transforms"); public static final ParseField COUNT = new ParseField("count"); + public static final ParseField GROUP_BY = new ParseField("group_by"); public static final ParseField TIMEOUT = new ParseField("timeout"); public static final ParseField WAIT_FOR_COMPLETION = new ParseField("wait_for_completion"); public static final ParseField STATS_FIELD = new ParseField("stats"); diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/dataframe/DataFrameMessages.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/dataframe/DataFrameMessages.java index e1b94425c3b06..a395dcdb3dfd9 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/dataframe/DataFrameMessages.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/dataframe/DataFrameMessages.java @@ -33,6 +33,10 @@ public class DataFrameMessages { "Failed to parse transform configuration for data frame transform [{0}]"; public static final String DATA_FRAME_TRANSFORM_CONFIGURATION_NO_TRANSFORM = "Data frame transform configuration must specify exactly 1 function"; + public static final String DATA_FRAME_TRANSFORM_CONFIGURATION_PIVOT_NO_GROUP_BY = + "Data frame pivot transform configuration must specify at least 1 group_by"; + public static final String DATA_FRAME_TRANSFORM_CONFIGURATION_PIVOT_NO_AGGREGATION = + "Data frame pivot transform configuration must specify at least 1 aggregation"; public static final String DATA_FRAME_TRANSFORM_PIVOT_FAILED_TO_CREATE_COMPOSITE_AGGREGATION = "Failed to create composite aggregation from pivot function"; public static final String DATA_FRAME_TRANSFORM_CONFIGURATION_INVALID = @@ -40,6 +44,8 @@ public class DataFrameMessages { public static final String LOG_DATA_FRAME_TRANSFORM_CONFIGURATION_BAD_QUERY = "Failed to parse query for data frame transform"; + public static final String LOG_DATA_FRAME_TRANSFORM_CONFIGURATION_BAD_GROUP_BY = + "Failed to parse group_by for data frame pivot transform"; public static final String LOG_DATA_FRAME_TRANSFORM_CONFIGURATION_BAD_AGGREGATION = "Failed to parse aggregation for data frame pivot transform"; diff --git a/x-pack/plugin/data-frame/qa/single-node-tests/src/test/java/org/elasticsearch/xpack/dataframe/integration/DataFramePivotRestIT.java b/x-pack/plugin/data-frame/qa/single-node-tests/src/test/java/org/elasticsearch/xpack/dataframe/integration/DataFramePivotRestIT.java index 6cf07fd88e0c2..eb8203e1dd2e2 100644 --- a/x-pack/plugin/data-frame/qa/single-node-tests/src/test/java/org/elasticsearch/xpack/dataframe/integration/DataFramePivotRestIT.java +++ b/x-pack/plugin/data-frame/qa/single-node-tests/src/test/java/org/elasticsearch/xpack/dataframe/integration/DataFramePivotRestIT.java @@ -87,11 +87,11 @@ public void testHistogramPivot() throws Exception { config += " \"pivot\": {" - + " \"group_by\": [ {" + + " \"group_by\": {" + " \"every_2\": {" + " \"histogram\": {" + " \"interval\": 2,\"field\":\"stars\"" - + " } } } ]," + + " } } }," + " \"aggregations\": {" + " \"avg_rating\": {" + " \"avg\": {" @@ -125,11 +125,11 @@ public void testBiggerPivot() throws Exception { config += " \"pivot\": {" - + " \"group_by\": [ {" + + " \"group_by\": {" + " \"reviewer\": {" + " \"terms\": {" + " \"field\": \"user_id\"" - + " } } } ]," + + " } } }," + " \"aggregations\": {" + " \"avg_rating\": {" + " \"avg\": {" @@ -199,11 +199,11 @@ public void testDateHistogramPivot() throws Exception { config += " \"pivot\": {" - + " \"group_by\": [ {" + + " \"group_by\": {" + " \"by_day\": {" + " \"date_histogram\": {" + " \"interval\": \"1d\",\"field\":\"timestamp\",\"format\":\"yyyy-MM-DD\"" - + " } } } ]," + + " } } }," + " \"aggregations\": {" + " \"avg_rating\": {" + " \"avg\": {" diff --git a/x-pack/plugin/data-frame/qa/single-node-tests/src/test/java/org/elasticsearch/xpack/dataframe/integration/DataFrameRestTestCase.java b/x-pack/plugin/data-frame/qa/single-node-tests/src/test/java/org/elasticsearch/xpack/dataframe/integration/DataFrameRestTestCase.java index d31c63de54279..bd6812ae4896d 100644 --- a/x-pack/plugin/data-frame/qa/single-node-tests/src/test/java/org/elasticsearch/xpack/dataframe/integration/DataFrameRestTestCase.java +++ b/x-pack/plugin/data-frame/qa/single-node-tests/src/test/java/org/elasticsearch/xpack/dataframe/integration/DataFrameRestTestCase.java @@ -125,11 +125,11 @@ protected void createPivotReviewsTransform(String transformId, String dataFrameI } config += " \"pivot\": {" - + " \"group_by\": [ {" + + " \"group_by\": {" + " \"reviewer\": {" + " \"terms\": {" + " \"field\": \"user_id\"" - + " } } } ]," + + " } } }," + " \"aggregations\": {" + " \"avg_rating\": {" + " \"avg\": {" diff --git a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationConfig.java b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationConfig.java index d74b0cd36ffbb..54b6109520a5b 100644 --- a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationConfig.java +++ b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationConfig.java @@ -70,19 +70,27 @@ public static AggregationConfig fromXContent(final XContentParser parser, boolea NamedXContentRegistry registry = parser.getXContentRegistry(); Map source = parser.mapOrdered(); AggregatorFactories.Builder aggregations = null; - try (XContentBuilder xContentBuilder = XContentFactory.jsonBuilder().map(source); - XContentParser sourceParser = XContentType.JSON.xContent().createParser(registry, LoggingDeprecationHandler.INSTANCE, - BytesReference.bytes(xContentBuilder).streamInput())) { - sourceParser.nextToken(); - aggregations = AggregatorFactories.parseAggregators(sourceParser); - } catch (Exception e) { + + if (source.isEmpty()) { if (lenient) { - logger.warn(DataFrameMessages.LOG_DATA_FRAME_TRANSFORM_CONFIGURATION_BAD_AGGREGATION, e); + logger.warn(DataFrameMessages.DATA_FRAME_TRANSFORM_CONFIGURATION_PIVOT_NO_AGGREGATION); } else { - throw e; + throw new IllegalArgumentException(DataFrameMessages.DATA_FRAME_TRANSFORM_CONFIGURATION_PIVOT_NO_AGGREGATION); + } + } else { + try (XContentBuilder xContentBuilder = XContentFactory.jsonBuilder().map(source); + XContentParser sourceParser = XContentType.JSON.xContent().createParser(registry, LoggingDeprecationHandler.INSTANCE, + BytesReference.bytes(xContentBuilder).streamInput())) { + sourceParser.nextToken(); + aggregations = AggregatorFactories.parseAggregators(sourceParser); + } catch (Exception e) { + if (lenient) { + logger.warn(DataFrameMessages.LOG_DATA_FRAME_TRANSFORM_CONFIGURATION_BAD_AGGREGATION, e); + } else { + throw e; + } } } - return new AggregationConfig(source, aggregations); } diff --git a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationResultUtils.java b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationResultUtils.java index 234bbb8626c1a..f301e64053664 100644 --- a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationResultUtils.java +++ b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationResultUtils.java @@ -27,22 +27,22 @@ final class AggregationResultUtils { * Extracts aggregation results from a composite aggregation and puts it into a map. * * @param agg The aggregation result - * @param sources The original sources used for querying + * @param groups The original groupings used for querying * @param aggregationBuilders the aggregation used for querying * @param dataFrameIndexerTransformStats stats collector * @return a map containing the results of the aggregation in a consumable way */ public static Stream> extractCompositeAggregationResults(CompositeAggregation agg, - Iterable sources, Collection aggregationBuilders, + GroupConfig groups, Collection aggregationBuilders, DataFrameIndexerTransformStats dataFrameIndexerTransformStats) { return agg.getBuckets().stream().map(bucket -> { dataFrameIndexerTransformStats.incrementNumDocuments(bucket.getDocCount()); Map document = new HashMap<>(); - for (GroupConfig source : sources) { - String destinationFieldName = source.getDestinationFieldName(); + groups.getGroups().keySet().forEach(destinationFieldName -> { document.put(destinationFieldName, bucket.getKey().get(destinationFieldName)); - } + }); + for (AggregationBuilder aggregationBuilder : aggregationBuilders) { String aggName = aggregationBuilder.getName(); diff --git a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/DateHistogramGroupSource.java b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/DateHistogramGroupSource.java index 539b4d221304b..59efac481d4d1 100644 --- a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/DateHistogramGroupSource.java +++ b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/DateHistogramGroupSource.java @@ -83,6 +83,11 @@ public static DateHistogramGroupSource fromXContent(final XContentParser parser, return lenient ? LENIENT_PARSER.apply(parser, null) : STRICT_PARSER.apply(parser, null); } + @Override + public Type getType() { + return Type.DATE_HISTOGRAM; + } + public long getInterval() { return interval; } diff --git a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/GroupConfig.java b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/GroupConfig.java index 4792d59cdac59..8ace9d64d9737 100644 --- a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/GroupConfig.java +++ b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/GroupConfig.java @@ -6,17 +6,29 @@ package org.elasticsearch.xpack.dataframe.transforms.pivot; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import org.elasticsearch.common.ParsingException; +import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.common.io.stream.Writeable; +import org.elasticsearch.common.xcontent.LoggingDeprecationHandler; +import org.elasticsearch.common.xcontent.NamedXContentRegistry; import org.elasticsearch.common.xcontent.ToXContentObject; import org.elasticsearch.common.xcontent.XContentBuilder; +import org.elasticsearch.common.xcontent.XContentFactory; import org.elasticsearch.common.xcontent.XContentParser; +import org.elasticsearch.common.xcontent.XContentType; +import org.elasticsearch.xpack.core.dataframe.DataFrameField; +import org.elasticsearch.xpack.core.dataframe.DataFrameMessages; +import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper; import org.elasticsearch.xpack.dataframe.transforms.pivot.SingleGroupSource.Type; import java.io.IOException; +import java.util.LinkedHashMap; import java.util.Locale; +import java.util.Map; import java.util.Objects; import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken; @@ -26,58 +38,53 @@ */ public class GroupConfig implements Writeable, ToXContentObject { - private final String destinationFieldName; - private final SingleGroupSource.Type groupType; - private final SingleGroupSource groupSource; + private static final Logger logger = LogManager.getLogger(GroupConfig.class); - public GroupConfig(final String destinationFieldName, final SingleGroupSource.Type groupType, final SingleGroupSource groupSource) { - this.destinationFieldName = Objects.requireNonNull(destinationFieldName); - this.groupType = Objects.requireNonNull(groupType); - this.groupSource = Objects.requireNonNull(groupSource); + private final Map source; + private final Map> groups; + + public GroupConfig(final Map source, final Map> groups) { + this.source = ExceptionsHelper.requireNonNull(source, DataFrameField.GROUP_BY.getPreferredName()); + this.groups = groups; } public GroupConfig(StreamInput in) throws IOException { - destinationFieldName = in.readString(); - groupType = Type.fromId(in.readByte()); - switch (groupType) { - case TERMS: - groupSource = in.readOptionalWriteable(TermsGroupSource::new); - break; - case HISTOGRAM: - groupSource = in.readOptionalWriteable(HistogramGroupSource::new); - break; - case DATE_HISTOGRAM: - groupSource = in.readOptionalWriteable(DateHistogramGroupSource::new); - break; - default: - throw new IOException("Unknown group type"); - } + source = in.readMap(); + groups = in.readMap(StreamInput::readString, (stream) -> { + Type groupType = Type.fromId(stream.readByte()); + switch (groupType) { + case TERMS: + return new TermsGroupSource(stream); + case HISTOGRAM: + return new HistogramGroupSource(stream); + case DATE_HISTOGRAM: + return new DateHistogramGroupSource(stream); + default: + throw new IOException("Unknown group type"); + } + }); } - @Override - public void writeTo(StreamOutput out) throws IOException { - out.writeString(destinationFieldName); - out.writeByte(groupType.getId()); - out.writeOptionalWriteable(groupSource); + public Map > getGroups() { + return groups; } - @Override - public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { - builder.startObject(); - builder.startObject(destinationFieldName); - - builder.field(groupType.value(), groupSource); - builder.endObject(); - builder.endObject(); - return builder; + public boolean isValid() { + return this.groups != null; } - public String getDestinationFieldName() { - return destinationFieldName; + @Override + public void writeTo(StreamOutput out) throws IOException { + out.writeMap(source); + out.writeMap(groups, StreamOutput::writeString, (stream, value) -> { + stream.writeByte(value.getType().getId()); + value.writeTo(stream); + }); } - public SingleGroupSource getGroupSource() { - return groupSource; + @Override + public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { + return builder.map(source); } @Override @@ -92,19 +99,44 @@ public boolean equals(Object other) { final GroupConfig that = (GroupConfig) other; - return Objects.equals(this.destinationFieldName, that.destinationFieldName) && Objects.equals(this.groupType, that.groupType) - && Objects.equals(this.groupSource, that.groupSource); + return Objects.equals(this.source, that.source) && Objects.equals(this.groups, that.groups); } @Override public int hashCode() { - return Objects.hash(destinationFieldName, groupType, groupSource); + return Objects.hash(source, groups); } public static GroupConfig fromXContent(final XContentParser parser, boolean lenient) throws IOException { - String destinationFieldName; - Type groupType; - SingleGroupSource groupSource; + NamedXContentRegistry registry = parser.getXContentRegistry(); + Map source = parser.mapOrdered(); + Map> groups = null; + + if (source.isEmpty()) { + if (lenient) { + logger.warn(DataFrameMessages.DATA_FRAME_TRANSFORM_CONFIGURATION_PIVOT_NO_GROUP_BY); + } else { + throw new IllegalArgumentException(DataFrameMessages.DATA_FRAME_TRANSFORM_CONFIGURATION_PIVOT_NO_GROUP_BY); + } + } else { + try (XContentBuilder xContentBuilder = XContentFactory.jsonBuilder().map(source); + XContentParser sourceParser = XContentType.JSON.xContent().createParser(registry, LoggingDeprecationHandler.INSTANCE, + BytesReference.bytes(xContentBuilder).streamInput())) { + groups = parseGroupConfig(sourceParser, lenient); + } catch (Exception e) { + if (lenient) { + logger.warn(DataFrameMessages.LOG_DATA_FRAME_TRANSFORM_CONFIGURATION_BAD_GROUP_BY, e); + } else { + throw e; + } + } + } + return new GroupConfig(source, groups); + } + + private static Map> parseGroupConfig(final XContentParser parser, + boolean lenient) throws IOException { + LinkedHashMap> groups = new LinkedHashMap<>(); // be parsing friendly, whether the token needs to be advanced or not (similar to what ObjectParser does) XContentParser.Token token; @@ -116,19 +148,21 @@ public static GroupConfig fromXContent(final XContentParser parser, boolean leni throw new ParsingException(parser.getTokenLocation(), "Failed to parse object: Expected START_OBJECT but was: " + token); } } - token = parser.nextToken(); - ensureExpectedToken(XContentParser.Token.FIELD_NAME, token, parser::getTokenLocation); - destinationFieldName = parser.currentName(); - token = parser.nextToken(); - ensureExpectedToken(XContentParser.Token.START_OBJECT, token, parser::getTokenLocation); - token = parser.nextToken(); - ensureExpectedToken(XContentParser.Token.FIELD_NAME, token, parser::getTokenLocation); - groupType = SingleGroupSource.Type.valueOf(parser.currentName().toUpperCase(Locale.ROOT)); - - token = parser.nextToken(); - ensureExpectedToken(XContentParser.Token.START_OBJECT, token, parser::getTokenLocation); - - switch (groupType) { + + while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) { + + ensureExpectedToken(XContentParser.Token.FIELD_NAME, token, parser::getTokenLocation); + String destinationFieldName = parser.currentName(); + token = parser.nextToken(); + ensureExpectedToken(XContentParser.Token.START_OBJECT, token, parser::getTokenLocation); + token = parser.nextToken(); + ensureExpectedToken(XContentParser.Token.FIELD_NAME, token, parser::getTokenLocation); + Type groupType = SingleGroupSource.Type.valueOf(parser.currentName().toUpperCase(Locale.ROOT)); + + token = parser.nextToken(); + ensureExpectedToken(XContentParser.Token.START_OBJECT, token, parser::getTokenLocation); + SingleGroupSource groupSource; + switch (groupType) { case TERMS: groupSource = TermsGroupSource.fromXContent(parser, lenient); break; @@ -140,11 +174,12 @@ public static GroupConfig fromXContent(final XContentParser parser, boolean leni break; default: throw new ParsingException(parser.getTokenLocation(), "invalid grouping type: " + groupType); - } + } - parser.nextToken(); - parser.nextToken(); + parser.nextToken(); - return new GroupConfig(destinationFieldName, groupType, groupSource); + groups.put(destinationFieldName, groupSource); + } + return groups; } } diff --git a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/HistogramGroupSource.java b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/HistogramGroupSource.java index 2e6101368619e..3c75dcdedc1b2 100644 --- a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/HistogramGroupSource.java +++ b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/HistogramGroupSource.java @@ -49,6 +49,11 @@ private static ConstructingObjectParser createParser return parser; } + @Override + public Type getType() { + return Type.HISTOGRAM; + } + public static HistogramGroupSource fromXContent(final XContentParser parser, boolean lenient) throws IOException { return lenient ? LENIENT_PARSER.apply(parser, null) : STRICT_PARSER.apply(parser, null); } diff --git a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/Pivot.java b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/Pivot.java index 26760d6f167cf..ca4a7ec8eb4fb 100644 --- a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/Pivot.java +++ b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/Pivot.java @@ -76,10 +76,11 @@ public SearchRequest buildSearchRequest(Map position) { public Stream> extractResults(CompositeAggregation agg, DataFrameIndexerTransformStats dataFrameIndexerTransformStats) { - Iterable sources = config.getGroups(); + + GroupConfig groups = config.getGroupConfig(); Collection aggregationBuilders = config.getAggregationConfig().getAggregatorFactories(); - return AggregationResultUtils.extractCompositeAggregationResults(agg, sources, aggregationBuilders, dataFrameIndexerTransformStats); + return AggregationResultUtils.extractCompositeAggregationResults(agg, groups, aggregationBuilders, dataFrameIndexerTransformStats); } private void runTestQuery(Client client, final ActionListener listener) { diff --git a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/PivotConfig.java b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/PivotConfig.java index 06fca1eea2d3d..086268b169fbf 100644 --- a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/PivotConfig.java +++ b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/PivotConfig.java @@ -6,7 +6,6 @@ package org.elasticsearch.xpack.dataframe.transforms.pivot; -import org.elasticsearch.common.ParseField; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.common.io.stream.Writeable; @@ -15,10 +14,11 @@ import org.elasticsearch.common.xcontent.XContentBuilder; import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.search.aggregations.bucket.composite.CompositeAggregationBuilder; +import org.elasticsearch.xpack.core.dataframe.DataFrameField; import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper; import java.io.IOException; -import java.util.List; +import java.util.Map.Entry; import java.util.Objects; import static org.elasticsearch.common.xcontent.ConstructingObjectParser.constructorArg; @@ -27,12 +27,7 @@ public class PivotConfig implements Writeable, ToXContentObject { private static final String NAME = "data_frame_transform_pivot"; - private static final ParseField GROUP_BY = new ParseField("group_by"); - private static final ParseField AGGREGATIONS = new ParseField("aggregations"); - private static final ParseField AGGS = new ParseField("aggs"); - - - private final List groups; + private final GroupConfig groups; private final AggregationConfig aggregationConfig; private static final ConstructingObjectParser STRICT_PARSER = createParser(false); @@ -41,8 +36,7 @@ public class PivotConfig implements Writeable, ToXContentObject { private static ConstructingObjectParser createParser(boolean lenient) { ConstructingObjectParser parser = new ConstructingObjectParser<>(NAME, lenient, args -> { - @SuppressWarnings("unchecked") - List groups = (List) args[0]; + GroupConfig groups = (GroupConfig) args[0]; // allow "aggs" and "aggregations" but require one to be specified // if somebody specifies both: throw @@ -64,30 +58,30 @@ private static ConstructingObjectParser createParser(boolean return new PivotConfig(groups, aggregationConfig); }); - parser.declareObjectArray(constructorArg(), - (p, c) -> (GroupConfig.fromXContent(p, lenient)), GROUP_BY); + parser.declareObject(constructorArg(), + (p, c) -> (GroupConfig.fromXContent(p, lenient)), DataFrameField.GROUP_BY); - parser.declareObject(optionalConstructorArg(), (p, c) -> AggregationConfig.fromXContent(p, lenient), AGGREGATIONS); - parser.declareObject(optionalConstructorArg(), (p, c) -> AggregationConfig.fromXContent(p, lenient), AGGS); + parser.declareObject(optionalConstructorArg(), (p, c) -> AggregationConfig.fromXContent(p, lenient), DataFrameField.AGGREGATIONS); + parser.declareObject(optionalConstructorArg(), (p, c) -> AggregationConfig.fromXContent(p, lenient), DataFrameField.AGGS); return parser; } - public PivotConfig(final List groups, final AggregationConfig aggregationConfig) { - this.groups = ExceptionsHelper.requireNonNull(groups, GROUP_BY.getPreferredName()); - this.aggregationConfig = ExceptionsHelper.requireNonNull(aggregationConfig, AGGREGATIONS.getPreferredName()); + public PivotConfig(final GroupConfig groups, final AggregationConfig aggregationConfig) { + this.groups = ExceptionsHelper.requireNonNull(groups, DataFrameField.GROUP_BY.getPreferredName()); + this.aggregationConfig = ExceptionsHelper.requireNonNull(aggregationConfig, DataFrameField.AGGREGATIONS.getPreferredName()); } public PivotConfig(StreamInput in) throws IOException { - this.groups = in.readList(GroupConfig::new); + this.groups = new GroupConfig(in); this.aggregationConfig = new AggregationConfig(in); } @Override public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { builder.startObject(); - builder.field(GROUP_BY.getPreferredName(), groups); - builder.field(AGGREGATIONS.getPreferredName(), aggregationConfig); + builder.field(DataFrameField.GROUP_BY.getPreferredName(), groups); + builder.field(DataFrameField.AGGREGATIONS.getPreferredName(), aggregationConfig); builder.endObject(); return builder; } @@ -96,16 +90,22 @@ public void toCompositeAggXContent(XContentBuilder builder, Params params) throw builder.startObject(); builder.field(CompositeAggregationBuilder.SOURCES_FIELD_NAME.getPreferredName()); builder.startArray(); - for (GroupConfig group : groups) { - group.toXContent(builder, params); + + for (Entry> groupBy : groups.getGroups().entrySet()) { + builder.startObject(); + builder.startObject(groupBy.getKey()); + builder.field(groupBy.getValue().getType().value(), groupBy.getValue()); + builder.endObject(); + builder.endObject(); } + builder.endArray(); builder.endObject(); // sources } @Override public void writeTo(StreamOutput out) throws IOException { - out.writeList(groups); + groups.writeTo(out); aggregationConfig.writeTo(out); } @@ -113,7 +113,7 @@ public AggregationConfig getAggregationConfig() { return aggregationConfig; } - public Iterable getGroups() { + public GroupConfig getGroupConfig() { return groups; } @@ -138,7 +138,7 @@ public int hashCode() { } public boolean isValid() { - return aggregationConfig.isValid(); + return groups.isValid() && aggregationConfig.isValid(); } public static PivotConfig fromXContent(final XContentParser parser, boolean lenient) throws IOException { diff --git a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/SchemaUtil.java b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/SchemaUtil.java index 55df001d6cc63..619e4514d7674 100644 --- a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/SchemaUtil.java +++ b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/SchemaUtil.java @@ -34,8 +34,8 @@ public static void deduceMappings(final Client client, final PivotConfig config, // collects the fieldnames and target fieldnames used for grouping Map fieldNamesForGrouping = new HashMap<>(); - config.getGroups().forEach(group -> { - fieldNamesForGrouping.put(group.getDestinationFieldName(), group.getGroupSource().getField()); + config.getGroupConfig().getGroups().forEach((destinationFieldName, group) -> { + fieldNamesForGrouping.put(destinationFieldName, group.getField()); }); for (AggregationBuilder agg : config.getAggregationConfig().getAggregatorFactories()) { diff --git a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/SingleGroupSource.java b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/SingleGroupSource.java index 5cd65124f0650..9b309e59af4c3 100644 --- a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/SingleGroupSource.java +++ b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/SingleGroupSource.java @@ -45,6 +45,10 @@ public static Type fromId(byte id) { switch (id) { case 0: return TERMS; + case 1: + return HISTOGRAM; + case 2: + return DATE_HISTOGRAM; default: throw new IllegalArgumentException("unknown type"); } @@ -89,6 +93,8 @@ public void writeTo(StreamOutput out) throws IOException { out.writeOptionalString(field); } + public abstract Type getType(); + public String getField() { return field; } diff --git a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/TermsGroupSource.java b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/TermsGroupSource.java index 5518c8eb5052f..b3073f0e1de21 100644 --- a/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/TermsGroupSource.java +++ b/x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/TermsGroupSource.java @@ -39,6 +39,11 @@ public TermsGroupSource(StreamInput in) throws IOException { super(in); } + @Override + public Type getType() { + return Type.TERMS; + } + public static TermsGroupSource fromXContent(final XContentParser parser, boolean lenient) throws IOException { return lenient ? LENIENT_PARSER.apply(parser, null) : STRICT_PARSER.apply(parser, null); } diff --git a/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/DataFrameTransformConfigTests.java b/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/DataFrameTransformConfigTests.java index 0bac3de558c2d..daabe1cccaa39 100644 --- a/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/DataFrameTransformConfigTests.java +++ b/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/DataFrameTransformConfigTests.java @@ -68,11 +68,11 @@ public void testDefaultMatchAll( ) throws IOException { + " \"source\" : \"src\"," + " \"dest\" : \"dest\"," + " \"pivot\" : {" - + " \"group_by\": [ {" + + " \"group_by\": {" + " \"id\": {" + " \"terms\": {" + " \"field\": \"id\"" - + "} } } ]," + + "} } }," + " \"aggs\": {" + " \"avg\": {" + " \"avg\": {" diff --git a/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationConfigTests.java b/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationConfigTests.java index 9328a11f049f6..ccf9090182349 100644 --- a/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationConfigTests.java +++ b/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationConfigTests.java @@ -40,7 +40,7 @@ public static AggregationConfig randomAggregationConfig() { // ensure that the unlikely does not happen: 2 aggs share the same name Set names = new HashSet<>(); - for (int i = 1; i < randomIntBetween(1, 20); ++i) { + for (int i = 0; i < randomIntBetween(1, 20); ++i) { AggregationBuilder aggBuilder = getRandomSupportedAggregation(); if (names.add(aggBuilder.getName())) { builder.addAggregator(aggBuilder); @@ -88,6 +88,21 @@ protected Reader instanceReader() { return AggregationConfig::new; } + public void testEmptyAggregation() throws IOException { + String source = "{}"; + + // lenient, passes but reports invalid + try (XContentParser parser = createParser(JsonXContent.jsonXContent, source)) { + AggregationConfig aggregationConfig = AggregationConfig.fromXContent(parser, true); + assertFalse(aggregationConfig.isValid()); + } + + // strict throws + try (XContentParser parser = createParser(JsonXContent.jsonXContent, source)) { + expectThrows(IllegalArgumentException.class, () -> AggregationConfig.fromXContent(parser, false)); + } + } + public void testFailOnStrictPassOnLenient() throws IOException { String source = "{\n" + " \"avg_rating\": { \"some_removed_agg\": { \"field\": \"rating\" } }\n" + diff --git a/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationResultUtilsTests.java b/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationResultUtilsTests.java index a0d5c4851212e..49829750e954a 100644 --- a/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationResultUtilsTests.java +++ b/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/AggregationResultUtilsTests.java @@ -8,6 +8,7 @@ import org.elasticsearch.common.ParseField; import org.elasticsearch.common.xcontent.ContextParser; +import org.elasticsearch.common.xcontent.DeprecationHandler; import org.elasticsearch.common.xcontent.NamedXContentRegistry; import org.elasticsearch.common.xcontent.XContentBuilder; import org.elasticsearch.common.xcontent.XContentFactory; @@ -54,7 +55,6 @@ import java.util.stream.Collectors; import static java.util.Arrays.asList; -import static org.elasticsearch.xpack.dataframe.transforms.pivot.SingleGroupSource.Type.TERMS; public class AggregationResultUtilsTests extends ESTestCase { @@ -93,9 +93,10 @@ protected NamedXContentRegistry xContentRegistry() { public void testExtractCompositeAggregationResults() throws IOException { String targetField = randomAlphaOfLengthBetween(5, 10); - List sources = Collections.singletonList( - new GroupConfig(targetField, TERMS, new TermsGroupSource("doesn't_matter_for_this_test")) - ); + GroupConfig groupBy = parseGroupConfig("{ \"" + targetField + "\" : {" + + "\"terms\" : {" + + " \"field\" : \"doesn't_matter_for_this_test\"" + + "} } }"); String aggName = randomAlphaOfLengthBetween(5, 10); String aggTypedName = "avg#" + aggName; @@ -139,17 +140,23 @@ aggTypedName, asMap( ) ); - executeTest(sources, aggregationBuilders, input, expected, 20); + executeTest(groupBy, aggregationBuilders, input, expected, 20); } public void testExtractCompositeAggregationResultsMultiSources() throws IOException { String targetField = randomAlphaOfLengthBetween(5, 10); String targetField2 = randomAlphaOfLengthBetween(5, 10) + "_2"; - List sources = asList( - new GroupConfig(targetField, TERMS, new TermsGroupSource("doesn't_matter_for_this_test")), - new GroupConfig(targetField2, TERMS, new TermsGroupSource("doesn't_matter_for_this_test")) - ); + GroupConfig groupBy = parseGroupConfig("{" + + "\"" + targetField + "\" : {" + + " \"terms\" : {" + + " \"field\" : \"doesn't_matter_for_this_test\"" + + " } }," + + "\"" + targetField2 + "\" : {" + + " \"terms\" : {" + + " \"field\" : \"doesn't_matter_for_this_test\"" + + " } }" + + "}"); String aggName = randomAlphaOfLengthBetween(5, 10); String aggTypedName = "avg#" + aggName; @@ -214,15 +221,16 @@ aggTypedName, asMap( aggName, 12.55 ) ); - executeTest(sources, aggregationBuilders, input, expected, 10); + executeTest(groupBy, aggregationBuilders, input, expected, 10); } public void testExtractCompositeAggregationResultsMultiAggregations() throws IOException { String targetField = randomAlphaOfLengthBetween(5, 10); - List sources = Collections.singletonList( - new GroupConfig(targetField, TERMS, new TermsGroupSource("doesn't_matter_for_this_test")) - ); + GroupConfig groupBy = parseGroupConfig("{\"" + targetField + "\" : {" + + "\"terms\" : {" + + " \"field\" : \"doesn't_matter_for_this_test\"" + + "} } }"); String aggName = randomAlphaOfLengthBetween(5, 10); String aggTypedName = "avg#" + aggName; @@ -278,10 +286,10 @@ aggTypedName2, asMap( aggName2, -2.44 ) ); - executeTest(sources, aggregationBuilders, input, expected, 200); + executeTest(groupBy, aggregationBuilders, input, expected, 200); } - private void executeTest(Iterable sources, Collection aggregationBuilders, Map input, + private void executeTest(GroupConfig groups, Collection aggregationBuilders, Map input, List> expected, long expectedDocCounts) throws IOException { DataFrameIndexerTransformStats stats = new DataFrameIndexerTransformStats(); XContentBuilder builder = XContentFactory.contentBuilder(randomFrom(XContentType.values())); @@ -290,13 +298,19 @@ private void executeTest(Iterable sources, Collection> result = AggregationResultUtils - .extractCompositeAggregationResults(agg, sources, aggregationBuilders, stats).collect(Collectors.toList()); + .extractCompositeAggregationResults(agg, groups, aggregationBuilders, stats).collect(Collectors.toList()); assertEquals(expected, result); assertEquals(expectedDocCounts, stats.getNumDocuments()); } } + private GroupConfig parseGroupConfig(String json) throws IOException { + final XContentParser parser = XContentType.JSON.xContent().createParser(xContentRegistry(), + DeprecationHandler.THROW_UNSUPPORTED_OPERATION, json); + return GroupConfig.fromXContent(parser, false); + } + static Map asMap(Object... fields) { assert fields.length % 2 == 0; final Map map = new HashMap<>(); diff --git a/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/GroupConfigTests.java b/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/GroupConfigTests.java index e3d1ed3901558..72b0af31c6d81 100644 --- a/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/GroupConfigTests.java +++ b/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/GroupConfigTests.java @@ -6,19 +6,56 @@ package org.elasticsearch.xpack.dataframe.transforms.pivot; +import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.io.stream.Writeable.Reader; +import org.elasticsearch.common.xcontent.ToXContent; +import org.elasticsearch.common.xcontent.XContentBuilder; +import org.elasticsearch.common.xcontent.XContentFactory; +import org.elasticsearch.common.xcontent.XContentHelper; import org.elasticsearch.common.xcontent.XContentParser; +import org.elasticsearch.common.xcontent.XContentType; +import org.elasticsearch.common.xcontent.json.JsonXContent; import org.elasticsearch.test.AbstractSerializingTestCase; +import org.elasticsearch.xpack.dataframe.transforms.pivot.SingleGroupSource.Type; import java.io.IOException; - -import static org.elasticsearch.xpack.dataframe.transforms.pivot.SingleGroupSource.Type.TERMS; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Set; public class GroupConfigTests extends AbstractSerializingTestCase { public static GroupConfig randomGroupConfig() { - String targetFieldName = randomAlphaOfLengthBetween(1, 20); - return new GroupConfig(targetFieldName, TERMS, TermsGroupSourceTests.randomTermsGroupSource()); + Map source = new LinkedHashMap<>(); + Map> groups = new LinkedHashMap<>(); + + // ensure that the unlikely does not happen: 2 group_by's share the same name + Set names = new HashSet<>(); + for (int i = 0; i < randomIntBetween(1, 20); ++i) { + String targetFieldName = randomAlphaOfLengthBetween(1, 20); + if (names.add(targetFieldName)) { + SingleGroupSource groupBy; + Type type = randomFrom(SingleGroupSource.Type.values()); + switch (type) { + case TERMS: + groupBy = TermsGroupSourceTests.randomTermsGroupSource(); + break; + case HISTOGRAM: + groupBy = HistogramGroupSourceTests.randomHistogramGroupSource(); + break; + case DATE_HISTOGRAM: + default: + groupBy = DateHistogramGroupSourceTests.randomDateHistogramGroupSource(); + } + + source.put(targetFieldName, Collections.singletonMap(type.value(), getSource(groupBy))); + groups.put(targetFieldName, groupBy); + } + } + + return new GroupConfig(source, groups); } @Override @@ -35,4 +72,30 @@ protected GroupConfig createTestInstance() { protected Reader instanceReader() { return GroupConfig::new; } + + public void testEmptyGroupBy() throws IOException { + String source = "{}"; + + // lenient, passes but reports invalid + try (XContentParser parser = createParser(JsonXContent.jsonXContent, source)) { + GroupConfig groupConfig = GroupConfig.fromXContent(parser, true); + assertFalse(groupConfig.isValid()); + } + + // strict throws + try (XContentParser parser = createParser(JsonXContent.jsonXContent, source)) { + expectThrows(IllegalArgumentException.class, () -> GroupConfig.fromXContent(parser, false)); + } + } + + private static Map getSource(SingleGroupSource groupSource) { + try (XContentBuilder xContentBuilder = XContentFactory.jsonBuilder()) { + XContentBuilder content = groupSource.toXContent(xContentBuilder, ToXContent.EMPTY_PARAMS); + return XContentHelper.convertToMap(BytesReference.bytes(content), true, XContentType.JSON).v2(); + } catch (IOException e) { + // should not happen + fail("failed to create random single group source"); + } + return null; + } } diff --git a/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/PivotConfigTests.java b/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/PivotConfigTests.java index 0ae59315e69f6..2397c088293f4 100644 --- a/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/PivotConfigTests.java +++ b/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/PivotConfigTests.java @@ -13,29 +13,15 @@ import org.elasticsearch.xpack.dataframe.transforms.AbstractSerializingDataFrameTestCase; import java.io.IOException; -import java.util.ArrayList; -import java.util.List; public class PivotConfigTests extends AbstractSerializingDataFrameTestCase { public static PivotConfig randomPivotConfig() { - List groups = new ArrayList<>(); - - for (int i = 0; i < randomIntBetween(1, 10); ++i) { - groups.add(GroupConfigTests.randomGroupConfig()); - } - - return new PivotConfig(groups, AggregationConfigTests.randomAggregationConfig()); + return new PivotConfig(GroupConfigTests.randomGroupConfig(), AggregationConfigTests.randomAggregationConfig()); } public static PivotConfig randomInvalidPivotConfig() { - List groups = new ArrayList<>(); - - for (int i = 0; i < randomIntBetween(1, 10); ++i) { - groups.add(GroupConfigTests.randomGroupConfig()); - } - - return new PivotConfig(groups, AggregationConfigTests.randomInvalidAggregationConfig()); + return new PivotConfig(GroupConfigTests.randomGroupConfig(), AggregationConfigTests.randomInvalidAggregationConfig()); } @Override @@ -55,42 +41,86 @@ protected Reader instanceReader() { public void testAggsAbbreviations() throws IOException { String pivotAggs = "{" - + " \"group_by\": [ {" + + " \"group_by\": {" + " \"id\": {" + " \"terms\": {" + " \"field\": \"id\"" - + "} } } ]," + + "} } }," + " \"aggs\": {" + " \"avg\": {" + " \"avg\": {" + " \"field\": \"points\"" + "} } } }"; - PivotConfig p1 = createPivotConfigFromString(pivotAggs); + PivotConfig p1 = createPivotConfigFromString(pivotAggs, false); String pivotAggregations = pivotAggs.replace("aggs", "aggregations"); assertNotEquals(pivotAggs, pivotAggregations); - PivotConfig p2 = createPivotConfigFromString(pivotAggregations); + PivotConfig p2 = createPivotConfigFromString(pivotAggregations, false); assertEquals(p1,p2); } public void testMissingAggs() throws IOException { String pivot = "{" - + " \"group_by\": [ {" + + " \"group_by\": {" + + " \"id\": {" + + " \"terms\": {" + + " \"field\": \"id\"" + + "} } } }"; + + expectThrows(IllegalArgumentException.class, () -> createPivotConfigFromString(pivot, false)); + } + + public void testEmptyAggs() throws IOException { + String pivot = "{" + + " \"group_by\": {" + " \"id\": {" + " \"terms\": {" + " \"field\": \"id\"" - + "} } } ] }"; + + "} } }," + + "\"aggs\": {}" + + " }"; + + expectThrows(IllegalArgumentException.class, () -> createPivotConfigFromString(pivot, false)); + + // lenient passes but reports invalid + PivotConfig pivotConfig = createPivotConfigFromString(pivot, true); + assertFalse(pivotConfig.isValid()); + } + + public void testEmptyGroupBy() throws IOException { + String pivot = "{" + + " \"group_by\": {}," + + " \"aggs\": {" + + " \"avg\": {" + + " \"avg\": {" + + " \"field\": \"points\"" + + "} } } }"; + + expectThrows(IllegalArgumentException.class, () -> createPivotConfigFromString(pivot, false)); + + // lenient passes but reports invalid + PivotConfig pivotConfig = createPivotConfigFromString(pivot, true); + assertFalse(pivotConfig.isValid()); + } - expectThrows(IllegalArgumentException.class, () -> createPivotConfigFromString(pivot)); + public void testMissingGroupBy() throws IOException { + String pivot = "{" + + " \"aggs\": {" + + " \"avg\": {" + + " \"avg\": {" + + " \"field\": \"points\"" + + "} } } }"; + + expectThrows(IllegalArgumentException.class, () -> createPivotConfigFromString(pivot, false)); } public void testDoubleAggs() throws IOException { String pivot = "{" - + " \"group_by\": [ {" + + " \"group_by\": {" + " \"id\": {" + " \"terms\": {" + " \"field\": \"id\"" - + "} } } ]," + + "} } }," + " \"aggs\": {" + " \"avg\": {" + " \"avg\": {" @@ -103,12 +133,12 @@ public void testDoubleAggs() throws IOException { + "} } }" + "}"; - expectThrows(IllegalArgumentException.class, () -> createPivotConfigFromString(pivot)); + expectThrows(IllegalArgumentException.class, () -> createPivotConfigFromString(pivot, false)); } - private PivotConfig createPivotConfigFromString(String json) throws IOException { + private PivotConfig createPivotConfigFromString(String json, boolean lenient) throws IOException { final XContentParser parser = XContentType.JSON.xContent().createParser(xContentRegistry(), DeprecationHandler.THROW_UNSUPPORTED_OPERATION, json); - return PivotConfig.fromXContent(parser, false); + return PivotConfig.fromXContent(parser, lenient); } } diff --git a/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/PivotTests.java b/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/PivotTests.java index c25d42cf07261..4845085eba337 100644 --- a/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/PivotTests.java +++ b/x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/PivotTests.java @@ -42,9 +42,7 @@ import java.util.stream.Collectors; import java.util.stream.Stream; -import static java.util.Arrays.asList; import static java.util.Collections.emptyList; -import static org.elasticsearch.xpack.dataframe.transforms.pivot.SingleGroupSource.Type.TERMS; import static org.hamcrest.Matchers.equalTo; public class PivotTests extends ESTestCase { @@ -161,21 +159,11 @@ protected void } private PivotConfig getValidPivotConfig() throws IOException { - List sources = asList( - new GroupConfig("terms", TERMS, new TermsGroupSource("terms")), - new GroupConfig("terms", TERMS, new TermsGroupSource("terms")) - ); - - return new PivotConfig(sources, getValidAggregationConfig()); + return new PivotConfig(GroupConfigTests.randomGroupConfig(), getValidAggregationConfig()); } private PivotConfig getValidPivotConfig(AggregationConfig aggregationConfig) throws IOException { - List sources = asList( - new GroupConfig("terms", TERMS, new TermsGroupSource("terms")), - new GroupConfig("terms", TERMS, new TermsGroupSource("terms")) - ); - - return new PivotConfig(sources, aggregationConfig); + return new PivotConfig(GroupConfigTests.randomGroupConfig(), aggregationConfig); } private AggregationConfig getValidAggregationConfig() throws IOException {