Aggregations Refactor: Refactor Cumulative Sum Aggregation

Author: colings86

core/src/main/java/org/elasticsearch/search/aggregations/pipeline/cumulativesum/CumulativeSumParser.java

@@ -24,8 +24,6 @@
 import org.elasticsearch.search.SearchParseException;
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregatorFactory;
-import org.elasticsearch.search.aggregations.support.format.ValueFormat;
-import org.elasticsearch.search.aggregations.support.format.ValueFormatter;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -85,20 +83,16 @@ public PipelineAggregatorFactory parse(String pipelineAggregatorName, XContentPa
                     + "] for derivative aggregation [" + pipelineAggregatorName + "]", parser.getTokenLocation());
         }
 
-        ValueFormatter formatter = null;
+        CumulativeSumPipelineAggregator.Factory factory = new CumulativeSumPipelineAggregator.Factory(pipelineAggregatorName, bucketsPaths);
         if (format != null) {
-            formatter = ValueFormat.Patternable.Number.format(format).formatter();
-        } else {
-            formatter = ValueFormatter.RAW;
+            factory.format(format);
         }
-
-        return new CumulativeSumPipelineAggregator.Factory(pipelineAggregatorName, bucketsPaths, formatter);
+        return factory;
     }
 
-    // NORELEASE implement this method when refactoring this aggregation
     @Override
     public PipelineAggregatorFactory getFactoryPrototype() {
-        return null;
+        return new CumulativeSumPipelineAggregator.Factory(null, null);
     }
 
 }

core/src/main/java/org/elasticsearch/search/aggregations/pipeline/cumulativesum/CumulativeSumPipelineAggregator.java

@@ -21,6 +21,7 @@
 
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.InternalAggregation;
 import org.elasticsearch.search.aggregations.InternalAggregation.ReduceContext;
@@ -33,13 +34,16 @@
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregatorFactory;
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregatorStreams;
+import org.elasticsearch.search.aggregations.pipeline.bucketmetrics.BucketMetricsParser;
+import org.elasticsearch.search.aggregations.support.format.ValueFormat;
 import org.elasticsearch.search.aggregations.support.format.ValueFormatter;
 import org.elasticsearch.search.aggregations.support.format.ValueFormatterStreams;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
@@ -109,16 +113,37 @@ public void doWriteTo(StreamOutput out) throws IOException {
 
     public static class Factory extends PipelineAggregatorFactory {
 
-        private final ValueFormatter formatter;
+        private String format;
 
-        public Factory(String name, String[] bucketsPaths, ValueFormatter formatter) {
+        public Factory(String name, String[] bucketsPaths) {
             super(name, TYPE.name(), bucketsPaths);
-            this.formatter = formatter;
+        }
+
+        /**
+         * Sets the format to use on the output of this aggregation.
+         */
+        public void format(String format) {
+            this.format = format;
+        }
+
+        /**
+         * Gets the format to use on the output of this aggregation.
+         */
+        public String format() {
+            return format;
+        }
+
+        protected ValueFormatter formatter() {
+            if (format != null) {
+                return ValueFormat.Patternable.Number.format(format).formatter();
+            } else {
+                return ValueFormatter.RAW;
+            }
         }
 
         @Override
         protected PipelineAggregator createInternal(Map<String, Object> metaData) throws IOException {
-            return new CumulativeSumPipelineAggregator(name, bucketsPaths, formatter, metaData);
+            return new CumulativeSumPipelineAggregator(name, bucketsPaths, formatter(), metaData);
         }
 
         @Override
@@ -139,5 +164,35 @@ public void doValidate(AggregatorFactory parent, AggregatorFactory[] aggFactorie
             }
         }
 
+        @Override
+        protected final XContentBuilder internalXContent(XContentBuilder builder, Params params) throws IOException {
+            if (format != null) {
+                builder.field(BucketMetricsParser.FORMAT.getPreferredName(), format);
+            }
+            return builder;
+        }
+
+        @Override
+        protected final PipelineAggregatorFactory doReadFrom(String name, String[] bucketsPaths, StreamInput in) throws IOException {
+            Factory factory = new Factory(name, bucketsPaths);
+            factory.format = in.readOptionalString();
+            return factory;
+        }
+
+        @Override
+        protected final void doWriteTo(StreamOutput out) throws IOException {
+            out.writeOptionalString(format);
+        }
+
+        @Override
+        protected int doHashCode() {
+            return Objects.hash(format);
+        }
+
+        @Override
+        protected boolean doEquals(Object obj) {
+            Factory other = (Factory) obj;
+            return Objects.equals(format, other.format);
+        }
     }
 }

core/src/test/java/org/elasticsearch/search/aggregations/pipeline/CumulativeSumTests.java

@@ -0,0 +1,40 @@
+/*
+ * 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;
+
+import org.elasticsearch.search.aggregations.BasePipelineAggregationTestCase;
+import org.elasticsearch.search.aggregations.pipeline.cumulativesum.CumulativeSumPipelineAggregator;
+import org.elasticsearch.search.aggregations.pipeline.cumulativesum.CumulativeSumPipelineAggregator.Factory;
+
+public class CumulativeSumTests extends BasePipelineAggregationTestCase<CumulativeSumPipelineAggregator.Factory> {
+
+    @Override
+    protected Factory createTestAggregatorFactory() {
+        String name = randomAsciiOfLengthBetween(3, 20);
+        String[] bucketsPaths = new String[1];
+        bucketsPaths[0] = randomAsciiOfLengthBetween(3, 20);
+        Factory factory = new Factory(name, bucketsPaths);
+        if (randomBoolean()) {
+            factory.format(randomAsciiOfLengthBetween(1, 10));
+        }
+        return factory;
+    }
+
+}