[all] make distributed filtering aggregations work as intended

aggregation/simple/src/main/java/com/spotify/heroic/aggregation/simple/AboveK.java

@@ -29,33 +29,29 @@
 
 import java.util.Optional;
 
-import static com.google.common.base.Preconditions.checkNotNull;
-
 @Data
 public class AboveK implements Aggregation {
     public static final String NAME = "abovek";
 
     private final double k;
-    private final Aggregation of;
 
     @JsonCreator
-    public AboveK(@JsonProperty("k") double k, @JsonProperty("of") Aggregation of) {
+    public AboveK(@JsonProperty("k") double k) {
         this.k = k;
-        this.of = checkNotNull(of, "of");
     }
 
     @Override
     public Optional<Long> size() {
-        return of.size();
+        return Optional.empty();
     }
 
     @Override
     public Optional<Long> extent() {
-        return of.extent();
+        return Optional.empty();
     }
 
     @Override
     public AboveKInstance apply(final AggregationContext context) {
-        return new AboveKInstance(k, of.apply(context));
+        return new AboveKInstance(k);
     }
 }

aggregation/simple/src/main/java/com/spotify/heroic/aggregation/simple/AboveKInstance.java

@@ -23,11 +23,12 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.spotify.heroic.aggregation.AggregationInstance;
+import lombok.Data;
 
+@Data
 public class AboveKInstance extends FilterAggregation {
     @JsonCreator
-    public AboveKInstance(@JsonProperty("k") double k, @JsonProperty("of") AggregationInstance of) {
-        super(new FilterKThresholdStrategy(FilterKThresholdType.ABOVE, k), of);
+    public AboveKInstance(@JsonProperty("k") double k) {
+        super(new FilterKThresholdStrategy(FilterKThresholdType.ABOVE, k));
     }
 }

aggregation/simple/src/main/java/com/spotify/heroic/aggregation/simple/BelowK.java

@@ -29,33 +29,29 @@
 
 import java.util.Optional;
 
-import static com.google.common.base.Preconditions.checkNotNull;
-
 @Data
 public class BelowK implements Aggregation {
     public static final String NAME = "belowk";
 
     private final double k;
-    private final Aggregation of;
 
     @JsonCreator
-    public BelowK(@JsonProperty("k") double k, @JsonProperty("of") Aggregation of) {
+    public BelowK(@JsonProperty("k") double k) {
         this.k = k;
-        this.of = checkNotNull(of, "of");
     }
 
     @Override
     public Optional<Long> size() {
-        return of.size();
+        return Optional.empty();
     }
 
     @Override
     public Optional<Long> extent() {
-        return of.extent();
+        return Optional.empty();
     }
 
     @Override
     public BelowKInstance apply(final AggregationContext context) {
-        return new BelowKInstance(k, of.apply(context));
+        return new BelowKInstance(k);
     }
 }

aggregation/simple/src/main/java/com/spotify/heroic/aggregation/simple/BelowKInstance.java

@@ -23,11 +23,12 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.spotify.heroic.aggregation.AggregationInstance;
+import lombok.Data;
 
+@Data
 public class BelowKInstance extends FilterAggregation {
     @JsonCreator
-    public BelowKInstance(@JsonProperty("k") double k, @JsonProperty("of") AggregationInstance of) {
-        super(new FilterKThresholdStrategy(FilterKThresholdType.BELOW, k), of);
+    public BelowKInstance(@JsonProperty("k") double k) {
+        super(new FilterKThresholdStrategy(FilterKThresholdType.BELOW, k));
     }
 }

aggregation/simple/src/main/java/com/spotify/heroic/aggregation/simple/BottomK.java

@@ -29,33 +29,29 @@
 
 import java.util.Optional;
 
-import static com.google.common.base.Preconditions.checkNotNull;
-
 @Data
 public class BottomK implements Aggregation {
     public static final String NAME = "bottomk";
 
     private final long k;
-    private final Aggregation of;
 
     @JsonCreator
-    public BottomK(@JsonProperty("k") long k, @JsonProperty("of") Aggregation of) {
+    public BottomK(@JsonProperty("k") long k) {
         this.k = k;
-        this.of = checkNotNull(of, "of");
     }
 
     @Override
     public Optional<Long> size() {
-        return of.size();
+        return Optional.empty();
     }
 
     @Override
     public Optional<Long> extent() {
-        return of.extent();
+        return Optional.empty();
     }
 
     @Override
     public BottomKInstance apply(final AggregationContext context) {
-        return new BottomKInstance(k, of.apply(context));
+        return new BottomKInstance(k);
     }
 }

aggregation/simple/src/main/java/com/spotify/heroic/aggregation/simple/BottomKInstance.java

@@ -23,11 +23,12 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.spotify.heroic.aggregation.AggregationInstance;
+import lombok.Data;
 
+@Data
 public class BottomKInstance extends FilterAggregation {
     @JsonCreator
-    public BottomKInstance(@JsonProperty("k") long k, @JsonProperty("of") AggregationInstance of) {
-        super(new FilterKAreaStrategy(FilterKAreaType.BOTTOM, k), of);
+    public BottomKInstance(@JsonProperty("k") long k) {
+        super(new FilterKAreaStrategy(FilterKAreaType.BOTTOM, k));
     }
 }

aggregation/simple/src/main/java/com/spotify/heroic/aggregation/simple/CountInstance.java

@@ -24,8 +24,6 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.spotify.heroic.aggregation.AggregationInstance;
-import com.spotify.heroic.aggregation.AggregationSession;
-import com.spotify.heroic.common.DateRange;
 import com.spotify.heroic.metric.MetricType;
 import com.spotify.heroic.metric.Point;
 
@@ -53,7 +51,7 @@ public AggregationInstance distributed() {
     }
 
     @Override
-    public AggregationSession reducer(DateRange range) {
-        return new SumInstance(size, extent).session(range);
+    public AggregationInstance reducer() {
+        return new SumInstance(size, extent);
     }
 }

aggregation/simple/src/main/java/com/spotify/heroic/aggregation/simple/DistributedBucketInstance.java

@@ -21,23 +21,12 @@
 
 package com.spotify.heroic.aggregation.simple;
 
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
-import com.spotify.heroic.aggregation.AggregationCombiner;
 import com.spotify.heroic.aggregation.AggregationInstance;
-import com.spotify.heroic.aggregation.AggregationOutput;
-import com.spotify.heroic.aggregation.AggregationResult;
-import com.spotify.heroic.aggregation.AggregationSession;
 import com.spotify.heroic.aggregation.Bucket;
 import com.spotify.heroic.aggregation.BucketAggregationInstance;
-import com.spotify.heroic.common.DateRange;
 import com.spotify.heroic.metric.MetricType;
-import com.spotify.heroic.metric.SeriesValues;
-import com.spotify.heroic.metric.ShardedResultGroup;
 
-import java.util.List;
-import java.util.Map;
 import java.util.Set;
 
 public abstract class DistributedBucketInstance<B extends Bucket>
@@ -53,34 +42,4 @@ public DistributedBucketInstance(
     public AggregationInstance distributed() {
         return new SpreadInstance(getSize(), getExtent());
     }
-
-    @Override
-    public AggregationCombiner combiner(final DateRange range) {
-        return all -> {
-            final Map<String, String> key = ImmutableMap.of();
-            final AggregationSession session = reducer(range);
-
-            final SeriesValues.Builder series = SeriesValues.builder();
-
-            for (final List<ShardedResultGroup> groups : all) {
-                for (final ShardedResultGroup g : groups) {
-                    g.getGroup().updateAggregation(session, key, ImmutableSet.of());
-                    series.addSeriesValues(g.getSeries());
-                }
-            }
-
-            final SeriesValues s = series.build();
-
-            final ImmutableList.Builder<ShardedResultGroup> groups = ImmutableList.builder();
-
-            final AggregationResult result = session.result();
-
-            for (final AggregationOutput out1 : result.getResult()) {
-                groups.add(new ShardedResultGroup(ImmutableMap.of(), key, s, out1.getMetrics(),
-                    getSize()));
-            }
-
-            return groups.build();
-        };
-    }
 }

aggregation/simple/src/main/java/com/spotify/heroic/aggregation/simple/FilterAggregation.java

@@ -21,17 +21,16 @@
 
 package com.spotify.heroic.aggregation.simple;
 
-import com.spotify.heroic.aggregation.AggregationCombiner;
 import com.spotify.heroic.aggregation.AggregationInstance;
 import com.spotify.heroic.aggregation.AggregationOutput;
 import com.spotify.heroic.aggregation.AggregationResult;
 import com.spotify.heroic.aggregation.AggregationSession;
+import com.spotify.heroic.aggregation.EmptyInstance;
 import com.spotify.heroic.common.DateRange;
 import com.spotify.heroic.common.Series;
 import com.spotify.heroic.metric.Event;
 import com.spotify.heroic.metric.MetricGroup;
 import com.spotify.heroic.metric.Point;
-import com.spotify.heroic.metric.ShardedResultGroup;
 import com.spotify.heroic.metric.Spread;
 
 import java.util.List;
@@ -42,16 +41,10 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 
 public class FilterAggregation implements AggregationInstance {
-    private final AggregationInstance of;
     private final FilterStrategy filterStrategy;
 
-    public FilterAggregation(final FilterStrategy filterStrategy, final AggregationInstance of) {
+    public FilterAggregation(final FilterStrategy filterStrategy) {
         this.filterStrategy = checkNotNull(filterStrategy, "filterStrategy");
-        this.of = checkNotNull(of, "of");
-    }
-
-    public AggregationInstance getOf() {
-        return of;
     }
 
     @Override
@@ -61,27 +54,31 @@ public long estimate(DateRange range) {
 
     @Override
     public long cadence() {
-        return of.cadence();
+        return -1;
     }
 
     @Override
-    public AggregationSession session(DateRange range) {
-        final AggregationSession child = of.session(range);
-        return new Session(filterStrategy, child);
+    public AggregationInstance distributed() {
+        return EmptyInstance.INSTANCE;
     }
 
     @Override
-    public AggregationCombiner combiner(DateRange range) {
-        return all -> {
-            final List<FilterableMetrics<ShardedResultGroup>> filterableMetrics = of
-                .combiner(range)
-                .combine(all)
-                .stream()
-                .map(s -> new FilterableMetrics<>(s, s::getGroup))
-                .collect(Collectors.toList());
+    public AggregationInstance reducer() {
+        return EmptyInstance.INSTANCE;
+    }
 
-            return filterStrategy.filter(filterableMetrics);
-        };
+    /**
+     * Filtering aggregations are by definition <em>not</em> distributable since they are incapable
+     * of making a complete local decision.
+     */
+    @Override
+    public boolean distributable() {
+        return false;
+    }
+
+    @Override
+    public AggregationSession session(DateRange range) {
+        return new Session(filterStrategy, EmptyInstance.INSTANCE.session(range));
     }
 
     private class Session implements AggregationSession {