diff --git a/lucene/core/src/java/org/apache/lucene/codecs/lucene70/Lucene70DocValuesProducer.java b/lucene/core/src/java/org/apache/lucene/codecs/lucene70/Lucene70DocValuesProducer.java
index 386655e8c77..0bfcb8ae91f 100644
--- a/lucene/core/src/java/org/apache/lucene/codecs/lucene70/Lucene70DocValuesProducer.java
+++ b/lucene/core/src/java/org/apache/lucene/codecs/lucene70/Lucene70DocValuesProducer.java
@@ -29,7 +29,9 @@ import org.apache.lucene.index.DocValues;
 import org.apache.lucene.index.FieldInfo;
 import org.apache.lucene.index.FieldInfos;
 import org.apache.lucene.index.IndexFileNames;
+import org.apache.lucene.index.LongValuesCollector;
 import org.apache.lucene.index.NumericDocValues;
+import org.apache.lucene.index.OrdStatsCollector;
 import org.apache.lucene.index.PostingsEnum;
 import org.apache.lucene.index.SegmentReadState;
 import org.apache.lucene.index.SortedDocValues;
@@ -37,6 +39,7 @@ import org.apache.lucene.index.SortedNumericDocValues;
 import org.apache.lucene.index.SortedSetDocValues;
 import org.apache.lucene.index.TermsEnum;
 import org.apache.lucene.index.TermsEnum.SeekStatus;
+import org.apache.lucene.search.DocIdSetIterator;
 import org.apache.lucene.store.ChecksumIndexInput;
 import org.apache.lucene.store.IndexInput;
 import org.apache.lucene.store.RandomAccessInput;
@@ -801,6 +804,10 @@ final class Lucene70DocValuesProducer extends DocValuesProducer implements Close
     }
   }
 
+  private LongValuesCollector asLongValuesCollector(OrdStatsCollector collector) {
+    return (index, value) -> collector.collectOrd(value);
+  }
+
   @Override
   public SortedDocValues getSorted(FieldInfo field) throws IOException {
     SortedEntry entry = sorted.get(field.name);
@@ -864,6 +871,26 @@ final class Lucene70DocValuesProducer extends DocValuesProducer implements Close
         public int ordValue() {
           return (int) ords.get(doc);
         }
+
+        @Override
+        public void forEach(DocIdSetIterator disi, OrdStatsCollector collector) throws IOException {
+          int next = disi.nextDoc();
+          LongValuesCollector lvCollector = asLongValuesCollector(collector);
+
+          while (next != NO_MORE_DOCS) {
+            int runBegin = next;
+            int runEnd = next + 1;
+
+            next = disi.nextDoc();
+
+            while (next != NO_MORE_DOCS && next - runEnd == 0) {
+              runEnd = next + 1;
+              next = disi.nextDoc();
+            }
+
+            ords.forRange(runBegin, runEnd, lvCollector);
+          }
+        }
       };
     } else {
       // sparse
@@ -899,6 +926,65 @@ final class Lucene70DocValuesProducer extends DocValuesProducer implements Close
         public int ordValue() {
           return (int) ords.get(disi.index());
         }
+
+        @Override
+        public void forEach(DocIdSetIterator disi2, OrdStatsCollector collector) throws IOException {
+          LongValuesCollector lvCollector = asLongValuesCollector(collector);
+
+          int next = disi2.nextDoc();
+          int inext = disi.nextDoc();
+
+          long runEnd = -1;
+          long runBegin = -1;
+
+          while (next != NO_MORE_DOCS) {
+            int head = next;
+            int tail = next;
+
+            next = disi2.nextDoc();
+            //formal guard, can fire only if max matched doc is NO_MORE_DOCS - 1
+            while (next != NO_MORE_DOCS && next - tail == 1) {
+              tail = next;
+              next = disi2.nextDoc();
+            }
+
+            if (inext < head) {
+              inext = disi.advance(head);
+            }
+
+            int begin = disi.index();
+            int end = -1;
+            int missing = Math.min(inext - head, tail - head + 1);
+
+            while (inext <= tail) {
+              end = disi.index() + 1;
+              head = inext + 1;
+              inext = disi.nextDoc(); // == disi.advance(head) == disi.advance(inext + 1)
+              missing += Math.min(inext - head, tail - head + 1);
+            }
+
+            collector.collectMissing(missing);
+
+            if (end == -1) {
+              continue;
+            }
+
+            if (runBegin == -1) {
+              runBegin = begin;
+              runEnd = end;
+            } else if (runEnd == begin) {
+              runEnd = end;
+            } else {
+              ords.forRange(runBegin, runEnd, lvCollector);
+              runBegin = begin;
+              runEnd = end;
+            }
+          }
+
+          if (runBegin != -1) {
+            ords.forRange(runBegin, runEnd, lvCollector);
+          }
+        }
       };
     }
   }
