Remove virtual file and fix duplicate clone (#6345)

A "virtual file" in this context is a small file whose entire contents
is stored in the snapshot metadata as opposed to in a discrete file in
the remote repository. OnDemandVirtualFileSnapshotIndexInput was a
complicated wrapper that ultimately returned a ByteArrayIndexInput
wrapping the file contents pulled from the metadata data. This change
simplifies things a lot and just creates the ByteArrayIndexInput
directly.

The other change (which led to removal of the virtual file) is to remove
a duplicate clone() call of the index input. The file cache design calls
for keeping the "origin" IndexInput instance in the cache and always
[returning clones][1]. OnDemandBlockIndexInput was incorrectly
duplicating this clone operation.

[1]: https://github.com/opensearch-project/OpenSearch/blob/0ca51a774211184835c4825dfeff38b23198352e/server/src/main/java/org/opensearch/index/store/remote/utils/TransferManager.java#L105

Signed-off-by: Andrew Ross <andrross@amazon.com>

server/src/main/java/org/opensearch/index/store/remote/directory/RemoteSnapshotDirectory.java

@@ -24,9 +24,9 @@
 import org.apache.lucene.store.NoLockFactory;
 import org.opensearch.LegacyESVersion;
 import org.opensearch.Version;
+import org.opensearch.common.lucene.store.ByteArrayIndexInput;
 import org.opensearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot;
 import org.opensearch.index.store.remote.file.OnDemandBlockSnapshotIndexInput;
-import org.opensearch.index.store.remote.file.OnDemandVirtualFileSnapshotIndexInput;
 import org.opensearch.index.store.remote.utils.TransferManager;
 import org.opensearch.repositories.blobstore.BlobStoreRepository;
 
@@ -72,7 +72,7 @@ public IndexInput openInput(String name, IOContext context) throws IOException {
         final BlobStoreIndexShardSnapshot.FileInfo fileInfo = fileInfoMap.get(name);
 
         if (fileInfo.name().startsWith(VIRTUAL_FILE_PREFIX)) {
-            return new OnDemandVirtualFileSnapshotIndexInput(fileInfo, localStoreDir, transferManager);
+            return new ByteArrayIndexInput(fileInfo.physicalName(), fileInfo.metadata().hash().bytes);
         }
         return new OnDemandBlockSnapshotIndexInput(fileInfo, localStoreDir, transferManager);
     }

server/src/main/java/org/opensearch/index/store/remote/file/OnDemandBlockIndexInput.java

@@ -338,7 +338,7 @@ private void demandBlock(int blockId) throws IOException {
             currentBlock.close();
         }
 
-        currentBlock = fetchBlock(blockId).clone();
+        currentBlock = fetchBlock(blockId);
         currentBlockId = blockId;
     }
 

server/src/main/java/org/opensearch/index/store/remote/utils/TransferManager.java

@@ -62,7 +62,7 @@ public CompletableFuture<IndexInput> asyncFetchBlob(BlobFetchRequest blobFetchRe
         });
     }
 
-    public CompletableFuture<IndexInput> asyncFetchBlob(Path path, Supplier<IndexInput> indexInputSupplier) {
+    private CompletableFuture<IndexInput> asyncFetchBlob(Path path, Supplier<IndexInput> indexInputSupplier) {
         return invocationLinearizer.linearize(path, p -> indexInputSupplier.get());
     }