
package dbmodel

/** Version 11.1
 *
 * Fancy stuff so that programmers can write programs like this:
 *
 *  {{{
 *        for 
 *          (a, bs, cs, ds, es) <- A join B on CondAB
 *                                   join C using condAC
 *                                   join D on CondAD where condAD'
 *                                   join E on condAE where condAE'
 *            ... blah blah ...
 *        yield ... more blah ...
 *  }}}
 *
 *  Here, for each [[a]] in [[A]],
 *  [[bs]] is result of [[B.filter(b => CondAB(b,a))]],
 *  [[cs]] is result of [[C.filter(c => condAC(c,a))]], 
 *  [[ds]] is result of [[D.filter(d => CondAD(d,a) && condAD'(d,a))]], 
 *  [[es]] is result of [[E.filter(e => condAE(e,a) && condAE'(e,a))]]. 
 *
 *  Moreover, the complexity is O(|A| + k|B| + k|C| + k|D| + k|E|),
 *  where k is the selectivity of the (non-equi)join predicates 
 *  condAB, condAC, and condAD.
 * 
 *  This is much much better than the naive execution complexity of 
 *  O(|A| * (|B| + |C| + |D| + |E|)). Note that usual techniques for
 *  implementing equijoins via grouping (ala Wadler and Peyton-Jones) 
 *  and indexed table (ala Gibbons) don't work here, because these
 *  are non-equijoins.
 *
 *  The supported syntax has the forms below and can be repeated
 *  up to four times in the current implementation:
 *
 *  {{{
 *        A join B on CondAB
 *        A join B on CondAB where g 
 *        A join B using condAB
 *        A join B using condAB where g 
 *  }}}
 *
 *  [[CondAB]] in "on ..." has type [[ Pred[K] ]];
 *  [[condAB]] in "using ..." has type [[(K,K) => Bool]] and
 *     is required to be right-bounded and antimonotonic;
 *  [[g]] has type [[(B,A) => Bool]];
 *  where [[K]] is the type of the indexing field of [[A]] and [[B]].
 *
 *  The [[A join B on CondAB]] and [[A join B on CondAB where g]]
 *  forms are probably more suitable for an occasional user who
 *  may be less carefully about the antimonotonicity requirement.
 *
 *  The [[A join B using condAB]] and [[A join B using condAB where g]]
 *  forms are probably more suited for an expert user who is careful
 *  to use only right-bounded antimonotonic functions as [[condAB]].
 *
 *  The implementation is a bit long winded due to limitation of
 *  Scala's type system. The acrobatics needed here is akin to that
 *  for implementing zipN, where you need to implement zip2, zip3,
 *  zip4, etc. because the type system makes zipN for zipping arbitrary
 *  number of lists of arbitrarily different types not expressible.
 *
 *  Wong Limsoon
 *  13 May 2023
 */


  object DBSQL: 


    import scala.language.implicitConversions
    import dbmodel.Synchrony.{ SIterator, siterator }
    import dbmodel.KEY.Key
    import dbmodel.OrderedCollection.{ OColl, OSeq }
    import dbmodel.Synchronizable.CBI
    import dbmodel.Predicates.{ *, given }

    type Bool    = Boolean       
    type Ord[K]  = Ordering[K]
    type SI[A,B,K] = SIterator[B,A,K]
    type Vec[A]  = Vector[A]


    case class SynchronyJoinException(msg: String) extends Throwable


    case class Inner[B,A,K](
            bs: OColl[B,K], 
            p: Pred[K] = null, 
            f: (B,A) => Bool = null):

      def addP(_p: Pred[K]) = 
        if p == null then copy(p = _p) else copy(p = p AND _p)

      def addF(_f: (B,A) => Bool) = 
        if f == null then copy(f = _f) else copy(f = f and _f)

      def siterator(using kA: Key[A,K]) = 
        SIterator(bs.cbi, bs.key, p, f)(using kA)
    


    extension [A,K:HasOrder](as: OColl[A,K])
      def join[B1](bs1: OColl[B1,K]) = OColl2[A,B1,K](as, Inner[B1,A,K](bs1))


    case class OColl2[A,B1,K:HasOrder](as: OColl[A,K], bi1: Inner[B1,A,K]):

      def on(p: Pred[K]): OColl2[A,B1,K] = OColl2(as, bi1.addP(p))

      def using(f: (K,K) => Bool): OColl2[A,B1,K] = 
        /** @require [[f]] is antimonotonic & right-bounded */
        val p = Pred.mkWithOrder(pred = f, csr = f)
        OColl2(as, bi1.addP(p))

      def where(f: (B1,A) => Bool): OColl2[A,B1,K] = OColl2(as, bi1.addF(f))

      def join[B2](bs2: OColl[B2,K]): OColl3[A,B1,B2,K] = 
        OColl3(as, bi1, Inner[B2,A,K](bs2))



    case class OColl3[A,B1,B2,K:HasOrder](
            as: OColl[A,K], 
            bi1: Inner[B1,A,K], bi2: Inner[B2,A,K]):

      def on(p: Pred[K]): OColl3[A,B1,B2,K] = OColl3(as, bi1, bi2.addP(p))

      def using(f: (K,K) => Bool): OColl3[A,B1,B2,K] =
        /** @require [[f]] is antimonotonic & right-bounded */
        val p = Pred.mkWithOrder(pred = f, csr = f)
        OColl3(as, bi1, bi2.addP(p))

      def where(f: (B2,A) => Bool): OColl3[A,B1,B2,K] = 
        OColl3(as, bi1, bi2.addF(f))

      def join[B3](bs3: OColl[B3,K]): OColl4[A,B1,B2,B3,K] =
        OColl4(as, bi1, bi2, Inner[B3,A,K](bs3))



    case class OColl4[A,B1,B2,B3,K:HasOrder](
            as: OColl[A,K], 
            bi1: Inner[B1,A,K], bi2: Inner[B2,A,K], bi3: Inner[B3,A,K]): 

      def on(p: Pred[K]): OColl4[A,B1,B2,B3,K] =
        OColl4(as, bi1, bi2, bi3.addP(p))

      def using(f: (K,K) => Bool): OColl4[A,B1,B2,B3,K] =
        /** @require [[f]] is antimonotonic & right-bounded */
        val p = Pred.mkWithOrder(pred = f, csr = f)
        OColl4(as, bi1, bi2, bi3.addP(p))

      def where(f: (B3,A) => Bool): OColl4[A,B1,B2,B3,K] = 
        OColl4(as, bi1, bi2, bi3.addF(f))

      def join[B4](bs4: OColl[B4,K]): OColl5[A,B1,B2,B3,B4,K] =
        OColl5(as, bi1, bi2, bi3, Inner[B4,A,K](bs4))



    case class OColl5[A,B1,B2,B3,B4,K:HasOrder](
           as: OColl[A,K], 
           bi1: Inner[B1,A,K], bi2: Inner[B2,A,K], 
           bi3: Inner[B3,A,K], bi4: Inner[B4,A,K]):

      def on(p: Pred[K]): OColl5[A,B1,B2,B3,B4,K] =
        OColl5(as, bi1, bi2, bi3, bi4.addP(p))

      def using(f: (K,K) => Bool): OColl5[A,B1,B2,B3,B4,K] =
        /** @require [[f]] is antimonotonic & right-bounded */
        val p = Pred.mkWithOrder(pred = f, csr = f)
        OColl5(as, bi1, bi2, bi3, bi4.addP(p))

      def where(f: (B4,A) => Bool): OColl5[A,B1,B2,B3,B4,K] = 
        OColl5(as, bi1, bi2, bi3, bi4.addF(f))



    type OSeq2[A,B,K]  = OSeq[(A,Vec[B]),K]

    given si2toseq[A,B,K]: Conversion[OColl2[A,B,K], OSeq2[A,B,K]]
      with
        def apply(si: OColl2[A,B,K]): OSeq2[A,B,K] = new OSeq2[A,B,K] {
          val ai = si.as.cbi
          val i1 = si.bi1.siterator(using si.as.key)

          override val elems = CBI[(A,Vec[B])] { 
            ai map { (a) => (a, i1.syncedWith(a)) }
          }.userev(i1, ai)
              
          override val key = 
            val get = si.as.key.get
            val ord = si.as.key.ord
            Key[(A,Vec[B]),K](x => get(x._1), ord)

          init()
          use(i1, ai)
        }



    type OSeq3[A,B1,B2,K]  = OSeq[(A,Vec[B1],Vec[B2]),K]

    given si3toseq[A,B1,B2,K]: Conversion[OColl3[A,B1,B2,K], OSeq3[A,B1,B2,K]]
      with
        def apply(si: OColl3[A,B1,B2,K]): OSeq3[A,B1,B2,K] =
          new OSeq3[A,B1,B2,K] {
            val ai = si.as.cbi
            val i1 = si.bi1.siterator(using si.as.key)
            val i2 = si.bi2.siterator(using si.as.key)

            override val elems = CBI[(A,Vec[B1],Vec[B2])] {
              ai map { (a) =>
                val b1 = i1.syncedWith(a)
                val b2 = i2.syncedWith(a)
                (a, b1, b2)
              }
            }.userev(i2, i1, ai)
    
            override val key =  
              val get = si.as.key.get
              val ord = si.as.key.ord 
              Key[(A,Vec[B1],Vec[B2]),K](x => get(x._1), ord) 

            init()
            use(i2, i1, ai)
         }



    type OSeq4[A,B1,B2,B3,K]  = OSeq[(A,Vec[B1],Vec[B2],Vec[B3]),K]

    given si4toseq[A,B1,B2,B3,K]: Conversion[OColl4[A,B1,B2,B3,K],
                                             OSeq4[A,B1,B2,B3,K]]
      with
        def apply(si: OColl4[A,B1,B2,B3,K]): OSeq4[A,B1,B2,B3,K] =
          new OSeq4[A,B1,B2,B3,K] {
            import si.{ as, bi1, bi2, bi3 }
            val ai = as.cbi
            val i1 = bi1.siterator(using as.key)
            val i2 = bi2.siterator(using as.key)
            val i3 = bi3.siterator(using as.key)

            override val elems = CBI[(A,Vec[B1],Vec[B2],Vec[B3])] {
              ai map { (a) =>
                val b1 = i1.syncedWith(a)
                val b2 = i2.syncedWith(a)
                val b3 = i3.syncedWith(a)
                (a, b1, b2, b3)
              }
            }.userev(i3, i2, i1, ai)
    
            override val key =  
              val get = as.key.get
              val ord = as.key.ord 
              Key[(A,Vec[B1],Vec[B2],Vec[B3]),K](x => get(x._1), ord) 

            init()
            use(i3, i2, i1, ai)
         }



    type OSeq5[A,B1,B2,B3,B4,K] = OSeq[(A,Vec[B1],Vec[B2],Vec[B3],Vec[B4]),K]

    given si5toseq[A,B1,B2,B3,B4,K]: Conversion[OColl5[A,B1,B2,B3,B4,K],
                                                OSeq5[A,B1,B2,B3,B4,K]]
      with
        def apply(si: OColl5[A,B1,B2,B3,B4,K]): OSeq5[A,B1,B2,B3,B4,K] =
          new OSeq5[A,B1,B2,B3,B4,K] {
            import si.{ as, bi1, bi2, bi3, bi4 }
            val ai = as.cbi
            val i1 = bi1.siterator(using as.key)
            val i2 = bi2.siterator(using as.key)
            val i3 = bi3.siterator(using as.key)
            val i4 = bi4.siterator(using as.key)

            override val elems = CBI[(A,Vec[B1],Vec[B2],Vec[B3],Vec[B4])] {
              ai map { (a) =>
                val b1 = i1.syncedWith(a)
                val b2 = i2.syncedWith(a)
                val b3 = i3.syncedWith(a)
                val b4 = i4.syncedWith(a)
                (a, b1, b2, b3, b4)
              }
            }.userev(i4, i3, i2, i1, ai)
    
            override val key =  
              val get = as.key.get
              val ord = as.key.ord 
              Key[(A,Vec[B1],Vec[B2],Vec[B3],Vec[B4]),K](x => get(x._1), ord) 

            init()
            use(i4, i3, i2, i1, ai)
         }


  end DBSQL 






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