@@ -1353,6 +1439,25 @@ final class Lucene70DocValuesProducer extends DocValuesProducer implements Close
           return ords.get(start++);
         }
 
+        @Override
+        public void forEach(DocIdSetIterator disi, OrdStatsCollector collector) throws IOException {
+          int next = disi.nextDoc();
+          LongValuesCollector lvCollector = asLongValuesCollector(collector);
+
+          while (next != NO_MORE_DOCS) {
+            int runBegin = next;
+            int runEnd = next + 1;
+
+            next = disi.nextDoc();
+
+            while (next != NO_MORE_DOCS && next - runEnd == 0) {
+              runEnd = next + 1;
+              next = disi.nextDoc();
+            }
+
+            ords.forRange(addresses.get(runBegin), addresses.get(runEnd), lvCollector);
+          }
+        }
       };
     } else {
       // sparse
@@ -1407,6 +1512,64 @@ final class Lucene70DocValuesProducer extends DocValuesProducer implements Close
           }
         }
 
+        @Override
+        public void forEach(DocIdSetIterator disi2, OrdStatsCollector collector) throws IOException {
+          LongValuesCollector lvCollector = asLongValuesCollector(collector);
+
+          int next = disi2.nextDoc();
+          int inext = disi.nextDoc();
+
+          long runEnd = -1;
+          long runBegin = -1;
+
+          while (next != NO_MORE_DOCS) {
+            int head = next;
+            int tail = next;
+
+            next = disi2.nextDoc();
+            //formal guard, can fire only if max matched doc is NO_MORE_DOCS - 1
+            while (next != NO_MORE_DOCS && next - tail == 1) {
+              tail = next;
+              next = disi2.nextDoc();
+            }
+
+            if (inext < head) {
+              inext = disi.advance(head);
+            }
+
+            int begin = disi.index();
+            int end = -1;
+            int missing = Math.min(inext - head, tail - head + 1);
+
+            while (inext <= tail) {
+              end = disi.index() + 1;
+              head = inext + 1;
+              inext = disi.nextDoc(); // == disi.advance(head) == disi.advance(inext + 1)
+              missing += Math.min(inext - head, tail - head + 1);
+            }
+
+            collector.collectMissing(missing);
+
+            if (end == -1) {
+              continue;
+            }
+
+            if (runBegin == -1) {
+              runBegin = begin;
+              runEnd = end;
+            } else if (runEnd == begin) {
+              runEnd = end;
+            } else {
+              ords.forRange(addresses.get(runBegin), addresses.get(runEnd), lvCollector);
+              runBegin = begin;
+              runEnd = end;
+            }
+          }
+
+          if (runBegin != -1) {
+            ords.forRange(addresses.get(runBegin), addresses.get(runEnd), lvCollector);
+          }
+        }
       };
     }
   }
diff --git a/lucene/core/src/java/org/apache/lucene/index/LongValuesCollector.java b/lucene/core/src/java/org/apache/lucene/index/LongValuesCollector.java
new file mode 100644
index 00000000000..a15a5df2c3e
--- /dev/null
+++ b/lucene/core/src/java/org/apache/lucene/index/LongValuesCollector.java
@@ -0,0 +1,28 @@
+/*
+ * 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.
+ */
+
+package org.apache.lucene.index;
+
+/**
+ * Collector for LongValues. Iterator replacement for bulk operations.
+ */
+public interface LongValuesCollector {
+  /**
+   * Called for each visited long value.
+   */
+  void collectValue(long index, long value);
+}
diff --git a/lucene/core/src/java/org/apache/lucene/index/OrdStatsCollector.java b/lucene/core/src/java/org/apache/lucene/index/OrdStatsCollector.java
new file mode 100644
index 00000000000..5021c886f3f
--- /dev/null
+++ b/lucene/core/src/java/org/apache/lucene/index/OrdStatsCollector.java
@@ -0,0 +1,34 @@
+/*
+ * 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.
+ */
+
+package org.apache.lucene.index;
+
+/**
+ * Collector for counting {@link SortedDocValues} and {@link SortedSetDocValues} ords and documents without values
+ */
+public interface OrdStatsCollector {
+  /**
+   * Called once for each ord instance.
+   */
+  void collectOrd(long ord);
+
+  /**
+   * Called multiple times for documents without values. Can be called with zero or not called at all.
+   * Implementation should sum all received counts to obtain total missing count.
+   */
+  void collectMissing(int count);
+}
diff --git a/lucene/core/src/java/org/apache/lucene/index/SortedDocValues.java b/lucene/core/src/java/org/apache/lucene/index/SortedDocValues.java
index bd234159179..2177a521c01 100644
--- a/lucene/core/src/java/org/apache/lucene/index/SortedDocValues.java
+++ b/lucene/core/src/java/org/apache/lucene/index/SortedDocValues.java
@@ -19,6 +19,7 @@ package org.apache.lucene.index;
 
 import java.io.IOException;
 
