
package dbmodel

  /** Version 11.1
   *
   *  This module describes an implementation of Synchrony iterator.
   *
   *  Wong Limsoon
   *  12 May 2023
   */
  
  /** For supporting searching an ordered collection using a reference 
   *  object in some general way, several concepts of convexity and
   *  boundedness are pertinent. These are defined below.
   *
   *  Given a list [[B]] of objects [[b1, b2, ..., bn]] sorted
   *  wrt an ordering [[<=]] such that [[bi <= bj]] if i < j.
   *
   *  A "canSee" predicate [[csl(b,a)]] is LEFT-CONVEX wrt [[<=]] iff
   *     for each [[a]] and [[b' <= b]],
   *        [[b < a && !csl(b,a)]] implies [[!csl(b',a)]]
   *
   *  Further, a left-convex [[csl(b,a)]] is said to be (LEFT-)BOUNDED
   *  wrt [[bf]] iff for each [[a]], there is a [[b]], such that
   *  [[bf(b,a) && !csl(b,a)]].
   *
   *  A "canSee" predicate [[csr(b,a)]] is RIGHT-CONVEX wrt [[<=]] iff
   *     for each [[a]] and [[b <= b']],
   *        [[b >= a && !csr(b,a)]] implies [[!csr(b',a)]]
   *
   *  Further, a right-convex [[csr(b,a)]] is said to be (RIGHT-)BOUNDED
   *  wrt [[bf]] iff for each [[a]], there is a [[b]], such that
   *  [[!bf(b,a) && !csr(b,a)]].
   *
   *  A "canSee" predicate [[cs(b,a)]] is BI-CONVEX wrt [[<=]] iff
   *  it is both left- and right-convex wrt [[bf]].
   *
   *  A predicate [[cs]] is CONVEX wrt [[<=]] iff 
   *     for each [[a]], and [[b <= b' <= b'']],
   *        [[cs(b, a) && cs(b'', a)]] implies [[cs(b', a)]].
   *
   *  Clearly, a predicate [[cs]] which is convex is also bi-convex.
   *  However, there is a difference between points satisfying a 
   *  convex predicate vs satisfying simultaneously both a left-
   *  and a right-convex predicate. To see this: 
   *
   *  Given a sorted list of objects [[a1 <= a2 <= ... <= am]], 
   *  and a sorted list of objects [[b1 <= b2 <= ... <= bn]]. 
   *  Let [[csl]] be a left-convex predicate,
   *  [[csr]] be a right convex predicate, and
   *  [[cs]] be a convex predicate. 
   *
   *  Let [[BCV(a) = { bi | 1 <= i <= n, csl(bi, a) && csr(bi, a) }]].
   *  Then [[BCV(a)]] is not necessarily convex in the sense that
   *  [[ai in BCV(a)]] and [[aj in BCV(a)]] does not imply [[ak in BCV(a)]],
   *  where 1 <= i <= k <= j <= m.
   *
   *  Whereas, let [[CV(a) = { bi | 1 <= i <= n, cs(bi, a) }]].
   *  Then [[CV(a)]] is convex in the sense that
   *  [[ai in CV(a)]] and [[aj in CV(a)]] implies [[ak in CV(a)],
   *  where 1 <= i <= k <= j <= m.
   */

  /** For supporting searching an ordered collection using a sorted
   *  list of reference objects, the concepts of monotonicity, 
   *  anti-monotonicity, and synchronized iteration are also needed
   *  and defined below.
   *
   *  Given two sorted lists [[B = { b1 <= b2, ..., <= bn }]] and 
   *  [[A = { a1 <= a2 <= ... <= am }]].  
   *
   *  An "isBefore" predicate [[bf]] is MONOTONIC wrt [[(B,A)]] iff
   *     [[ai <= aj && bf(b,ai)]] implies [[bf(b,aj)]], and
   *     [[bi <= bj && bf(bj,a)]] implies [[bf(bi,a)]]
   *
   *  A predicate [[anti]] is ANTI-MONOTONIC wrt a [[bf]], 
   *  which is monotonic wrt [[(B,A)]], iff 
   *     [[ai <= aj && bf(b, ai) && !anti(b, ai)]] implies [[!anti(b, aj)]].
   *
   *  A Synchrony iterator [[siterator(pred)]] on [[(B,A)]] enables
   *  an efficient synchronized iteration on [[(B,A)]] to compute
   *  a "join" conceptually equivalent to
   *     [[ for a <- A yield (a, B filter { b => pred(b, a) }) ]]

   *  which is equivalent to this SQL query, albeit without flattening
   *  out the [[b]]'s that correspond to an [[a]]:
   *     [[ SELECT a, b FROM A as a, B as b where pred(b,a) GROUPBY a ]]
   *
   *  The efficiency is enabled by exploiting left-, right-, bi-
   *  convexity/boundedness, monotonicity and anti-monotonicity
   *  properties that [[pred(b,a)]] might enjoy wrt [[(B,A)]] and the
   *  associated orderings. In particular, the predicate [[pred(b,a)]]
   *  is conceptually decomposed into a quintet of predicates:
   *     [[bf(b,a)]]   - monotonic wrt [[(B,A)]],
   *     [[csl(b,a)]]  - left-convex & preferably left-bounded wrt [[bf]],
   *     [[csr(b,a)]]  - right-convex & preferably right-bounded wrt [[bf]],
   *     [[anti(b,a)]] - anti-monotonic wrt [[bf]], and
   *     [[sc(b,a)]]   - a residual predicate; 
   *   such that
   *     [[pred(b,a)]] iff [[csl(b,a) && csr(b,a) && sc(b,a)]] and
   *     [[!anti(b,a)]] implies [[!csl(b,a) || !csr(b,a)]]
   *
   *  Intuitively, given an [[a <- A]] and we want to find
   *  all [[b <- B]] that satisfy [[pred(b,a)]]. Then, 
   *     [[!csl(b,a)]] means no need to look at [[b' <= b]];
   *     [[!csr(b,a)]] means no need to look at [[b <= b']];
   *     [[!anti(b,a)]] means no need to consider this [[b]] for [[a <= a']].
   *  A Synchrony iterator uses these properties to prune useless
   *  steps from the iteration on [[(B,A)]] to achieve efficiency.
   */
 




  object Synchrony:

    import scala.collection.mutable.Builder
    import scala.collection.BufferedIterator
    import scala.util.Using
    import scala.annotation.tailrec
    import dbmodel.RESOURCE.Resource
    import dbmodel.KEY.Key
    import dbmodel.Predicates.{ Pred, and, HasOrder }
    import dbmodel.Predicates.Pred.mkWithOrder
    import dbmodel.Synchronizable.CBI
    

    type ISBEFORE[B,A] = (B,A) => Boolean   // Type of bf predicate
    type CANSEE[B,A]   = (B,A) => Boolean   // Type of cs predicate
    type ANTI[B,A]     = (B,A) => Boolean   // Type of anti predicate
    type SCREEN[B,A]   = (B,A) => Boolean   // Type of screening predicate



    /** Synchrony iterator on synchronizable ordered collection
     *
     *  @param oit   the synchronizable ordered collection
     *  @param bf    monotonic wrt [[ovec.key.ord]]
     *  @param anti  antimonotonic wrt [[bf]] and
     *               [[!anti]] implies [[!csl]] or [[!csr]].
     *  @param csl   left-convex & preferably left-bounded wrt [[bf]]
     *  @param csr   right-convex & preferably right-bounded wrt [[bf]]
     */

    trait SIterator[B,A,K](
          oit: CBI[B],
          kB: Key[B,K],
          bf: ISBEFORE[K,K], anti: ANTI[K,K],
          csl: CANSEE[K,K],  csr: CANSEE[K,K])
      extends Resource:

      /** Register [[oit]] as a resource held.
       *  Normally, [[oit]] and [[kB]] are not null. 
       *  However, the [[addFilter]] method produces a [[SIterator]]
       *  which has null [[oit]] and [[kB]].
       */

      if oit != null then use(oit)
      val getkB = if kB != null then kB.get else null
                                                

      /** Memoised the immediate past result & key
       */

      private var ores: Vector[B] = Vector()
      private var oa: Option[K] = None

      /** Collect selectivity stats 
       */
      var collectStats = false
      val selectivity = Vector.newBuilder[Int]


      /** Directly access & manage [[oit]]'s buffer for greater efficiency.
       */

      var buf: Vector[B] = if oit != null then oit.bs() else Vector()
      var bx = buf.length
      var hs: Array[Boolean] = Array.fill(bx)(true)
      var p = 0
      var q = 0
      def b = buf(p)
      def discard() = hs(p) = false
      def rewind() = { while (q < p) && !hs(q) do { q += 1}; p = q }
      def isEmpty = p >= bx && oit.isEmpty
      def shift() = 
        p += 1 
        while p < bx && !hs(p) do p += 1 
        if p >= bx then
           val tmp = Vector.newBuilder[B]
           var j = -1
           for i <- q to p - 1 do if hs(i) then { tmp += buf(i); j += 1 }
           q = 0
           p = j
           if collectStats then selectivity += j
           buf = tmp.result ++ oit.bs()
           bx = buf.length
           hs = Array.fill(bx)(true)
           p += 1



      /** Defining [[syncedWith]] via [[_syncedWith]] is a
       *  trick to make it possible for inheriting classes
       *  to more easily customize [[syncedWith]] 
       */

      val syncedWith: A => Vector[B] 


      protected def _syncedWith(ka: K): Vector[B] = {

        // import oit.{ b, isEmpty, shift, discard, rewind  }

        var zs: Builder[B,Vector[B]] = Vector.newBuilder[B]

        @tailrec def run: Builder[B,Vector[B]] =
          if isEmpty then { rewind(); zs }
          else
            val kb = getkB(b)
            val (isBF, isAnti) = (bf(kb, ka), anti(kb, ka))
            val (isCL, isCR) = (csl(kb, ka), csr(kb, ka)) 
            //
            // isBF && isCL && isCR => a kb we are looking for.
            // So, include into result.
            if isBF && isCL && isCR then { zs += b; shift(); run } 
            //
            // isBF && !isAnti => !anti(kb, ka') for ka << ka'.
            // Thus, !csl(kb, ka') or !csr(kb, ka') for ka << ka'.
            // So, this kb can be discarded.
            else if isBF && !isAnti then { discard(); shift(); run }
            //
            // This kb is not what we are looking for, but
            // may still match some ka' >> ka 
            else if isBF then { shift(); run }
            //
            // !isBF && isCL && isCR => a kb we are looking for.
            // So, include into result.
            else if isCL && isCR then { zs += b; shift(); run }
            //
            // !isBF && !isCL && isCR => we are not looking for this kb,
            // but it may still match some ka' >> ka
            else if isCR then { shift(); run }
            //
            // !isBF && !isCR => !csr(kb', ka) for kb << kb'.
            // Thus, there is no kb' >> kb that matches ka.
            // So, stop and return answer in zs.
            else { rewind(); zs }

        if oa == Some(ka) then ores
        else { oa = Some(ka); ores = run.result(); ores }
      }


      def addFilter(screen: SCREEN[B,A]): SIterator[B,A,K] =
        val me = this
        if screen == null then me
        else new SIterator[B,A,K](null, null, null, null, null, null) {
          val syncedWith: A => Vector[B] = (a) =>
            me.syncedWith(a).filter { (b) => screen(b,a) }
        }


      override def close(): Unit =
        if collectStats then
          val stats = dbmodel.OpG.STATS((x:Int)=> x) of selectivity.result
          dbmodel.DEBUG.message(s"** selectivity = $stats")
        oit.close()



    object SIterator: 

      /** @return a Synchrony iterator on [[oit]].
       *
       *  @require  [[pred]] decomposes into 
       *            a left-convex & preferably left-bounded [[csl]],
       *            a right-convex & preferably right-bounded [[csr]], 
       *            a residual filter [[screen]], and
       *            an antimonotonic [[anti]],
       *            such that [[pred iff csl && csr && screen]] and
       *            [[!anti]] implies [[!csl]] or [[!csr]].
       *            The convexity, boundedness & antimonotonicity are
       *            wrt a predicate [[bf]] which is monotonic wrt
       *            ordering [[ovec.key.ord]].
       */

      def apply[B,K](
                oit: CBI[B], 
                kB: Key[B,K], 
                pred: Pred[K])
           : SIterator[B,K,K] =
        import pred.{ bf, anti, csl, csr, ord, screen => sc }
        require(kB.ord == ord)
        // val getkB = kB.get
        new SIterator[B,K,K](oit, kB, bf, anti, csl, csr) {
          val syncedWith: K => Vector[B] =
            if sc == null then (ka: K) => _syncedWith(ka)
            else (ka: K) => _syncedWith(ka).filter { (b) => sc(getkB(b), ka) }
      }
  

      /** @return a Synchrony iterator on [[oit]].
       *
       *  @require  [[pred]] decomposes into 
       *            a left-convex & preferably left-bounded [[csl]],
       *            a right-convex & preferably right-bounded [[csr]], 
       *            a residual filter [[screen]], and
       *            an antimonotonic [[anti]],
       *            such that [[pred iff csl && csr && screen]] and
       *            [[!anti]] implies [[!csl]] or [[!csr]].
       *            The convexity, boundedness & antimonotonicity are
       *            wrt a predicate [[bf]] which is monotonic wrt
       *            ordering [[ovec.key.ord]].
       *
       *  @param screen is an additional filtering function
       *  @param kA   is the sorting key to be used for extracting
       *                the key from the synchronizing request [[a: A]]
       */

      def apply[B,A,K]
               (oit: CBI[B], kB: Key[B,K], pred: Pred[K], screen: SCREEN[B,A])
               (using kA: Key[A,K])
           : SIterator[B,A,K] =
        import pred.{ bf, anti, csl, csr, ord }
        require(kB.ord == ord && kA.ord == ord)
        val getkA = kA.get
        val getkB = kB.get
        val sk: SCREEN[B,A] =
          if pred.screen == null then null
          else (b, a) => pred.screen(getkB(b), getkA(a)) 
        val sc: SCREEN[B,A] = screen and sk
        //
        new SIterator[B,A,K](oit, kB, bf, anti, csl, csr) {
          val syncedWith: A => Vector[B] =
            if sc == null then (a: A) => _syncedWith(getkA(a))
            else (a: A) => _syncedWith(getkA(a)).filter { sc(_, a) }
      }

    end SIterator



    /** Endow ordered collections with Synchrony iterators
     */

    import dbmodel.OrderedCollection.{ OColl }


    extension [B,K](ocoll: OColl[B,K])

      def siterator(pred: Pred[K]): SIterator[B,K,K] =
        SIterator(ocoll.cbi, ocoll.key, pred)

      def siterator[A](pred: Pred[K], kA: Key[A,K]): SIterator[B,A,K] =
        SIterator(ocoll.cbi, ocoll.key, pred, null)(using kA)

      def siterator[A](pred: Pred[K], getkA: A => K): SIterator[B,A,K] =
        val kA = Key(getkA, ocoll.key.ord)
        SIterator(ocoll.cbi, ocoll.key, pred, null)(using kA)
     
      def siterator[A](kA: Key[A,K], pred: Pred[K]): SIterator[B,A,K] =
        siterator(pred, kA)

      def siterator[A](getkA: A => K, pred: Pred[K]): SIterator[B,A,K] =
        siterator(pred, getkA)



    extension [B,K:HasOrder](ocoll: OColl[B,K])

      /** @require [[csr]] is right-convex, -bounded, & antimonotonic wrt [[bf]]
       *  @require [[bf]]  is monotonic wrt [[key]] of this collection
       */

      def siterator[A](
              kA: Key[A,K],
              csr: CANSEE[K,K], 
              bf: ISBEFORE[K,K]) 
          : SIterator[B,A,K] =
        val pred = mkWithOrder(pred = csr, csr = csr, bf = bf)
        siterator(pred, kA)


      def siterator[A](kA: Key[A,K], csr: CANSEE[K,K]): SIterator[B,A,K] =
        val pred = mkWithOrder(pred = csr, csr = csr)
        siterator(pred, kA)


  end Synchrony



