Extract files stored within metadata hash into ByteArrayIndexInputs (#54072)

We can now detect blobs that are stored within the metadata hash field and 
open them directly as ByteArrayIndexInput.

Relates #53860

Author: tlrx

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

@@ -7,12 +7,10 @@
 
 import org.apache.lucene.store.BufferedIndexInput;
 import org.apache.lucene.store.IOContext;
-import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot.FileInfo;
 import org.elasticsearch.index.snapshots.blobstore.SlicedInputStream;
 
-import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Objects;
@@ -28,6 +26,8 @@ public BaseSearchableSnapshotIndexInput(String resourceDesc, BlobContainer blobC
         this.blobContainer = Objects.requireNonNull(blobContainer);
         this.fileInfo = Objects.requireNonNull(fileInfo);
         this.context = Objects.requireNonNull(context);
+        assert fileInfo.metadata().hashEqualsContents() == false
+            : "this method should only be used with blobs that are NOT stored in metadata's hash field (fileInfo: " + fileInfo + ')';
     }
 
     public BaseSearchableSnapshotIndexInput(
@@ -42,18 +42,6 @@ public BaseSearchableSnapshotIndexInput(
     }
 
     protected InputStream openInputStream(final long position, final long length) throws IOException {
-        // TODO move this at the Directory level
-        if (fileInfo.metadata().hashEqualsContents()) {
-            // extract blob content from metadata hash
-            final BytesRef data = fileInfo.metadata().hash();
-            if ((position < 0L) || (length < 0L) || (position + length > data.bytes.length)) {
-                throw new IllegalArgumentException(
-                    "Invalid arguments (pos=" + position + ", length=" + length + ") for hash content (length=" + data.bytes.length + ')'
-                );
-            }
-            return new ByteArrayInputStream(data.bytes, Math.toIntExact(position), Math.toIntExact(length));
-        }
-
         final long startPart = getPartNumberForPosition(position);
         final long endPart = getPartNumberForPosition(position + length);
         if ((startPart == endPart) || fileInfo.numberOfParts() == 1L) {

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

@@ -13,8 +13,10 @@
 import org.apache.lucene.store.IndexInput;
 import org.apache.lucene.store.IndexOutput;
 import org.apache.lucene.store.SingleInstanceLockFactory;
+import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.blobstore.BlobContainer;
+import org.elasticsearch.common.lucene.store.ByteArrayIndexInput;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.index.IndexSettings;
@@ -231,7 +233,13 @@ public CacheFile getCacheFile(CacheKey cacheKey, long fileLength) throws Excepti
     @Override
     public IndexInput openInput(final String name, final IOContext context) throws IOException {
         ensureOpen();
+
         final BlobStoreIndexShardSnapshot.FileInfo fileInfo = fileInfo(name);
+        if (fileInfo.metadata().hashEqualsContents()) {
+            final BytesRef content = fileInfo.metadata().hash();
+            return new ByteArrayIndexInput("ByteArrayIndexInput(" + name + ')', content.bytes, content.offset, content.length);
+        }
+
         final IndexInputStats inputStats = stats.computeIfAbsent(name, n -> createIndexInputStats(fileInfo.length()));
         if (useCache && isExcludedFromCache(name) == false) {
             return new CacheBufferedIndexInput(this, fileInfo, context, inputStats);

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

@@ -8,14 +8,12 @@
 import org.apache.lucene.store.BufferedIndexInput;
 import org.apache.lucene.store.IOContext;
 import org.apache.lucene.store.IndexInput;
-import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.common.CheckedRunnable;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot.FileInfo;
 
-import java.io.ByteArrayInputStream;
 import java.io.Closeable;
 import java.io.EOFException;
 import java.io.IOException;
@@ -259,25 +257,7 @@ public String toString() {
     }
 
     private InputStream openBlobStream(int part, long pos, long length) throws IOException {
-        final InputStream stream;
-        if (fileInfo.metadata().hashEqualsContents() == false) {
-            stream = blobContainer.readBlob(fileInfo.partName(part), pos, length);
-        } else {
-            // extract blob content from metadata hash
-            final BytesRef data = fileInfo.metadata().hash();
-            if (part > 0) {
-                assert fileInfo.numberOfParts() >= part;
-                for (int i = 0; i < part; i++) {
-                    pos += fileInfo.partBytes(i);
-                }
-            }
-            if ((pos < 0L) || (length < 0L) || (pos + length > data.bytes.length)) {
-                throw new IllegalArgumentException("Invalid arguments (pos=" + pos + ", length=" + length
-                    + ") for hash content (length=" + data.bytes.length + ')');
-            }
-            stream = new ByteArrayInputStream(data.bytes, Math.toIntExact(pos), Math.toIntExact(length));
-        }
-        return stream;
+        return blobContainer.readBlob(fileInfo.partName(part), pos, length);
     }
 
     /**