Index stale operations to Lucene to have complete history (#29679)

Today, when processing out of order operations, we only add it into
translog but skip adding into Lucene. Translog, therefore, has a
complete history of sequence numbers while Lucene does not.

Since we would like to have a complete history in Lucene, this change
makes sure that stale operations will be added to Lucene as soft-deleted
documents if required.

Relates #29530

Author: dnhatn

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 all live docs.
+     * The wrapped reader can be used to query all documents.
+     *
+     * @param in the input directory reader
+     * @return the wrapped reader
+     */
+    public static DirectoryReader wrapAllDocsLive(DirectoryReader in) throws IOException {
+        return new DirectoryReaderWithAllLiveDocs(in);
+    }
+
+    private static final class DirectoryReaderWithAllLiveDocs extends FilterDirectoryReader {
+        static final class SubReaderWithAllLiveDocs extends FilterLeafReader {
+            SubReaderWithAllLiveDocs(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
+            }
+        }
+        DirectoryReaderWithAllLiveDocs(DirectoryReader in) throws IOException {
+            super(in, new FilterDirectoryReader.SubReaderWrapper() {
+                @Override
+                public LeafReader wrap(LeafReader leaf) {
+                    return new SubReaderWithAllLiveDocs(leaf);
+                }
+            });
+        }
+        @Override
+        protected DirectoryReader doWrapDirectoryReader(DirectoryReader in) throws IOException {
+            return wrapAllDocsLive(in);
+        }
+
+        @Override
+        public CacheHelper getReaderCacheHelper() {
+            return null; // Modifying liveDocs
+        }
+    }
+
     /**
      * Returns a numeric docvalues which can be used to soft-delete documents.
      */