Deprecate serializing PipelineAggregators

server/src/main/java/org/elasticsearch/plugins/SearchPlugin.java

@@ -368,8 +368,47 @@ public AggregationSpec setAggregatorRegistrar(Consumer<ValuesSourceRegistry> agg
     class PipelineAggregationSpec extends SearchExtensionSpec<PipelineAggregationBuilder,
             ContextParser<String, ? extends PipelineAggregationBuilder>> {
         private final Map<String, Writeable.Reader<? extends InternalAggregation>> resultReaders = new TreeMap<>();
+        /**
+         * Read the aggregator from a stream.
+         * @deprecated Pipelines implemented after 7.8.0 do not need to be sent across the wire
+         */
+        @Deprecated
         private final Writeable.Reader<? extends PipelineAggregator> aggregatorReader;
 
+        /**
+         * Specification of a {@link PipelineAggregator}.
+         *
+         * @param name holds the names by which this aggregation might be parsed. The {@link ParseField#getPreferredName()} is special as it
+         *        is the name by under which the readers are registered. So it is the name that the {@link PipelineAggregationBuilder} and
+         *        {@link PipelineAggregator} should return from {@link NamedWriteable#getWriteableName()}.
+         * @param builderReader the reader registered for this aggregation's builder. Typically a reference to a constructor that takes a
+         *        {@link StreamInput}
+         * @param parser reads the aggregation builder from XContent
+         */
+        public PipelineAggregationSpec(ParseField name,
+                Writeable.Reader<? extends PipelineAggregationBuilder> builderReader,
+                ContextParser<String, ? extends PipelineAggregationBuilder> parser) {
+            super(name, builderReader, parser);
+            this.aggregatorReader = null;
+        }
+
+        /**
+         * Specification of a {@link PipelineAggregator}.
+         *
+         * @param name name by which this aggregation might be parsed or deserialized. Make sure it is the name that the
+         *        {@link PipelineAggregationBuilder} and {@link PipelineAggregator} should return from
+         *        {@link NamedWriteable#getWriteableName()}.
+         * @param builderReader the reader registered for this aggregation's builder. Typically a reference to a constructor that takes a
+         *        {@link StreamInput}
+         * @param parser reads the aggregation builder from XContent
+         */
+        public PipelineAggregationSpec(String name,
+                Writeable.Reader<? extends PipelineAggregationBuilder> builderReader,
+                ContextParser<String, ? extends PipelineAggregationBuilder> parser) {
+            super(name, builderReader, parser);
+            this.aggregatorReader = null;
+        }
+
         /**
          * Specification of a {@link PipelineAggregator}.
          *
@@ -380,7 +419,10 @@ class PipelineAggregationSpec extends SearchExtensionSpec<PipelineAggregationBui
          *        {@link StreamInput}
          * @param aggregatorReader reads the {@link PipelineAggregator} from a stream
          * @param parser reads the aggregation builder from XContent
+         * @deprecated Use {@link PipelineAggregationSpec#PipelineAggregationSpec(ParseField, Writeable.Reader, ContextParser)} for
+         *             pipelines implemented after 7.8.0
          */
+        @Deprecated
         public PipelineAggregationSpec(ParseField name,
                 Writeable.Reader<? extends PipelineAggregationBuilder> builderReader,
                 Writeable.Reader<? extends PipelineAggregator> aggregatorReader,
@@ -399,7 +441,10 @@ public PipelineAggregationSpec(ParseField name,
          *        {@link StreamInput}
          * @param aggregatorReader reads the {@link PipelineAggregator} from a stream
          * @param parser reads the aggregation builder from XContent
+         * @deprecated Use {@link PipelineAggregationSpec#PipelineAggregationSpec(String, Writeable.Reader, ContextParser)} for pipelines
+         *             implemented after 7.8.0
          */
+        @Deprecated
         public PipelineAggregationSpec(String name,
                 Writeable.Reader<? extends PipelineAggregationBuilder> builderReader,
                 Writeable.Reader<? extends PipelineAggregator> aggregatorReader,
@@ -466,8 +511,10 @@ public PipelineAggregationSpec addResultReader(String writeableName, Writeable.R
         }
 
         /**
-         * The reader for the {@link PipelineAggregator}.
+         * Read the aggregator from a stream.
+         * @deprecated Pipelines implemented after 7.8.0 do not need to be sent across the wire
          */
+        @Deprecated
         public Writeable.Reader<? extends PipelineAggregator> getAggregatorReader() {
             return aggregatorReader;
         }

server/src/main/java/org/elasticsearch/search/SearchModule.java

@@ -588,8 +588,10 @@ private void registerPipelineAggregation(PipelineAggregationSpec spec) {
                 (p, c) -> spec.getParser().parse(p, (String) c)));
         namedWriteables.add(
                 new NamedWriteableRegistry.Entry(PipelineAggregationBuilder.class, spec.getName().getPreferredName(), spec.getReader()));
-        namedWriteables.add(
-                new NamedWriteableRegistry.Entry(PipelineAggregator.class, spec.getName().getPreferredName(), spec.getAggregatorReader()));
+        if (spec.getAggregatorReader() != null) {
+            namedWriteables.add(new NamedWriteableRegistry.Entry(
+                PipelineAggregator.class, spec.getName().getPreferredName(), spec.getAggregatorReader()));
+        }
         for (Map.Entry<String, Writeable.Reader<? extends InternalAggregation>> resultReader : spec.getResultReaders().entrySet()) {
             namedWriteables
                     .add(new NamedWriteableRegistry.Entry(InternalAggregation.class, resultReader.getKey(), resultReader.getValue()));

server/src/main/java/org/elasticsearch/search/aggregations/AggregatorFactories.java

@@ -57,6 +57,9 @@
 import static java.util.stream.Collectors.toList;
 import static java.util.stream.Collectors.toMap;
 
+/**
+ * An immutable collection of {@link AggregatorFactories}.
+ */
 public class AggregatorFactories {
     public static final Pattern VALID_AGG_NAME = Pattern.compile("[^\\[\\]>]+");
 
@@ -155,26 +158,16 @@ private static AggregatorFactories.Builder parseAggregators(XContentParser parse
         return factories;
     }
 
-    public static final AggregatorFactories EMPTY = new AggregatorFactories(new AggregatorFactory[0], new ArrayList<>());
+    public static final AggregatorFactories EMPTY = new AggregatorFactories(new AggregatorFactory[0]);
 
     private AggregatorFactory[] factories;
-    private List<PipelineAggregationBuilder> pipelineAggregatorFactories;
 
     public static Builder builder() {
         return new Builder();
     }
 
-    private AggregatorFactories(AggregatorFactory[] factories, List<PipelineAggregationBuilder> pipelineAggregators) {
+    private AggregatorFactories(AggregatorFactory[] factories) {
         this.factories = factories;
-        this.pipelineAggregatorFactories = pipelineAggregators;
-    }
-
-    public List<PipelineAggregator> createPipelineAggregators() {
-        List<PipelineAggregator> pipelineAggregators = new ArrayList<>(this.pipelineAggregatorFactories.size());
-        for (PipelineAggregationBuilder factory : this.pipelineAggregatorFactories) {
-            pipelineAggregators.add(factory.create());
-        }
-        return pipelineAggregators;
     }
 
     /**
@@ -216,13 +209,16 @@ public Aggregator[] createTopLevelAggregators(SearchContext searchContext) throw
     }
 
     /**
-     * @return the number of sub-aggregator factories not including pipeline
-     *         aggregator factories
+     * @return the number of sub-aggregator factories
      */
     public int countAggregators() {
         return factories.length;
     }
 
+    /**
+     * A mutable collection of {@link AggregationBuilder}s and
+     * {@link PipelineAggregationBuilder}s.
+     */
     public static class Builder implements Writeable, ToXContentObject {
         private final Set<String> names = new HashSet<>();
 
@@ -333,16 +329,13 @@ public AggregatorFactories build(QueryShardContext queryShardContext, Aggregator
             if (aggregationBuilders.isEmpty() && pipelineAggregatorBuilders.isEmpty()) {
                 return EMPTY;
             }
-            List<PipelineAggregationBuilder> orderedPipelineAggregators =
-                    resolvePipelineAggregatorOrder(pipelineAggregatorBuilders, aggregationBuilders);
             AggregatorFactory[] aggFactories = new AggregatorFactory[aggregationBuilders.size()];
-
             int i = 0;
             for (AggregationBuilder agg : aggregationBuilders) {
                 aggFactories[i] = agg.build(queryShardContext, parent);
                 ++i;
             }
-            return new AggregatorFactories(aggFactories, orderedPipelineAggregators);
+            return new AggregatorFactories(aggFactories);
         }
 
         private List<PipelineAggregationBuilder> resolvePipelineAggregatorOrder(

server/src/main/java/org/elasticsearch/search/aggregations/pipeline/PipelineAggregator.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.search.aggregations.pipeline;
 
 
+import org.elasticsearch.Version;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.io.stream.NamedWriteable;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -113,22 +114,54 @@ protected PipelineAggregator(String name, String[] bucketsPaths, Map<String, Obj
 
     /**
      * Read from a stream.
+     * @deprecated pipeline aggregations added after 7.8.0 shouldn't call this
      */
+    @Deprecated
     protected PipelineAggregator(StreamInput in) throws IOException {
-        name = in.readString();
-        bucketsPaths = in.readStringArray();
-        metadata = in.readMap();
+        if (in.getVersion().before(Version.V_7_8_0)) {
+            name = in.readString();
+            bucketsPaths = in.readStringArray();
+            metadata = in.readMap();
+        } else {
+           throw new IllegalStateException("Cannot deserialize pipeline [" + getClass() + "] from before 7.8.0");
+        }
     }
 
+    /**
+     * {@inheritDoc}
+     * @deprecated pipeline aggregations added after 7.8.0 shouldn't call this
+     */
     @Override
+    @Deprecated
     public final void writeTo(StreamOutput out) throws IOException {
-        out.writeString(name);
-        out.writeStringArray(bucketsPaths);
-        out.writeMap(metadata);
-        doWriteTo(out);
+        if (out.getVersion().before(Version.V_7_8_0)) {
+            out.writeString(name);
+            out.writeStringArray(bucketsPaths);
+            out.writeMap(metadata);
+            doWriteTo(out);
+        } else {
+            throw new IllegalArgumentException("[" + name + "] is not supported on versions before 7.8.0");
+        }
+    }
+
+    /**
+     * Write the body of the aggregation to the wire.
+     * @deprecated pipeline aggregations added after 7.8.0 don't need to implement this
+     */
+    @Deprecated
+    protected void doWriteTo(StreamOutput out) throws IOException {
+    }
+
+    /**
+     * The name of the writeable object.
+     * @deprecated pipeline aggregations added after 7.8.0 don't need to implement this
+     */
+    @Override
+    @Deprecated
+    public String getWriteableName() {
+        throw new IllegalArgumentException("[" + name + "] is not supported on versions before 7.8.0");
     }
 
-    protected abstract void doWriteTo(StreamOutput out) throws IOException;
 
     public String name() {
         return name;