{{{

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


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


// aa is a sorted table.

   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.

   for (a, bs) <- aa join aa on DL(3) where { (b,a) => (b-a) > 1 }
   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.


   for (a, bs, cs) <- aa join aa on DL(3) join aa on DL(3) do
     println((a, bs, cs.length))


   {
      val bi = aa.siterator(aa.key, dInt.dl(3))
      val ci = aa.siterator(aa.key, dInt.dl(3))

      for
        a <- aa
        bs = bi.syncedWith(a)
        cs = ci.syncedWith(a)
      do
        println((a, bs, cs.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


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

   {
     val pred = DL(3).addFilter(gteq[Int])
     for (a, bs, cs) <- aa join aa on pred join aa on pred
     do println((a, bs, cs.length))
   }


   {
      val bi = aa.siterator(aa.key, dInt.dl(3)).addFilter(dInt.gteq)
      val ci = aa.siterator(aa.key, dInt.dl(3)).addFilter(dInt.gteq)

      for
        a <- aa
        bs = bi.syncedWith(a)
        cs =ci.syncedWith(a)
      do
        println((a, bs, cs.length))
   }


   {
      val bi = aa.siterator(aa.key, dInt.dl(3))
      val ci = aa.siterator(aa.key, dInt.dl(3))

      for
        a <- aa
        bs = bi.syncedWith(a).filter { dInt.gteq(_, a) }
        cs = ci.syncedWith(a).filter { dInt.gteq(_, a) }
      do
        println((a, bs, cs.length))
   }


// Output should be:

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

// 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.


   for (x, ys) <- xx join yy on OVERLAP(1) do println((x, ys))


// Output should be:

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

// End output
 


//==================================================
// Let's do some relational DB queries...
//==================================================


   case class Phone(model: String, price: Int, brand: String)

   val s21    = Phone("S21", 1000, "Samsung")
   val a52    = Phone("A52", 550, "Samsung")
   val a32    = Phone("A32", 350, "Samsung")
   val n10    = Phone("N10", 360, "OnePlus")
   val a94    = Phone("A94", 400, "Oppo")
   val m40    = Phone("Mate 40", 1200, "Huawei")
   val pix5   = Phone("Pixel 5", 1300, "Google")
   val pix4   = Phone("Pixel 4", 500, "Google")
   val phones = Vector(s21, a52, a32, n10,a94, m40, pix5, pix4)

   val kPrice = Key.asc[Phone,Int](x => x.price)
   val kBrand = Key.asc[Phone,String](x => x.brand)


// Phones, sorted by brands

   val OByBrand = phones orderedBy kBrand


// Phones grouped by brands


   for (b, ps) <- OByBrand.elemSeq.groupBy(_.brand)
   do println(s"brand = ${b}\n  ${ps}\n")


// Output should be something like below.
// Note that it is not ordered:

brand = Oppo
  Vector(Phone(A94,400,Oppo))

brand = Huawei
  Vector(Phone(Mate 40,1200,Huawei))

brand = Google
  Vector(Phone(Pixel 5,1300,Google), Phone(Pixel 4,500,Google))

brand = OnePlus
  Vector(Phone(N10,360,OnePlus))

brand = Samsung
  Vector(Phone(S21,1000,Samsung), Phone(A52,550,Samsung), Phone(A32,350,Samsung))

// End output




// Phones, re-sorted by price 


   val OByPrice = OByBrand orderedBy kPrice


// Phones and their price competitors from
// other brands within +/- $150.
// Non-equijoin alert!


   for 
     (p, cs) <- OByPrice join OByPrice on DLEQ(150)
     ob = cs.filter(_.brand != p.brand) 
   do 
     println(s"${p}\n  ${ob}\n") 


// Output should look like:

Phone(A32,350,Samsung)
  List(Phone(N10,360,OnePlus), Phone(A94,400,Oppo), Phone(Pixel 4,500,Google))

Phone(N10,360,OnePlus)
  List(Phone(A32,350,Samsung), Phone(A94,400,Oppo), Phone(Pixel 4,500,Google))

Phone(A94,400,Oppo)
  List(Phone(A32,350,Samsung), Phone(N10,360,OnePlus), Phone(Pixel 4,500,Google), Phone(A52,550,Samsung))

Phone(Pixel 4,500,Google)
  List(Phone(A32,350,Samsung), Phone(N10,360,OnePlus), Phone(A94,400,Oppo), Phone(A52,550,Samsung))

Phone(A52,550,Samsung)
  List(Phone(A94,400,Oppo), Phone(Pixel 4,500,Google))

Phone(S21,1000,Samsung)
  List()

Phone(Mate 40,1200,Huawei)
  List(Phone(Pixel 5,1300,Google))

Phone(Pixel 5,1300,Google)
  List(Phone(Mate 40,1200,Huawei))

// End output



// Samsung phones and their price competitors 
// from Google, within 20%.
// Non-equijoin alert!


   {
     val samsung = OByPrice.filter(_.brand == "Samsung") 
     val google = OByPrice.filter(_.brand == "Google")
  
     for (p, cs) <- samsung join google on SZPERCENT(1.2) 
     do println(s"${p}\n  ${cs}\n") 
   }


// Output should be:

Phone(A32,350,Samsung)
  List()

Phone(A52,550,Samsung)
  List(Phone(Pixel 4,500,Google))

Phone(S21,1000,Samsung)
  List()

//  End output



//======================================================
// Timing study to compare the standard groupBy with
// sortedBy followed by clustered, which is our way of 
// implementing groupby using Synchrony iterator. 
//======================================================


// [[bm]] runs [[f]] as many times as possible
// within [[duration]] number of milliseconds.


   def bm(duration: Long)(f: => Unit) = {
     val end = System.currentTimeMillis + duration
     var count = 0
     while System.currentTimeMillis < end do { f; count += 1 }
     count
   }


// Standard groupBy


   def test1 = 
     (for (b,ps) <- OByBrand.elemSeq.groupBy(_.brand) yield ps.length).max


// Synchrony-based implementation of groupby.


   def test3 = (for (b,ps) <- OByBrand.clustered yield ps.length).max


// Timing test.


   bm(500)(test1)  // standard groupBy
   bm(500)(test3)  // clustered


}}}

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


