Index: modules/grouping/src/java/org/apache/lucene/search/grouping/indexdocvalues/IndexDocValuesAllGroupsCollector.java
===================================================================
--- modules/grouping/src/java/org/apache/lucene/search/grouping/indexdocvalues/IndexDocValuesAllGroupsCollector.java	(revision )
+++ modules/grouping/src/java/org/apache/lucene/search/grouping/indexdocvalues/IndexDocValuesAllGroupsCollector.java	(revision )
@@ -0,0 +1,197 @@
+package org.apache.lucene.search.grouping.indexdocvalues;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.
+ */
+
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.values.IndexDocValues;
+import org.apache.lucene.index.values.ValueType;
+import org.apache.lucene.search.grouping.AbstractAllGroupsCollector;
+import org.apache.lucene.search.grouping.SentinelIntSet;
+import org.apache.lucene.util.BytesRef;
+
+import java.io.IOException;
+import java.util.*;
+
+/**
+ * Implementation of {@link AbstractAllGroupsCollector} that groups documents based on
+ * {@link IndexDocValues} fields.
+ */
+public abstract class IndexDocValuesAllGroupsCollector<GROUP_VALUE_TYPE> extends AbstractAllGroupsCollector<GROUP_VALUE_TYPE> {
+
+
+  /**
+   * Constructs a {@link IndexDocValuesAllGroupsCollector}.
+   * Selects and constructs the most optimal all groups collector implementation for grouping by {@link IndexDocValues}.
+   * 
+   *
+   * @param groupField  The field to group by
+   * @param type The {@link ValueType} which is used to select a concrete implementation.
+   * @param initialSize The initial allocation size of the
+   *                    internal int set and group list
+   *                    which should roughly match the total
+   *                    number of expected unique groups. Be aware that the
+   *                    heap usage is 4 bytes * initialSize. Not all concrete implementions use this!
+   * @return the most optimal all groups collector implementation for grouping by {@link IndexDocValues}
+   */
+  public static IndexDocValuesAllGroupsCollector create(String groupField, ValueType type, int initialSize) {
+    switch (type) {
+      case VAR_INTS:
+      case FIXED_INTS_8:
+      case FIXED_INTS_16:
+      case FIXED_INTS_32:
+      case FIXED_INTS_64:
+        return new Lng(groupField);
+      case FLOAT_32:
+      case FLOAT_64:
+        return new Dbl(groupField);
+      case BYTES_FIXED_STRAIGHT:
+      case BYTES_FIXED_DEREF:
+      case BYTES_VAR_STRAIGHT:
+      case BYTES_VAR_DEREF:
+        return new BR(groupField);
+      case BYTES_VAR_SORTED:
+      case BYTES_FIXED_SORTED:
+        return new SortedBR(groupField, initialSize);
+      default:
+        throw new IllegalArgumentException(String.format("ValueType %s not supported", type));
+    }
+  }
+
+  final String groupField;
+  final Collection<GROUP_VALUE_TYPE> groups;
+
+  IndexDocValuesAllGroupsCollector(String groupField, Collection<GROUP_VALUE_TYPE> groups) {
+    this.groupField = groupField;
+    this.groups = groups;
+  }
+
+  static class Lng extends IndexDocValuesAllGroupsCollector<Long> {
+
+    private IndexDocValues.Source source;
+
+    Lng(String groupField) {
+      super(groupField, new TreeSet<Long>());
+    }
+
+    public void collect(int doc) throws IOException {
+      long value = source.getInt(doc);
+      if (!groups.contains(value)) {
+        groups.add(value);
+      }
+    }
+
+    public Collection<Long> getGroups() {
+      return groups;
+    }
+
+    public void setNextReader(IndexReader.AtomicReaderContext context) throws IOException {
+      source = context.reader.perDocValues().docValues(groupField).getSource();
+    }
+
+  }
+
+  static class Dbl extends IndexDocValuesAllGroupsCollector<Double> {
+
+    private IndexDocValues.Source source;
+
+    Dbl(String groupField) {
+      super(groupField, new TreeSet<Double>());
+    }
+
+    public void collect(int doc) throws IOException {
+      double value = source.getFloat(doc);
+      if (!groups.contains(value)) {
+        groups.add(value);
+      }
+    }
+
+    public Collection<Double> getGroups() {
+      return groups;
+    }
+
+    public void setNextReader(IndexReader.AtomicReaderContext context) throws IOException {
+      source = context.reader.perDocValues().docValues(groupField).getSource();
+    }
+
+  }
+
+  static class BR extends IndexDocValuesAllGroupsCollector<BytesRef> {
+
+    private final BytesRef spare = new BytesRef();
+
+    private IndexDocValues.Source source;
+
+    BR(String groupField) {
+      super(groupField, new TreeSet<BytesRef>());
+    }
+
+    public void collect(int doc) throws IOException {
+      BytesRef value = source.getBytes(doc, spare);
+      if (!groups.contains(value)) {
+        groups.add(new BytesRef(value));
+      }
+    }
+
+    public Collection<BytesRef> getGroups() {
+      return groups;
+    }
+
+    public void setNextReader(IndexReader.AtomicReaderContext context) throws IOException {
+      source = context.reader.perDocValues().docValues(groupField).getSource();
+    }
+
+  }
+
+  static class SortedBR extends IndexDocValuesAllGroupsCollector<BytesRef> {
+
+    private final SentinelIntSet ordSet;
+    private final BytesRef spare = new BytesRef();
+
+    private IndexDocValues.SortedSource source;
+
+    SortedBR(String groupField, int initialSize) {
+      super(groupField, new ArrayList<BytesRef>(initialSize));
+      ordSet = new SentinelIntSet(initialSize, -1);
+    }
+
+    public void collect(int doc) throws IOException {
+      int ord = source.ord(doc);
+      if (!ordSet.exists(ord)) {
+        ordSet.put(ord);
+        BytesRef value = ord == 0 ? null : source.getBytes(doc, new BytesRef());
+        groups.add(value);
+      }
+    }
+
+    public Collection<BytesRef> getGroups() {
+      return groups;
+    }
+
+    public void setNextReader(IndexReader.AtomicReaderContext context) throws IOException {
+      source = context.reader.perDocValues().docValues(groupField).getSource().asSortedSource();
+      ordSet.clear();
+      for (BytesRef countedGroup : groups) {
+        int ord = countedGroup == null ? 0 : source.getByValue(countedGroup, spare);
+        if (ord >= 0) {
+          ordSet.put(ord);
+        }
+      }
+    }
+  }
+
+}