perf(offsets): stop walking every offset when only distance-based encoders are active

docs/inflight/bug-162-offset-state-truncation.md

@@ -59,10 +59,12 @@ warning or `maybeTruncateBelowOrAbove` - the other references are javadoc in
 no commit data is never asserted to start quietly. A test wants both cases: first offset 0, and first offset
 above 0.
 
-<!-- post-merge: checked -->
+<!-- post-merge: checked-begin -->
 astubbs#106 (stop walking every offset), astubbs#306 (encoding density) and astubbs#207
 (`invalidOffsetMetadataPolicy` reachability) all touch this area and address neither case. The mirror
-body implies astubbs#106 might; it does not.
+body implies astubbs#106 might; it does not - re-confirmed against its merged tree, which changes only
+the offsets/ encoders and leaves `PartitionState#maybeTruncateBelowOrAbove` untouched.
+<!-- post-merge: checked-end -->
 
 ## Draft replacement for the mirror's `## Fork status` (NOT posted)
 

docs/todo-index.md

@@ -136,8 +136,6 @@ that line, leave it in the code - it will show up here.
 
 - TODO: optimisation - inline this into the partition iteration loop in {@link WorkManager}
 - TODO: optimisation - could double the run-length range from Short.MAX_VALUE (~33,000) to Short.MAX_VALUE * 2
-- TODO VERY large offset ranges is slow (Integer.MAX_VALUE) - encoding scans could be avoided if passing in map of incompletes which should already be known
-- todo refactor this loop into the encoders (or sequential vs non sequential encoders) as RunLength doesn't need
 
 **`parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/state/PartitionState.java`**
 

parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/offsets/OffsetEncoder.java

