PMM-347 coordinator support (nearly full), Scala 2.13

Author: trolley813

build.sbt

@@ -7,8 +7,7 @@ dockerUsername in Docker := Some("clovergrp")
 dockerUpdateLatest := true
 dockerAlias in Docker := dockerAlias.value.withTag(dockerAlias.value.tag.map(_.replace("+", "_")))
 
-// Flink currently does not work with Scala 2.12.8+
-scalaVersion in ThisBuild := "2.12.7"
+scalaVersion in ThisBuild := "2.13.8"
 resolvers in ThisBuild ++= Seq(
   "Apache Development Snapshot Repository" at "https://repository.apache.org/content/repositories/snapshots/",
   Resolver.mavenLocal,
@@ -217,7 +216,7 @@ unmanagedResourceDirectories in Compile -= (resourceDirectory in Compile).value
 // Kind projector
 resolvers += Resolver.sonatypeRepo("releases")
 //addCompilerPlugin("org.spire-math" %% "kind-projector" % Version.kindProjector)
-addCompilerPlugin("org.scalamacros" % "paradise" % "2.1.1" cross CrossVersion.full)
+//addCompilerPlugin("org.scalamacros" % "paradise" % "2.1.1" cross CrossVersion.full)
 
 
 // Git-specific settings

core/src/main/scala/ru/itclover/tsp/core/PQueue.scala

@@ -73,6 +73,7 @@ object PQueue {
       val buffer = scala.collection.mutable.ArrayBuffer.empty[IdxValue[T]]
       import scala.collection.convert.ImplicitConversionsToScala._
       buffer ++= queue.iterator()
+      buffer.toSeq
     }
     override def size: Int = queue.size
 
@@ -140,7 +141,7 @@ object PQueue {
     )
     override def clean(): PQueue[T] = this.copy(queue = queue.clean())
 
-    override def toSeq: Seq[IdxValue[T]] = queue.toSeq.view.map(x => x.map(_ => func(x)))
+    override def toSeq: Seq[IdxValue[T]] = queue.toSeq.view.map(x => x.map(_ => func(x))).toSeq
 
     override def rewindTo(newStart: Idx): PQueue[T] = this.copy(queue = queue.rewindTo(newStart))
   }

core/src/main/scala/ru/itclover/tsp/core/QueueUtils.scala

