Added filters aggregation

docs/reference/search/aggregations/bucket.asciidoc

@@ -4,6 +4,8 @@ include::bucket/global-aggregation.asciidoc[]
 
 include::bucket/filter-aggregation.asciidoc[]
 
+include::bucket/filters-aggregation.asciidoc[]
+
 include::bucket/missing-aggregation.asciidoc[]
 
 include::bucket/nested-aggregation.asciidoc[]

docs/reference/search/aggregations/bucket/filters-aggregation.asciidoc

@@ -0,0 +1,60 @@
+[[search-aggregations-bucket-filters-aggregation]]
+=== Filters
+
+Defines a multi bucket aggregations where each bucket is associated with a filter. Each bucket will collect all
+documents that match its associated filter.
+
+Example:
+
+[source,js]
+--------------------------------------------------
+{
+    "aggs" : {
+        "messages" : {
+            "filters" : {
+                "filters" : {
+                    "errors" : { "query" : { "match" : { "body" : "error" } } },
+                    "warnings" : { "query" : { "match" : { "body" : "error" } } }
+                }
+            },
+            "aggs" : { "monthly" : { "histogram" : { "field" : "timestamp", "interval" : "1M" } } }
+        }
+    }
+}
+--------------------------------------------------
+
+In the above example, we analyze log messages. The aggregation wll build two collection (buckets) of log messages - one
+for all those containing an error, and another for all those containing a warning. And for each of these buckets it will
+break them down by month.
+
+Response:
+
+[source,js]
+--------------------------------------------------
+{
+    ...
+
+    "aggs" : {
+        "messages" : {
+            "buckets" : {
+                "errors" : {
+                    "doc_count" : 34,
+                    "monthly" : {
+                        "buckets : [
+                            ... // the histogram monthly breakdown
+                        ]
+                    }
+                },
+                "warnings" : {
+                    "doc_count" : 439,
+                    "monthly" : {
+                        "buckets : [
+                            ... // the histogram monthly breakdown
+                        ]
+                    }
+                }
+            }
+        }
+    }
+}
+--------------------------------------------------

src/main/java/org/elasticsearch/search/aggregations/AggregationBuilders.java

@@ -19,6 +19,7 @@
 package org.elasticsearch.search.aggregations;
 
 import org.elasticsearch.search.aggregations.bucket.filter.FilterAggregationBuilder;
+import org.elasticsearch.search.aggregations.bucket.filters.FiltersAggregationBuilder;
 import org.elasticsearch.search.aggregations.bucket.geogrid.GeoHashGridBuilder;
 import org.elasticsearch.search.aggregations.bucket.global.GlobalBuilder;
 import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramBuilder;
@@ -82,6 +83,10 @@ public static FilterAggregationBuilder filter(String name) {
         return new FilterAggregationBuilder(name);
     }
 
+    public static FiltersAggregationBuilder filters(String name) {
+        return new FiltersAggregationBuilder(name);
+    }
+
     public static GlobalBuilder global(String name) {
         return new GlobalBuilder(name);
     }

src/main/java/org/elasticsearch/search/aggregations/AggregationModule.java

@@ -22,6 +22,7 @@
 import org.elasticsearch.common.inject.AbstractModule;
 import org.elasticsearch.common.inject.multibindings.Multibinder;
 import org.elasticsearch.search.aggregations.bucket.filter.FilterParser;
+import org.elasticsearch.search.aggregations.bucket.filters.FiltersParser;
 import org.elasticsearch.search.aggregations.bucket.geogrid.GeoHashGridParser;
 import org.elasticsearch.search.aggregations.bucket.global.GlobalParser;
 import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramParser;
