Faster sliding for IndexedSeq

The default implementation of `Iterable.sliding` via `Iterator.sliding`
is rather inefficient and can benefit greatly from `IndexedSeq`-specific
optimizations. The new `IndexedSeqSlidingIterator` provides a
`slice`-based implementation of the basic functionality without any of
the additional features of `GroupedIterator` (which is only exposed as a
return type in `Iterator` but not in `Iterable`).

Co-authored-by: Som Snytt <som.snytt@gmail.com>

Author: szeiger

library/src/scala/collection/IndexedSeq.scala

@@ -91,6 +91,11 @@ trait IndexedSeqOps[+A, +CC[_], +C] extends Any with SeqOps[A, CC, C] { self =>
 
   override def slice(from: Int, until: Int): C = fromSpecific(new IndexedSeqView.Slice(this, from, until))
 
+  override def sliding(size: Int, step: Int): Iterator[C] = {
+    require(size >= 1 && step >= 1, f"size=$size%d and step=$step%d, but both must be positive")
+    new IndexedSeqSlidingIterator[A, CC, C](this, size, step)
+  }
+
   override def head: A =
     if (!isEmpty) apply(0)
     else throw new NoSuchElementException(s"head of empty ${
@@ -145,3 +150,26 @@ trait IndexedSeqOps[+A, +CC[_], +C] extends Any with SeqOps[A, CC, C] { self =>
     }
   }
 }
+
+/** A fast sliding iterator for IndexedSeqs which uses the underlying `slice` operation. */
+private final class IndexedSeqSlidingIterator[A, CC[_], C](s: IndexedSeqOps[A, CC, C], size: Int, step: Int)
+  extends AbstractIterator[C] {
+
+  private[this] val len = s.length
+  private[this] var pos = 0
+  private def chklen: Boolean = len == s.length || {
+    throw new java.util.ConcurrentModificationException("collection size changed during iteration")
+    false
+  }
+
+  def hasNext: Boolean = chklen && pos < len
+
+  def next(): C = if (!chklen || !hasNext) Iterator.empty.next() else {
+    val end = { val x = pos + size; if (x < 0 || x > len) len else x } // (pos.toLong + size).min(len).toInt
+    val slice = s.slice(pos, end)
+    pos =
+      if (end >= len) len
+      else { val x = pos + step; if (x < 0 || x > len) len else x } // (pos.toLong + step).min(len).toInt
+    slice
+  }
+}

library/src/scala/collection/mutable/ArrayBuffer.scala

@@ -269,9 +269,16 @@ class ArrayBuffer[A] private (initialElements: Array[AnyRef], initialSize: Int)
 
   override def foldRight[B](z: B)(op: (A, B) => B): B = foldr(0, length, z, op)
 
-  override def reduceLeft[B >: A](op: (B, A) => B): B = if (length > 0) foldl(1, length, array(0).asInstanceOf[B], op) else super.reduceLeft(op)
+  override def reduceLeft[B >: A](op: (B, A) => B): B =
+    if (length > 0) foldl(1, length, array(0).asInstanceOf[B], op)
+    else super.reduceLeft(op)
 
-  override def reduceRight[B >: A](op: (A, B) => B): B = if (length > 0) foldr(0, length - 1, array(length - 1).asInstanceOf[B], op) else super.reduceRight(op)
+  override def reduceRight[B >: A](op: (A, B) => B): B =
+    if (length > 0) foldr(0, length - 1, array(length - 1).asInstanceOf[B], op)
+    else super.reduceRight(op)
+
+  override def sliding(size: Int, step: Int): Iterator[ArrayBuffer[A]] =
+    new MutationTracker.CheckedIterator(super.sliding(size = size, step = step), mutationCount)
 }
 
 /**

library/src/scala/collection/mutable/ArrayDeque.scala

@@ -633,16 +633,8 @@ trait ArrayDequeOps[A, +CC[_], +C <: AnyRef] extends StrictOptimizedSeqOps[A, CC
     }
   }
 
-  override def sliding(window: Int, step: Int): Iterator[C] = {
-    require(window > 0 && step > 0, s"window=$window and step=$step, but both must be positive")
-    length match {
-      case 0 => Iterator.empty
-      case n if n <= window => Iterator.single(slice(0, length))
-      case n =>
-        val lag = if (window > step) window - step else 0
-        Iterator.range(start = 0, end = n - lag, step = step).map(i => slice(i, i + window))
-    }
-  }
+  override def sliding(@deprecatedName("window") size: Int, step: Int): Iterator[C] =
+    super.sliding(size = size, step = step)
 
   override def grouped(n: Int): Iterator[C] = sliding(n, n)
 }