Index stale operations to Lucene to have complete history

server/src/main/java/org/elasticsearch/common/lucene/Lucene.java

@@ -30,6 +30,7 @@
 import org.apache.lucene.document.NumericDocValuesField;
 import org.apache.lucene.index.CorruptIndexException;
 import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.FilterDirectoryReader;
 import org.apache.lucene.index.FilterLeafReader;
 import org.apache.lucene.index.IndexCommit;
 import org.apache.lucene.index.IndexFileNames;
@@ -833,6 +834,58 @@ public int length() {
         };
     }
 
+    /**
+     * Wraps a directory reader to include soft-deleted documents.
+     * This should be only used to query the history of documents rather than the documents.
+     *
+     * @param in the input directory reader
+     * @return the wrapped reader including soft-deleted documents.
+     */
+    public static DirectoryReader includeSoftDeletes(DirectoryReader in) throws IOException {
+        return new DirectoryReaderWithSoftDeletes(in);
+    }
+
+    private static final class DirectoryReaderWithSoftDeletes extends FilterDirectoryReader {
+        static final class SubReaderWithSoftDeletes extends FilterLeafReader {
+            SubReaderWithSoftDeletes(LeafReader in) {
+                super(in);
+            }
+            @Override
+            public Bits getLiveDocs() {
+                return null;
+            }
+            @Override
+            public int numDocs() {
+                return maxDoc();
+            }
+            @Override
+            public CacheHelper getCoreCacheHelper() {
+                return in.getCoreCacheHelper();
+            }
+            @Override
+            public CacheHelper getReaderCacheHelper() {
+                return null; // Modifying liveDocs
+            }
+        }
+        DirectoryReaderWithSoftDeletes(DirectoryReader in) throws IOException {
+            super(in, new FilterDirectoryReader.SubReaderWrapper() {
+                @Override
+                public LeafReader wrap(LeafReader leaf) {
+                    return new SubReaderWithSoftDeletes(leaf);
+                }
+            });
+        }
+        @Override
+        protected DirectoryReader doWrapDirectoryReader(DirectoryReader in) throws IOException {
+            return includeSoftDeletes(in);
+        }
+
+        @Override
+        public CacheHelper getReaderCacheHelper() {
+            return null; // Modifying liveDocs
+        }
+    }
+
     /**
      * Returns a numeric docvalues which can be used to soft-delete documents.
      */

server/src/main/java/org/elasticsearch/common/lucene/uid/PerThreadIDVersionAndSeqNoLookup.java

@@ -145,4 +145,15 @@ DocIdAndSeqNo lookupSeqNo(BytesRef id, LeafReaderContext context) throws IOExcep
             return null;
         }
     }
+
+    /**
+     * Returns an internal posting list of the given uid
+     */
+    PostingsEnum getPostingsOrNull(BytesRef id) throws IOException {
+        if (termsEnum.seekExact(id)) {
+            docsEnum = termsEnum.postings(docsEnum, 0);
+            return docsEnum;
+        }
+        return null;
+    }
 }

server/src/main/java/org/elasticsearch/common/lucene/uid/VersionsAndSeqNoResolver.java

@@ -23,7 +23,9 @@
 import org.apache.lucene.index.LeafReader;
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.index.NumericDocValues;
+import org.apache.lucene.index.PostingsEnum;
 import org.apache.lucene.index.Term;
+import org.apache.lucene.search.DocIdSetIterator;
 import org.apache.lucene.util.CloseableThreadLocal;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.index.mapper.SeqNoFieldMapper;
@@ -193,4 +195,32 @@ public static long loadVersion(IndexReader reader, Term term) throws IOException
         final DocIdAndVersion docIdAndVersion = loadDocIdAndVersion(reader, term);
         return docIdAndVersion == null ? NOT_FOUND : docIdAndVersion.version;
     }
+
+    /**
+     * Checks for the existence of the history of a pair SeqNo/PrimaryTerm in Lucene. The checking pair is considered as existed
+     * if there is a pair such as the seqNo equals to the checking seqNo and the primary term is at least the checking term.
+     */
+    public static boolean hasHistoryInLucene(IndexReader reader, Term idTerm, long seqNo, long primaryTerm) throws IOException {
+        final PerThreadIDVersionAndSeqNoLookup[] lookups = getLookupState(reader, idTerm.field());
+        final List<LeafReaderContext> leaves = reader.leaves();
+        // iterate backwards to optimize for the frequently updated documents which are likely to be in the last segments
+        for (int i = leaves.size() - 1; i >= 0; i--) {
+            final LeafReaderContext leaf = leaves.get(i);
+            final PerThreadIDVersionAndSeqNoLookup lookup = lookups[leaf.ord];
+            final PostingsEnum postingsEnum = lookup.getPostingsOrNull(idTerm.bytes());
+            if (postingsEnum == null) {
+                continue;
+            }
+            final NumericDocValues seqNoDV = leaf.reader().getNumericDocValues(SeqNoFieldMapper.NAME);
+            final NumericDocValues primaryTermDV = leaf.reader().getNumericDocValues(SeqNoFieldMapper.PRIMARY_TERM_NAME);
+            for (int docId = postingsEnum.nextDoc(); docId != DocIdSetIterator.NO_MORE_DOCS; docId = postingsEnum.nextDoc()) {
+                if (seqNoDV != null && seqNoDV.advanceExact(docId) && primaryTermDV != null && primaryTermDV.advanceExact(docId)) {
+                    if (seqNoDV.longValue() == seqNo && primaryTermDV.longValue() >= primaryTerm) {
+                        return true;
+                    }
+                }
+            }
+        }
+        return false;
+    }
 }