Create SparseFileTracker with already completed/available ranges (ela…

…stic#65501)

This commit allows to create SparseFileTracker instances with 
already completed/available ranges. Creating non empty sparse 
file tracker will be required for the searchable snapshots cache 
to be initialized with existing information on cache files.

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

@@ -37,12 +37,50 @@ public class SparseFileTracker {
 
     private final long length;
 
+    /**
+     * Creates a new empty {@link SparseFileTracker}
+     *
+     * @param description a description for the sparse file tracker
+     * @param length      the length of the file tracked by the sparse file tracker
+     */
     public SparseFileTracker(String description, long length) {
+        this(description, length, Collections.emptySortedSet());
+    }
+
+    /**
+     * Creates a {@link SparseFileTracker} with some ranges already present
+     *
+     * @param description a description for the sparse file tracker
+     * @param length      the length of the file tracked by the sparse file tracker
+     * @param ranges      the set of ranges to be considered present
+     */
+    public SparseFileTracker(String description, long length, SortedSet<Tuple<Long, Long>> ranges) {
         this.description = description;
         this.length = length;
         if (length < 0) {
             throw new IllegalArgumentException("Length [" + length + "] must be equal to or greater than 0 for [" + description + "]");
         }
+        if (ranges.isEmpty() == false) {
+            synchronized (mutex) {
+                Range previous = null;
+                for (Tuple<Long, Long> next : ranges) {
+                    final Range range = new Range(next.v1(), next.v2(), null);
+                    if (range.end <= range.start) {
+                        throw new IllegalArgumentException("Range " + range + " cannot be empty");
+                    }
+                    if (length < range.end) {
+                        throw new IllegalArgumentException("Range " + range + " is exceeding maximum length [" + length + ']');
+                    }
+                    if (previous != null && range.start <= previous.end) {
+                        throw new IllegalArgumentException("Range " + range + " is overlapping a previous range " + previous);
+                    }
+                    final boolean added = this.ranges.add(range);
+                    assert added : range + " already exist in " + this.ranges;
+                    previous = range;
+                }
+                assert invariant();
+            }
+        }
     }
 
     public long getLength() {

x-pack/plugin/searchable-snapshots/src/test/java/org/elasticsearch/index/store/cache/SparseFileTrackerTests.java

@@ -34,6 +34,7 @@
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
+import static org.hamcrest.Matchers.nullValue;
 
 public class SparseFileTrackerTests extends ESTestCase {
 
@@ -414,7 +415,33 @@ public void testThreadSafety() throws InterruptedException {
         checkThread.join();
     }
 
-    public void testCompletedRanges() {
+    public void testSparseFileTrackerCreatedWithCompletedRanges() {
+        final long fileLength = between(0, 1000);
+        final SortedSet<Tuple<Long, Long>> completedRanges = randomRanges(fileLength);
+
+        final SparseFileTracker sparseFileTracker = new SparseFileTracker("test", fileLength, completedRanges);
+        assertThat(sparseFileTracker.getCompletedRanges(), equalTo(completedRanges));
+
+        for (Tuple<Long, Long> completedRange : completedRanges) {
+            assertThat(sparseFileTracker.getAbsentRangeWithin(completedRange.v1(), completedRange.v2()), nullValue());
+
+            final AtomicBoolean listenerCalled = new AtomicBoolean();
+            assertThat(sparseFileTracker.waitForRange(completedRange, completedRange, new ActionListener<>() {
+                @Override
+                public void onResponse(Void aVoid) {
+                    listenerCalled.set(true);
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    throw new AssertionError(e);
+                }
+            }), hasSize(0));
+            assertThat(listenerCalled.get(), is(true));
+        }
+    }
+
+    public void testGetCompletedRanges() {
         final byte[] fileContents = new byte[between(0, 1000)];
         final SparseFileTracker sparseFileTracker = new SparseFileTracker("test", fileContents.length);
 
@@ -536,4 +563,18 @@ private static boolean processGap(byte[] fileContents, SparseFileTracker.Gap gap
             return true;
         }
     }
+
+    /**
+     * Generates a sorted set of non-empty and non-contiguous random ranges that could fit into a file of a given maximum length.
+     */
+    private static SortedSet<Tuple<Long, Long>> randomRanges(long length) {
+        final SortedSet<Tuple<Long, Long>> randomRanges = new TreeSet<>(Comparator.comparingLong(Tuple::v1));
+        for (long i = 0L; i < length;) {
+            long start = randomLongBetween(i, Math.max(0L, length - 1L));
+            long end = randomLongBetween(start + 1L, length); // +1 for non empty ranges
+            randomRanges.add(Tuple.tuple(start, end));
+            i = end + 1L + randomLongBetween(0L, Math.max(0L, length - end)); // +1 for non contiguous ranges
+        }
+        return randomRanges;
+    }
 }