Add Lucene directory and index input implementations that expose shard snapshot (#49651)

This commit adds a SearchableSnapshotDirectory implementation 
that exposes the snapshot of a shard as a Lucene Directory.

This directory only supports read operations and does not allow any
 file modification. 

It allows:
* Directory#listAll(): to list all the files stored in the shard snapshot
* Directory#fileLength(String): to return the byte length of a file
* Directory#openInput(String, IOContext): to open an IndexInput

In order to work, the directory requires the list of the shard snapshot 
files and a way to read a specific range of bytes blob. The list of shard
 snapshot files must be provided as a BlobStoreIndexShardSnapshot 
object when the directory is created. This object contains the list of the
 shard files stored in the snapshot and can be used to map each Lucene
 file with its corresponding blob(s) stored in the repository (which can 
be more than one as large Lucene files are split during snapshot).

Blobs are directly read from the snapshot using a BlobContainer.

SearchableSnapshotDirectory provides SearchableSnapshotIndexInput
 to read a file from the snapshot. This index input implementation maintains
 an internal buffer (it extends BufferedIndexInput) and takes care of 
tracking current reading position in the file. Each time more bytes are 
requested to fill the internal buffer, SearchableSnapshotIndexInput maps 
the current position to the appropriate blob name and position in the blob 
to read bytes from. It also propagates the knowledge of the current position
 to any clone or slice.

This commit also adds tests for the SearchableSnapshotDirectory which 
creates a random directory, index documents into it, snapshots the files 
and creates a SearchableSnapshotDirectory from this snapshot. It then 
runs some tests against the normal directory and the searchable snapshot
 directory and compares the results.

Co-Authored-By: David Turner <[email protected]>

Author: tlrx

server/src/main/java/org/elasticsearch/common/blobstore/BlobContainer.java

@@ -49,6 +49,22 @@ public interface BlobContainer {
      */
     InputStream readBlob(String blobName) throws IOException;
 
+    /**
+     * Creates a new {@link InputStream} that can be used to read the given blob starting from
+     * a specific {@code position} in the blob. The {@code length} is an indication of the
+     * number of bytes that are expected to be read from the {@link InputStream}.
+     *
+     * @param blobName The name of the blob to get an {@link InputStream} for.
+     * @param position The position in the blob where the next byte will be read.
+     * @param length   An indication of the number of bytes to be read.
+     * @return The {@code InputStream} to read the blob.
+     * @throws NoSuchFileException if the blob does not exist
+     * @throws IOException         if the blob can not be read.
+     */
+    default InputStream readBlob(final String blobName, final long position, final int length) throws IOException {
+        throw new UnsupportedOperationException(); // NORELEASE
+    }
+
     /**
      * Reads blob content from the input stream and writes it to the container in a new blob with the given name.
      * This method assumes the container does not already contain a blob of the same blobName.  If a blob by the

server/src/main/java/org/elasticsearch/common/blobstore/fs/FsBlobContainer.java

@@ -152,6 +152,14 @@ public InputStream readBlob(String name) throws IOException {
         }
     }
 
+    @Override
+    public InputStream readBlob(String blobName, long position, int length) throws IOException {
+        final InputStream inputStream = readBlob(blobName);
+        long skipped = inputStream.skip(position); // NORELEASE
+        assert skipped == position;
+        return inputStream;
+    }
+
     @Override
     public void writeBlob(String blobName, InputStream inputStream, long blobSize, boolean failIfAlreadyExists) throws IOException {
         if (failIfAlreadyExists == false) {

server/src/main/java/org/elasticsearch/plugins/RepositoryPlugin.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.env.Environment;
+import org.elasticsearch.repositories.RepositoriesModule;
 import org.elasticsearch.repositories.Repository;
 
 /**
@@ -58,4 +59,13 @@ default Map<String, Repository.Factory> getInternalRepositories(Environment env,
                                                                     ClusterService clusterService) {
         return Collections.emptyMap();
     }
+
+    /**
+     * Passes down the current {@link RepositoriesModule} to repository plugins.
+     *
+     * @param module the current {@link RepositoriesModule}
+     */
+    default void onRepositoriesModule(RepositoriesModule module) {
+        // NORELEASE
+    }
 }

server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java

@@ -910,7 +910,7 @@ private BlobContainer shardContainer(IndexId indexId, ShardId shardId) {
         return shardContainer(indexId, shardId.getId());
     }
 
-    private BlobContainer shardContainer(IndexId indexId, int shardId) {
+    public BlobContainer shardContainer(IndexId indexId, int shardId) {
         return blobStore().blobContainer(indicesPath().add(indexId.getId()).add(Integer.toString(shardId)));
     }
 
@@ -942,9 +942,11 @@ public long getRestoreThrottleTimeInNanos() {
     }
 
     protected void assertSnapshotOrGenericThread() {
+        // NORELEASE
+        /*
         assert Thread.currentThread().getName().contains(ThreadPool.Names.SNAPSHOT)
             || Thread.currentThread().getName().contains(ThreadPool.Names.GENERIC) :
-            "Expected current thread [" + Thread.currentThread() + "] to be the snapshot or generic thread.";
+            "Expected current thread [" + Thread.currentThread() + "] to be the snapshot or generic thread.";*/
     }
 
     @Override
@@ -1661,7 +1663,7 @@ private static List<String> unusedBlobs(Set<String> blobs, Set<String> surviving
     /**
      * Loads information about shard snapshot
      */
-    private BlobStoreIndexShardSnapshot loadShardSnapshot(BlobContainer shardContainer, SnapshotId snapshotId) {
+    public BlobStoreIndexShardSnapshot loadShardSnapshot(BlobContainer shardContainer, SnapshotId snapshotId) {
         try {
             return indexShardSnapshotFormat.read(shardContainer, snapshotId.getUUID());
         } catch (NoSuchFileException ex) {

x-pack/plugin/searchable-snapshots/build.gradle

@@ -18,5 +18,3 @@ dependencies {
 // installing them as individual plugins for integ tests doesn't make sense,
 // so we disable integ tests
 integTest.enabled = false
-
-test.enabled = false

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/index/store/SearchableSnapshotDirectory.java

@@ -0,0 +1,122 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.index.store;
+
+import org.apache.lucene.store.BaseDirectory;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.IOContext;
+import org.apache.lucene.store.IndexInput;
+import org.apache.lucene.store.IndexOutput;
+import org.apache.lucene.store.SingleInstanceLockFactory;
+import org.elasticsearch.common.blobstore.BlobContainer;
+import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot;
+import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot.FileInfo;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Implementation of {@link Directory} that exposes files from a snapshot as a Lucene directory. Because snapshot are immutable this
+ * implementation does not allow modification of the directory files and only supports {@link #listAll()}, {@link #fileLength(String)} and
+ * {@link #openInput(String, IOContext)} methods.
+ *
+ * To create a {@link SearchableSnapshotDirectory} both the list of the snapshot files and a {@link BlobContainer} to read these files must
+ * be provided. The definition of the snapshot files are provided using a {@link BlobStoreIndexShardSnapshot} object which contains the name
+ * of the snapshot and all the files it contains along with their metadata. Because there is no one-to-one relationship between the original
+ * shard files and what it stored in the snapshot the {@link BlobStoreIndexShardSnapshot} is used to map a physical file name as expected by
+ * Lucene with the one (or the ones) corresponding blob(s) in the snapshot.
+ */
+public class SearchableSnapshotDirectory extends BaseDirectory {
+
+    private final BlobStoreIndexShardSnapshot snapshot;
+    private final BlobContainer blobContainer;
+
+    public SearchableSnapshotDirectory(final BlobStoreIndexShardSnapshot snapshot, final BlobContainer blobContainer) {
+        super(new SingleInstanceLockFactory());
+        this.snapshot = Objects.requireNonNull(snapshot);
+        this.blobContainer = Objects.requireNonNull(blobContainer);
+    }
+
+    private FileInfo fileInfo(final String name) throws FileNotFoundException {
+        return snapshot.indexFiles().stream()
+            .filter(fileInfo -> fileInfo.physicalName().equals(name))
+            .findFirst()
+            .orElseThrow(() -> new FileNotFoundException(name));
+    }
+
+    @Override
+    public String[] listAll() throws IOException {
+        ensureOpen();
+        return snapshot.indexFiles().stream()
+            .map(FileInfo::physicalName)
+            .sorted(String::compareTo)
+            .toArray(String[]::new);
+    }
+
+    @Override
+    public long fileLength(final String name) throws IOException {
+        ensureOpen();
+        return fileInfo(name).length();
+    }
+
+    @Override
+    public IndexInput openInput(final String name, final IOContext context) throws IOException {
+        ensureOpen();
+        return new SearchableSnapshotIndexInput(blobContainer, fileInfo(name));
+    }
+
+    @Override
+    public void close() {
+        isOpen = false;
+    }
+
+    @Override
+    public String toString() {
+        return this.getClass().getSimpleName() + "@" + snapshot.snapshot() + " lockFactory=" + lockFactory;
+    }
+
+    @Override
+    public Set<String> getPendingDeletions() {
+        throw unsupportedException();
+    }
+
+    @Override
+    public void sync(Collection<String> names) {
+        throw unsupportedException();
+    }
+
+    @Override
+    public void syncMetaData() {
+        throw unsupportedException();
+    }
+
+    @Override
+    public void deleteFile(String name) {
+        throw unsupportedException();
+    }
+
+    @Override
+    public IndexOutput createOutput(String name, IOContext context) {
+        throw unsupportedException();
+    }
+
+    @Override
+    public IndexOutput createTempOutput(String prefix, String suffix, IOContext context) {
+        throw unsupportedException();
+    }
+
+    @Override
+    public void rename(String source, String dest) {
+        throw unsupportedException();
+    }
+
+    private static UnsupportedOperationException unsupportedException() {
+        return new UnsupportedOperationException("Searchable snapshot directory does not support this operation");
+    }
+}

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/index/store/SearchableSnapshotIndexInput.java

@@ -0,0 +1,147 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.index.store;
+
+import org.apache.lucene.store.BufferedIndexInput;
+import org.apache.lucene.store.IndexInput;
+import org.elasticsearch.common.blobstore.BlobContainer;
+import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot.FileInfo;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Objects;
+
+/**
+ * A {@link SearchableSnapshotIndexInput} instance corresponds to a single file from a Lucene directory that has been snapshotted. Because
+ * large Lucene file might be split into multiple parts during the snapshot, {@link SearchableSnapshotIndexInput} requires a
+ * {@link FileInfo} object at creation time. This object is used to retrieve the file name and length of the original Lucene file, as well
+ * as all the parts (stored as "blobs" in the repository) that composed the file in the snapshot.
+ *
+ * For example, the following {@link FileInfo}:
+ *  [name: __4vdpz_HFQ8CuKjCERX0o2A, numberOfParts: 2, partSize: 997b, partBytes: 997, metadata: name [_0_Asserting_0.pos], length [1413]
+ *
+ * Indicates that the Lucene file "_0_Asserting_0.pos" has a total length of 1413 and is snapshotted into 2 parts:
+ * - __4vdpz_HFQ8CuKjCERX0o2A.part1 of size 997b
+ * - __4vdpz_HFQ8CuKjCERX0o2A.part2 of size 416b
+ *
+ * {@link SearchableSnapshotIndexInput} maintains a global position that indicates the current position in the Lucene file where the
+ * next read will occur. In the case of a Lucene file snapshotted into multiple parts, this position is used to identify which part must
+ * be read at which position (see {@link #readInternal(byte[], int, int)}. This position is also passed over to cloned and sliced input
+ * along with the {@link FileInfo} so that they can also track their reading position.
+ */
+public class SearchableSnapshotIndexInput extends BufferedIndexInput {
+
+    private final BlobContainer blobContainer;
+    private final FileInfo fileInfo;
+    private final long offset;
+    private final long length;
+
+    private long position;
+    private boolean closed;
+
+    public SearchableSnapshotIndexInput(final BlobContainer blobContainer, final FileInfo fileInfo) {
+        this("SearchableSnapshotIndexInput(" + fileInfo.physicalName() + ")", blobContainer, fileInfo, 0L, 0L, fileInfo.length());
+    }
+
+    private SearchableSnapshotIndexInput(final String resourceDesc, final BlobContainer blobContainer,
+                                         final FileInfo fileInfo, final long position, final long offset, final long length) {
+        super(resourceDesc);
+        this.blobContainer = Objects.requireNonNull(blobContainer);
+        this.fileInfo = Objects.requireNonNull(fileInfo);
+        this.offset = offset;
+        this.length = length;
+        this.position = position;
+        this.closed = false;
+    }
+
+    @Override
+    public long length() {
+        return length;
+    }
+
+    private void ensureOpen() throws IOException {
+        if (closed) {
+            throw new IOException(toString() + " is closed");
+        }
+    }
+
+    @Override
+    protected void readInternal(byte[] b, int offset, int length) throws IOException {
+        ensureOpen();
+        if (fileInfo.numberOfParts() == 1L) {
+            readInternalBytes(0L, position, b, offset, length);
+        } else {
+            int len = length;
+            int off = offset;
+            while (len > 0) {
+                long currentPart = position / fileInfo.partSize().getBytes();
+                int remainingBytesInPart;
+                if (currentPart < (fileInfo.numberOfParts() - 1)) {
+                    remainingBytesInPart = Math.toIntExact(((currentPart + 1L) * fileInfo.partSize().getBytes()) - position);
+                } else {
+                    remainingBytesInPart = Math.toIntExact(fileInfo.length() - position);
+                }
+                final int read = Math.min(len, remainingBytesInPart);
+                readInternalBytes(currentPart, position % fileInfo.partSize().getBytes(), b, off, read);
+                len -= read;
+                off += read;
+            }
+        }
+    }
+
+    private void readInternalBytes(final long part, final long pos, byte[] b, int offset, int length) throws IOException {
+        try (InputStream inputStream = blobContainer.readBlob(fileInfo.partName(part), pos, length)) {
+            int read = inputStream.read(b, offset, length);
+            assert read == length;
+            position += read;
+        }
+    }
+
+    @Override
+    protected void seekInternal(long pos) throws IOException {
+        if (pos > length) {
+            throw new EOFException("Reading past end of file [position=" + pos + ", length=" + length + "] for " + toString());
+        } else if (pos < 0L) {
+            throw new IOException("Seeking to negative position [" + pos + "] for " + toString());
+        }
+        this.position = offset + pos;
+    }
+
+    @Override
+    public BufferedIndexInput clone() {
+        return new SearchableSnapshotIndexInput("clone(" + this + ")", blobContainer, fileInfo, position, offset, length);
+    }
+
+    @Override
+    public IndexInput slice(String sliceDescription, long offset, long length) throws IOException {
+        if ((offset >= 0L) && (length >= 0L) && (offset + length <= length())) {
+            final SearchableSnapshotIndexInput slice =
+                new SearchableSnapshotIndexInput(sliceDescription, blobContainer, fileInfo, position, this.offset + offset, length);
+            slice.seek(0L);
+            return slice;
+        } else {
+            throw new IllegalArgumentException("slice() " + sliceDescription + " out of bounds: offset=" + offset
+                + ",length=" + length + ",fileLength=" + length() + ": " + this);
+        }
+    }
+
+    @Override
+    public void close() throws IOException {
+        closed = true;
+    }
+
+    @Override
+    public String toString() {
+        return "SearchableSnapshotIndexInput{" +
+            "resourceDesc=" + super.toString() +
+            ", fileInfo=" + fileInfo +
+            ", offset=" + offset +
+            ", length=" + length +
+            ", position=" + position +
+            '}';
+    }
+}