@@ -47,6 +47,25 @@ protected OffsetEncoder(OffsetSimultaneousEncoder offsetSimultaneousEncoder, Off
 
     abstract int getEncodedSize();
 
+    /**
+     * Whether this encoder must be shown <em>every</em> offset in the range being encoded, or whether it can be shown
+     * only the offsets at which the completion state changes (plus the range boundaries).
+     * <p>
+     * Encoders which write one unit of output per call (e.g. {@link BitSetEncoder}, {@link ByteBufferEncoder}) require
+     * every offset. Encoders which derive their output from the <em>distance</em> between the offsets they are shown
+     * (e.g. {@link RunLengthEncoder}) do not - see {@link OffsetSimultaneousEncoder#invoke()} for how that is exploited
+     * to avoid walking enormous offset ranges.
+     * <p>
+     * Defaults to the safe answer ({@code true}): a new encoder gets correct behaviour without having to know anything
+     * about the sparse iteration optimisation. Only override this to {@code false} if the encoder is genuinely
+     * distance-based.
+     *
+     * @return true if every offset in the range must be passed to this encoder, in ascending order
+     */
+    boolean requiresEveryOffset() {
+        return true;
+    }
+
     boolean quiteSmall() {
         return this.getEncodedSize() < OffsetSimultaneousEncoder.LARGE_ENCODED_SIZE_THRESHOLD_BYTES;
     }

parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/offsets/OffsetSimultaneousEncoder.java

@@ -9,6 +9,7 @@
 import bz.stub.parallelconsumer.ParallelConsumerOptions;
 import bz.stub.parallelconsumer.state.PartitionState;
 import bz.stub.parallelconsumer.state.WorkManager;
+import lombok.AccessLevel;
 import lombok.Getter;
 import lombok.ToString;
 import lombok.extern.slf4j.Slf4j;
@@ -85,7 +86,7 @@
      * Visible for testing.
      */
     @ToString.Include
     public static boolean compressionForced = false;
 
     /**
      * Used to prevent tests running in parallel that depends on setting static state in this class. Manipulation of
@@ -98,6 +99,17 @@
      */
     private final ConcurrentHashMap.KeySetView<OffsetEncoder, Boolean> activeEncoders;
 
+    /**
+     * Which iteration strategy {@link #invoke()} actually used - true if the sparse (transition-only) walk was taken,
+     * false if every offset in the range was visited.
+     * <p>
+     * Visible for testing.
+     *
+     * @see #invoke(boolean)
+     */
+    @Getter(AccessLevel.PACKAGE)
+    private boolean sparseIterationUsed = false;
+
     public OffsetSimultaneousEncoder(long baseOffsetToCommit, long highestSucceededOffset, SortedSet<Long> incompleteOffsets) {
         this.lowWaterMark = baseOffsetToCommit;
         this.incompleteOffsets = incompleteOffsets;
@@ -215,23 +227,40 @@
      * TODO: optimisation - could double the run-length range from Short.MAX_VALUE (~33,000) to Short.MAX_VALUE * 2
      *  (~66,000) by using unsigned shorts instead (highest representable relative offset is Short.MAX_VALUE because each
      *  run-length entry is a Short)
-     * <p>
-     *  TODO VERY large offset ranges is slow (Integer.MAX_VALUE) - encoding scans could be avoided if passing in map of incompletes which should already be known
+     *
+     * @see #buildSparseRelativeOffsetsToVisit()
      */
     public OffsetSimultaneousEncoder invoke() {
+        /*
+         * Decide the iteration strategy ONCE, up front, from the encoders that are actually active. Encoders which
+         * write one unit of output per call (BitSet, ByteBuffer) must see every offset; if any such encoder is active
+         * we have to walk the whole range anyway, so there is nothing to gain from the sparse walk.
+         */
+        boolean canIterateSparsely = activeEncoders.stream().noneMatch(OffsetEncoder::requiresEveryOffset);
+        return invoke(canIterateSparsely);
+    }
+
+    /**
+     * Visible for testing ONLY - production code must use {@link #invoke()}, which chooses the strategy itself.
+     *
+     * @param useSparseIteration if true, visit only the offsets needed by distance-based encoders (see
+     *                           {@link #buildSparseRelativeOffsetsToVisit()}); if false, visit every offset in the
+     *                           range. Only safe to pass true when no active encoder
+     *                           {@link OffsetEncoder#requiresEveryOffset()}.
+     */
+    OffsetSimultaneousEncoder invoke(boolean useSparseIteration) {
         log.debug("Starting encode of incompletes, base offset is: {}, end offset is: {}", lowWaterMark, getEndOffsetExclusive());
         log.trace("Incompletes are: {}", this.incompleteOffsets);
 
         //
-        log.debug("Encode loop offset start,end: [{},{}] length: {}", this.lowWaterMark, getEndOffsetExclusive(), lengthBetweenBaseAndHighOffset);
-        /*
-         * todo refactor this loop into the encoders (or sequential vs non sequential encoders) as RunLength doesn't need
-         *  to look at every offset in the range, only the ones that change from 0 to 1. BitSet however needs to iterate
-         *  the entire range. So when BitSet can't be used, the encoding would be potentially a lot faster as RunLength
-         *  didn't need the whole loop.
-         */
-        Range relativeOffsetsLongRange = range(lengthBetweenBaseAndHighOffset);
-        relativeOffsetsLongRange.forEach(relativeOffset -> {
+        log.debug("Encode loop offset start,end: [{},{}] length: {} sparse: {}", this.lowWaterMark, getEndOffsetExclusive(), lengthBetweenBaseAndHighOffset, useSparseIteration);
+
+        this.sparseIterationUsed = useSparseIteration;
+        Iterable<Long> relativeOffsetsToVisit = useSparseIteration
+                ? buildSparseRelativeOffsetsToVisit()
+                : range(lengthBetweenBaseAndHighOffset);
+
+        relativeOffsetsToVisit.forEach(relativeOffset -> {
             // range index (relativeOffset) is used as we don't actually encode offsets, we encode the relative offset from the base offset
             final long actualOffset = this.lowWaterMark + relativeOffset;
             final boolean isIncomplete = this.incompleteOffsets.contains(actualOffset);
@@ -257,6 +286,93 @@
         return this;
     }
 
+    /**
+     * The (ascending, deduplicated) set of relative offsets that a purely distance-based encoder needs to be shown in
+     * order to produce exactly the same output as walking the entire range.
+     * <p>
+     * <b>Why this is correct - do not "simplify" this away.</b> {@link RunLengthEncoder} accumulates the open run by the
+     * <em>delta</em> to the previously seen offset ({@code currentRunLengthSize += relativeOffset - previousRangeIndex}),
+     * it does not count calls. So for a maximal run of same-state offsets {@code [a,b]}:
+     * <ul>
+     *     <li>the call at {@code a} sees a state change, closes the previous run and opens a new one of size 1;</li>
+     *     <li>the call at {@code b} adds delta {@code b - a}, giving the correct length {@code b - a + 1};</li>
+     *     <li>any additional calls <em>inside</em> {@code (a,b)} are harmless - the deltas simply telescope to the
+     *     same total.</li>
+     * </ul>
+     * Therefore it suffices to visit the first AND last offset of every maximal run. Both are needed: visiting only run
+     * starts would close each run at length 1.
+     * <p>
+     * Because a state change can only happen at an incomplete offset or immediately after one, the run boundaries are
+     * covered by the union of:
+     * <ul>
+     *     <li>{@code 0} (the first run always starts here, and it anchors the initial delta), and {@code length - 1}
+     *     (the last run always ends here),</li>
+     *     <li>for each incomplete offset in range, its relative offset and its two neighbours, clamped to the range.</li>
+     * </ul>
+     * A few redundant offsets are included (per the "harmless inside a run" property above) in exchange for an
+     * obviously-correct construction.
+     * <p>
+     * Note the returned collection is sized by the number of <em>incompletes</em>, never by the range - the whole point
+     * is to not touch a structure proportional to a range that can be ~2.1 billion wide.
+     */
+    private List<Long> buildSparseRelativeOffsetsToVisit() {
+        final long lastRelativeOffset = lengthBetweenBaseAndHighOffset - 1;
+        if (lastRelativeOffset < 0) {
+            // empty range - the full scan would visit nothing either
+            return Collections.emptyList();
+        }
+
+        SortedSet<Long> relativeOffsetsToVisit = new TreeSet<>();
+        relativeOffsetsToVisit.add(0L);
+        relativeOffsetsToVisit.add(lastRelativeOffset);
+
+        // Only incompletes that actually fall inside the encoded range matter - the full scan never looks outside it.
+        // Filtered by hand rather than via subSet(): this set is caller-supplied, and SortedSet#subSet throws
+        // IllegalArgumentException when the bounds fall outside an already-restricted view. That would surface as a
+        // failure to encode a commit, so it is not worth the risk.
+        //
+        // There is deliberately NO early exit, and the bound is the INCLUSIVE last offset. Three reasons, the first
+        // being the one that stops a reader putting the break back:
+        //  - it would never fire. OffsetMapCodecManager#encodeOffsetsCompressed samples highestSucceeded ONCE and
+        //    derives both this set (getIncompleteOffsetsBelow, which keeps x < that sample) and this encoder's range
+        //    top from it, so the set's maximum is highestSucceeded - 1 while an above-range test needs
+        //    > highestSucceeded. The two cannot overlap. The scan over entries below lowWaterMark is the `continue`
+        //    below, and is there with or without an early exit.
+        //  - incompleteOffsets is caller-supplied and its comparator need not be the natural one (the field's own
+        //    javadoc promises no order requirement), so breaking on the first out-of-range entry can skip in-range
+        //    ones. Note subSet WOULD have honoured the set's comparator - avoiding it must not import an ordering
+        //    assumption it never had.
+        //  - an exclusive bound is lowWaterMark + length, i.e. highestSucceededOffset + 1, which wraps to
+        //    Long.MIN_VALUE at Long.MAX_VALUE and silently drops every incomplete - encoding them as completed. The
+        //    inclusive form reconstructs highestSucceededOffset, a long that already existed, so it cannot wrap.
+        final long lastOffsetInclusive = lowWaterMark + lastRelativeOffset; // == highestSucceededOffset, cannot wrap
+        for (Long incompleteOffset : this.incompleteOffsets) {
+            if (incompleteOffset < lowWaterMark || incompleteOffset > lastOffsetInclusive) {
+                continue; // outside the encoded range - the full scan never looks there either
+            }
+            final long relativeOffset = incompleteOffset - lowWaterMark;
+            relativeOffsetsToVisit.add(Math.max(0, relativeOffset - 1));
+            relativeOffsetsToVisit.add(relativeOffset);
+            relativeOffsetsToVisit.add(Math.min(lastRelativeOffset, relativeOffset + 1));
+        }
+
+        log.debug("Sparse encode: visiting {} relative offsets instead of {}", relativeOffsetsToVisit.size(), lengthBetweenBaseAndHighOffset);
+
+        // ascending order is required - the run-length delta must always be positive
+        return new ArrayList<>(relativeOffsetsToVisit);
+    }
+
+    /**
+     * Visible for testing ONLY. Drops every encoder that {@link OffsetEncoder#requiresEveryOffset()}, reproducing the
+     * production situation in which the sparse path in {@link #invoke()} is actually taken: an offset range so large
+     * that {@link BitSetEncoder} cannot even be constructed, leaving only the {@link RunLengthEncoder}s.
+     * <p>
+     * Lets tests exercise the sparse path over small, fast ranges.
+     */
+    void dropEncodersRequiringEveryOffset() {
+        activeEncoders.removeIf(OffsetEncoder::requiresEveryOffset);
+    }
+
     private void registerEncodings(final Set<? extends OffsetEncoder> encoders) {
         List<OffsetEncoder> toRemove = new ArrayList<>();
         for (OffsetEncoder encoder : encoders) {

parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/offsets/RunLengthEncoder.java

@@ -78,6 +78,18 @@ public void encodeCompletedOffset(final long relativeOffset) throws EncodingNotS
         encodeRunLength(true, relativeOffset);
     }
 
+    /**
+     * Run-length encoding is <em>distance</em> based, not call based: {@link #encodeRunLength} grows the open run by
+     * {@code relativeOffset - previousRangeIndex}, so it does not care how many offsets in between it was shown.
+     * <p>
+     * It therefore only needs to see the first and last offset of each maximal run of same-state offsets - see
+     * {@link OffsetSimultaneousEncoder#invoke()}.
+     */
+    @Override
+    boolean requiresEveryOffset() {
+        return false;
+    }
+
     @Override
     public byte[] serialise() throws EncodingNotSupportedException {
         addTail();