Add SharedBytes.copyToCacheFileAligned without length method (#106193)

This change adds a variant of the existing SharedBytes.copyToCacheFileAligned 
that is not limited to copy only a given length of bytes but copies all bytes that 
can be read from the InputStream.

It also adds a variant of the existing BlobCacheUtils.computeRange() method 
that computes range of bytes to expand over a complete regions, instead of 
being limited to the length of the blob.

Finally, it removes an assertion that a CacheFile cannot write a range larger 
than the blob length.

Author: tlrx

x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/BlobCacheUtils.java

@@ -63,6 +63,10 @@ public static ByteRange computeRange(long rangeSize, long position, long size, l
         );
     }
 
+    public static ByteRange computeRange(long rangeSize, long position, long size) {
+        return ByteRange.of((position / rangeSize) * rangeSize, (((position + size - 1) / rangeSize) + 1) * rangeSize);
+    }
+
     public static void ensureSlice(String sliceName, long sliceOffset, long sliceLength, IndexInput input) {
         if (sliceOffset < 0 || sliceLength < 0 || sliceOffset + sliceLength > input.length()) {
             throw new IllegalArgumentException(

x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/shared/SharedBlobCacheService.java

@@ -1006,7 +1006,8 @@ public int populateAndRead(
             final RangeAvailableHandler reader,
             final RangeMissingHandler writer
         ) throws Exception {
-            assert assertOffsetsWithinFileLength(rangeToWrite.start(), rangeToWrite.length(), length);
+            // some cache files can grow after being created, so rangeToWrite can be larger than the initial {@code length}
+            assert rangeToWrite.start() >= 0 : rangeToWrite;
             assert assertOffsetsWithinFileLength(rangeToRead.start(), rangeToRead.length(), length);
             // We are interested in the total time that the system spends when fetching a result (including time spent queuing), so we start
             // our measurement here.

x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/shared/SharedBytes.java

@@ -14,6 +14,7 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.core.AbstractRefCounted;
 import org.elasticsearch.core.IOUtils;
+import org.elasticsearch.core.Streams;
 import org.elasticsearch.core.SuppressForbidden;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
@@ -184,6 +185,37 @@ public static void copyToCacheFileAligned(
         }
     }
 
+    /**
+     * Copy all bytes from {@code input} to {@code fc}, only doing writes aligned along {@link #PAGE_SIZE}.
+     *
+     * @param fc output cache file reference
+     * @param input stream to read from
+     * @param fileChannelPos position in {@code fc} to write to
+     * @param progressUpdater callback to invoke with the number of copied bytes as they are copied
+     * @param buffer bytebuffer to use for writing
+     * @return the number of bytes copied
+     * @throws IOException on failure
+     */
+    public static int copyToCacheFileAligned(IO fc, InputStream input, int fileChannelPos, IntConsumer progressUpdater, ByteBuffer buffer)
+        throws IOException {
+        int bytesCopied = 0;
+        while (true) {
+            final int bytesRead = Streams.read(input, buffer, buffer.remaining());
+            if (bytesRead <= 0) {
+                break;
+            }
+            if (buffer.hasRemaining()) {
+                // ensure that last write is aligned on 4k boundaries (= page size)
+                final int remainder = buffer.position() % PAGE_SIZE;
+                final int adjustment = remainder == 0 ? 0 : PAGE_SIZE - remainder;
+                buffer.position(buffer.position() + adjustment);
+            }
+            bytesCopied += positionalWrite(fc, fileChannelPos + bytesCopied, buffer);
+            progressUpdater.accept(bytesCopied);
+        }
+        return bytesCopied;
+    }
+
     private static int positionalWrite(IO fc, int start, ByteBuffer byteBuffer) throws IOException {
         byteBuffer.flip();
         int written = fc.write(byteBuffer, start);