@@ -9,26 +9,26 @@ object QueueUtils {
 
   private val trueFunction = (_: Any) => true
 
-  def takeWhileFromQueue[A](queue: m.Queue[A])(predicate: A => Boolean = trueFunction): (m.Queue[A], m.Queue[A]) =
-    if (predicate.eq(trueFunction)) (queue, m.Queue.empty)
+  def takeWhileFromQueue[A](queue: m.ArrayDeque[A])(predicate: A => Boolean = trueFunction): (m.ArrayDeque[A], m.ArrayDeque[A]) =
+    if (predicate.eq(trueFunction)) (queue, m.ArrayDeque.empty)
     else {
 
       @tailrec
-      def inner(result: m.Queue[A], q: m.Queue[A]): (m.Queue[A], m.Queue[A]) =
+      def inner(result: m.ArrayDeque[A], q: m.ArrayDeque[A]): (m.ArrayDeque[A], m.ArrayDeque[A]) =
         q.headOption match {
-          case Some(x) if predicate(x) => inner({ result.enqueue(x); result }, { q.dequeue(); q })
+          case Some(x) if predicate(x) => inner({ result.append(x); result }, { q.removeHead(); q })
           case _                       => (result, q)
         }
 
-      inner(m.Queue.empty, queue)
+      inner(m.ArrayDeque.empty, queue)
     }
 
   /**
     * Splits inner `q` at point idx, so all records with id < idx are in first returned queue, and all with id >= idx are in second.
     */
-  def splitAtIdx(q: m.Queue[(Idx, Time)], idx: Idx, marginToFirst: Boolean = false)(
+  def splitAtIdx(q: m.ArrayDeque[(Idx, Time)], idx: Idx, marginToFirst: Boolean = false)(
     implicit ord: Order[Idx]
-  ): (m.Queue[(Idx, Time)], m.Queue[(Idx, Time)]) = {
+  ): (m.ArrayDeque[(Idx, Time)], m.ArrayDeque[(Idx, Time)]) = {
     takeWhileFromQueue(q) {
       if (marginToFirst) {
         case (idx1: Idx, _: Time) => ord.lteqv(idx1, idx)

core/src/main/scala/ru/itclover/tsp/core/Result.scala

@@ -1,6 +1,6 @@
 package ru.itclover.tsp.core
 
-sealed trait Result[+A] {
+sealed trait Result[+A] extends Serializable {
   private def get: A = this match {
     case Fail    => throw new RuntimeException("Illegal get on Fail")
     case Succ(t) => t

core/src/main/scala/ru/itclover/tsp/core/aggregators/AccumPattern.scala

@@ -19,7 +19,7 @@ case class AggregatorPState[InnerState, InnerOut, AState](
   innerState: InnerState,
   innerQueue: PQueue[InnerOut],
   astate: AState,
-  indexTimeMap: m.Queue[(Idx, Time)]
+  indexTimeMap: m.ArrayDeque[(Idx, Time)]
 )
 
 abstract class AccumPattern[Event: IdxExtractor: TimeExtractor, InnerState, InnerOut, Out, AState <: AccumState[
@@ -39,7 +39,7 @@ abstract class AccumPattern[Event: IdxExtractor: TimeExtractor, InnerState, Inne
   ): F[(AggregatorPState[InnerState, InnerOut, AState], PQueue[Out])] = {
 
     val idxTimeMapWithNewEvents =
-      event.foldLeft(state.indexTimeMap) { case (a, b) => a.enqueue(b.index -> b.time); a }
+      event.foldLeft(state.indexTimeMap) { case (a, b) => a.append(b.index -> b.time); a }
 
     inner
       .apply[F, Cont](state.innerState, state.innerQueue, event)
@@ -63,8 +63,8 @@ abstract class AccumPattern[Event: IdxExtractor: TimeExtractor, InnerState, Inne
     innerQueue: QI[InnerOut],
     accumState: AState,
     results: QI[Out],
-    indexTimeMap: m.Queue[(Idx, Time)]
-  ): (QI[InnerOut], AState, QI[Out], m.Queue[(Idx, Time)]) = {
+    indexTimeMap: m.ArrayDeque[(Idx, Time)]
+  ): (QI[InnerOut], AState, QI[Out], m.ArrayDeque[(Idx, Time)]) = {
     innerQueue.dequeueOption() match {
       case None                                               => (innerQueue, accumState, results, indexTimeMap)
       case Some((iv @ IdxValue(start, end, _), updatedQueue)) =>
@@ -91,5 +91,5 @@ trait AccumState[In, Out, Self <: AccumState[In, Out, Self]] extends Product wit
     * @param idxValue - result from inner pattern.
     * @return Tuple of updated state and queue of results to be emitted from this pattern.
     */
-  def updated(window: Window, times: m.Queue[(Idx, Time)], idxValue: IdxValue[In]): (Self, QI[Out])
+  def updated(window: Window, times: m.ArrayDeque[(Idx, Time)], idxValue: IdxValue[In]): (Self, QI[Out])
 }

core/src/main/scala/ru/itclover/tsp/core/aggregators/GroupPattern.scala

@@ -33,12 +33,12 @@ case class GroupPattern[Event: IdxExtractor: TimeExtractor, S, T: Group](
     )
 }
 
-case class GroupAccumState[T: Group](lastValue: Option[GroupAccumResult[T]], windowQueue: m.Queue[GroupAccumValue[T]])
+case class GroupAccumState[T: Group](lastValue: Option[GroupAccumResult[T]], windowQueue: m.ArrayDeque[GroupAccumValue[T]])
     extends AccumState[T, GroupAccumResult[T], GroupAccumState[T]] {
 
   override def updated(
     window: Window,
-    times: m.Queue[(Idx, Time)],
+    times: m.ArrayDeque[(Idx, Time)],
     idxValue: IdxValue[T]
   ): (GroupAccumState[T], QI[GroupAccumResult[T]]) = {
 
@@ -57,9 +57,9 @@ case class GroupAccumState[T: Group](lastValue: Option[GroupAccumResult[T]], win
     window: Window,
     value: Result[T],
     lastValue: Option[GroupAccumResult[T]],
-    windowQueue: m.Queue[GroupAccumValue[T]],
+    windowQueue: m.ArrayDeque[GroupAccumValue[T]],
     outputQueue: QI[GroupAccumResult[T]]
-  ): (Option[GroupAccumResult[T]], m.Queue[GroupAccumValue[T]], QI[GroupAccumResult[T]]) = {
+  ): (Option[GroupAccumResult[T]], m.ArrayDeque[GroupAccumValue[T]], QI[GroupAccumResult[T]]) = {
     value
       .map { t =>
         val newLastValue = lastValue
@@ -77,7 +77,7 @@ case class GroupAccumState[T: Group](lastValue: Option[GroupAccumResult[T]], win
         }
 
         // add new element to queue
-        val finalWindowQueue = { updatedWindowQueue.enqueue(GroupAccumValue(idx, time, t)); updatedWindowQueue }
+        val finalWindowQueue = { updatedWindowQueue.append(GroupAccumValue(idx, time, t)); updatedWindowQueue }
 
         Tuple3(
           finalNewLastValue,

core/src/main/scala/ru/itclover/tsp/core/aggregators/PreviousValue.scala

@@ -27,7 +27,7 @@ case class PreviousValue[Event: IdxExtractor: TimeExtractor, State, Out](
 case class PreviousValueAccumState[T](queue: QI[(Time, T)]) extends AccumState[T, T, PreviousValueAccumState[T]] {
   override def updated(
     window: Window,
-    times: m.Queue[(Idx, Time)],
+    times: m.ArrayDeque[(Idx, Time)],
     idxValue: IdxValue[T]
   ): (PreviousValueAccumState[T], QI[T]) = {
     val (newQueue, newOutputQueue) =

core/src/main/scala/ru/itclover/tsp/core/aggregators/TimerPattern.scala

@@ -26,7 +26,7 @@ case class TimerPattern[Event: IdxExtractor: TimeExtractor, S, T](
 }
 
 case class TimerAccumState[T](
-  windowQueue: m.Queue[(Idx, Time)],
+  windowQueue: m.ArrayDeque[(Idx, Time)],
   lastEnd: (Idx, Time),
   lastValue: Result[T],
   eventsMaxGapMs: Long
@@ -35,7 +35,7 @@ case class TimerAccumState[T](
   @inline
   override def updated(
     window: Window,
-    times: m.Queue[(Idx, Time)],
+    times: m.ArrayDeque[(Idx, Time)],
     idxValue: IdxValue[T]
   ): (TimerAccumState[T], QI[Boolean]) = {
 
@@ -66,7 +66,7 @@ case class TimerAccumState[T](
         val end: Time = times.last._2 // time corresponding to the idxValue.end
 
         // don't use ++ here, slow!
-        val windowQueueWithNewPoints = times.foldLeft(windowQueue) { case (a, b) => a.enqueue(b); a }
+        val windowQueueWithNewPoints = times.foldLeft(windowQueue) { case (a, b) => a.append(b); a }
 
         // output fail on older points (before the end of the window)
         // but don't clean the whole queue
@@ -86,7 +86,7 @@ case class TimerAccumState[T](
         // if event chunk is shorter than the window, and the next window is sufficiently close,
         // then save it in the queue, and return empty state (since the later events can continue the window)
         if (cleanedWindowQueue.isEmpty && times.head._2.toMillis - lastEnd._2.toMillis < eventsMaxGapMs) {
-          updatedWindowQueue.enqueue(failOutputs: _*)
+          updatedWindowQueue.appendAll(failOutputs)
           (
             TimerAccumState(updatedWindowQueue, times.last, idxValue.value, eventsMaxGapMs),
             PQueue.empty[Boolean]

core/src/main/scala/ru/itclover/tsp/core/aggregators/TimestampsAdderPattern.scala

@@ -27,7 +27,7 @@ protected case class TimestampAdderAccumState[T]() extends AccumState[T, Segment
   @inline
   override def updated(
     window: Window,
-    times: m.Queue[(Idx, Time)],
+    times: m.ArrayDeque[(Idx, Time)],
     idxValue: IdxValue[T]
   ): (TimestampAdderAccumState[T], QI[Segment]) = {
     if (times.isEmpty) (this, PQueue.empty)

core/src/main/scala/ru/itclover/tsp/core/aggregators/WaitPattern.scala

@@ -26,7 +26,7 @@ case class WaitPattern[Event: IdxExtractor: TimeExtractor, S, T](
 
 // Here, head and last are guaranteed to work, so suppress warnings for them
 @SuppressWarnings(Array("org.wartremover.warts.TraversableOps"))
-case class WaitAccumState[T](windowQueue: m.Queue[(Idx, Time)], lastFail: Boolean, lastTime: (Idx, Time))
+case class WaitAccumState[T](windowQueue: m.ArrayDeque[(Idx, Time)], lastFail: Boolean, lastTime: (Idx, Time))
     extends AccumState[T, T, WaitAccumState[T]] {
 
   /** This method is called for each IdxValue produced by inner patterns.
@@ -40,7 +40,7 @@ case class WaitAccumState[T](windowQueue: m.Queue[(Idx, Time)], lastFail: Boolea
   @inline
   override def updated(
     window: Window,
-    times: m.Queue[(Idx, Time)],
+    times: m.ArrayDeque[(Idx, Time)],
     idxValue: IdxValue[T]
   ): (WaitAccumState[T], QI[T]) = {
 
@@ -50,7 +50,7 @@ case class WaitAccumState[T](windowQueue: m.Queue[(Idx, Time)], lastFail: Boolea
       val end = if (idxValue.value.isFail) times.last._2.minus(window) else times.last._2
 
       // don't use ++ here, slow!
-      val windowQueueWithNewPoints = times.foldLeft(windowQueue) { case (a, b) => a.enqueue(b); a }
+      val windowQueueWithNewPoints = times.foldLeft(windowQueue) { case (a, b) => a.append(b); a }
 
       val cleanedWindowQueue = windowQueueWithNewPoints.dropWhile {
         case (_, t) => t < start