/** Examples ***********************************************

{{{

   import scala.language.implicitConversions
   import dbmodel.OrderedCollection.{ *, given }
   import dbmodel.KEY.{ *, given }
   import dbmodel.Predicates.{ *, given }
   import dbmodel.Synchrony.{*, given }


//==============================================
// Simple 1-D test
//==============================================


   val aa = OSeq(1,2,3,4,5,6,7,7,8,8,9)


// for each a <- aa, 
// return items b in aa s.t. 1 < (b - a) < 3.
//
// Note that this cansee predicate 1 < (b - a) < 3 
// is not antimonotonic wrt <.  However, it is 
// antimonotonic wrt an isBefore predicate defined as
// bf(b,a) iff b - 2 < a.  So, we can use this bf
// instead of < as the isBefore predicate.


   {
     val kA = aa.key
     def bf(b: Int, a: Int) = b - 2 < a
     def cs(b: Int, a: Int) = (1 < b - a) && (b - a < 3)
     val bi = aa.siterator(kA = kA, csr = cs, bf = bf)
     for 
       a <- aa
       bs = bi.syncedWith(a)
     do 
       println((a, bs))
   }


// Output should be: 

(1,Vector(3))
(2,Vector(4))
(3,Vector(5))
(4,Vector(6))
(5,Vector(7, 7))
(6,Vector(8, 8))
(7,Vector(9))
(7,Vector(9))
(8,Vector())
(8,Vector())
(9,Vector())

// End output


// for each a <- aa,
// return items b in aa s.t. abs(b - a) < 3.
// return also a count of such b's.


   {
     def cs(b: Int, a: Int) = (b - a).abs < 3
     val bi = aa.siterator(aa.key, cs)
     for 
       a <- aa
       bs = bi.syncedWith(a)
     do
       println((a, bs, bs.length))
   }


// Output should be:

(1,Vector(1, 2, 3),3)
(2,Vector(1, 2, 3, 4),4)
(3,Vector(1, 2, 3, 4, 5),5)
(4,Vector(2, 3, 4, 5, 6),5)
(5,Vector(3, 4, 5, 6, 7, 7),6)
(6,Vector(4, 5, 6, 7, 7, 8, 8),7)
(7,Vector(5, 6, 7, 7, 8, 8, 9),7)
(7,Vector(5, 6, 7, 7, 8, 8, 9),7)
(8,Vector(6, 7, 7, 8, 8, 9),6)
(8,Vector(6, 7, 7, 8, 8, 9),6)
(9,Vector(7, 7, 8, 8, 9),5)

// End output



//==============================================
// Simple 2-D test. Non-equijoin alert!
//==============================================


// xx and yy are tables of intervals/line segments.
// Line segments are ordered by (start, end).


   case class Ev(start: Int, end: Int, id: String)

   val ky = Key.asc((x: Ev) => (x.start, x.end))
   val xx = OSeq(Seq(Ev(60,90,"d")), ky)
   val yy = OSeq(Seq(Ev(10,70,"a"), Ev(20,30,"b"), Ev(40,80,"c")), ky)


// Show overlapping line segment in xx and yy.


   def overlap(b:(Int,Int), a: (Int, Int)) = (a._1 < b._2) && (b._1 < a._2)

   val yi = yy.siterator(xx.key, overlap)

   for 
     x <- xx
     ys = yi.syncedWith(x)
   do
     println((x, ys))


// Output should be:

(Ev(60,90,d),Vector(Ev(10,70,a), Ev(40,80,c)))

// End output
 


// Show line segments enclosing another.


   case class Ev(start: Int, end: Int, id: String) {
     def key = (start, end)
   }
 
   val a = Ev(10, 70, "a")
   val b = Ev(13, 22, "b")
   val c = Ev(15, 25, "c")
   val d = Ev(15, 55, "d")
   val i = Ev(16, 25, "i")
   val j = Ev(16, 55, "j")
   val e = Ev(20, 30, "e")
   val f = Ev(25, 50, "f")
   val g = Ev(40, 80, "g")
   val h = Ev(60, 90, "h")
   val events = List(a, b, c, d, i, j, e, f, g, h)
   
   val ky = Key.asc((x: Ev) => (x.start, x.end))
   val xx = OSeq(events, ky)
   val yy = OSeq(events, ky)

   {
     val bi = yy.siterator(dbmodel.Predicates.ENCLOSE, ky)
     for x <- xx do println((x, bi.syncedWith(x)))
   }


// Output should be:

(Ev(10,70,a),Vector())
(Ev(13,22,b),Vector(Ev(10,70,a)))
(Ev(15,25,c),Vector(Ev(10,70,a)))
(Ev(15,55,d),Vector(Ev(10,70,a)))
(Ev(16,25,d),Vector(Ev(10,70,a), Ev(15,55,d)))
(Ev(16,55,d),Vector(Ev(10,70,a)))
(Ev(20,30,e),Vector(Ev(10,70,a), Ev(15,55,d), Ev(16,55,d)))
(Ev(25,50,f),Vector(Ev(10,70,a), Ev(15,55,d), Ev(16,55,d)))
(Ev(40,80,g),Vector())
(Ev(60,90,h),Vector())

// End output


}}}

*************************************************************/



