Rebuild version map when opening internal engine (#43202)

With this change, we will rebuild the live version map and local
checkpoint using documents (including soft-deleted) from the safe commit
when opening an internal engine. This allows us to safely prune away _id
of all soft-deleted documents as the version map is always in-sync with
the Lucene index.

Relates #40741
Supersedes #42979

Author: dnhatn

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

@@ -27,7 +27,6 @@
 import org.apache.lucene.codecs.DocValuesFormat;
 import org.apache.lucene.codecs.PostingsFormat;
 import org.apache.lucene.document.LatLonDocValuesField;
-import org.apache.lucene.document.LongPoint;
 import org.apache.lucene.document.NumericDocValuesField;
 import org.apache.lucene.index.BinaryDocValues;
 import org.apache.lucene.index.CorruptIndexException;
@@ -95,7 +94,6 @@
 import org.elasticsearch.index.analysis.AnalyzerScope;
 import org.elasticsearch.index.analysis.NamedAnalyzer;
 import org.elasticsearch.index.fielddata.IndexFieldData;
-import org.elasticsearch.index.mapper.SeqNoFieldMapper;
 
 import java.io.IOException;
 import java.text.ParseException;
@@ -105,7 +103,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.function.LongConsumer;
 
 public class Lucene {
     public static final String LATEST_DOC_VALUES_FORMAT = "Lucene70";
@@ -1050,39 +1047,4 @@ public CacheHelper getReaderCacheHelper() {
             }
         };
     }
-
-    /**
-     * Scans sequence numbers (i.e., {@link SeqNoFieldMapper#NAME}) between {@code fromSeqNo}(inclusive) and {@code toSeqNo}(inclusive)
-     * in the provided directory reader. This method invokes the callback {@code onNewSeqNo} whenever a sequence number value is found.
-     *
-     * @param directoryReader the directory reader to scan
-     * @param fromSeqNo       the lower bound of a range of seq_no to scan (inclusive)
-     * @param toSeqNo         the upper bound of a range of seq_no to scan (inclusive)
-     * @param onNewSeqNo      the callback to be called whenever a new valid sequence number is found
-     */
-    public static void scanSeqNosInReader(DirectoryReader directoryReader, long fromSeqNo, long toSeqNo,
-                                          LongConsumer onNewSeqNo) throws IOException {
-        final DirectoryReader reader = Lucene.wrapAllDocsLive(directoryReader);
-        final IndexSearcher searcher = new IndexSearcher(reader);
-        searcher.setQueryCache(null);
-        final Query query = LongPoint.newRangeQuery(SeqNoFieldMapper.NAME, fromSeqNo, toSeqNo);
-        final Weight weight = searcher.createWeight(query, ScoreMode.COMPLETE_NO_SCORES, 1.0f);
-        for (LeafReaderContext leaf : reader.leaves()) {
-            final Scorer scorer = weight.scorer(leaf);
-            if (scorer == null) {
-                continue;
-            }
-            final DocIdSetIterator docIdSetIterator = scorer.iterator();
-            final NumericDocValues seqNoDocValues = leaf.reader().getNumericDocValues(SeqNoFieldMapper.NAME);
-            int docId;
-            while ((docId = docIdSetIterator.nextDoc()) != DocIdSetIterator.NO_MORE_DOCS) {
-                if (seqNoDocValues == null || seqNoDocValues.advanceExact(docId) == false) {
-                    throw new IllegalStateException("seq_no doc_values not found for doc_id=" + docId);
-                }
-                final long seqNo = seqNoDocValues.longValue();
-                assert fromSeqNo <= seqNo && seqNo <= toSeqNo : "from_seq_no=" + fromSeqNo + " seq_no=" + seqNo + " to_seq_no=" + toSeqNo;
-                onNewSeqNo.accept(seqNo);
-            }
-        }
-    }
 }

server/src/main/java/org/elasticsearch/index/engine/CombinedDocValues.java

@@ -0,0 +1,90 @@
+/*
+ * 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.engine;
+
+import org.apache.lucene.index.LeafReader;
+import org.apache.lucene.index.NumericDocValues;
+import org.elasticsearch.index.mapper.SeqNoFieldMapper;
+import org.elasticsearch.index.mapper.SourceFieldMapper;
+import org.elasticsearch.index.mapper.VersionFieldMapper;
+
+import java.io.IOException;
+import java.util.Objects;
+
+final class CombinedDocValues {
+    private final NumericDocValues versionDV;
+    private final NumericDocValues seqNoDV;
+    private final NumericDocValues primaryTermDV;
+    private final NumericDocValues tombstoneDV;
+    private final NumericDocValues recoverySource;
+
+    CombinedDocValues(LeafReader leafReader) throws IOException {
+        this.versionDV = Objects.requireNonNull(leafReader.getNumericDocValues(VersionFieldMapper.NAME), "VersionDV is missing");
+        this.seqNoDV = Objects.requireNonNull(leafReader.getNumericDocValues(SeqNoFieldMapper.NAME), "SeqNoDV is missing");
+        this.primaryTermDV = Objects.requireNonNull(
+            leafReader.getNumericDocValues(SeqNoFieldMapper.PRIMARY_TERM_NAME), "PrimaryTermDV is missing");
+        this.tombstoneDV = leafReader.getNumericDocValues(SeqNoFieldMapper.TOMBSTONE_NAME);
+        this.recoverySource = leafReader.getNumericDocValues(SourceFieldMapper.RECOVERY_SOURCE_NAME);
+    }
+
+    long docVersion(int segmentDocId) throws IOException {
+        assert versionDV.docID() < segmentDocId;
+        if (versionDV.advanceExact(segmentDocId) == false) {
+            throw new IllegalStateException("DocValues for field [" + VersionFieldMapper.NAME + "] is not found");
+        }
+        return versionDV.longValue();
+    }
+
+    long docSeqNo(int segmentDocId) throws IOException {
+        assert seqNoDV.docID() < segmentDocId;
+        if (seqNoDV.advanceExact(segmentDocId) == false) {
+            throw new IllegalStateException("DocValues for field [" + SeqNoFieldMapper.NAME + "] is not found");
+        }
+        return seqNoDV.longValue();
+    }
+
+    long docPrimaryTerm(int segmentDocId) throws IOException {
+        if (primaryTermDV == null) {
+            return -1L;
+        }
+        assert primaryTermDV.docID() < segmentDocId;
+        // Use -1 for docs which don't have primary term. The caller considers those docs as nested docs.
+        if (primaryTermDV.advanceExact(segmentDocId) == false) {
+            return -1;
+        }
+        return primaryTermDV.longValue();
+    }
+
+    boolean isTombstone(int segmentDocId) throws IOException {
+        if (tombstoneDV == null) {
+            return false;
+        }
+        assert tombstoneDV.docID() < segmentDocId;
+        return tombstoneDV.advanceExact(segmentDocId) && tombstoneDV.longValue() > 0;
+    }
+
+    boolean hasRecoverySource(int segmentDocId) throws IOException {
+        if (recoverySource == null) {
+            return false;
+        }
+        assert recoverySource.docID() < segmentDocId;
+        return recoverySource.advanceExact(segmentDocId);
+    }
+}