Speed up merging when source is disabled. (#62443) (#62474)

The CodecReader wrapper we use to remove the `_recovery_source` field
doesn't override `StoredFieldsreader#getMergeInstance`, which has the
undesired side-effect of preventing the wrapped stored fields reader
from optimizing merging.

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

@@ -45,6 +45,7 @@ import org.apache.lucene.util.BitSetIterator;
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.Objects;
 import java.util.function.Supplier;
 
 final class RecoverySourcePruneMergePolicy extends OneMergeWrappingMergePolicy {
@@ -133,25 +134,8 @@ final class RecoverySourcePruneMergePolicy extends OneMergeWrappingMergePolicy {
 
         @Override
         public StoredFieldsReader getFieldsReader() {
-            StoredFieldsReader fieldsReader = super.getFieldsReader();
+            return new RecoverySourcePruningStoredFieldsReader(
-            return new FilterStoredFieldsReader(fieldsReader) {
+                    super.getFieldsReader(), recoverySourceToKeep, recoverySourceField);
-                @Override
-                public void visitDocument(int docID, StoredFieldVisitor visitor) throws IOException {
-                    if (recoverySourceToKeep != null && recoverySourceToKeep.get(docID)) {
-                        super.visitDocument(docID, visitor);
-                    } else {
-                        super.visitDocument(docID, new FilterStoredFieldVisitor(visitor) {
-                            @Override
-                            public Status needsField(FieldInfo fieldInfo) throws IOException {
-                                if (recoverySourceField.equals(fieldInfo.name)) {
-                                    return Status.NO;
-                                }
-                                return super.needsField(fieldInfo);
-                            }
-                        });
-                    }
-                }
-            };
         }
 
         @Override
@@ -212,38 +196,76 @@ final class RecoverySourcePruneMergePolicy extends OneMergeWrappingMergePolicy {
             }
         }
 
-        private static class FilterStoredFieldsReader extends StoredFieldsReader {
+        private abstract static class FilterStoredFieldsReader extends StoredFieldsReader {
 
-            private final StoredFieldsReader fieldsReader;
+            protected final StoredFieldsReader in;
 
             FilterStoredFieldsReader(StoredFieldsReader fieldsReader) {
-                this.fieldsReader = fieldsReader;
+                this.in = fieldsReader;
             }
 
             @Override
             public long ramBytesUsed() {
-                return fieldsReader.ramBytesUsed();
+                return in.ramBytesUsed();
             }
 
             @Override
             public void close() throws IOException {
-                fieldsReader.close();
+                in.close();
             }
 
             @Override
             public void visitDocument(int docID, StoredFieldVisitor visitor) throws IOException {
-                fieldsReader.visitDocument(docID, visitor);
+                in.visitDocument(docID, visitor);
+            }
+
+            @Override
+            public abstract StoredFieldsReader clone();
+
+            @Override
+            public void checkIntegrity() throws IOException {
+                in.checkIntegrity();
+            }
+        }
+
+        private static class RecoverySourcePruningStoredFieldsReader extends FilterStoredFieldsReader {
+
+            private final BitSet recoverySourceToKeep;
+            private final String recoverySourceField;
+
+            RecoverySourcePruningStoredFieldsReader(StoredFieldsReader in, BitSet recoverySourceToKeep, String recoverySourceField) {
+                super(in);
+                this.recoverySourceToKeep = recoverySourceToKeep;
+                this.recoverySourceField = Objects.requireNonNull(recoverySourceField);
+            }
+
+            @Override
+            public void visitDocument(int docID, StoredFieldVisitor visitor) throws IOException {
+                if (recoverySourceToKeep != null && recoverySourceToKeep.get(docID)) {
+                    super.visitDocument(docID, visitor);
+                } else {
+                    super.visitDocument(docID, new FilterStoredFieldVisitor(visitor) {
+                        @Override
+                        public Status needsField(FieldInfo fieldInfo) throws IOException {
+                            if (recoverySourceField.equals(fieldInfo.name)) {
+                                return Status.NO;
+                            }
+                            return super.needsField(fieldInfo);
+                        }
+                    });
+                }
+            }
+
+            @Override
+            public StoredFieldsReader getMergeInstance() {
+                return new RecoverySourcePruningStoredFieldsReader(in.getMergeInstance(), recoverySourceToKeep, recoverySourceField);
             }
 
             @Override
             public StoredFieldsReader clone() {
-                return fieldsReader.clone();
+                return new RecoverySourcePruningStoredFieldsReader(in.clone(), recoverySourceToKeep, recoverySourceField);
             }
 
-            @Override
-            public void checkIntegrity() throws IOException {
-                fieldsReader.checkIntegrity();
-            }
         }
 
         private static class FilterStoredFieldVisitor extends StoredFieldVisitor {