+import org.apache.lucene.search.DocIdSetIterator;
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.automaton.CompiledAutomaton;
 
@@ -132,4 +133,17 @@ public abstract class SortedDocValues extends BinaryDocValues {
     }
   }
 
+  /**
+   * Applies {@link OrdStatsCollector} for each document matched by {@link DocIdSetIterator}
+   */
+  public void forEach(DocIdSetIterator disi, OrdStatsCollector collector) throws IOException {
+    int doc;
+    while ((doc = disi.nextDoc()) != DocIdSetIterator.NO_MORE_DOCS) {
+      if (advanceExact(doc)) {
+        collector.collectOrd(ordValue());
+      } else {
+        collector.collectMissing(1);
+      }
+    }
+  }
 }
diff --git a/lucene/core/src/java/org/apache/lucene/index/SortedSetDocValues.java b/lucene/core/src/java/org/apache/lucene/index/SortedSetDocValues.java
index 9e1c6a395f9..efb88107ce7 100644
--- a/lucene/core/src/java/org/apache/lucene/index/SortedSetDocValues.java
+++ b/lucene/core/src/java/org/apache/lucene/index/SortedSetDocValues.java
@@ -19,6 +19,7 @@ package org.apache.lucene.index;
 
 import java.io.IOException;
 
+import org.apache.lucene.search.DocIdSetIterator;
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.automaton.CompiledAutomaton;
 
@@ -124,4 +125,21 @@ public abstract class SortedSetDocValues extends DocValuesIterator {
         throw new RuntimeException("unhandled case");
     }
   }
+
+  /**
+   * Applies {@link OrdStatsCollector} for each document matched by {@link DocIdSetIterator}
+   */
+  public void forEach(DocIdSetIterator disi, OrdStatsCollector collector) throws IOException {
+    int doc;
+    while ((doc = disi.nextDoc()) != DocIdSetIterator.NO_MORE_DOCS) {
+      if (advanceExact(doc)) {
+        long ord = nextOrd();
+        do {
+          collector.collectOrd(ord);
+        } while ((ord = nextOrd()) != NO_MORE_ORDS);
+      } else {
+        collector.collectMissing(1);
+      }
+    }
+  }
 }
diff --git a/lucene/core/src/java/org/apache/lucene/util/LongValues.java b/lucene/core/src/java/org/apache/lucene/util/LongValues.java
index 04fbf81dc1c..99034fd6a1c 100644
--- a/lucene/core/src/java/org/apache/lucene/util/LongValues.java
+++ b/lucene/core/src/java/org/apache/lucene/util/LongValues.java
@@ -16,6 +16,8 @@
  */
 package org.apache.lucene.util;
 
+import org.apache.lucene.index.LongValuesCollector;
+
 /** Abstraction over an array of longs.
  *  @lucene.internal */
 public abstract class LongValues  {
@@ -42,4 +44,13 @@ public abstract class LongValues  {
   /** Get value at <code>index</code>. */
   public abstract long get(long index);
 
+  /**
+   * Visits all values between begin (inclusive) and end (exclusive) indexes.
+   * {@link LongValuesCollector#collectValue(long, long)} is called for each value.
+   */
+  public void forRange(long begin, long end, LongValuesCollector collector) {
+    for (long i = begin; i < end; ++i) {
+      collector.collectValue(i, get(i));
+    }
+  }
 }
diff --git a/lucene/core/src/java/org/apache/lucene/util/packed/DirectReader.java b/lucene/core/src/java/org/apache/lucene/util/packed/DirectReader.java
index a0949505af1..b0f8ba8424b 100644
--- a/lucene/core/src/java/org/apache/lucene/util/packed/DirectReader.java
+++ b/lucene/core/src/java/org/apache/lucene/util/packed/DirectReader.java
@@ -19,6 +19,7 @@ package org.apache.lucene.util.packed;
 
 import java.io.IOException;
 
+import org.apache.lucene.index.LongValuesCollector;
 import org.apache.lucene.store.RandomAccessInput;
 import org.apache.lucene.util.LongValues;
 
@@ -70,13 +71,72 @@ public class DirectReader {
     }
   }
   
