Sequence Numbers Lucene commit data uses Iterable interface

Sequence number related data (maximum sequence number, local checkpoint,
and global checkpoint) gets stored in Lucene on each commit. The logical
place to store this data is on each Lucene commit's user commit data
structure (see IndexWriter#setCommitData and the new version
IndexWriter#setLiveCommitData). However, previously we did not store the
maximum sequence number in the commit data because the commit data got
copied over before the Lucene IndexWriter flushed the documents to segments
in the commit.  This means that between the time that the commit data was
set on the IndexWriter and the time that the IndexWriter completes the commit,
documents with higher sequence numbers could have entered the commit.
Hence, we would use FieldStats on the _seq_no field in the documents to get
the maximum sequence number value, but this suffers the drawback that if the
last sequence number in the commit corresponded to a delete document action,
that sequence number would not show up in FieldStats as there would be no
corresponding document in Lucene.

In Lucene 6.2, the commit data was changed to take an Iterable interface, so
that the commit data can be calculated and retrieved *after* all documents
have been flushed, while the commit data itself is being set on the Lucene commit.
This commit changes max_seq_no so it is stored in the commit data instead of
being calculated from FieldStats, taking advantage of the deferred calculation
of the max_seq_no through passing an Iterable that dynamically sets the iterator
data.

Author: Ali Beyad

core/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -43,7 +43,6 @@
 import org.apache.lucene.util.InfoStream;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.Version;
-import org.elasticsearch.action.fieldstats.FieldStats;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.lease.Releasable;
@@ -59,7 +58,6 @@
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.index.mapper.Uid;
-import org.elasticsearch.index.mapper.internal.SeqNoFieldMapper;
 import org.elasticsearch.index.merge.MergeStats;
 import org.elasticsearch.index.merge.OnGoingMerge;
 import org.elasticsearch.index.seqno.SeqNoStats;
@@ -86,8 +84,6 @@
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Function;
 
-import static org.elasticsearch.index.seqno.SequenceNumbersService.NO_OPS_PERFORMED;
-
 public class InternalEngine extends Engine {
 
     /**
@@ -121,6 +117,7 @@ public class InternalEngine extends Engine {
     private final SequenceNumbersService seqNoService;
     static final String LOCAL_CHECKPOINT_KEY = "local_checkpoint";
     static final String GLOBAL_CHECKPOINT_KEY = "global_checkpoint";
+    static final String MAX_SEQ_NO = "max_seq_no";
 
     // How many callers are currently requesting index throttling.  Currently there are only two situations where we do this: when merges
     // are falling behind and when writing indexing buffer to disk is too slow.  When this is 0, there is no throttling, else we throttling
@@ -326,32 +323,20 @@ private Translog.TranslogGeneration loadTranslogIdFromCommit(IndexWriter writer)
     }
 
     private SeqNoStats loadSeqNoStatsFromCommit(IndexWriter writer) throws IOException {
-        final long maxSeqNo;
-        try (IndexReader reader = DirectoryReader.open(writer)) {
-            final FieldStats stats = SeqNoFieldMapper.Defaults.FIELD_TYPE.stats(reader);
-            if (stats != null) {
-                maxSeqNo = (long) stats.getMaxValue();
-            } else {
-                maxSeqNo = NO_OPS_PERFORMED;
+        long maxSeqNo = SequenceNumbersService.NO_OPS_PERFORMED;
+        long localCheckpoint = SequenceNumbersService.NO_OPS_PERFORMED;
+        long globalCheckpoint = SequenceNumbersService.UNASSIGNED_SEQ_NO;
+        for (Map.Entry<String, String> entry : writer.getLiveCommitData()) {
+            final String key = entry.getKey();
+            if (key.equals(LOCAL_CHECKPOINT_KEY)) {
+                localCheckpoint = Long.parseLong(entry.getValue());
+            } else if (key.equals(GLOBAL_CHECKPOINT_KEY)) {
+                globalCheckpoint = Long.parseLong(entry.getValue());
+            } else if (key.equals(MAX_SEQ_NO)) {
+                maxSeqNo = Long.parseLong(entry.getValue());
             }
         }
 
-        final Map<String, String> commitUserData = writer.getCommitData();
-
-        final long localCheckpoint;
-        if (commitUserData.containsKey(LOCAL_CHECKPOINT_KEY)) {
-            localCheckpoint = Long.parseLong(commitUserData.get(LOCAL_CHECKPOINT_KEY));
-        } else {
-            localCheckpoint = SequenceNumbersService.NO_OPS_PERFORMED;
-        }
-
-        final long globalCheckpoint;
-        if (commitUserData.containsKey(GLOBAL_CHECKPOINT_KEY)) {
-            globalCheckpoint = Long.parseLong(commitUserData.get(GLOBAL_CHECKPOINT_KEY));
-        } else {
-            globalCheckpoint = SequenceNumbersService.UNASSIGNED_SEQ_NO;
-        }
-
         return new SeqNoStats(maxSeqNo, localCheckpoint, globalCheckpoint);
     }
 
@@ -1336,10 +1321,26 @@ private void commitIndexWriter(IndexWriter writer, Translog translog, String syn
             }
 
             if (logger.isTraceEnabled()) {
-                logger.trace("committing writer with commit data [{}]", commitData);
+                logger.trace("committing writer with commit data (max_seq_no excluded) [{}]", commitData);
             }
 
-            indexWriter.setCommitData(commitData);
+            indexWriter.setLiveCommitData(() -> {
+                /**
+                 * The user data in the commitData map contains data that must be evaluated *before*
+                 * Lucene flushes segments, including the local and global checkpoints amongst other values.
+                 * The maximum sequence number is different - we never want the maximum sequence number to be
+                 * less than the last sequence number to go into a Lucene commit, otherwise we run the risk
+                 * of re-using a sequence number for two different documents when restoring from this commit
+                 * point and subsequently writing new documents to the index.  Since we only know which Lucene
+                 * documents made it into the final commit after the {@link IndexWriter#commit()} call flushes
+                 * all documents, we defer computation of the max_seq_no to the time of invocation of the commit
+                 * data iterator (which occurs after all documents have been flushed to Lucene).
+                 */
+                final Map<String, String> deferredCommitData = new HashMap<>(commitData.size() + 1);
+                deferredCommitData.putAll(commitData);
+                deferredCommitData.put(MAX_SEQ_NO, Long.toString(seqNoService().getMaxSeqNo()));
+                return deferredCommitData.entrySet().iterator();
+            });
             writer.commit();
         } catch (Exception ex) {
             try {
@@ -1395,7 +1396,8 @@ public MergeStats getMergeStats() {
     public SequenceNumbersService seqNoService() {
         return seqNoService;
     }
-        @Override
+
+    @Override
     public DocsStats getDocStats() {
         final int numDocs = indexWriter.numDocs();
         final int maxDoc = indexWriter.maxDoc();

core/src/main/java/org/elasticsearch/index/seqno/SequenceNumbersService.java

@@ -81,6 +81,13 @@ public long generateSeqNo() {
         return localCheckpointService.generateSeqNo();
     }
 
+    /**
+     * Gets the maximum sequence number seen so far.  See {@link LocalCheckpointService#getMaxSeqNo()} for details.
+     */
+    public long getMaxSeqNo() {
+        return localCheckpointService.getMaxSeqNo();
+    }
+
     /**
      * marks the given seqNo as completed. See {@link LocalCheckpointService#markSeqNoAsCompleted(long)}
      * more details

core/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java

@@ -583,6 +583,10 @@ public SequenceNumbersService seqNoService() {
             assertThat(
                 Long.parseLong(stats1.getUserData().get(InternalEngine.GLOBAL_CHECKPOINT_KEY)),
                 equalTo(SequenceNumbersService.UNASSIGNED_SEQ_NO));
+            assertThat(stats1.getUserData(), hasKey(InternalEngine.MAX_SEQ_NO));
+            assertThat(
+                Long.parseLong(stats1.getUserData().get(InternalEngine.MAX_SEQ_NO)),
+                equalTo(SequenceNumbersService.NO_OPS_PERFORMED));
 
             maxSeqNo.set(rarely() ? SequenceNumbersService.NO_OPS_PERFORMED : randomIntBetween(0, 1024));
             localCheckpoint.set(
@@ -608,6 +612,8 @@ public SequenceNumbersService seqNoService() {
             assertThat(
                 Long.parseLong(stats2.getUserData().get(InternalEngine.GLOBAL_CHECKPOINT_KEY)),
                 equalTo(globalCheckpoint.get()));
+            assertThat(stats2.getUserData(), hasKey(InternalEngine.MAX_SEQ_NO));
+            assertThat(Long.parseLong(stats2.getUserData().get(InternalEngine.MAX_SEQ_NO)), equalTo(maxSeqNo.get()));
         } finally {
             IOUtils.close(engine);
         }
@@ -1625,6 +1631,7 @@ public void testSeqNoAndCheckpoints() throws IOException {
         long localCheckpoint = SequenceNumbersService.NO_OPS_PERFORMED;
         long replicaLocalCheckpoint = SequenceNumbersService.NO_OPS_PERFORMED;
         long globalCheckpoint = SequenceNumbersService.UNASSIGNED_SEQ_NO;
+        long maxSeqNo = SequenceNumbersService.NO_OPS_PERFORMED;
         InternalEngine initialEngine = null;
 
         try {
@@ -1653,6 +1660,7 @@ public void testSeqNoAndCheckpoints() throws IOException {
 
                 if (rarely()) {
                     localCheckpoint = primarySeqNo;
+                    maxSeqNo = primarySeqNo;
                     globalCheckpoint = replicaLocalCheckpoint;
                     initialEngine.seqNoService().updateGlobalCheckpointOnPrimary();
                     initialEngine.flush(true, true);
@@ -1661,6 +1669,7 @@ public void testSeqNoAndCheckpoints() throws IOException {
 
             initialEngine.seqNoService().updateGlobalCheckpointOnPrimary();
 
+            assertEquals(primarySeqNo, initialEngine.seqNoService().getMaxSeqNo());
             assertThat(initialEngine.seqNoService().stats().getMaxSeqNo(), equalTo(primarySeqNo));
             assertThat(initialEngine.seqNoService().stats().getLocalCheckpoint(), equalTo(primarySeqNo));
             assertThat(initialEngine.seqNoService().stats().getGlobalCheckpoint(), equalTo(replicaLocalCheckpoint));
@@ -1671,6 +1680,9 @@ public void testSeqNoAndCheckpoints() throws IOException {
             assertThat(
                 Long.parseLong(initialEngine.commitStats().getUserData().get(InternalEngine.GLOBAL_CHECKPOINT_KEY)),
                 equalTo(globalCheckpoint));
+            assertThat(
+                Long.parseLong(initialEngine.commitStats().getUserData().get(InternalEngine.MAX_SEQ_NO)),
+                equalTo(maxSeqNo));
 
         } finally {
             IOUtils.close(initialEngine);
@@ -1681,12 +1693,17 @@ public void testSeqNoAndCheckpoints() throws IOException {
             recoveringEngine = new InternalEngine(copy(initialEngine.config(), EngineConfig.OpenMode.OPEN_INDEX_AND_TRANSLOG));
             recoveringEngine.recoverFromTranslog();
 
+            assertEquals(primarySeqNo, recoveringEngine.seqNoService().getMaxSeqNo());
             assertThat(
                 Long.parseLong(recoveringEngine.commitStats().getUserData().get(InternalEngine.LOCAL_CHECKPOINT_KEY)),
                 equalTo(primarySeqNo));
             assertThat(
                 Long.parseLong(recoveringEngine.commitStats().getUserData().get(InternalEngine.GLOBAL_CHECKPOINT_KEY)),
                 equalTo(globalCheckpoint));
+            assertThat(
+                Long.parseLong(recoveringEngine.commitStats().getUserData().get(InternalEngine.MAX_SEQ_NO)),
+                // after recovering from translog, all docs have been flushed to Lucene segments, so check against primarySeqNo
+                equalTo(primarySeqNo));
             assertThat(recoveringEngine.seqNoService().stats().getLocalCheckpoint(), equalTo(primarySeqNo));
             assertThat(recoveringEngine.seqNoService().stats().getMaxSeqNo(), equalTo(primarySeqNo));
             assertThat(recoveringEngine.seqNoService().generateSeqNo(), equalTo(primarySeqNo + 1));