content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package accord.utils;
  
  import java.util.Arrays;
  import java.util.Comparator;
  import java.util.Objects;
  import java.util.TreeSet;
  import java.util.stream.Collectors;
  import javax.annotation.Nonnull;
  
  import static accord.utils.ArrayBuffers.cachedInts;
  import static accord.utils.CheckpointIntervalArray.MAX_SCAN_DISTANCE;
  import static accord.utils.SortedArrays.Search.CEIL;
  
  public class CheckpointIntervalArrayBuilder<Ranges, Range, RoutingKey>
  {
      public enum Strategy
      {
          /**
           * Do not tenure any ranges that are scannable from the currently in-effect max scan distance.
           * This means we probably do less work on construction, but that our measure of the match count
           * at any point is inaccurate, and so our heuristics for when to write checkpoints may be wrong,
           * leading to more checkpoints than necessary.
           */
          FAST,
  
          /**
           * Tenure every range covering more than goalScanDistance. Any within max scan distance will also
           * update the scan distance, so that they will be filtered when a checkpoint is written. But
           * in the meantime they permit accurate tracking of the number of matches a query can return,
           * permitting our complexity calculations (that determine when checkpoints should be written, and
           * what our maximum scan distance should be), to be accurate. This can avoid bouncing between
           * two extremes, where a low max scan distance tenures and correctly detects a desirable larger scan
           * distance, which we rollover and this prevents us tenuring and tracking the number of matches, so
           * that we then pick a low max scan distance (and thereby also write a new checkpoint)
           */
          ACCURATE
      }
  
      /**
       * Should we maintain pointers to prior checkpoints that we may reference instead of reserializing
       * the remaining contents. This is cheap to visit as we stop enumerating as soon as we encounter
       * an entry that no longer covers us. We use some simple heuristics when deciding whether to do
       * this, namely that there are at least two entries (so we save one checkpoint entry) and that
       * there is at least one direct entry for each indirect/link entry in the range we will link.
       */
      public enum Links
      {
          LINKS,
          NO_LINKS
      }
  
      public interface Accessor<Ranges, Range, RoutingKey>
      {
          int size(Ranges ranges);
          Range get(Ranges ranges, int index);
          RoutingKey start(Ranges ranges, int index);
          RoutingKey start(Range range);
          RoutingKey end(Ranges ranges, int index);
          RoutingKey end(Range range);
          Comparator<RoutingKey> keyComparator();
          int compareEndTo(Range range, RoutingKey key);
          int compareStartTo(Range range, RoutingKey key);
          boolean endInclusive(Ranges ranges);
          int binarySearch(Ranges ranges, int from, int to, RoutingKey find, AsymmetricComparator<RoutingKey, Range> comparator, SortedArrays.Search op);
      }
  
      private static final int BIT31 = 0x80000000;
      private static final int BIT30 = 0x40000000;
      private static final int BIT29 = 0x20000000;
      private static final int BIT28 = 0x10000000;
      static final int MIN_INDIRECT_LINK_LENGTH = 2;
  
      final Accessor<Ranges, Range, RoutingKey> accessor;
      final boolean isAccurate;
      final boolean withLinks;
      final Ranges ranges;
  
      int[] bounds;
      int[] headers;
      int[] lists;
      int checkpointCount, headerPointer, listCount;
  
      final Scan<Ranges, Range, RoutingKey> scan;
      final TenuredSet<Ranges, Range, RoutingKey> tenured;
      final PendingCheckpoint<Ranges, Range, RoutingKey> pending = new PendingCheckpoint<>();
  
      // track the maximum possible number of entries we can match with both a scan + checkpoint lookup
      // this is an over-estimate and may be used by consumers to allocate out-of-order buffers for visitations
      int maxScanAndCheckpointMatches;
  
      public CheckpointIntervalArrayBuilder(Accessor<Ranges, Range, RoutingKey> accessor,
                                            Ranges ranges,
                                            Strategy strategy, Links links)
      {
          this(accessor, ranges, Math.min(MAX_SCAN_DISTANCE, 34 - Integer.numberOfLeadingZeros(accessor.size(ranges))), strategy, links);
      }
  
      public CheckpointIntervalArrayBuilder(Accessor<Ranges, Range, RoutingKey> accessor,
                                            Ranges ranges,
                                            int goalScanDistance,
                                            Strategy strategy, Links links)
      {
          this.accessor = accessor;
          this.isAccurate = strategy == Strategy.ACCURATE;
          this.withLinks = links == Links.LINKS;
          Invariants.requireArgument(goalScanDistance <= MAX_SCAN_DISTANCE);
          Invariants.requireArgument(goalScanDistance > 0);
          this.ranges = ranges;
          this.scan = new Scan<>(accessor);
          this.tenured = new TenuredSet<>(accessor);
          init(ranges, goalScanDistance);
      }
  
      void init(Ranges ranges, int goalScanDistance)
      {
          // we write checkpoints at least goalScanDistance apart
          scan.init(goalScanDistance);
          ArrayBuffers.IntBuffers cachedInts = cachedInts();
          // ask for int buffers in descending order of size
          int size = accessor.size(ranges);
          this.lists = cachedInts.getInts(size); // this one might need to grow
          // +2 to round-up each division, and +2 to account for the final entry (which might require an empty scan distance header)
          this.headers = cachedInts.getInts(((size / goalScanDistance) * 5) / 4 + 4);
          this.bounds = cachedInts.getInts(size / goalScanDistance + 1);
      }
  
      public interface Factory<T, Ranges>
      {
          T build(Ranges ranges, int[] bounds, int[] headers, int[] lists, int maxScanAndCheckpointMatches);
      }
  
      /**
       * Walk over each range, looking ahead by {@link #maxScanDistance} to decide if a range should
       * be tenured (written to a checkpoint) or scanned; the maximum scan distance is determined by the
       * number of open tenured entries, i.e. the minimum number of results we can expect to be returned
       * (or, if greater, the logarithm of the number of ranges in the collection).
       * <p>
       * Once we encounter a range that should be tenured, either write a checkpoint immediately
       * or make a note of the position we must scan to from the last entry in this checkpoint
       * and wait until it is permitted to write a checkpoint. This range will be tenured either
       * way for the following checkpoint.
       * <p>
       * The only reason not to write a checkpoint immediately is in the case we would breach
       * our linear space complexity limit, which is imposed by ensuring we have a space between
       * checkpoints at least as large as the number of entries written to the last checkpoint,
       * discounted by the number of entries we have removed from the tenured collection since
       * the last checkpoint.
       */
      public CheckpointIntervalArray<Ranges, Range, RoutingKey> build()
      {
          return build((ranges, bounds, headers, lists, maxScanAndCheckpointMatches) -> new CheckpointIntervalArray<>(accessor, ranges, bounds, headers, lists, maxScanAndCheckpointMatches));
      }
  
      public <T> T build(Factory<T, Ranges> factory)
      {
          int size = accessor.size(ranges);
          for (int ri = 0 ; ri < size ; ++ri)
          {
              // write a checkpoint if we meet our linear space complexity requirements
              // and we either have a tenured range that we must scan,
              // or the scan distance is now much larger than the minimum number of search results
              if (shouldWriteCheckpoint(ri))
                  writeCheckpoint(ri);
  
              // either tenure or update scan distance, potentially writing a checkpoint
              tenureOrScan(ri);
              tenured.untenure(ri);
          }
  
          // write our final pending checkpoint
          writeCheckpoint(size);
          closeHeaders();
  
          ArrayBuffers.IntBuffers cachedInts = cachedInts();
          int[] lists = cachedInts.completeAndDiscard(this.lists, listCount);
          int[] headers = cachedInts.completeAndDiscard(this.headers, headerPointer);
          int[] bounds = cachedInts.completeAndDiscard(this.bounds, checkpointCount);
          return factory.build(ranges, bounds, headers, lists, maxScanAndCheckpointMatches);
      }
  
      /**
       * Categorise the candidateIdx as either scannable, and if so update the scan distance;
       * or unscannable, in which case add it to the {@link #tenured} collection.
       * Note, that in ACCURATE mode we tenure the item if it is outside of the goalScanDistance
       * so we may track O(k) accurately above the O(lg2(N)) search and default scan distance,
       * but we still update the scan distance so that the checkpoint will exclude this entry.
       */
      private void tenureOrScan(int index)
      {
          Invariants.requireArgument(index >= 0);
  
          // then either migrate the index to pendingTenured, or ensure it will be scanned
          RoutingKey end = accessor.end(ranges, index);
          int scanLimit = scanLimit(index, isAccurate ? scan.goal : maxScanDistance());
          if (shouldTenure(end, scanLimit))
          {
              int lastIndex = tenured.tenure(end, index, ranges, scanLimit + 1);
              if (lastIndex - index > maxScanDistance()) scan.tenured(index);
              else if (!isAccurate) throw new IllegalStateException();
              else scan.updateScanDistance(index, lastIndex - index, this);
          }
          else
          {
              // TODO (low priority, efficiency): if the prior checkpoint has a scan distance >= this one,
              //  and <= 50% more than this one and there's no scanMustReachIndex nor tenuredRanges, don't
              //  write a new checkpoint (perhaps split shouldWriteCheckpoint logic in two)
              scan.update(end, index, ranges, scanLimit, this);
          }
      }
  
      /**
       * We are forbidden from writing a checkpoint nearer than this to a prior checkpoint.
       * This imposes our linear space complexity bounds, while not harming our O(log2(N) + K)
       * complexity bounds, as we guarantee minimumSpan is never more than the number of query
       * results.
       */
      private int minimumSpan()
      {
          return Math.max(scan.goal(), tenured.minimumSpan());
      }
  
      private int maxScanDistance()
      {
          // minimumSpan() reduces overtime, but there is no reason to reduce our scan distance
          // for tenuring below the scan distance we will write
          return Math.max(scan.watermark(), minimumSpan());
      }
  
      /**
       * The index after the last index we can scan from {@code atIndex} with at most {@code maxScanDistance}.
       */
      private int scanLimit(int atIndex, int maxScanDistance)
      {
          return Math.min(1 + atIndex + maxScanDistance, accessor.size(ranges));
      }
  
      private boolean shouldTenure(RoutingKey end, int scanLimit)
      {
          return scanLimit < accessor.size(ranges) && accessor.keyComparator().compare(end, accessor.start(ranges, scanLimit)) > 0;
      }
  
      private boolean canWriteCheckpoint(int atIndex)
      {
          return atIndex - pending.atIndex >= minimumSpan();
      }
  
      private boolean shouldWriteCheckpoint(int atIndex)
      {
          if (!canWriteCheckpoint(atIndex))
              return false;
  
          // TODO (desired, efficiency): consider these triggers
          if (scan.mustCheckpointToScanTenured(atIndex, maxScanDistance()))
              return true;
  
          return scan.hasMaybeDivergedFromMatchSize(tenured);
      }
  
      /**
       * Write a checkpoint for ranges[prevCheckpointIndex...ri)
       *
       * 1) Finalise the scan distance
       * 2) Write the header
       * 3) Filter the pending tenured ranges to remove those we can scan
       * 4) Write this list out
       * 5) Setup a link to this list, if it is large enough
       * 6) Rollover the scan, tenured and pending structures for the new pending checkpoint
       */
      private void writeCheckpoint(int nextCheckpointIndex)
      {
          int lastIndex = nextCheckpointIndex - 1;
          int scanDistance = scan.finalise(lastIndex);
          scanDistance = extendScanDistance(lastIndex, scanDistance);
  
          if (pending.atIndex < 0)
          {
              // we don't have any checkpoints pending, so don't try to finalise it
              // but if the new checkpoint doesn't cover index 0, insert a new empty
              // checkpoint for the scan distance
              if (nextCheckpointIndex > 0)
              {
                  // setup an initial empty checkpoint to store the first scan distance
                  maxScanAndCheckpointMatches = scanDistance;
                  writeHeader(scanDistance, 0);
              }
          }
          else
          {
              writeHeader(scanDistance, pending.atIndex);
              int maxCheckpointMatchCount = pending.filter(scanDistance, lastIndex);
              int listIndex = writeList(pending);
              if (withLinks)
                  pending.setupLinkChain(tenured, listIndex, listCount);
              maxScanAndCheckpointMatches = Math.max(maxScanAndCheckpointMatches, scanDistance + maxCheckpointMatchCount);
          }
  
          savePendingCheckpointAndResetScanDistance(nextCheckpointIndex);
      }
  
      private void savePendingCheckpointAndResetScanDistance(int checkpointIndex)
      {
          // use the tail of checkpointListBuf to buffer ranges we plan to tenure
          ensureCapacity(tenured.count() + scan.watermark());
  
          scan.reset();
  
          if (isAccurate)
          {
              // TODO (low priority, efficiency): we can shift back the existing scanDistance if it's far enough from
              //  the next checkpoint. this might permit us to skip some comparisons
              scan.resetPeakMax(tenured);
              for (Tenured<Ranges, Range, RoutingKey> tenured : this.tenured)
              {
                  int distanceToEnd = (tenured.lastIndex - checkpointIndex);
                  if (distanceToEnd >= scan.peakMax)
                      break;
  
                  int scanDistance = tenured.lastIndex - tenured.index;
                  if (scanDistance <= scan.peakMax)
                      scan.updateScanDistance(tenured.index, scanDistance, null);
              }
  
              if (scan.watermark() < scan.goal)
              {
                  int ri = Scan.minScanIndex(checkpointIndex, scan.goal);
                  while (ri < checkpointIndex)
                  {
                      RoutingKey end = accessor.end(ranges, ri);
                      int scanLimit = scanLimit(ri, scan.peakMax);
                      if (!shouldTenure(end, scanLimit))
                          scan.update(end, ri, ranges, scanLimit, null);
                      ++ri;
                  }
              }
          }
          else
          {
              // the maximum scan distance that could ever have been adopted for last chunk
              int oldPeakMax = scan.peakMax();
              // the minimum scan distance we will start with for processing the proceeding ranges
              // note: this may increase if we decide to tenure additional ranges, at which point it will be the actual newPeakMax
              int newMinPeakMax = scan.newPeakMax(tenured);
              int minUntenuredIndex = scan.minUntenuredIndex(checkpointIndex, tenured);
              int minScanIndex = Scan.minScanIndex(checkpointIndex, newMinPeakMax);
  
              // we now make sure tenured and scan are correct for the new parameters.
              // 1) if our peakMax is lower then we need to go back and find items to tenure that we previously marked for scanning
              // 2) we must also reset our scan distances
  
              // since our peakMax is determined by tenured.count(), but we are tenuring items here we keep things simple
              // and do not account for those items we tenure but would later permit to scan as our peakMax grows
  
              int ri = Math.min(minUntenuredIndex, minScanIndex);
              while (ri < checkpointIndex)
              {
                  RoutingKey end = accessor.end(ranges, ri);
                  int newPeakMax = scan.newPeakMax(tenured);
                  int scanLimit = scanLimit(ri, newPeakMax);
                  if (shouldTenure(end, scanLimit))
                  {
                      // note: might have already been tenured
                      // in this case our untenureLimit may be incorrect, but we won't use it
                      if (ri >= minUntenuredIndex && newPeakMax < oldPeakMax)
                          tenured.tenure(end, ri, ranges, scanLimit + 1, scanLimit(ri, oldPeakMax));
                  }
                  else
                  {
                      // this might effectively remove a previously tenured item
                      scan.update(end, ri, ranges, scanLimit, null);
                  }
                  ++ri;
              }
  
              scan.resetPeakMax(tenured);
          }
  
          pending.atIndex = checkpointIndex;
          pending.clear();
          tenured.rollover(pending);
      }
  
      private int extendScanDistance(int lastIndex, int scanDistance)
      {
          // now we've established our lower bound on scan distance, see how many checkpoints we can remove
          // by increasing our scan distance so that it remains proportional to the number of results returned
          // TODO (low priority, efficiency): can reduce cost here by using scanDistances array for upper bounds to scan distance
          int maxScanDistance = scan.goal() + 2 * Math.min(tenured.count(), tenured.countAtPrevCheckpoint());
          if (maxScanDistance >= 1 + scanDistance + scanDistance/4 && pending.count() >= (maxScanDistance - scanDistance)/2)
          {
              int removeCount = 0;
              int extendedScanDistance = scanDistance;
              int target = (maxScanDistance - scanDistance)/2;
              for (int i = 0 ; i < pending.count() ; ++i)
              {
                  Tenured<Ranges, Range, RoutingKey> t = pending.get(i);
                  if (t.index < 0)
                      continue;
  
                  int distance = Math.min(lastIndex, t.lastIndex) - t.index;
                  if (distance <= scanDistance)
                      continue; // already scanned or untenured
  
                  if (distance <= maxScanDistance)
                  {
                      ++removeCount;
                      extendedScanDistance = Math.max(extendedScanDistance, distance);
                      if (extendedScanDistance == maxScanDistance && removeCount >= target)
                          break;
                  }
              }
  
              // TODO (low priority, efficiency): should perhaps also gate this decision on the span we're covering
              //  algorithmically, however, so long as we are under maxScanDistance we are fine
              if (removeCount >= (extendedScanDistance - scanDistance)/2)
                  scanDistance = extendedScanDistance;
          }
          return scanDistance;
      }
  
      int writeList(PendingCheckpoint<Ranges, Range, RoutingKey> pending)
      {
          int startIndex = listCount;
          for (int i = pending.count() - 1 ; i >= 0 ; --i)
          {
              Tenured<Ranges, Range, RoutingKey> t = pending.get(i);
              if (t.index >= 0)
              {
                  lists[listCount++] = t.index;
              }
              else
              {
                  int index = t.index & ~BIT31;
                  int length = t.linkLength & ~BIT31;
                  Invariants.require(length < BIT30);
                  if (length <= 0xff && index <= 0xfffff)
                  {
                      lists[listCount++] = BIT31 | BIT30 | (length << 20) | index;
                  }
  //                This logic has been disabled.
  //                It is possible for some pathological cases to have the checkpoint lists run into one another.
  //                So that the search is not guaranteed to stop based on comparison, and an end index can only be computed
  //                by scanning the checkpoint list array, which is fine but not worth the complexity or saving 4 bytes
  //                else if (t.linkLength >= 0)
  //                {
  //                    lists[listCount++] = BIT31 | BIT29 | index;
  //                }
                  else
                  {
                      lists[listCount++] = BIT31 | length;
                      lists[listCount++] = BIT31 | index;
                  }
              }
          }
          return startIndex;
      }
  
      void writeHeader(int scanDistance, int lowerBound)
      {
          int headerScanDistance = Math.min(scanDistance, MAX_SCAN_DISTANCE);
  
          if ((checkpointCount & 3) == 0)
              headers[headerPointer++] = headerScanDistance;
          else
              headers[headerPointer - (1 + (checkpointCount & 3))] |= headerScanDistance << (8 * (checkpointCount & 3));
  
          bounds[checkpointCount++] = lowerBound;
          headers[headerPointer++] = listCount;
  
          if (scanDistance >= MAX_SCAN_DISTANCE)
              lists[listCount++] = -scanDistance; // serialize as a negative value so we ignore it in most cases automatically
      }
  
      void closeHeaders()
      {
          // write our final checkpoint header
          if ((checkpointCount & 3) == 0) headers[headerPointer++] = 0;
          headers[headerPointer++] = listCount;
      }
  
      void ensureCapacity(int maxPendingSize)
      {
          if (listCount + maxPendingSize >= lists.length)
              lists = cachedInts().resize(lists, listCount, lists.length + lists.length/2 + maxPendingSize);
      }
  
      static class Scan<Ranges, Range, RoutingKey>
      {
          final Accessor<Ranges, Range, RoutingKey> accessor;
          /** the scan distance we are aiming for; should be proportional to log2(N) */
          int goal;
  
          /** the indexes at which we increased the scan distance, and the new scan distance */
          int[] distances = new int[16];
          /** the number of unique scan distances we have adopted since the last checkpoint */
          int count;
          /** the highest scan distance we have adopted (==scanDistance(scanDistanceCount-1)) */
          int watermark;
          /**
           * the first index we have tenured a range from, but for which we did not immediately write a new checkpoint
           * we *must* scan at least from the last index in the checkpoint to here
           */
          int scanTenuredAtIndex = -1;
  
          /** The maximum (i.e. initial) scan distance limit we have used since the last attempted checkpoint write */
          int peakMax;
  
          Scan(Accessor<Ranges, Range, RoutingKey> accessor)
          {
              this.accessor = accessor;
          }
  
          void init(int goalScanDistance)
          {
              goal = peakMax = goalScanDistance;
          }
  
          private void update(RoutingKey end, int atIndex, Ranges ranges, int scanLimit, CheckpointIntervalArrayBuilder<Ranges, Range, RoutingKey> checkpoint)
          {
              int newScanDistance = find(end, atIndex, ranges, scanLimit, watermark);
              updateScanDistance(atIndex, newScanDistance, checkpoint);
          }
  
          private void updateScanDistance(int atIndex, int newScanDistance, CheckpointIntervalArrayBuilder<Ranges, Range, RoutingKey> checkpoint)
          {
              if (newScanDistance > watermark)
              {
                  // TODO (desired, efficiency): we don't mind slight increases to the watermark;
                  //  should really look at scan distance history and ensure we haven't e.g. doubled since
                  //  some earlier point (and should track the match count + scan distance at each bump
                  //  to check overall work hasn't increased too much)
                  if (checkpoint != null && checkpoint.canWriteCheckpoint(atIndex))
                      checkpoint.writeCheckpoint(atIndex);
  
                  watermark = newScanDistance;
                  if (count * 2 == distances.length)
                      distances = Arrays.copyOf(distances, distances.length * 2);
                  distances[count * 2] = newScanDistance;
                  distances[count * 2 + 1] = atIndex;
                  ++count;
              }
          }
  
          private int find(RoutingKey end, int atIndex, Ranges ranges, int scanLimit, int currentScanDistance)
          {
              var c = accessor.keyComparator();
              int lowerIndex = accessor.binarySearch(ranges, atIndex + currentScanDistance, scanLimit, end, (e, s) -> c.compare(e, accessor.start(s)), CEIL);
              if (lowerIndex < 0) lowerIndex = -2 - lowerIndex;
              else lowerIndex -= 1;
              return lowerIndex - atIndex;
          }
  
          boolean isAboveGoal()
          {
              return watermark > goal;
          }
  
          int watermark()
          {
              return watermark;
          }
  
          int goal()
          {
              return goal;
          }
  
          int distanceToTenured(int lastIndex)
          {
              return scanTenuredAtIndex >= 0 ? lastIndex - scanTenuredAtIndex : 0;
          }
  
          boolean mustCheckpointToScanTenured(int checkpointIndex, int maxScanDistance)
          {
              return scanTenuredAtIndex >= 0 && checkpointIndex - scanTenuredAtIndex >= maxScanDistance;
          }
  
          /**
           * Are we scanning a much longer distance than the minimum number of matches we know a query will return?
           * Note: with Strategy.FAST, {@code tenured.count()} gets less accurate as scan distance increases, so this
           * will bounce around triggering checkpoints due to the larger scan distance, resetting the scan distance
           * and starting again
           */
          boolean hasMaybeDivergedFromMatchSize(TenuredSet<Ranges, Range, RoutingKey> tenured)
          {
              return isAboveGoal() && tenured.count() < watermark()/2;
          }
  
          private int distance(int i)
          {
              return distances[i*2];
          }
  
          private int index(int i)
          {
              return distances[i*2+1];
          }
  
          int finalise(int lastIndex)
          {
              Invariants.require(distanceToTenured(lastIndex) <= Math.max(watermark(), peakMax()));
  
              int scanDistance = watermark;
              // then, compute the minimum scan distance implied by any tenured ranges we did not immediately
              // write a checkpoint for - we *must* scan back as far as this record
              int minScanDistance = scanTenuredAtIndex >= 0 ? lastIndex - scanTenuredAtIndex : 0;
              if (minScanDistance > scanDistance)
              {
                  // if this minimum is larger than the largest scan distance we picked up for non-tenured ranges
                  // then we are done, as there's nothing we can save
                  scanDistance = minScanDistance;
              }
              else if (scanDistance > 0)
              {
                  // otherwise, we can look to see if any of the scan distances we computed overflow the checkpoint,
                  // i.e. where no records served by this checkpoint need to scan the full distance to reach it
                  int distanceToLastScanIndex = lastIndex - index(count -1);
                  // if the distance to the last scan index is larger than its scan distance, we have overflowed;
                  if (distanceToLastScanIndex < scanDistance)
                  {
                      minScanDistance = Math.max(distanceToLastScanIndex, minScanDistance);
                      // loop until we find one that doesn't overflow, as this is another minimum scan distance
                      int i = count - 1;
                      while (--i >= 0)
                      {
                          int distance = lastIndex - index(i);
                          if (distance >= distance(i)) break;
                          else if (distance > minScanDistance) minScanDistance = distance;
                      }
                      if (i >= 0) scanDistance = Math.max(minScanDistance, distance(i));
                      else scanDistance = minScanDistance;
                  }
              }
  
              return scanDistance;
          }
  
          void reset()
          {
              // we could in theory reset our scan distance using the contents of scanDistance[]
              // but it's a bit complicated, as we want to have the first item to increment the scan distance
              // so that we can use it in writeScanDistance to shrink the scan distance;
              // jumping straight to the highest scan distance breaks this
              count = 0;
              scanTenuredAtIndex = -1;
              watermark = 0;
          }
  
          void resetPeakMax(TenuredSet<Ranges, Range, RoutingKey> tenured)
          {
              peakMax = newPeakMax(tenured);
          }
  
          int peakMax()
          {
              return peakMax;
          }
  
          int newPeakMax(TenuredSet<Ranges, Range, RoutingKey> tenured)
          {
              return Math.max(goal, tenured.count());
          }
  
          /**
           * The minimum index containing a range that might need to be tenured, if we have a smaller max scan distance than before
           */
          int minUntenuredIndex(int checkpointIndex, TenuredSet<Ranges, Range, RoutingKey> tenured)
          {
              int minUntenuredIndex = Math.max(0, (checkpointIndex - 1) - watermark());
              // the maximum scan distance that cxould ever have been adopted for the ranges processed since last checkpoint
              int oldPeakMax = peakMax;
              int newMinPeakMax = newPeakMax(tenured);
              if (newMinPeakMax < oldPeakMax)
              {
                  // minimise range we unnecessarily re-tenure over
                  // TODO (low priority, efficiency): see if can also use to reduce range we re-scan e.g. can recycle
                  //  scanDistances contents if we know we won't need to step back further at next checkpoint
                  for (int i = count - 1; i >= 0 ; --i)
                  {
                      if (index(i) <= minUntenuredIndex)
                          break;
                      if (distance(i) <= newMinPeakMax)
                          return i + 1 == count ? index(i) : index(i + 1) - 1;
                  }
              }
              return minUntenuredIndex;
          }
  
          /**
           * Record that a range at this index has been tenured, so that we can track how far back
           * we need to scan to determine how long we can defer writing a new checkpoint while still
           * being able to scan it.
           *
           * TODO (low priority, efficiency): when a checkpoint is written, we should consider moving it
           *  earlier if the scan distance is increased primarily because of this index, and the tenured
           *  collection is otherwise unchanged (so can be written with minimal overhead)
           */
          void tenured(int atIndex)
          {
              if (scanTenuredAtIndex < 0)
                  scanTenuredAtIndex = atIndex;
          }
  
          static int minScanIndex(int checkpointIndex, int scanDistance)
          {
              return Math.max(0, (checkpointIndex - 1) - scanDistance);
          }
  
          @Override
          public String toString()
          {
              return "Scan{watermark=" + watermark + ", tenured=" + scanTenuredAtIndex + '}';
          }
      }
  
      /**
       * Record-keeping for a range we have decided is not scannable
       */
      static class Tenured<Ranges, Range, RoutingKey> implements Comparable<Tenured<Ranges, Range, RoutingKey>>
      {
          final Accessor<Ranges, Range, RoutingKey> accessor;
          /**
           * The end of the tenured range covered by the contents referred to be {@link #index}
           */
          RoutingKey end;
  
          /**
           * <ul>
           * <li>If positive, this points to {@code ranges[index]}</li>
           * <li>If negative, this points to an entry in {@link #lists};
           * see {@link SearchableRangeList#checkpointLists}</li>
           * </ul>
           */
          int index;
  
          /**
           * The last index in {@link #ranges} covered by this tenured range
           */
          int lastIndex;
  
          /**
           * set when this record is serialized in a checkpoint list to either:
           * <ul>
           * <li>point to itself, in which case no action should be
           *     taken on removal (it is only retained for size bookkeeping); or</li>
           * <li>point to the next item in the checkpoint list; the first
           *     such element removed triggers the clearing of the checkpoint
           *     list so that its entries are re-inserted in the next checkpoint</li>
           * </ul>
           */
          Tenured<Ranges, Range, RoutingKey> next;
  
          /**
           * Only set for link entries, i.e. where {@code index < 0}.
           * <ul>
           * <li>if positive, the length is optional as we will terminate safely using the end bound filtering</li>
           * <li>if negative, the low 31 bits <b>must</b> be retrieved as the length for safe iteration</li>
           * </ul>
           */
          int linkLength;
  
          Tenured(Accessor<Ranges, Range, RoutingKey> accessor, RoutingKey end, int index)
          {
              this.accessor = accessor;
              this.end = end;
              this.index = index;
          }
  
          @Override
          public int compareTo(@Nonnull Tenured<Ranges, Range, RoutingKey> that)
          {
              int c = accessor.keyComparator().compare(this.end, that.end);
              // we sort indexes in reverse order so later tenured items find the earlier ones with same end when searching
              // for higher entries for the range of indexes to search, and
              if (c == 0) c = -Integer.compare(this.index, that.index);
              return c;
          }
  
          @Override
          public String toString()
          {
              return "Tenured{end=" + end + ", index=" + index + '}';
          }
      }
  
      /**
       * The set of ranges that we intend to write to checkpoints that remain open at the current point in the iteration
       * This collection may be filtered before serialization, but every member will be visited either by scanning
       * or visiting the checkpoint list
       * TODO (low priority, efficiency): save garbage by using an insertion-sorted array for collections where
       *  this is sufficient. later, introduce a mutable b-tree supporting object recycling. we would also like
       *  to use a collection that permits us to insert and return a finger into the tree so we can find the
       *  successor as part of insertion, and that permits constant-time first() calls
       */
      static class TenuredSet<Ranges, Range, RoutingKey> extends TreeSet<Tenured<Ranges, Range, RoutingKey>>
      {
          final Accessor<Ranges, Range, RoutingKey> accessor;
          /**
           * the number of direct tenured entries (i.e. ignoring link entries)
           * this is used to provide a minimum bound on the number of results a range query can return
           * note: with Strategy.FAST this gets less accurate as the span distance increases
           */
          int directCount;
          int directCountAtPrevCheckpoint;
          int minSpan;
  
          // a stack of recently used EndAndIndex objects - used only for the duration of a single build
          Tenured<Ranges, Range, RoutingKey> reuse, pendingReuse, pendingReuseTail;
  
          TenuredSet(Accessor<Ranges, Range, RoutingKey> accessor)
          {
              this.accessor = accessor;
          }
  
          int count()
          {
              return directCount;
          }
  
          int countAtPrevCheckpoint()
          {
              return directCountAtPrevCheckpoint;
          }
  
          /**
           * We require a checkpoint to cover a distance at least as large as the number of tenured ranges leftover
           * since the prior checkpoint, to ensure these require at most linear additional space, while not requiring
           * more than O(k) additional complexity on search (i.e., we will scan a number of elements at most equal
           * to the number we have to visit in the checkpoint).
           *
           * We achieve this by recording the minimum number of match results as of the prior checkpoint (i.e. {@link #count()})
           * and discounting it by one each time we untenure a range, so that for each tenured range from the prior checkpoint
           * we have either untenured a range or processed at least one additional input.
           */
          int minimumSpan()
          {
              return minSpan;
          }
  
          private int tenure(RoutingKey end, int index, Ranges ranges, int minUntenureIndex)
          {
              return tenure(newTenured(end, index), ranges, minUntenureIndex, accessor.size(ranges));
          }
  
          private void tenure(RoutingKey end, int index, Ranges ranges, int minUntenureIndex, int untenureLimit)
          {
              tenure(newTenured(end, index), ranges, minUntenureIndex, untenureLimit);
          }
  
          private int tenure(Tenured<Ranges, Range, RoutingKey> tenure, Ranges ranges, int untenureMinIndex, int untenureLimit)
          {
              if (!add(tenure))
                  return tenure.lastIndex;
  
              Tenured<Ranges, Range, RoutingKey> next = higher(tenure);
              if (next != null)
                  untenureLimit = Math.min(untenureLimit, next.lastIndex + 1);
              var c = accessor.keyComparator();
              int untenureIndex = accessor.binarySearch(ranges, untenureMinIndex, untenureLimit, tenure.end, (e, s) -> c.compare(e, accessor.start(s)), CEIL);
              if (untenureIndex < 0) untenureIndex = -1 - untenureIndex;
              tenure.lastIndex = untenureIndex - 1;
              Invariants.require(c.compare(tenure.end, accessor.start(ranges, tenure.lastIndex)) > 0);
              Invariants.require(tenure.lastIndex + 1 == accessor.size(ranges) || c.compare(tenure.end, accessor.start(ranges, tenure.lastIndex + 1)) <= 0);
              ++directCount;
              return untenureIndex - 1;
          }
  
          private Tenured<Ranges, Range, RoutingKey> newTenured(RoutingKey end, int index)
          {
              Tenured<Ranges, Range, RoutingKey> result = reuse;
              if (result == null)
                  return new Tenured<>(accessor, end, index);
  
              reuse = result.next;
              result.end = end;
              result.index = index;
              result.lastIndex = 0;
              result.next = null;
              return result;
          }
  
          private Tenured<Ranges, Range, RoutingKey> addLinkEntry(RoutingKey end, int index, int lastIndex, int length)
          {
              Invariants.requireArgument(index < 0);
              Tenured<Ranges, Range, RoutingKey> result = newTenured(end, index);
              result.linkLength = length;
              result.lastIndex = lastIndex;
              add(result);
              return result;
          }
  
          /**
           * Retire any active tenured ranges that no longer cover the pointer into ranges;
           * if this crosses our checkpoint threshold, write a new checkpoint.
           */
          void untenure(int index)
          {
              while (!isEmpty() && first().lastIndex < index)
              {
                  Tenured<Ranges, Range, RoutingKey> removed = pollFirst();
  
                  // if removed.next == null, this is not referenced by a link
                  // if removed.next == removed, it is referenced by a link but does not modify the link on removal
                  if (removed.next != null && removed.next != removed)
                  {
                      // this is a member of a link's chain, which may serve one of two purposes:
                      // 1) it may be the entry nominated to invalidate the link, due to the link
                      //    membership shrinking below the required threshold; in which case we
                      //    must clear the chain to reactivate its members for insertion into the
                      //    next checkpoint, and remove the chain link itself
                      // 2) it may be nominated as an entry to update the chain link info, to make
                      //    it more succinct: if every entry of the chain remains active, and there
                      //    are *many* entries then we need two integers to represent the chain, but
                      //    as soon as any entry is invalid we can rely on this entry to terminate
                      //    iteration, so we update the bookkeeping on the first entry we remove in
                      //    this case
  
                      // first clear the chain starting at the removed entry
                      Tenured<Ranges, Range, RoutingKey> prev = removed, next = removed.next;
                      while (next.next != null)
                      {
                          prev = next;
                          next = next.next;
                          prev.next = null;
                      }
                      Invariants.require(next.index < 0);
                      if (prev.end == next.end)
                      {
                          // if this is the last entry in the link, the link is expired and should be removed/reused
                          remove(next);
                          if (pendingReuseTail == null)
                              pendingReuseTail = next;
                          next.next = pendingReuse;
                          pendingReuse = next;
                      }
                      else if (next.linkLength < 0)
                      {
                          // otherwise, flag the link as safely consumed without knowing the length
                          next.linkLength = next.linkLength & Integer.MAX_VALUE;
                      }
                  }
  
                  // this was not a link reference; update our bookkeeping and save it for reuse
                  Invariants.require(removed.index >= 0);
                  --directCount;
                  --minSpan;
                  if (pendingReuseTail == null)
                      pendingReuseTail = removed;
                  removed.next = pendingReuse;
                  pendingReuse = removed;
              }
          }
  
          /**
           * Write out any direct entries that are not pointed to by a chain entry, and any chain entries;
           * rollover any per-checkpoint data and free up for reuse discarded Tenured objects
           */
          void rollover(PendingCheckpoint<Ranges, Range, RoutingKey> pending)
          {
              for (Tenured<Ranges, Range, RoutingKey> tenured : this)
              {
                  if (tenured.next == null)
                      pending.add(tenured);
              }
              // make freed Tenured objects available for reuse
              if (pendingReuse != null)
              {
                  pendingReuseTail.next = reuse;
                  reuse = pendingReuse;
                  pendingReuseTail = pendingReuse = null;
              }
              directCountAtPrevCheckpoint = minSpan = directCount;
          }
      }
  
      /**
       * we write checkpoints out before knowing the scan distance needed for the range, as a checkpoint precedes
       * the ranges it covers; so we record the position and contents of the checkpoint, and once the scan distance is
       * known (i.e. when the next checkpoint is written) we re-process the list to remove items we can now scan before
       * serializing to checkpointListsBuf.
       */
      static class PendingCheckpoint<Ranges, Range, RoutingKey>
      {
          int atIndex = -1;
          int count;
  
          Tenured<Ranges, Range, RoutingKey>[] contents = new Tenured[10];
  
          int openDirectCount, firstOpenDirect, openIndirectCount;
          boolean hasClosedDirect;
  
          int count()
          {
              return count;
          }
  
          Tenured<Ranges, Range, RoutingKey> get(int i)
          {
              return contents[i];
          }
  
          void add(Tenured<Ranges, Range, RoutingKey> tenured)
          {
              if (contents.length == count)
                  contents = Arrays.copyOf(contents, 2 * contents.length);
              contents[count++] = tenured;
          }
  
          void clear()
          {
              count = 0;
          }
  
          /**
           * Remove pending entries that will be scanned by the scanDistance, and update
           * our bookkeeping for creating links
           */
          int filter(int scanDistance, int lastIndex)
          {
              int matchCountModifier = 0;
              int maxi = count;
              count = 0;
              openDirectCount = 0;
              openIndirectCount = 0;
              firstOpenDirect = -1;
  //            lastClosedDirect = -1;
  
              for (int i = 0; i < maxi ; ++i)
              {
                  Tenured<Ranges, Range, RoutingKey> t = get(i);
                  if (t.index >= 0)
                  {
                      if (t.index + scanDistance >= lastIndex)
                          continue; // last index will find it with a scan
  
                      if (t.lastIndex <= t.index + scanDistance)
                          continue; // all indexes will find it with a scan
  
                      if (t.lastIndex > lastIndex)
                      {
                          // this range remains open for the next checkpoint;
                          // we may want to reference this list from there
                          // so track count and position of first one to make a determination
                          ++openDirectCount;
                          if (firstOpenDirect < 0)
                              firstOpenDirect = count;
                      }
                      else hasClosedDirect = true;
                  }
                  else
                  {
                      // note: we over count here, as we count pointers within the chain
                      matchCountModifier += (t.linkLength & Integer.MAX_VALUE) - 1; // (subtract 1 to discount the pointer)
                      if (t.lastIndex > lastIndex)
                          ++openIndirectCount;
                  }
  
                  if (i == count) ++count;
                  else contents[count++] = t;
              }
  
              return count + matchCountModifier;
          }
  
          /**
           * Setup a link for referencing this chain later, if permitted.
           * Must have at least two items, and at least as many direct records as indirect
           */
          void setupLinkChain(TenuredSet<Ranges, Range, RoutingKey> tenured, int startIndex, int endIndex)
          {
              int minSizeToReference = openIndirectCount + MIN_INDIRECT_LINK_LENGTH;
              if (openDirectCount >= minSizeToReference)
              {
                  int i = firstOpenDirect;
                  Tenured<Ranges, Range, RoutingKey> prev = get(i++);
  
                  while (openDirectCount > minSizeToReference)
                  {
                      Tenured<Ranges, Range, RoutingKey> e = get(i++);
                      if (e.index < 0)
                      {
                          --minSizeToReference;
                          continue;
                      }
  
                      Invariants.require(prev.next == null);
                      prev.next = prev;
                      prev = e;
                      --openDirectCount;
                  }
  
                  while (i < count)
                  {
                      Tenured<Ranges, Range, RoutingKey> next = get(i++);
                      if (next.index < 0)
                          continue;
  
                      Invariants.require(prev.next == null);
                      prev.next = next;
                      prev = next;
                  }
  
                  // may be more than one entry per item (though usually not)
                  int length = endIndex - startIndex;
                  Tenured<Ranges, Range, RoutingKey> chainEntry = tenured.addLinkEntry(prev.end, BIT31 | startIndex, prev.lastIndex, length);
                  prev.next = chainEntry;
                  if (hasClosedDirect && (startIndex > 0xfffff || (length > 0xff)))
                  {
                      // TODO (expected, testing): make sure this is tested, as not a common code path (may never be executed in normal operation)
                      // we have no closed ranges so iteration needs to know the end bound, but we cannot encode our bounds cheaply
                      // so link the first bound to the chain entry, so that on removal it triggers an update of endIndex to note
                      // that it can be iterated safely without an end bound
                      get(firstOpenDirect).next = chainEntry;
                  }
              }
          }
  
          @Override
          public String toString()
          {
              return Arrays.stream(contents, 0, count)
                           .map(Objects::toString)
                           .collect(Collectors.joining(",", "[", "]"));
          }
      }
  }
  