Frozen index input clone copy cache file (#98930)

Frozen index input clone now copies the cache file object ensuring that the
fast path optimizations work.

docs/changelog/98930.yaml

@@ -0,0 +1,5 @@
+pr: 98930
+summary: Frozen index input clone copy cache file
+area: Snapshot/Restore
+type: bug
+issues: []

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/store/input/FrozenIndexInput.java

@@ -25,7 +25,7 @@ import org.elasticsearch.xpack.searchablesnapshots.store.SearchableSnapshotDirec
 import java.io.InputStream;
 import java.nio.ByteBuffer;
 
-public class FrozenIndexInput extends MetadataCachingIndexInput {
+public final class FrozenIndexInput extends MetadataCachingIndexInput {
 
     private static final Logger logger = LogManager.getLogger(FrozenIndexInput.class);
 
@@ -94,6 +94,14 @@ public class FrozenIndexInput extends MetadataCachingIndexInput {
         this.cacheFile = cacheFile.copy();
     }
 
+    /**
+     * Clone constructor, will mark this as cloned.
+     */
+    private FrozenIndexInput(FrozenIndexInput input) {
+        super(input);
+        this.cacheFile = input.cacheFile.copy();
+    }
+
     @Override
     protected void readWithoutBlobCache(ByteBuffer b) throws Exception {
         final long position = getAbsolutePosition();
@@ -196,4 +204,13 @@ public class FrozenIndexInput extends MetadataCachingIndexInput {
         );
     }
 
+    @Override
+    public FrozenIndexInput clone() {
+        return new FrozenIndexInput(this);
+    }
+
+    // for tests only
+    SharedBlobCacheService<CacheKey>.CacheFile cacheFile() {
+        return cacheFile;
+    }
 }

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/store/input/MetadataCachingIndexInput.java

@@ -39,6 +39,7 @@ import org.elasticsearch.xpack.searchablesnapshots.store.SearchableSnapshotDirec
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.util.Objects;
@@ -145,6 +146,35 @@ public abstract class MetadataCachingIndexInput extends BlobCacheBufferedIndexIn
         assert getBufferSize() <= BlobStoreCacheService.DEFAULT_CACHED_BLOB_SIZE; // must be able to cache at least one buffer's worth
     }
 
+    /**
+     * Clone constructor, will mark this as cloned.
+     */
+    protected MetadataCachingIndexInput(MetadataCachingIndexInput input) {
+        this(
+            input.logger,
+            "(clone of) " + input,
+            input.directory,
+            input.fileInfo,
+            input.context,
+            input.stats,
+            input.offset,
+            input.compoundFileOffset,
+            input.length,
+            input.cacheFileReference,
+            input.defaultRangeSize,
+            input.recoveryRangeSize,
+            input.headerBlobCacheByteRange,
+            input.footerBlobCacheByteRange
+        );
+        this.isClone = true;
+        try {
+            seek(input.getFilePointer());
+        } catch (IOException e) {
+            assert false : e;
+            throw new UncheckedIOException(e);
+        }
+    }
+
     /**
      * Detects read operations that are executed on the last 16 bytes of the index input which is where Lucene stores the footer checksum
      * of Lucene files. If such a read is detected this method tries to complete the read operation by reading the checksum from the

x-pack/plugin/searchable-snapshots/src/test/java/org/elasticsearch/xpack/searchablesnapshots/store/input/FrozenIndexInputTests.java

@@ -41,7 +41,9 @@ import java.util.List;
 
 import static org.elasticsearch.core.IOUtils.WINDOWS;
 import static org.elasticsearch.xpack.searchablesnapshots.cache.full.CacheService.resolveSnapshotCache;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.not;
 
 public class FrozenIndexInputTests extends AbstractSearchableSnapshotsTestCase {
 
@@ -130,6 +132,13 @@ public class FrozenIndexInputTests extends AbstractSearchableSnapshotsTestCase {
 
             final byte[] result = randomReadAndSlice(indexInput, fileData.length);
             assertArrayEquals(fileData, result);
+
+            // validate clone copies cache file object
+            indexInput.seek(randomLongBetween(0, fileData.length - 1));
+            FrozenIndexInput clone = (FrozenIndexInput) indexInput.clone();
+            assertThat(clone.cacheFile(), not(equalTo(((FrozenIndexInput) indexInput).cacheFile())));
+            assertThat(clone.getFilePointer(), equalTo(indexInput.getFilePointer()));
+
             indexInput.close();
         }
     }