[Searchable Snapshot] Fix bug of Searchable Snapshot Dependency on repository chunk_size

server/src/internalClusterTest/java/org/opensearch/snapshots/SearchableSnapshotIT.java

@@ -88,7 +88,7 @@ protected Settings.Builder randomRepositorySettings() {
     private Settings.Builder chunkedRepositorySettings() {
         final Settings.Builder settings = Settings.builder();
         settings.put("location", randomRepoPath()).put("compress", randomBoolean());
-        settings.put("chunk_size", 2 << 23, ByteSizeUnit.BYTES);
+        settings.put("chunk_size", randomBoolean() ? (2 << 23) : 1000, ByteSizeUnit.BYTES);
         return settings;
     }
 

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

@@ -139,18 +139,13 @@ protected IndexInput fetchBlock(int blockId) throws IOException {
         // If the snapshot file is chunked, we must account for this by
         // choosing the appropriate file part and updating the position
         // accordingly.
-        final int part = (int) (blockStart / partSize);
-        final long partStart = part * partSize;
-
-        final long position = blockStart - partStart;
-        final long length = blockEnd - blockStart;
 
         BlobFetchRequest blobFetchRequest = BlobFetchRequest.builder()
-            .position(position)
-            .length(length)
-            .blobName(fileInfo.partName(part))
+            .position(blockStart)
+            .length(blockEnd)
             .directory(directory)
             .fileName(blockFileName)
+            .fileInfo(fileInfo)
             .build();
         return transferManager.fetchBlob(blobFetchRequest);
     }

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

@@ -10,6 +10,7 @@
 
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.FSDirectory;
+import org.opensearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot;
 
 import java.nio.file.Path;
 
@@ -26,6 +27,8 @@ public class BlobFetchRequest {
 
     private final String blobName;
 
+    private final BlobStoreIndexShardSnapshot.FileInfo fileInfo;
+
     private final Path filePath;
 
     private final Directory directory;
@@ -36,6 +39,7 @@ private BlobFetchRequest(Builder builder) {
         this.position = builder.position;
         this.length = builder.length;
         this.blobName = builder.blobName;
+        this.fileInfo = builder.fileInfo;
         this.fileName = builder.fileName;
         this.filePath = builder.directory.getDirectory().resolve(fileName);
         this.directory = builder.directory;
@@ -53,6 +57,10 @@ public String getBlobName() {
         return blobName;
     }
 
+    public BlobStoreIndexShardSnapshot.FileInfo getFileInfo() {
+        return fileInfo;
+    }
+
     public Path getFilePath() {
         return filePath;
     }
@@ -96,6 +104,8 @@ public static final class Builder {
         private long position;
         private long length;
         private String blobName;
+
+        private BlobStoreIndexShardSnapshot.FileInfo fileInfo;
         private FSDirectory directory;
         private String fileName;
 
@@ -119,6 +129,11 @@ public Builder blobName(String blobName) {
             return this;
         }
 
+        public Builder fileInfo(BlobStoreIndexShardSnapshot.FileInfo fileInfo) {
+            this.fileInfo = fileInfo;
+            return this;
+        }
+
         public Builder directory(FSDirectory directory) {
             this.directory = directory;
             return this;

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

@@ -21,11 +21,15 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.io.SequenceInputStream;
 import java.io.UncheckedIOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -84,17 +88,47 @@ private static FileCachedIndexInput createIndexInput(FileCache fileCache, BlobCo
         return AccessController.doPrivileged((PrivilegedAction<FileCachedIndexInput>) () -> {
             try {
                 if (Files.exists(request.getFilePath()) == false) {
-                    try (
+                    List<InputStream> inputStreamList = new ArrayList<>();
+                    long partSize = request.getFileInfo().partSize().getBytes();
+                    long blockStart = request.getPosition();
+                    long blockEnd = request.getLength();
+                    int partNum = (int) (blockStart / partSize);
+                    long pos = blockStart;
+                    long diff = (blockEnd - pos);
+                    while (diff > 0) {
+                        long partStart = pos % partSize;
+                        long partEnd;
+                        if ((partStart + diff) > partSize) {
+                            partEnd = partSize;
+                        } else {
+                            partEnd = (partStart + diff);
+                        }
+                        long fetchBytes = partEnd - partStart;
                         InputStream snapshotFileInputStream = blobContainer.readBlob(
-                            request.getBlobName(),
-                            request.getPosition(),
-                            request.getLength()
+                            request.getFileInfo().partName(partNum),
+                            partStart,
+                            fetchBytes
                         );
-                        OutputStream fileOutputStream = Files.newOutputStream(request.getFilePath());
-                        OutputStream localFileOutputStream = new BufferedOutputStream(fileOutputStream)
-                    ) {
-                        snapshotFileInputStream.transferTo(localFileOutputStream);
+                        inputStreamList.add(snapshotFileInputStream);
+                        partNum++;
+                        pos = pos + fetchBytes;
+                        diff = (blockEnd - pos);
+
+                    }
+                    OutputStream fileOutputStream = Files.newOutputStream(request.getFilePath());
+                    OutputStream localFileOutputStream = new BufferedOutputStream(fileOutputStream);
+                    SequenceInputStream sis = new SequenceInputStream(Collections.enumeration(inputStreamList));
+                    int ch;
+                    while (true) {
+                        try {
+                            if (!((ch = sis.read()) != -1)) break;
+                            localFileOutputStream.write(ch);
+                        } catch (IOException e) {
+                            throw new RuntimeException(e);
+                        }
                     }
+                    sis.close();
+                    localFileOutputStream.close();
                 }
                 final IndexInput luceneIndexInput = request.getDirectory().openInput(request.getFileName(), IOContext.READ);
                 return new FileCachedIndexInput(fileCache, request.getFilePath(), luceneIndexInput);