Do not wrap soft-deletes reader for segment stats (#51331)

IndexWriter might not filter out fully deleted segments if retention 
leases exist or the number of the retaining operations is non-zero.
SoftDeletesDirectoryReaderWrapper, however, always filters out fully
deleted segments.

This change uses the original directory reader when calculating segment 
stats instead.

Relates #51192
Closes #51303

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

@@ -54,7 +54,8 @@ public NoOpEngine(EngineConfig config) {
         super(config, null, null, true, Function.identity());
         this.stats = new SegmentsStats();
         Directory directory = store.directory();
-        try (DirectoryReader reader = openDirectory(directory)) {
+        // Do not wrap soft-deletes reader when calculating segment stats as the wrapper might filter out fully deleted segments.
+        try (DirectoryReader reader = openDirectory(directory, false)) {
             for (LeafReaderContext ctx : reader.getContext().leaves()) {
                 SegmentReader segmentReader = Lucene.segmentReader(ctx.reader());
                 fillSegmentStats(segmentReader, true, stats);

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

@@ -516,7 +516,12 @@ public void advanceMaxSeqNoOfUpdatesOrDeletes(long maxSeqNoOfUpdatesOnPrimary) {
             maxSeqNoOfUpdatesOnPrimary + ">" + getMaxSeqNoOfUpdatesOrDeletes();
     }
 
-    protected static DirectoryReader openDirectory(Directory dir) throws IOException {
-        return new SoftDeletesDirectoryReaderWrapper(DirectoryReader.open(dir, OFF_HEAP_READER_ATTRIBUTES), Lucene.SOFT_DELETES_FIELD);
+    protected static DirectoryReader openDirectory(Directory directory, boolean wrapSoftDeletes) throws IOException {
+        final DirectoryReader reader = DirectoryReader.open(directory, OFF_HEAP_READER_ATTRIBUTES);
+        if (wrapSoftDeletes) {
+            return new SoftDeletesDirectoryReaderWrapper(reader, Lucene.SOFT_DELETES_FIELD);
+        } else {
+            return reader;
+        }
     }
 }

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

@@ -21,6 +21,7 @@
 import org.apache.lucene.index.PointValues;
 import org.apache.lucene.index.SegmentCommitInfo;
 import org.apache.lucene.index.SegmentReader;
+import org.apache.lucene.index.SoftDeletesDirectoryReaderWrapper;
 import org.apache.lucene.index.SortedDocValues;
 import org.apache.lucene.index.SortedNumericDocValues;
 import org.apache.lucene.index.SortedSetDocValues;
@@ -77,16 +78,18 @@ public FrozenEngine(EngineConfig config) {
 
         boolean success = false;
         Directory directory = store.directory();
-        try (DirectoryReader reader = openDirectory(directory)) {
-            canMatchReader = ElasticsearchDirectoryReader.wrap(new RewriteCachingDirectoryReader(directory, reader.leaves()),
-                config.getShardId());
+        // Do not wrap soft-deletes reader when calculating segment stats as the wrapper might filter out fully deleted segments.
+        try (DirectoryReader reader = openDirectory(directory, false)) {
             // we record the segment stats here - that's what the reader needs when it's open and it give the user
             // an idea of what it can save when it's closed
             this.stats = new SegmentsStats();
             for (LeafReaderContext ctx : reader.getContext().leaves()) {
                 SegmentReader segmentReader = Lucene.segmentReader(ctx.reader());
                 fillSegmentStats(segmentReader, true, stats);
             }
+            final DirectoryReader wrappedReader = new SoftDeletesDirectoryReaderWrapper(reader, Lucene.SOFT_DELETES_FIELD);
+            canMatchReader = ElasticsearchDirectoryReader.wrap(
+                new RewriteCachingDirectoryReader(directory, wrappedReader.leaves()), config.getShardId());
             success = true;
         } catch (IOException e) {
             throw new UncheckedIOException(e);
@@ -167,7 +170,7 @@ private synchronized ElasticsearchDirectoryReader getOrOpenReader() throws IOExc
                 for (ReferenceManager.RefreshListener listeners : config ().getInternalRefreshListener()) {
                     listeners.beforeRefresh();
                 }
-                final DirectoryReader dirReader = openDirectory(engineConfig.getStore().directory());
+                final DirectoryReader dirReader = openDirectory(engineConfig.getStore().directory(), true);
                 reader = lastOpenedReader = wrapReader(dirReader, Function.identity());
                 processReader(reader);
                 reader.getReaderCacheHelper().addClosedListener(this::onReaderClosed);