Lazily load soft-deletes for searchable snapshot shards (#69203)

Opening a Lucene index that supports soft-deletes currently creates the liveDocs bitset eagerly. This requires scanning
the doc values to materialize the liveDocs bitset from the soft-delete doc values. In order for searchable snapshot shards
to be available for searches as quickly as possible (i.e. on recovery, or in case of FrozenEngine whenever a search comes
in), they should read as little as possible from the Lucene files.

This commit introduces a LazySoftDeletesDirectoryReaderWrapper, a variant of Lucene's
SoftDeletesDirectoryReaderWrapper that loads the livedocs bitset lazily on first access. It is special-tailored to
ReadOnlyEngine / FrozenEngine as it only operates on non-NRT readers.

qa/smoke-test-http/src/test/java/org/elasticsearch/http/ClusterStatsRestCancellationIT.java

@@ -172,7 +172,7 @@ public class ClusterStatsRestCancellationIT extends HttpSmokeTestCase {
         final Semaphore statsBlock = new Semaphore(1);
 
         StatsBlockingEngine(EngineConfig config) {
-            super(config, null, new TranslogStats(), true, Function.identity(), true);
+            super(config, null, new TranslogStats(), true, Function.identity(), true, false);
         }
 
         @Override

qa/smoke-test-http/src/test/java/org/elasticsearch/http/IndicesSegmentsRestCancellationIT.java

@@ -169,7 +169,7 @@ public class IndicesSegmentsRestCancellationIT extends HttpSmokeTestCase {
         final Semaphore searcherBlock = new Semaphore(1);
 
         SearcherBlockingEngine(EngineConfig config) {
-            super(config, null, new TranslogStats(), true, Function.identity(), true);
+            super(config, null, new TranslogStats(), true, Function.identity(), true, false);
         }
 
         @Override

server/src/main/java/org/apache/lucene/index/LazySoftDeletesDirectoryReaderWrapper.java

@@ -0,0 +1,259 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.apache.lucene.index;
+
+import org.apache.lucene.document.Field;
+import org.apache.lucene.search.DocIdSetIterator;
+import org.apache.lucene.search.DocValuesFieldExistsQuery;
+import org.apache.lucene.util.Bits;
+import org.apache.lucene.util.FixedBitSet;
+import org.elasticsearch.common.lucene.Lucene;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * This is a modified version of {@link SoftDeletesDirectoryReaderWrapper} that materializes the liveDocs
+ * bitset lazily. In contrast to {@link SoftDeletesDirectoryReaderWrapper}, this wrapper can only be used
+ * for non-NRT readers.
+ *
+ * This reader filters out documents that have a doc values value in the given field and treat these
+ * documents as soft deleted. Hard deleted documents will also be filtered out in the live docs of this reader.
+ * @see IndexWriterConfig#setSoftDeletesField(String)
+ * @see IndexWriter#softUpdateDocument(Term, Iterable, Field...)
+ * @see SoftDeletesRetentionMergePolicy
+ */
+public final class LazySoftDeletesDirectoryReaderWrapper extends FilterDirectoryReader {
+    private final CacheHelper readerCacheHelper;
+    /**
+     * Creates a new soft deletes wrapper.
+     * @param in the incoming directory reader
+     * @param field the soft deletes field
+     */
+    public LazySoftDeletesDirectoryReaderWrapper(DirectoryReader in, String field) throws IOException {
+        super(in, new LazySoftDeletesSubReaderWrapper(field));
+        readerCacheHelper = in.getReaderCacheHelper() == null ? null : new DelegatingCacheHelper(in.getReaderCacheHelper());
+    }
+
+    @Override
+    protected DirectoryReader doWrapDirectoryReader(DirectoryReader in) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public CacheHelper getReaderCacheHelper() {
+        return readerCacheHelper;
+    }
+
+    private static class LazySoftDeletesSubReaderWrapper extends SubReaderWrapper {
+        private final String field;
+
+        LazySoftDeletesSubReaderWrapper(String field) {
+            Objects.requireNonNull(field, "Field must not be null");
+            this.field = field;
+        }
+
+        @Override
+        protected LeafReader[] wrap(List<? extends LeafReader> readers) {
+            List<LeafReader> wrapped = new ArrayList<>(readers.size());
+            for (LeafReader reader : readers) {
+                LeafReader wrap = wrap(reader);
+                assert wrap != null;
+                if (wrap.numDocs() != 0) {
+                    wrapped.add(wrap);
+                }
+            }
+            return wrapped.toArray(new LeafReader[0]);
+        }
+
+        @Override
+        public LeafReader wrap(LeafReader reader) {
+            return LazySoftDeletesDirectoryReaderWrapper.wrap(reader, field);
+        }
+    }
+
+    static LeafReader wrap(LeafReader reader, String field) {
+        final SegmentReader segmentReader = Lucene.segmentReader(reader);
+        assert segmentReader.isNRT == false : "expected non-NRT reader";
+        final SegmentCommitInfo segmentInfo = segmentReader.getSegmentInfo();
+        final int numSoftDeletes = segmentInfo.getSoftDelCount();
+        if (numSoftDeletes == 0) {
+            return reader;
+        }
+        final int maxDoc = reader.maxDoc();
+        final int numDocs = maxDoc - segmentInfo.getDelCount() - segmentInfo.getSoftDelCount();
+        final LazyBits lazyBits = new LazyBits(maxDoc, field, reader, numSoftDeletes, numDocs);
+        return reader instanceof CodecReader ? new LazySoftDeletesFilterCodecReader((CodecReader) reader, lazyBits, numDocs)
+            : new LazySoftDeletesFilterLeafReader(reader, lazyBits, numDocs);
+    }
+
+    public static final class LazySoftDeletesFilterLeafReader extends FilterLeafReader {
+        private final LeafReader reader;
+        private final LazyBits bits;
+        private final int numDocs;
+        private final CacheHelper readerCacheHelper;
+
+        public LazySoftDeletesFilterLeafReader(LeafReader reader, LazyBits bits, int numDocs) {
+            super(reader);
+            this.reader = reader;
+            this.bits = bits;
+            this.numDocs = numDocs;
+            this.readerCacheHelper = reader.getReaderCacheHelper() == null ? null :
+                new DelegatingCacheHelper(reader.getReaderCacheHelper());
+        }
+
+        @Override
+        public LazyBits getLiveDocs() {
+            return bits;
+        }
+
+        @Override
+        public int numDocs() {
+            return numDocs;
+        }
+
+        @Override
+        public CacheHelper getCoreCacheHelper() {
+            return reader.getCoreCacheHelper();
+        }
+
+        @Override
+        public CacheHelper getReaderCacheHelper() {
+            return readerCacheHelper;
+        }
+    }
+
+    public static final class LazySoftDeletesFilterCodecReader extends FilterCodecReader {
+        private final LeafReader reader;
+        private final LazyBits bits;
+        private final int numDocs;
+        private final CacheHelper readerCacheHelper;
+
+        public LazySoftDeletesFilterCodecReader(CodecReader reader, LazyBits bits, int numDocs) {
+            super(reader);
+            this.reader = reader;
+            this.bits = bits;
+            this.numDocs = numDocs;
+            this.readerCacheHelper = reader.getReaderCacheHelper() == null ? null :
+                new DelegatingCacheHelper(reader.getReaderCacheHelper());
+        }
+
+        @Override
+        public LazyBits getLiveDocs() {
+            return bits;
+        }
+
+        @Override
+        public int numDocs() {
+            return numDocs;
+        }
+
+        @Override
+        public CacheHelper getCoreCacheHelper() {
+            return reader.getCoreCacheHelper();
+        }
+
+        @Override
+        public CacheHelper getReaderCacheHelper() {
+            return readerCacheHelper;
+        }
+    }
+
+    private static class DelegatingCacheHelper implements CacheHelper {
+        private final CacheHelper delegate;
+        private final CacheKey cacheKey = new CacheKey();
+
+        DelegatingCacheHelper(CacheHelper delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public CacheKey getKey() {
+            return cacheKey;
+        }
+
+        @Override
+        public void addClosedListener(ClosedListener listener) {
+            // here we wrap the listener and call it with our cache key
+            // this is important since this key will be used to cache the reader and otherwise we won't free caches etc.
+            delegate.addClosedListener(unused -> listener.onClose(cacheKey));
+        }
+    }
+
+    public static class LazyBits implements Bits {
+
+        private final int maxDoc;
+        private final String field;
+        private final LeafReader reader;
+        private final int numSoftDeletes;
+        private final int numDocs;
+        volatile Bits materializedBits;
+
+        public LazyBits(int maxDoc, String field, LeafReader reader, int numSoftDeletes, int numDocs) {
+            this.maxDoc = maxDoc;
+            this.field = field;
+            this.reader = reader;
+            this.numSoftDeletes = numSoftDeletes;
+            this.numDocs = numDocs;
+            materializedBits = null;
+            assert numSoftDeletes > 0;
+        }
+
+        @Override
+        public boolean get(int index) {
+            if (materializedBits == null) {
+                synchronized (this) {
+                    try {
+                        if (materializedBits == null) {
+                            materializedBits = init();
+                        }
+                    } catch (IOException e) {
+                        throw new UncheckedIOException(e);
+                    }
+                }
+            }
+            return materializedBits.get(index);
+        }
+
+        @Override
+        public int length() {
+            return maxDoc;
+        }
+
+        private Bits init() throws IOException {
+            assert Thread.holdsLock(this);
+
+            DocIdSetIterator iterator = DocValuesFieldExistsQuery.getDocValuesDocIdSetIterator(field, reader);
+            assert iterator != null;
+            Bits liveDocs = reader.getLiveDocs();
+            final FixedBitSet bits;
+            if (liveDocs != null) {
+                bits = FixedBitSet.copyOf(liveDocs);
+            } else {
+                bits = new FixedBitSet(maxDoc);
+                bits.set(0, maxDoc);
+            }
+            int numComputedSoftDeletes = PendingSoftDeletes.applySoftDeletes(iterator, bits);
+            assert numComputedSoftDeletes == numSoftDeletes :
+                "numComputedSoftDeletes: " + numComputedSoftDeletes + " expected: " + numSoftDeletes;
+
+            int numDeletes = reader.numDeletedDocs() + numComputedSoftDeletes;
+            int computedNumDocs = reader.maxDoc() - numDeletes;
+            assert computedNumDocs == numDocs : "computedNumDocs: " + computedNumDocs + " expected: " + numDocs;
+            return bits;
+        }
+
+        public boolean initialized() {
+            return materializedBits != null;
+        }
+    }
+}

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

@@ -43,7 +43,7 @@ public final class NoOpEngine extends ReadOnlyEngine {
     private final DocsStats docsStats;
 
     public NoOpEngine(EngineConfig config) {
-        super(config, null, null, true, Function.identity(), true);
+        super(config, null, null, true, Function.identity(), true, true);
         this.segmentsStats = new SegmentsStats();
         Directory directory = store.directory();
         try (DirectoryReader reader = openDirectory(directory)) {

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

@@ -11,6 +11,7 @@ import org.apache.lucene.document.LongPoint;
 import org.apache.lucene.index.DirectoryReader;
 import org.apache.lucene.index.IndexCommit;
 import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.LazySoftDeletesDirectoryReaderWrapper;
 import org.apache.lucene.index.PointValues;
 import org.apache.lucene.index.SegmentCommitInfo;
 import org.apache.lucene.index.SegmentInfos;
@@ -52,7 +53,7 @@ import java.util.stream.Stream;
  * Note: this engine can be opened side-by-side with a read-write engine but will not reflect any changes made to the read-write
  * engine.
  *
- * @see #ReadOnlyEngine(EngineConfig, SeqNoStats, TranslogStats, boolean, Function, boolean)
+ * @see #ReadOnlyEngine(EngineConfig, SeqNoStats, TranslogStats, boolean, Function, boolean, boolean)
  */
 public class ReadOnlyEngine extends Engine {
 
@@ -69,6 +70,7 @@ public class ReadOnlyEngine extends Engine {
     private final SafeCommitInfo safeCommitInfo;
     private final CompletionStatsCache completionStatsCache;
     private final boolean requireCompleteHistory;
+    final boolean lazilyLoadSoftDeletes;
 
     protected volatile TranslogStats translogStats;
     protected final String commitId;
@@ -85,9 +87,11 @@ public class ReadOnlyEngine extends Engine {
      *                   the lock won't be obtained
      * @param readerWrapperFunction allows to wrap the index-reader for this engine.
      * @param requireCompleteHistory indicates whether this engine permits an incomplete history (i.e. LCP < MSN)
+     * @param lazilyLoadSoftDeletes indicates whether this engine should load the soft-delete based liveDocs eagerly, or on first access
      */
     public ReadOnlyEngine(EngineConfig config, SeqNoStats seqNoStats, TranslogStats translogStats, boolean obtainLock,
-                          Function<DirectoryReader, DirectoryReader> readerWrapperFunction, boolean requireCompleteHistory) {
+                          Function<DirectoryReader, DirectoryReader> readerWrapperFunction, boolean requireCompleteHistory,
+                          boolean lazilyLoadSoftDeletes) {
         super(config);
         this.refreshListener = new RamAccountingRefreshListener(engineConfig.getCircuitBreakerService());
         this.requireCompleteHistory = requireCompleteHistory;
@@ -110,6 +114,7 @@ public class ReadOnlyEngine extends Engine {
                 }
                 this.seqNoStats = seqNoStats;
                 this.indexCommit = Lucene.getIndexCommit(lastCommittedSegmentInfos, directory);
+                this.lazilyLoadSoftDeletes = lazilyLoadSoftDeletes;
                 reader = wrapReader(open(indexCommit), readerWrapperFunction);
                 readerManager = new ElasticsearchReaderManager(reader, refreshListener);
                 assert translogStats != null || obtainLock : "mutiple translogs instances should not be opened at the same time";
@@ -194,7 +199,11 @@ public class ReadOnlyEngine extends Engine {
 
     protected DirectoryReader open(IndexCommit commit) throws IOException {
         assert Transports.assertNotTransportThread("opening index commit of a read-only engine");
-        return new SoftDeletesDirectoryReaderWrapper(DirectoryReader.open(commit), Lucene.SOFT_DELETES_FIELD);
+        if (lazilyLoadSoftDeletes) {
+            return new LazySoftDeletesDirectoryReaderWrapper(DirectoryReader.open(commit), Lucene.SOFT_DELETES_FIELD);
+        } else {
+            return new SoftDeletesDirectoryReaderWrapper(DirectoryReader.open(commit), Lucene.SOFT_DELETES_FIELD);
+        }
     }
 
     @Override
@@ -512,10 +521,14 @@ public class ReadOnlyEngine extends Engine {
             maxSeqNoOfUpdatesOnPrimary + ">" + getMaxSeqNoOfUpdatesOrDeletes();
     }
 
-    protected static DirectoryReader openDirectory(Directory directory) throws IOException {
+    protected DirectoryReader openDirectory(Directory directory) throws IOException {
         assert Transports.assertNotTransportThread("opening directory reader of a read-only engine");
         final DirectoryReader reader = DirectoryReader.open(directory);
-        return new SoftDeletesDirectoryReaderWrapper(reader, Lucene.SOFT_DELETES_FIELD);
+        if (lazilyLoadSoftDeletes) {
+            return new LazySoftDeletesDirectoryReaderWrapper(reader, Lucene.SOFT_DELETES_FIELD);
+        } else {
+            return new SoftDeletesDirectoryReaderWrapper(reader, Lucene.SOFT_DELETES_FIELD);
+        }
     }
 
     @Override

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -3541,7 +3541,8 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
             // we must create both new read-only engine and new read-write engine under engineMutex to ensure snapshotStoreMetadata,
             // acquireXXXCommit and close works.
             final Engine readOnlyEngine =
-                new ReadOnlyEngine(newEngineConfig(replicationTracker), seqNoStats, translogStats, false, Function.identity(), true) {
+                new ReadOnlyEngine(newEngineConfig(replicationTracker), seqNoStats, translogStats, false, Function.identity(), true,
+                    false) {
                     @Override
                     public IndexCommitRef acquireLastIndexCommit(boolean flushFirst) {
                         synchronized (engineMutex) {

server/src/test/java/org/apache/lucene/index/LazySoftDeletesDirectoryReaderWrapperTests.java

@@ -0,0 +1,175 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.apache.lucene.index;
+
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.NumericDocValuesField;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.search.IndexSearcher;
+import org.apache.lucene.search.MatchAllDocsQuery;
+import org.apache.lucene.search.TermQuery;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.util.LuceneTestCase;
+import org.elasticsearch.core.internal.io.IOUtils;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class LazySoftDeletesDirectoryReaderWrapperTests extends LuceneTestCase {
+
+    public void testDropFullyDeletedSegments() throws IOException {
+        IndexWriterConfig indexWriterConfig = newIndexWriterConfig();
+        String softDeletesField = "soft_delete";
+        indexWriterConfig.setSoftDeletesField(softDeletesField);
+        indexWriterConfig.setMergePolicy(
+            new SoftDeletesRetentionMergePolicy(
+                softDeletesField, MatchAllDocsQuery::new, NoMergePolicy.INSTANCE));
+        try (Directory dir = newDirectory();
+             IndexWriter writer = new IndexWriter(dir, indexWriterConfig)) {
+
+            Document doc = new Document();
+            doc.add(new StringField("id", "1", Field.Store.YES));
+            doc.add(new StringField("version", "1", Field.Store.YES));
+            writer.addDocument(doc);
+            writer.commit();
+            doc = new Document();
+            doc.add(new StringField("id", "2", Field.Store.YES));
+            doc.add(new StringField("version", "1", Field.Store.YES));
+            writer.addDocument(doc);
+            writer.commit();
+
+            try (DirectoryReader reader =
+                     new LazySoftDeletesDirectoryReaderWrapper(DirectoryReader.open(dir), softDeletesField)) {
+                assertEquals(2, reader.leaves().size());
+                assertEquals(2, reader.numDocs());
+                assertEquals(2, reader.maxDoc());
+                assertEquals(0, reader.numDeletedDocs());
+            }
+            writer.updateDocValues(new Term("id", "1"), new NumericDocValuesField(softDeletesField, 1));
+            writer.commit();
+            try (DirectoryReader reader =
+                     new LazySoftDeletesDirectoryReaderWrapper(DirectoryReader.open(dir), softDeletesField)) {
+                assertEquals(1, reader.numDocs());
+                assertEquals(1, reader.maxDoc());
+                assertEquals(0, reader.numDeletedDocs());
+                assertEquals(1, reader.leaves().size());
+            }
+
+            try (DirectoryReader reader = DirectoryReader.open(dir)) {
+                assertEquals(2, reader.numDocs());
+                assertEquals(2, reader.maxDoc());
+                assertEquals(0, reader.numDeletedDocs());
+                assertEquals(2, reader.leaves().size());
+            }
+        }
+    }
+
+    public void testMixSoftAndHardDeletes() throws IOException {
+        Directory dir = newDirectory();
+        IndexWriterConfig indexWriterConfig = newIndexWriterConfig();
+        String softDeletesField = "soft_delete";
+        indexWriterConfig.setSoftDeletesField(softDeletesField);
+        IndexWriter writer = new IndexWriter(dir, indexWriterConfig);
+        Set<Integer> uniqueDocs = new HashSet<>();
+        for (int i = 0; i < 100; i++) {
+            int docId = random().nextInt(5);
+            uniqueDocs.add(docId);
+            Document doc = new Document();
+            doc.add(new StringField("id", String.valueOf(docId), Field.Store.YES));
+            if (docId % 2 == 0) {
+                writer.updateDocument(new Term("id", String.valueOf(docId)), doc);
+            } else {
+                writer.softUpdateDocument(
+                    new Term("id", String.valueOf(docId)),
+                    doc,
+                    new NumericDocValuesField(softDeletesField, 0));
+            }
+        }
+
+        writer.commit();
+        writer.close();
+        DirectoryReader reader =
+            new LazySoftDeletesDirectoryReaderWrapper(DirectoryReader.open(dir), softDeletesField);
+        assertEquals(uniqueDocs.size(), reader.numDocs());
+        IndexSearcher searcher = new IndexSearcher(reader);
+        for (Integer docId : uniqueDocs) {
+            assertEquals(1, searcher.count(new TermQuery(new Term("id", docId.toString()))));
+        }
+
+        IOUtils.close(reader, dir);
+    }
+
+    public void testReaderCacheKey() throws IOException {
+        Directory dir = newDirectory();
+        IndexWriterConfig indexWriterConfig = newIndexWriterConfig();
+        String softDeletesField = "soft_delete";
+        indexWriterConfig.setSoftDeletesField(softDeletesField);
+        indexWriterConfig.setMergePolicy(NoMergePolicy.INSTANCE);
+        IndexWriter writer = new IndexWriter(dir, indexWriterConfig);
+
+        Document doc = new Document();
+        doc.add(new StringField("id", "1", Field.Store.YES));
+        doc.add(new StringField("version", "1", Field.Store.YES));
+        writer.addDocument(doc);
+        doc = new Document();
+        doc.add(new StringField("id", "2", Field.Store.YES));
+        doc.add(new StringField("version", "1", Field.Store.YES));
+        writer.addDocument(doc);
+        writer.commit();
+        DirectoryReader reader =
+            new LazySoftDeletesDirectoryReaderWrapper(DirectoryReader.open(dir), softDeletesField);
+        IndexReader.CacheHelper readerCacheHelper =
+            reader.leaves().get(0).reader().getReaderCacheHelper();
+        AtomicInteger leafCalled = new AtomicInteger(0);
+        AtomicInteger dirCalled = new AtomicInteger(0);
+        readerCacheHelper.addClosedListener(
+            key -> {
+                leafCalled.incrementAndGet();
+                assertSame(key, readerCacheHelper.getKey());
+            });
+        IndexReader.CacheHelper dirReaderCacheHelper = reader.getReaderCacheHelper();
+        dirReaderCacheHelper.addClosedListener(
+            key -> {
+                dirCalled.incrementAndGet();
+                assertSame(key, dirReaderCacheHelper.getKey());
+            });
+        assertEquals(2, reader.numDocs());
+        assertEquals(2, reader.maxDoc());
+        assertEquals(0, reader.numDeletedDocs());
+
+        doc = new Document();
+        doc.add(new StringField("id", "1", Field.Store.YES));
+        doc.add(new StringField("version", "2", Field.Store.YES));
+        writer.softUpdateDocument(
+            new Term("id", "1"), doc, new NumericDocValuesField("soft_delete", 1));
+
+        doc = new Document();
+        doc.add(new StringField("id", "3", Field.Store.YES));
+        doc.add(new StringField("version", "1", Field.Store.YES));
+        writer.addDocument(doc);
+        writer.commit();
+        assertEquals(0, leafCalled.get());
+        assertEquals(0, dirCalled.get());
+        DirectoryReader newReader =
+            new LazySoftDeletesDirectoryReaderWrapper(DirectoryReader.open(dir), softDeletesField);
+        assertEquals(0, leafCalled.get());
+        assertEquals(0, dirCalled.get());
+        assertNotSame(
+            newReader.getReaderCacheHelper().getKey(), reader.getReaderCacheHelper().getKey());
+        assertNotSame(newReader, reader);
+        reader.close();
+        reader = newReader;
+        assertEquals(1, dirCalled.get());
+        assertEquals(1, leafCalled.get());
+        IOUtils.close(reader, writer, dir);
+    }
+}

server/src/test/java/org/elasticsearch/index/engine/ReadOnlyEngineTests.java

@@ -62,7 +62,7 @@ public class ReadOnlyEngineTests extends EngineTestCase {
                 globalCheckpoint.set(randomLongBetween(globalCheckpoint.get(), engine.getPersistedLocalCheckpoint()));
                 engine.flush();
                 readOnlyEngine = new ReadOnlyEngine(engine.engineConfig, engine.getSeqNoStats(globalCheckpoint.get()),
-                    engine.getTranslogStats(), false, Function.identity(), true);
+                    engine.getTranslogStats(), false, Function.identity(), true, randomBoolean());
                 lastSeqNoStats = engine.getSeqNoStats(globalCheckpoint.get());
                 lastDocIds = getDocIds(engine, true);
                 assertThat(readOnlyEngine.getPersistedLocalCheckpoint(), equalTo(lastSeqNoStats.getLocalCheckpoint()));
@@ -77,6 +77,11 @@ public class ReadOnlyEngineTests extends EngineTestCase {
                         engine.flush();
                     }
                 }
+                try (ReadOnlyEngine readOnlyEngineWithLazySoftDeletes = new ReadOnlyEngine(engine.engineConfig,
+                    engine.getSeqNoStats(globalCheckpoint.get()),
+                    engine.getTranslogStats(), false, Function.identity(), true, true)) {
+                    EngineTestCase.checkNoSoftDeletesLoaded(readOnlyEngineWithLazySoftDeletes);
+                }
                 Engine.Searcher external = readOnlyEngine.acquireSearcher("test", Engine.SearcherScope.EXTERNAL);
                 Engine.Searcher internal = readOnlyEngine.acquireSearcher("test", Engine.SearcherScope.INTERNAL);
                 assertSame(external.getIndexReader(), internal.getIndexReader());
@@ -128,7 +133,7 @@ public class ReadOnlyEngineTests extends EngineTestCase {
                 engine.flushAndClose();
 
                 IllegalStateException exception = expectThrows(IllegalStateException.class,
-                    () -> new ReadOnlyEngine(config, null, null, true, Function.identity(), true) {
+                    () -> new ReadOnlyEngine(config, null, null, true, Function.identity(), true, randomBoolean()) {
                         @Override
                         protected boolean assertMaxSeqNoEqualsToGlobalCheckpoint(final long maxSeqNo, final long globalCheckpoint) {
                             // we don't want the assertion to trip in this test
@@ -147,7 +152,8 @@ public class ReadOnlyEngineTests extends EngineTestCase {
         try (Store store = createStore()) {
             EngineConfig config = config(defaultSettings, store, createTempDir(), newMergePolicy(), null, null, globalCheckpoint::get);
             store.createEmpty();
-            try (ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(config, null , new TranslogStats(), true, Function.identity(), true)) {
+            try (ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(config, null , new TranslogStats(), true, Function.identity(), true,
+                randomBoolean())) {
                 Class<? extends Throwable> expectedException = LuceneTestCase.TEST_ASSERTS_ENABLED ? AssertionError.class :
                     UnsupportedOperationException.class;
                 expectThrows(expectedException, () -> readOnlyEngine.index(null));
@@ -167,7 +173,8 @@ public class ReadOnlyEngineTests extends EngineTestCase {
         try (Store store = createStore()) {
             EngineConfig config = config(defaultSettings, store, createTempDir(), newMergePolicy(), null, null, globalCheckpoint::get);
             store.createEmpty();
-            try (ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(config, null , new TranslogStats(), true, Function.identity(), true)) {
+            try (ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(config, null , new TranslogStats(), true, Function.identity(), true,
+                randomBoolean())) {
                 globalCheckpoint.set(randomNonNegativeLong());
                 try {
                     readOnlyEngine.verifyEngineBeforeIndexClosing();
@@ -198,7 +205,8 @@ public class ReadOnlyEngineTests extends EngineTestCase {
                 numSegments = engine.getLastCommittedSegmentInfos().size();
             }
 
-            try (ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(config, null , null, true, Function.identity(), true)) {
+            try (ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(config, null , null, true, Function.identity(), true,
+                randomBoolean())) {
                 if (numSegments > 1) {
                     final int target = between(1, numSegments - 1);
                     UnsupportedOperationException exception = expectThrows(UnsupportedOperationException.class,
@@ -234,7 +242,8 @@ public class ReadOnlyEngineTests extends EngineTestCase {
                 engine.syncTranslog();
                 engine.flushAndClose();
             }
-            try (ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(config, null , null, true, Function.identity(), true)) {
+            try (ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(config, null , null, true, Function.identity(), true,
+                randomBoolean())) {
                 final TranslogHandler translogHandler = new TranslogHandler(xContentRegistry(), config.getIndexSettings());
                 readOnlyEngine.recoverFromTranslog(translogHandler, randomNonNegativeLong());
 
@@ -276,7 +285,8 @@ public class ReadOnlyEngineTests extends EngineTestCase {
                 engine.flush(true, true);
             }
 
-            try (ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(config, null, null, true, Function.identity(), true)) {
+            try (ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(config, null, null, true, Function.identity(), true,
+                randomBoolean())) {
                 assertThat(readOnlyEngine.getTranslogStats().estimatedNumberOfOperations(), equalTo(softDeletesEnabled ? 0 : numDocs));
                 assertThat(readOnlyEngine.getTranslogStats().getUncommittedOperations(), equalTo(0));
                 assertThat(readOnlyEngine.getTranslogStats().getTranslogSizeInBytes(), greaterThan(0L));
@@ -313,7 +323,8 @@ public class ReadOnlyEngineTests extends EngineTestCase {
                 }
                 globalCheckpoint.set(engine.getProcessedLocalCheckpoint());
             }
-            try (ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(config, null, null, true, Function.identity(), true)) {
+            try (ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(config, null, null, true, Function.identity(), true,
+                randomBoolean())) {
                 try (Engine.SearcherSupplier searcher =
                          readOnlyEngine.acquireSearcherSupplier(Function.identity(), randomFrom(Engine.SearcherScope.values()))) {
                     assertThat(searcher.getSearcherId(), equalTo(lastSearcherId));

server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java

@@ -4028,7 +4028,7 @@ public class IndexShardTests extends IndexShardTestCase {
         ShardRouting readonlyShardRouting = newShardRouting(replicaRouting.shardId(), replicaRouting.currentNodeId(), true,
             ShardRoutingState.INITIALIZING, RecoverySource.ExistingStoreRecoverySource.INSTANCE);
         final IndexShard readonlyShard = reinitShard(shard, readonlyShardRouting, shard.indexSettings.getIndexMetadata(),
-            engineConfig -> new ReadOnlyEngine(engineConfig, null, null, true, Function.identity(), true) {
+            engineConfig -> new ReadOnlyEngine(engineConfig, null, null, true, Function.identity(), true, randomBoolean()) {
                 @Override
                 protected void ensureMaxSeqNoEqualsToGlobalCheckpoint(SeqNoStats seqNoStats) {
                     // just like a following shard, we need to skip this check for now.

test/framework/src/main/java/org/elasticsearch/index/engine/EngineTestCase.java

@@ -17,16 +17,19 @@ import org.apache.lucene.document.NumericDocValuesField;
 import org.apache.lucene.document.StoredField;
 import org.apache.lucene.document.TextField;
 import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.FilterCodecReader;
 import org.apache.lucene.index.FilterDirectoryReader;
 import org.apache.lucene.index.FilterLeafReader;
 import org.apache.lucene.index.IndexCommit;
 import org.apache.lucene.index.IndexWriter;
 import org.apache.lucene.index.IndexWriterConfig;
+import org.apache.lucene.index.LazySoftDeletesDirectoryReaderWrapper;
 import org.apache.lucene.index.LeafReader;
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.index.LiveIndexWriterConfig;
 import org.apache.lucene.index.MergePolicy;
 import org.apache.lucene.index.NumericDocValues;
+import org.apache.lucene.index.SegmentReader;
 import org.apache.lucene.index.Term;
 import org.apache.lucene.search.DocIdSetIterator;
 import org.apache.lucene.search.IndexSearcher;
@@ -1343,4 +1346,37 @@ public abstract class EngineTestCase extends ESTestCase {
             return searcher -> SearcherHelper.wrapSearcher(searcher, readerWrapper);
         }
     }
+
+    public static void checkNoSoftDeletesLoaded(ReadOnlyEngine readOnlyEngine) {
+        if (readOnlyEngine.lazilyLoadSoftDeletes == false) {
+            throw new IllegalStateException("method should only be called when lazily loading soft-deletes is enabled");
+        }
+        try (Engine.Searcher searcher = readOnlyEngine.acquireSearcher("soft-deletes-check", Engine.SearcherScope.INTERNAL)) {
+            for (LeafReaderContext ctx : searcher.getIndexReader().getContext().leaves()) {
+                LazySoftDeletesDirectoryReaderWrapper.LazyBits lazyBits = lazyBits(ctx.reader());
+                if (lazyBits != null && lazyBits.initialized()) {
+                    throw new IllegalStateException("soft-deletes loaded");
+                }
+            }
+        }
+    }
+
+    @Nullable
+    private static LazySoftDeletesDirectoryReaderWrapper.LazyBits lazyBits(LeafReader reader) {
+        if (reader instanceof LazySoftDeletesDirectoryReaderWrapper.LazySoftDeletesFilterLeafReader) {
+            return ((LazySoftDeletesDirectoryReaderWrapper.LazySoftDeletesFilterLeafReader) reader).getLiveDocs();
+        } else if (reader instanceof LazySoftDeletesDirectoryReaderWrapper.LazySoftDeletesFilterCodecReader) {
+            return ((LazySoftDeletesDirectoryReaderWrapper.LazySoftDeletesFilterCodecReader) reader).getLiveDocs();
+        } else if (reader instanceof FilterLeafReader) {
+            final FilterLeafReader fReader = (FilterLeafReader) reader;
+            return lazyBits(FilterLeafReader.unwrap(fReader));
+        } else if (reader instanceof FilterCodecReader) {
+            final FilterCodecReader fReader = (FilterCodecReader) reader;
+            return lazyBits(FilterCodecReader.unwrap(fReader));
+        } else if (reader instanceof SegmentReader) {
+            return null;
+        }
+        // hard fail - we can't get the lazybits
+        throw new IllegalStateException("Can not extract lazy bits from given index reader [" + reader + "]");
+    }
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/index/engine/FrozenEngine.java

@@ -49,14 +49,14 @@ public final class FrozenEngine extends ReadOnlyEngine {
     private volatile ElasticsearchDirectoryReader lastOpenedReader;
     private final ElasticsearchDirectoryReader canMatchReader;
 
-    public FrozenEngine(EngineConfig config, boolean requireCompleteHistory) {
-        this(config, null, null, true, Function.identity(), requireCompleteHistory);
+    public FrozenEngine(EngineConfig config, boolean requireCompleteHistory, boolean lazilyLoadSoftDeletes) {
+        this(config, null, null, true, Function.identity(), requireCompleteHistory, lazilyLoadSoftDeletes);
     }
 
     public FrozenEngine(EngineConfig config, SeqNoStats seqNoStats, TranslogStats translogStats, boolean obtainLock,
-                        Function<DirectoryReader, DirectoryReader> readerWrapperFunction, boolean requireCompleteHistory) {
-        super(config, seqNoStats, translogStats, obtainLock, readerWrapperFunction, requireCompleteHistory);
-
+                        Function<DirectoryReader, DirectoryReader> readerWrapperFunction, boolean requireCompleteHistory,
+                        boolean lazilyLoadSoftDeletes) {
+        super(config, seqNoStats, translogStats, obtainLock, readerWrapperFunction, requireCompleteHistory, lazilyLoadSoftDeletes);
         boolean success = false;
         Directory directory = store.directory();
         try (DirectoryReader reader = openDirectory(directory)) {

x-pack/plugin/core/src/main/java/org/elasticsearch/snapshots/SourceOnlySnapshotRepository.java

@@ -174,7 +174,7 @@ public final class SourceOnlySnapshotRepository extends FilterRepository {
      */
     public static EngineFactory getEngineFactory() {
         return config -> new ReadOnlyEngine(config, null, new TranslogStats(0, 0, 0, 0, 0), true,
-            readerWrapper(config), true);
+            readerWrapper(config), true, false);
     }
 
     public static Function<DirectoryReader, DirectoryReader> readerWrapper(EngineConfig engineConfig) {

x-pack/plugin/frozen-indices/src/main/java/org/elasticsearch/xpack/frozen/FrozenIndices.java

@@ -43,7 +43,7 @@ public class FrozenIndices extends Plugin implements ActionPlugin, EnginePlugin
     @Override
     public Optional<EngineFactory> getEngineFactory(IndexSettings indexSettings) {
         if (indexSettings.getValue(FrozenEngine.INDEX_FROZEN) && isSearchableSnapshotStore(indexSettings.getSettings()) == false) {
-            return Optional.of(config -> new FrozenEngine(config, true));
+            return Optional.of(config -> new FrozenEngine(config, true, false));
         } else {
             return Optional.empty();
         }

x-pack/plugin/frozen-indices/src/test/java/org/elasticsearch/index/engine/FrozenEngineTests.java

@@ -50,7 +50,7 @@ public class FrozenEngineTests extends EngineTestCase {
                 int numDocs = Math.min(10, addDocuments(globalCheckpoint, engine));
                 engine.flushAndClose();
                 listener.reset();
-                try (FrozenEngine frozenEngine = new FrozenEngine(engine.engineConfig, true)) {
+                try (FrozenEngine frozenEngine = new FrozenEngine(engine.engineConfig, true, randomBoolean())) {
                     assertFalse(frozenEngine.isReaderOpen());
                     try (Engine.SearcherSupplier reader = frozenEngine.acquireSearcherSupplier(Function.identity())) {
                         assertFalse(frozenEngine.isReaderOpen());
@@ -87,7 +87,7 @@ public class FrozenEngineTests extends EngineTestCase {
                 int numDocs = Math.min(10, addDocuments(globalCheckpoint, engine));
                 engine.flushAndClose();
                 listener.reset();
-                try (FrozenEngine frozenEngine = new FrozenEngine(engine.engineConfig, true)) {
+                try (FrozenEngine frozenEngine = new FrozenEngine(engine.engineConfig, true, randomBoolean())) {
                     assertFalse(frozenEngine.isReaderOpen());
                     Engine.SearcherSupplier reader1 = frozenEngine.acquireSearcherSupplier(Function.identity());
                     try (Engine.Searcher searcher1 = reader1.acquireSearcher("test")) {
@@ -130,7 +130,7 @@ public class FrozenEngineTests extends EngineTestCase {
                 addDocuments(globalCheckpoint, engine);
                 engine.flushAndClose();
                 listener.reset();
-                try (FrozenEngine frozenEngine = new FrozenEngine(engine.engineConfig, true)) {
+                try (FrozenEngine frozenEngine = new FrozenEngine(engine.engineConfig, true, randomBoolean())) {
                     try (Engine.SearcherSupplier reader = frozenEngine.acquireSearcherSupplier(Function.identity())) {
                         SegmentsStats segmentsStats = frozenEngine.segmentsStats(randomBoolean(), false);
                         try (Engine.Searcher searcher = reader.acquireSearcher("test")) {
@@ -172,7 +172,7 @@ public class FrozenEngineTests extends EngineTestCase {
                 engine.refresh("test"); // pull the reader to account for RAM in the breaker.
             }
             final long expectedUse;
-            try (ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(config, null, null, true, i -> i, true)) {
+            try (ReadOnlyEngine readOnlyEngine = new ReadOnlyEngine(config, null, null, true, i -> i, true, randomBoolean())) {
                 expectedUse = breaker.getUsed();
                 DocsStats docsStats = readOnlyEngine.docStats();
                 assertEquals(docs, docsStats.getCount());
@@ -180,7 +180,7 @@ public class FrozenEngineTests extends EngineTestCase {
             assertTrue(expectedUse > 0);
             assertEquals(0, breaker.getUsed());
             listener.reset();
-            try (FrozenEngine frozenEngine = new FrozenEngine(config, true)) {
+            try (FrozenEngine frozenEngine = new FrozenEngine(config, true, randomBoolean())) {
                 try (Engine.SearcherSupplier reader = frozenEngine.acquireSearcherSupplier(Function.identity())) {
                     try (Engine.Searcher searcher = reader.acquireSearcher("test")) {
                         assertEquals(expectedUse, breaker.getUsed());
@@ -228,7 +228,7 @@ public class FrozenEngineTests extends EngineTestCase {
                 int numDocsAdded = addDocuments(globalCheckpoint, engine);
                 engine.flushAndClose();
                 int numIters = randomIntBetween(100, 1000);
-                try (FrozenEngine frozenEngine = new FrozenEngine(engine.engineConfig, true)) {
+                try (FrozenEngine frozenEngine = new FrozenEngine(engine.engineConfig, true, randomBoolean())) {
                     int numThreads = randomIntBetween(2, 4);
                     Thread[] threads = new Thread[numThreads];
                     CyclicBarrier barrier = new CyclicBarrier(numThreads);
@@ -317,7 +317,7 @@ public class FrozenEngineTests extends EngineTestCase {
                 addDocuments(globalCheckpoint, engine);
                 engine.flushAndClose();
                 listener.reset();
-                try (FrozenEngine frozenEngine = new FrozenEngine(engine.engineConfig, true)) {
+                try (FrozenEngine frozenEngine = new FrozenEngine(engine.engineConfig, true, randomBoolean())) {
                     DirectoryReader dirReader;
                     try (Engine.SearcherSupplier reader = frozenEngine.acquireSearcherSupplier(Function.identity())) {
                         try (Engine.Searcher searcher = reader.acquireSearcher(Engine.CAN_MATCH_SEARCH_SOURCE)) {
@@ -366,7 +366,7 @@ public class FrozenEngineTests extends EngineTestCase {
                     }
                 }
             }
-            try (FrozenEngine frozenEngine = new FrozenEngine(config, true)) {
+            try (FrozenEngine frozenEngine = new FrozenEngine(config, true, randomBoolean())) {
                 try (Engine.SearcherSupplier reader = frozenEngine.acquireSearcherSupplier(Function.identity())) {
                     try (Engine.Searcher searcher = reader.acquireSearcher("test")) {
                         TopDocs topDocs = searcher.search(new MatchAllDocsQuery(), Integer.MAX_VALUE);

x-pack/plugin/frozen-indices/src/test/java/org/elasticsearch/index/engine/FrozenIndexShardTests.java

@@ -31,12 +31,12 @@ public class FrozenIndexShardTests extends IndexShardTestCase {
         final ShardRouting shardRouting = indexShard.routingEntry();
         IndexShard frozenShard = reinitShard(indexShard, ShardRoutingHelper.initWithSameId(shardRouting,
             shardRouting.primary() ? RecoverySource.ExistingStoreRecoverySource.INSTANCE : RecoverySource.PeerRecoverySource.INSTANCE
-        ), indexShard.indexSettings().getIndexMetadata(), config -> new FrozenEngine(config, true));
+        ), indexShard.indexSettings().getIndexMetadata(), config -> new FrozenEngine(config, true, randomBoolean()));
         recoverShardFromStore(frozenShard);
         assertThat(frozenShard.getMaxSeqNoOfUpdatesOrDeletes(), equalTo(frozenShard.seqNoStats().getMaxSeqNo()));
         assertDocCount(frozenShard, 3);
 
-        IndexShard replica = newShard(false, Settings.EMPTY, config -> new FrozenEngine(config, true));
+        IndexShard replica = newShard(false, Settings.EMPTY, config -> new FrozenEngine(config, true, randomBoolean()));
         recoverReplica(replica, frozenShard, true);
         assertDocCount(replica, 3);
         closeShards(frozenShard, replica);

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/BaseSearchableSnapshotsIntegTestCase.java

@@ -151,9 +151,11 @@ public abstract class BaseSearchableSnapshotsIntegTestCase extends AbstractSnaps
         }
         indexRandom(true, true, indexRequestBuilders);
         refresh(indexName);
-        assertThat(
-            client().admin().indices().prepareForceMerge(indexName).setOnlyExpungeDeletes(true).setFlush(true).get().getFailedShards(),
-            equalTo(0)
-        );
+        if (randomBoolean()) {
+            assertThat(
+                client().admin().indices().prepareForceMerge(indexName).setOnlyExpungeDeletes(true).setFlush(true).get().getFailedShards(),
+                equalTo(0)
+            );
+        }
     }
 }

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshotsIntegTests.java

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.searchablesnapshots;
 
 import com.carrotsearch.hppc.cursors.ObjectCursor;
 import org.apache.lucene.search.TotalHits;
+import org.apache.lucene.store.AlreadyClosedException;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
@@ -41,9 +42,15 @@ import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexModule;
 import org.elasticsearch.index.IndexNotFoundException;
+import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.engine.Engine;
+import org.elasticsearch.index.engine.EngineTestCase;
+import org.elasticsearch.index.engine.ReadOnlyEngine;
 import org.elasticsearch.index.mapper.DateFieldMapper;
 import org.elasticsearch.index.shard.IndexLongFieldRange;
+import org.elasticsearch.index.shard.IndexShard;
+import org.elasticsearch.index.shard.IndexShardTestCase;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.shard.ShardPath;
 import org.elasticsearch.indices.IndexClosedException;
@@ -103,6 +110,7 @@ import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.oneOf;
@@ -259,9 +267,11 @@ public class SearchableSnapshotsIntegTests extends BaseSearchableSnapshotsIntegT
         assertThat(IndexMetadata.INDEX_NUMBER_OF_REPLICAS_SETTING.get(settings), equalTo(expectedReplicas));
         assertThat(DataTierAllocationDecider.INDEX_ROUTING_PREFER_SETTING.get(settings), equalTo(expectedDataTiersPreference));
 
+        checkSoftDeletesNotEagerlyLoaded(restoredIndexName);
         assertTotalHits(restoredIndexName, originalAllHits, originalBarHits);
         assertRecoveryStats(restoredIndexName, preWarmEnabled);
         assertSearchableSnapshotStats(restoredIndexName, cacheEnabled, nonCachedExtensions);
+
         ensureGreen(restoredIndexName);
         assertShardFolders(restoredIndexName, true);
 
@@ -546,6 +556,7 @@ public class SearchableSnapshotsIntegTests extends BaseSearchableSnapshotsIntegT
         assertTrue(SearchableSnapshots.SNAPSHOT_PARTIAL_SETTING.get(settings));
         assertTrue(DiskThresholdDecider.SETTING_IGNORE_DISK_WATERMARKS.get(settings));
 
+        checkSoftDeletesNotEagerlyLoaded(restoredIndexName);
         assertTotalHits(restoredIndexName, originalAllHits, originalBarHits);
         assertRecoveryStats(restoredIndexName, false);
         // TODO: fix
@@ -700,6 +711,24 @@ public class SearchableSnapshotsIntegTests extends BaseSearchableSnapshotsIntegT
         assertTotalHits(aliasName, originalAllHits, originalBarHits);
     }
 
+    private void checkSoftDeletesNotEagerlyLoaded(String restoredIndexName) {
+        for (IndicesService indicesService : internalCluster().getDataNodeInstances(IndicesService.class)) {
+            for (IndexService indexService : indicesService) {
+                if (indexService.index().getName().equals(restoredIndexName)) {
+                    for (IndexShard indexShard : indexService) {
+                        try {
+                            Engine engine = IndexShardTestCase.getEngine(indexShard);
+                            assertThat(engine, instanceOf(ReadOnlyEngine.class));
+                            EngineTestCase.checkNoSoftDeletesLoaded((ReadOnlyEngine) engine);
+                        } catch (AlreadyClosedException ace) {
+                            // ok to ignore these
+                        }
+                    }
+                }
+            }
+        }
+    }
+
     private void assertShardFolders(String indexName, boolean snapshotDirectory) throws IOException {
         final Index restoredIndex = resolveIndex(indexName);
         final String customDataPath = resolveCustomDataPath(indexName);

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshots.java

@@ -404,7 +404,8 @@ public class SearchableSnapshots extends Plugin implements IndexStorePlugin, Eng
                         indexSettings.getValue(SourceOnlySnapshotRepository.SOURCE_ONLY)
                             ? SourceOnlySnapshotRepository.readerWrapper(engineConfig)
                             : Function.identity(),
-                        false
+                        false,
+                        true
                     )
                 );
             } else {
@@ -417,7 +418,8 @@ public class SearchableSnapshots extends Plugin implements IndexStorePlugin, Eng
                         indexSettings.getValue(SourceOnlySnapshotRepository.SOURCE_ONLY)
                             ? SourceOnlySnapshotRepository.readerWrapper(engineConfig)
                             : Function.identity(),
-                        false
+                        false,
+                        true
                     )
                 );
             }