-  static final class DirectPackedReader1 extends LongValues {
-    final RandomAccessInput in;
-    final long offset;
-    
-    DirectPackedReader1(RandomAccessInput in, long offset) {
+  abstract static class DirectBulkReader extends LongValues {
+    protected final RandomAccessInput in;
+    protected final long offset;
+    private final PackedInts.Decoder decoder;
+    private final int bitsPerValue;
+
+    DirectBulkReader(RandomAccessInput in, long offset, int bitsPerValue) {
       this.in = in;
       this.offset = offset;
+      this.decoder = PackedInts.getDecoder(PackedInts.Format.PACKED, PackedInts.VERSION_MONOTONIC_WITHOUT_ZIGZAG, bitsPerValue);
+      this.bitsPerValue = bitsPerValue;
+    }
+
+    @Override
+    public final void forRange(long begin, long end, LongValuesCollector collector) {
+      final int longsPerBlock = decoder.longBlockCount();
+      final int bytesPerBlock = longsPerBlock << 3;
+      final int valuesPerBlock = decoder.longValueCount();
+
+      assert Long.bitCount(valuesPerBlock) == 1; // must be power of two !
+
+      final long mask = -valuesPerBlock;
+
+      long i = begin;
+      long preBlock = Math.min((i + valuesPerBlock - 1) & mask, end);
+
+      for (; i < preBlock; ++i) {
+        collector.collectValue(i, get(i));
+      }
+
+      final long[] unpack = new long[valuesPerBlock];
+      final long[] pack = new long[longsPerBlock];
+
+      long bulkMax = i + ((end - i) & mask);
+      long base = this.offset + ((i * bitsPerValue) >>> 3);
+
+      for (; i < bulkMax; i += valuesPerBlock) {
+        readLongs(base, pack);
+        decoder.decode(pack, 0, unpack, 0, 1);
+
+        for (int j = 0; j < valuesPerBlock; ++j) {
+          collector.collectValue(i + j, unpack[j]);
+        }
+
+        base += bytesPerBlock;
+      }
+
+      for (; i < end; ++i) {
+        collector.collectValue(i, get(i));
+      }
+    }
+
+    private void readLongs(long base, long[] longs) {
+      try {
+        for (int i = 0; i < longs.length; ++i) {
+          longs[i] = in.readLong(base + i * 8);
+        }
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      }
+    }
+  }
+
+  static final class DirectPackedReader1 extends DirectBulkReader {
+    DirectPackedReader1(RandomAccessInput in, long offset) {
+      super(in, offset, 1);
     }
 
     @Override
@@ -90,13 +150,9 @@ public class DirectReader {
     }    
   }
   
-  static final class DirectPackedReader2 extends LongValues {
-    final RandomAccessInput in;
-    final long offset;
-    
+  static final class DirectPackedReader2 extends DirectBulkReader {
     DirectPackedReader2(RandomAccessInput in, long offset) {
-      this.in = in;
-      this.offset = offset;
+      super(in, offset, 2);
     }
 
     @Override
@@ -109,14 +165,10 @@ public class DirectReader {
       }
     }    
   }
-  
-  static final class DirectPackedReader4 extends LongValues {
-    final RandomAccessInput in;
-    final long offset;
 
+  static final class DirectPackedReader4 extends DirectBulkReader {
     DirectPackedReader4(RandomAccessInput in, long offset) {
-      this.in = in;
-      this.offset = offset;
+      super(in, offset, 4);
     }
 
     @Override
@@ -129,14 +181,10 @@ public class DirectReader {
       }
     }    
   }
-    
-  static final class DirectPackedReader8 extends LongValues {
-    final RandomAccessInput in;
-    final long offset;
 
+  static final class DirectPackedReader8 extends DirectBulkReader {
     DirectPackedReader8(RandomAccessInput in, long offset) {
-      this.in = in;
-      this.offset = offset;
+      super(in, offset, 8);
     }
 
     @Override
@@ -149,13 +197,9 @@ public class DirectReader {
     }    
   }
   
-  static final class DirectPackedReader12 extends LongValues {
-    final RandomAccessInput in;
-    final long offset;
-    
+  static final class DirectPackedReader12 extends DirectBulkReader {
     DirectPackedReader12(RandomAccessInput in, long offset) {
-      this.in = in;
-      this.offset = offset;
+      super(in, offset, 12);
     }
 
     @Override
@@ -169,14 +213,10 @@ public class DirectReader {
       }
     }    
   }
-  
-  static final class DirectPackedReader16 extends LongValues {
-    final RandomAccessInput in;
-    final long offset;
-    
+
+  static final class DirectPackedReader16 extends DirectBulkReader {
     DirectPackedReader16(RandomAccessInput in, long offset) {
-      this.in = in;
-      this.offset = offset;
+      super(in, offset, 16);
     }
 
     @Override
@@ -188,14 +228,10 @@ public class DirectReader {
       }
     }
   }
-  
-  static final class DirectPackedReader20 extends LongValues {
-    final RandomAccessInput in;
-    final long offset;
 
+  static final class DirectPackedReader20 extends DirectBulkReader {
     DirectPackedReader20(RandomAccessInput in, long offset) {
-      this.in = in;
-      this.offset = offset;
+      super(in, offset, 20);
     }
 
     @Override
@@ -212,13 +248,9 @@ public class DirectReader {
     }
   }
   
-  static final class DirectPackedReader24 extends LongValues {
-    final RandomAccessInput in;
-    final long offset;
-    
+  static final class DirectPackedReader24 extends DirectBulkReader {
     DirectPackedReader24(RandomAccessInput in, long offset) {
-      this.in = in;
-      this.offset = offset;
+      super(in, offset, 24);
     }
 
     @Override
@@ -231,13 +263,9 @@ public class DirectReader {
     }
   }
   
-  static final class DirectPackedReader28 extends LongValues {
-    final RandomAccessInput in;
-    final long offset;
-    
+  static final class DirectPackedReader28 extends DirectBulkReader {
     DirectPackedReader28(RandomAccessInput in, long offset) {
-      this.in = in;
-      this.offset = offset;
+      super(in, offset, 28);
     }
     
     @Override
@@ -252,6 +280,7 @@ public class DirectReader {
     }    
   }
   
+  // 32 bits is single int, no DirectBulkReader needed here
   static final class DirectPackedReader32 extends LongValues {
     final RandomAccessInput in;
     final long offset;
@@ -271,13 +300,9 @@ public class DirectReader {
     }    
   }
   
-  static final class DirectPackedReader40 extends LongValues {
-    final RandomAccessInput in;
-    final long offset;
-    
+  static final class DirectPackedReader40 extends DirectBulkReader {
     DirectPackedReader40(RandomAccessInput in, long offset) {
-      this.in = in;
-      this.offset = offset;
+      super(in, offset, 40);
     }
     
     @Override
@@ -290,13 +315,9 @@ public class DirectReader {
     }    
   }
   
-  static final class DirectPackedReader48 extends LongValues {
-    final RandomAccessInput in;
-    final long offset;
-    
+  static final class DirectPackedReader48 extends DirectBulkReader {
     DirectPackedReader48(RandomAccessInput in, long offset) {
-      this.in = in;
-      this.offset = offset;
+      super(in, offset, 48);
     }
     
     @Override
@@ -309,13 +330,9 @@ public class DirectReader {
     }    
   }
   
-  static final class DirectPackedReader56 extends LongValues {
-    final RandomAccessInput in;
-    final long offset;
-    
+  static final class DirectPackedReader56 extends DirectBulkReader {
     DirectPackedReader56(RandomAccessInput in, long offset) {
-      this.in = in;
-      this.offset = offset;
+      super(in, offset, 56);
     }
     
     @Override
diff --git a/lucene/core/src/test/org/apache/lucene/util/packed/TestDirectPacked.java b/lucene/core/src/test/org/apache/lucene/util/packed/TestDirectPacked.java
index 3c5d5d82a0f..7a269062bdb 100644
--- a/lucene/core/src/test/org/apache/lucene/util/packed/TestDirectPacked.java
+++ b/lucene/core/src/test/org/apache/lucene/util/packed/TestDirectPacked.java
@@ -19,6 +19,7 @@ package org.apache.lucene.util.packed;
 
 import java.util.Random;
 
+import org.apache.lucene.index.LongValuesCollector;
 import org.apache.lucene.store.ByteArrayDataInput;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.IOContext;
@@ -114,6 +115,23 @@ public class TestDirectPacked extends LuceneTestCase {
       for (int j = 0; j < original.length; j++) {
         assertEquals("bpv=" + bpv, original[j], reader.get(j));
       }
+
+      reader.forRange(0, original.length, new LongValuesCollector() {
+        @Override
+        public void collectValue(long index, long value) {
+          assertEquals(original[(int) index], value);
+        }
+      });
+
+      if (original.length > 1) {
+        reader.forRange(1, original.length - 1, new LongValuesCollector() {
+          @Override
+          public void collectValue(long index, long value) {
+            assertEquals(original[(int) index], value);
+          }
+        });
+      }
+
       input.close();
     }
   }
