Dedup translog operations by reading in reverse

Currently, translog operations are read and processed one by one.  This
may be a problem as stale operations in translogs may suddenly reappear
in recoveries. To make sure that stale operations won't be processed, we
read the translog files in a reverse order (eg. from the most recent
file to the oldest file) and only process an operation if its sequence
number was not seen before.

Author: dnhatn

core/src/main/java/org/elasticsearch/index/translog/MultiSnapshot.java

@@ -19,6 +19,11 @@
 
 package org.elasticsearch.index.translog;
 
+import com.carrotsearch.hppc.LongHashSet;
+import com.carrotsearch.hppc.LongObjectHashMap;
+import com.carrotsearch.hppc.LongSet;
+import org.apache.lucene.util.FixedBitSet;
+
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Arrays;
@@ -30,32 +35,44 @@ final class MultiSnapshot implements Translog.Snapshot {
 
     private final TranslogSnapshot[] translogs;
     private final int totalOperations;
+    private int skippedOperations;
     private final Closeable onClose;
     private int index;
+    private final SeqNumSet seenSeqNo;
 
     /**
      * Creates a new point in time snapshot of the given snapshots. Those snapshots are always iterated in-order.
      */
     MultiSnapshot(TranslogSnapshot[] translogs, Closeable onClose) {
         this.translogs = translogs;
-        totalOperations = Arrays.stream(translogs).mapToInt(TranslogSnapshot::totalOperations).sum();
+        this.totalOperations = Arrays.stream(translogs).mapToInt(TranslogSnapshot::totalOperations).sum();
+        this.skippedOperations = 0;
         this.onClose = onClose;
-        index = 0;
+        this.seenSeqNo = new SeqNumSet();
+        this.index = translogs.length - 1;
     }
 
-
     @Override
     public int totalOperations() {
         return totalOperations;
     }
 
+    @Override
+    public int skippedOperations() {
+        return skippedOperations;
+    }
+
     @Override
     public Translog.Operation next() throws IOException {
-        for (; index < translogs.length; index++) {
+        for (; index >= 0; index--) {
             final TranslogSnapshot current = translogs[index];
-            Translog.Operation op = current.next();
-            if (op != null) { // if we are null we move to the next snapshot
-                return op;
+            Translog.Operation op;
+            while ((op = current.next()) != null) {
+                if (op.seqNo() < 0 || seenSeqNo.getAndSet(op.seqNo()) == false) {
+                    return op;
+                }else {
+                    skippedOperations++;
+                }
             }
         }
         return null;
@@ -65,4 +82,65 @@ public Translog.Operation next() throws IOException {
     public void close() throws IOException {
         onClose.close();
     }
+
+    /**
+     * A wrapper of {@link FixedBitSet} but allows to check if all bits are set in O(1).
+     */
+    private static final class CountedBitSet {
+        private short onBits;
+        private final FixedBitSet bitset;
+
+        CountedBitSet(short numBits) {
+            assert numBits > 0;
+            this.onBits = 0;
+            this.bitset = new FixedBitSet(numBits);
+        }
+
+        boolean getAndSet(int index) {
+            assert index >= 0;
+            boolean wasOn = bitset.getAndSet(index);
+            if (wasOn == false) {
+                onBits++;
+            }
+            return wasOn;
+        }
+
+        boolean hasAllBitsOn() {
+            return onBits == bitset.length();
+        }
+    }
+
+    /**
+     * Sequence numbers from translog are likely to form contiguous ranges, thus using two tiers can reduce memory usage.
+     */
+    static final class SeqNumSet {
+        static final short BIT_SET_SIZE = 1024;
+        private final LongSet topTier = new LongHashSet();
+        private final LongObjectHashMap<CountedBitSet> bottomTier = new LongObjectHashMap<>();
+
+        /**
+         * Marks this sequence number and returns <tt>true</tt> if it is seen before.
+         */
+        boolean getAndSet(long value) {
+            assert value >= 0;
+            final long key = value / BIT_SET_SIZE;
+
+            if (topTier.contains(key)) {
+                return true;
+            }
+
+            CountedBitSet bitset = bottomTier.get(key);
+            if (bitset == null) {
+                bitset = new CountedBitSet(BIT_SET_SIZE);
+                bottomTier.put(key, bitset);
+            }
+
+            final boolean wasOn = bitset.getAndSet(Math.toIntExact(value % BIT_SET_SIZE));
+            if (bitset.hasAllBitsOn()) {
+                bottomTier.remove(key);
+                topTier.add(key);
+            }
+            return wasOn;
+        }
+    }
 }

core/src/main/java/org/elasticsearch/index/translog/Translog.java

@@ -832,10 +832,18 @@ public int hashCode() {
     public interface Snapshot extends Closeable {
 
         /**
-         * The total number of operations in the translog.
+         * The total estimated number of operations in the snapshot.
          */
         int totalOperations();
 
+        /**
+         * The number of operations has been skipped in the snapshot so far.
+         * Unlike {@link #totalOperations()}, this value is updated each time after {@link #next()}) is called.
+         */
+        default int skippedOperations() {
+            return 0;
+        }
+
         /**
          * Returns the next operation in the snapshot or <code>null</code> if we reached the end.
          */

core/src/main/java/org/elasticsearch/indices/recovery/RecoverySourceHandler.java

@@ -562,6 +562,9 @@ protected SendSnapshotResult sendSnapshot(final long startingSeqNo, final Transl
             }
         }
 
+        logger.trace("Translog skipped [{}] operations", snapshot.skippedOperations());
+        skippedOps += snapshot.skippedOperations();
+
         if (!operations.isEmpty() || totalSentOps == 0) {
             // send the leftover operations or if no operations were sent, request the target to respond with its local checkpoint
             cancellableThreads.executeIO(sendBatch);

core/src/test/java/org/elasticsearch/index/translog/MultiSnapshotTests.java

@@ -0,0 +1,82 @@
+/*
+ * 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.index.translog;
+
+import com.carrotsearch.hppc.LongHashSet;
+import com.carrotsearch.hppc.LongSet;
+import org.elasticsearch.common.Randomness;
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.LongStream;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+
+public class MultiSnapshotTests extends ESTestCase {
+    public void testTrackSeqNumRandomRanges() throws Exception {
+        final MultiSnapshot.SeqNumSet bitSet = new MultiSnapshot.SeqNumSet();
+        final LongSet normalSet = new LongHashSet();
+        IntStream.range(0, between(20_000, 50_000)).forEach(i -> {
+            long seq = randomNonNegativeLong();
+            boolean existed = normalSet.add(seq) == false;
+            assertThat("SeqNumSet != Set", bitSet.getAndSet(seq), equalTo(existed));
+        });
+    }
+
+    public void testTrackSeqNumDenseRanges() throws Exception {
+        final MultiSnapshot.SeqNumSet bitSet = new MultiSnapshot.SeqNumSet();
+        final LongSet normalSet = new LongHashSet();
+        IntStream.range(0, between(20_000, 50_000)).forEach(i -> {
+            long seq = between(0, 5000);
+            boolean existed = normalSet.add(seq) == false;
+            assertThat("SeqNumSet != Set" + seq, bitSet.getAndSet(seq), equalTo(existed));
+        });
+    }
+
+    public void testTrackSeqNumSparseRanges() throws Exception {
+        final MultiSnapshot.SeqNumSet bitSet = new MultiSnapshot.SeqNumSet();
+        final LongSet normalSet = new LongHashSet();
+        IntStream.range(0, between(20_000, 50_000)).forEach(i -> {
+            long seq = between(i * 10_000, i * 30_000);
+            boolean existed = normalSet.add(seq) == false;
+            assertThat("SeqNumSet != Set", bitSet.getAndSet(seq), equalTo(existed));
+        });
+    }
+
+    public void testSequenceNumMimicTranslog() throws Exception {
+        final MultiSnapshot.SeqNumSet bitSet = new MultiSnapshot.SeqNumSet();
+        final LongSet normalSet = new LongHashSet();
+        long currentSeq = between(10_000_000, 1_000_000_000);
+        final int iterations = between(100, 2000);
+        for (long i = 0; i < iterations; i++) {
+            List<Long> batch = LongStream.range(currentSeq, currentSeq + between(1, 1000))
+                .boxed()
+                .collect(Collectors.toList());
+            Randomness.shuffle(batch);
+            batch.forEach(seq -> {
+                boolean existed = normalSet.add(seq) == false;
+                assertThat("SeqNumSet != Set", bitSet.getAndSet(seq), equalTo(existed));
+            });
+            currentSeq -= batch.size();
+        }
+    }
+}

core/src/test/java/org/elasticsearch/index/translog/SnapshotMatchers.java

@@ -22,10 +22,14 @@
 import org.elasticsearch.ElasticsearchException;
 import org.hamcrest.Description;
 import org.hamcrest.Matcher;
+import org.hamcrest.TypeSafeDiagnosingMatcher;
 import org.hamcrest.TypeSafeMatcher;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.stream.Collectors;
 
 
 public final class SnapshotMatchers {
@@ -50,10 +54,14 @@ public static Matcher<Translog.Snapshot> equalsTo(Translog.Operation... ops) {
     /**
      * Consumes a snapshot and make sure it's content is as expected
      */
-    public static Matcher<Translog.Snapshot> equalsTo(ArrayList<Translog.Operation> ops) {
+    public static Matcher<Translog.Snapshot> equalsTo(List<Translog.Operation> ops) {
         return new EqualMatcher(ops.toArray(new Translog.Operation[ops.size()]));
     }
 
+    public static Matcher<Translog.Snapshot> containsOperationsInAnyOrder(Collection<Translog.Operation> expectedOperations) {
+        return new ContainingInAnyOrderMatcher(expectedOperations);
+    }
+
     public static class SizeMatcher extends TypeSafeMatcher<Translog.Snapshot> {
 
         private final int size;
@@ -128,4 +136,51 @@ public void describeTo(Description description) {
     }
 
 
+    public static class ContainingInAnyOrderMatcher extends TypeSafeDiagnosingMatcher<Translog.Snapshot> {
+        private final Collection<Translog.Operation> expectedOps;
+
+        static List<Translog.Operation> drainAll(Translog.Snapshot snapshot) throws IOException {
+            final List<Translog.Operation> actualOps = new ArrayList<>();
+            Translog.Operation op;
+            while ((op = snapshot.next()) != null) {
+                actualOps.add(op);
+            }
+            return actualOps;
+        }
+
+        public ContainingInAnyOrderMatcher(Collection<Translog.Operation> expectedOps) {
+            this.expectedOps = expectedOps;
+        }
+
+        @Override
+        protected boolean matchesSafely(Translog.Snapshot snapshot, Description mismatchDescription) {
+            try {
+                List<Translog.Operation> actualOps = drainAll(snapshot);
+
+                List<Translog.Operation> notFound = expectedOps.stream()
+                    .filter(o -> actualOps.contains(o) == false)
+                    .collect(Collectors.toList());
+                if (notFound.isEmpty() == false) {
+                    mismatchDescription
+                        .appendText(" Operations not found").appendValueList("[", ", ", "]", notFound);
+                }
+
+                List<Translog.Operation> notExpected = actualOps.stream()
+                    .filter(o -> expectedOps.contains(o) == false)
+                    .collect(Collectors.toList());
+                if (notExpected.isEmpty() == false) {
+                    mismatchDescription
+                        .appendText(" Operations not expected ").appendValueList("[", ", ", "]", notExpected);
+                }
+                return notFound.isEmpty() && notExpected.isEmpty();
+            } catch (IOException ex) {
+                throw new ElasticsearchException("failed to read snapshot content", ex);
+            }
+        }
+
+        @Override
+        public void describeTo(Description description) {
+
+        }
+    }
 }