@@ -68,6 +69,7 @@ public AggregationModule() {
         parsers.add(GlobalParser.class);
         parsers.add(MissingParser.class);
         parsers.add(FilterParser.class);
+        parsers.add(FiltersParser.class);
         parsers.add(TermsParser.class);
         parsers.add(SignificantTermsParser.class);
         parsers.add(RangeParser.class);

src/main/java/org/elasticsearch/search/aggregations/TransportAggregationModule.java

@@ -20,6 +20,7 @@
 
 import org.elasticsearch.common.inject.AbstractModule;
 import org.elasticsearch.search.aggregations.bucket.filter.InternalFilter;
+import org.elasticsearch.search.aggregations.bucket.filters.InternalFilters;
 import org.elasticsearch.search.aggregations.bucket.geogrid.InternalGeoHashGrid;
 import org.elasticsearch.search.aggregations.bucket.global.InternalGlobal;
 import org.elasticsearch.search.aggregations.bucket.histogram.InternalDateHistogram;
@@ -70,6 +71,7 @@ protected void configure() {
         // buckets
         InternalGlobal.registerStreams();
         InternalFilter.registerStreams();
+        InternalFilters.registerStream();
         InternalMissing.registerStreams();
         StringTerms.registerStreams();
         LongTerms.registerStreams();

src/main/java/org/elasticsearch/search/aggregations/bucket/filters/Filters.java

@@ -0,0 +1,43 @@
+/*
+ * 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.bucket.filters;
+
+import org.elasticsearch.search.aggregations.bucket.MultiBucketsAggregation;
+
+import java.util.Collection;
+
+/**
+ * A multi bucket aggregation where the buckets are defined by a set of filters (a bucket per filter). Each bucket
+ * will collect all documents matching its filter.
+ */
+public interface Filters extends MultiBucketsAggregation {
+
+    /**
+     * A bucket associated with a specific filter (identified by its key)
+     */
+    public static interface Bucket extends MultiBucketsAggregation.Bucket {
+    }
+
+    Collection<? extends Bucket> getBuckets();
+
+    @Override
+    Bucket getBucketByKey(String key);
+
+}

src/main/java/org/elasticsearch/search/aggregations/bucket/filters/FiltersAggregationBuilder.java

@@ -0,0 +1,57 @@
+/*
+ * 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.bucket.filters;
+
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.index.query.FilterBuilder;
+import org.elasticsearch.search.aggregations.AggregationBuilder;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ *
+ */
+public class FiltersAggregationBuilder extends AggregationBuilder<FiltersAggregationBuilder> {
+
+    private Map<String, FilterBuilder> filters = new HashMap<>();
+
+    public FiltersAggregationBuilder(String name) {
+        super(name, InternalFilters.TYPE.name());
+    }
+
+    public FiltersAggregationBuilder filter(String key, FilterBuilder filter) {
+        filters.put(key, filter);
+        return this;
+    }
+
+    @Override
+    protected XContentBuilder internalXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        builder.startObject("filters");
+        for (Map.Entry<String, FilterBuilder> entry : filters.entrySet()) {
+            builder.field(entry.getKey());
+            entry.getValue().toXContent(builder, params);
+        }
+        builder.endObject();
+        return builder.endObject();
+    }
+}

src/main/java/org/elasticsearch/search/aggregations/bucket/filters/FiltersAggregator.java

@@ -0,0 +1,132 @@
+/*
+ * 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.bucket.filters;
+
+import com.google.common.collect.Lists;
+import org.apache.lucene.index.AtomicReaderContext;
+import org.apache.lucene.search.Filter;
+import org.apache.lucene.util.Bits;
+import org.elasticsearch.common.lucene.docset.DocIdSets;
+import org.elasticsearch.search.aggregations.*;
+import org.elasticsearch.search.aggregations.bucket.BucketsAggregator;
+import org.elasticsearch.search.aggregations.support.AggregationContext;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ *
+ */
+public class FiltersAggregator extends BucketsAggregator {
+
+    static class KeyedFilter {
+
+        final String key;
+        final Filter filter;
+
+        KeyedFilter(String key, Filter filter) {
+            this.key = key;
+            this.filter = filter;
+        }
+    }
+
+    private final KeyedFilter[] filters;
+    private final Bits[] bits;
+    private final boolean keyed;
+
+    public FiltersAggregator(String name, AggregatorFactories factories, List<KeyedFilter> filters,
+                             boolean keyed, AggregationContext aggregationContext, Aggregator parent) {
+
+        super(name, BucketAggregationMode.MULTI_BUCKETS, factories, filters.size() * (parent == null ? 1 : parent.estimatedBucketCount()), aggregationContext, parent);
+        this.keyed = keyed;
+        this.filters = filters.toArray(new KeyedFilter[filters.size()]);
+        this.bits = new Bits[this.filters.length];
+    }
+
+    @Override
+    public boolean shouldCollect() {
+        return true;
+    }
+
+    @Override
+    public void setNextReader(AtomicReaderContext reader) {
+        try {
+            for (int i = 0; i < filters.length; i++) {
+                bits[i] = DocIdSets.toSafeBits(reader.reader(), filters[i].filter.getDocIdSet(reader, reader.reader().getLiveDocs()));
+            }
+        } catch (IOException ioe) {
+            throw new AggregationExecutionException("Failed to aggregate filter aggregator [" + name + "]", ioe);
+        }
+    }
+
+    @Override
+    public void collect(int doc, long owningBucketOrdinal) throws IOException {
+        for (int i = 0; i < bits.length; i++) {
+            if (bits[i].get(doc)) {
+                collectBucket(doc, bucketOrd(owningBucketOrdinal, i));
+            }
+        }
+    }
+
+    @Override
+    public InternalAggregation buildAggregation(long owningBucketOrdinal) {
+        List<InternalFilters.Bucket> buckets = Lists.newArrayListWithCapacity(filters.length);
+        for (int i = 0; i < filters.length; i++) {
+            KeyedFilter filter = filters[i];
+            long bucketOrd = bucketOrd(owningBucketOrdinal, i);
+            InternalFilters.Bucket bucket = new InternalFilters.Bucket(filter.key, bucketDocCount(bucketOrd), bucketAggregations(bucketOrd));
+            buckets.add(bucket);
+        }
+        return new InternalFilters(name, buckets, keyed);
+    }
+
+    @Override
+    public InternalAggregation buildEmptyAggregation() {
+        InternalAggregations subAggs = buildEmptySubAggregations();
+        List<InternalFilters.Bucket> buckets = Lists.newArrayListWithCapacity(filters.length);
+        for (int i = 0; i < filters.length; i++) {
+            InternalFilters.Bucket bucket = new InternalFilters.Bucket(filters[i].key, 0, subAggs);
+            buckets.add(bucket);
+        }
+        return new InternalFilters(name, buckets, keyed);
+    }
+
+    private final long bucketOrd(long owningBucketOrdinal, int filterOrd) {
+        return owningBucketOrdinal * filters.length + filterOrd;
+    }
+
+    public static class Factory extends AggregatorFactory {
+
+        private final List<KeyedFilter> filters;
+        private final boolean keyed;
+
+        public Factory(String name, List<KeyedFilter> filters, boolean keyed) {
+            super(name, InternalFilters.TYPE.name());
+            this.filters = filters;
+            this.keyed = keyed;
+        }
+
+        @Override
+        public Aggregator create(AggregationContext context, Aggregator parent, long expectedBucketsCount) {
+            return new FiltersAggregator(name, factories, filters, keyed, context, parent);
+        }
+    }
+
+}