ColumnLowering.scala

package relation
package compiler

import scala.collection.mutable

import ch.epfl.data.sc.pardis
import pardis.optimization.RecursiveRuleBasedTransformer
import pardis.quasi.TypeParameters._
import pardis.types._
import PardisTypeImplicits._
import pardis.ir._

import relation.deep.RelationDSLOpsPackaged
import relation.shallow._  

class ColumnStoreLowering(override val IR: RelationDSLOpsPackaged,
    override val schemaAnalysis: SchemaAnalysis) extends RelationLowering(IR,
    schemaAnalysis) {
  import IR.Predef._
  
  type Column = Array[String]
  type LoweredRelation = (Rep[Int], Array[Rep[Column]])
  
  def relationScan(scanner: Rep[RelationScanner], schema: Schema,
      size: Rep[Int], resultSchema: Schema): LoweredRelation = {
    val nbColumn = schema.size
    val res = new Array[Rep[Column]](nbColumn) 
    var i: Int = 0
    while (i < nbColumn) {
      res(i) = dsl"new Array[String]($size)"
      i += 1
    }
    i = 0
    val scanLine: (Rep[Int]) => Rep[Unit] = (u: Rep[Int]) => {
      for(j <- 0 until nbColumn) {
        var currCol = res(j)
        dsl"""
          $currCol($u) = $scanner.next_string()
        """
        res(j) = currCol
      }
      dsl"unit()"
    }
    dsl"""
      var u = 0
      while ($scanner.hasNext) {
        $scanLine(u)
        u += 1
      }
    """
    (size, res)
  }
  
  def relationProject(relation: Rep[Relation], schema: Schema,
      resultSchema: Schema): LoweredRelation = {
    val (size, rel) = getRelationLowered(relation)
    val inSchema = getRelationSchema(relation)
    val indexesWanted = resultSchema.columns.map(name =>
      inSchema.columns.indexOf(name)).toArray
    (size, indexesWanted.map(idx => rel(idx)))
  }
  
  def relationSelect(relation: Rep[Relation], field: String,
      value: Rep[String], resultSchema: Schema): LoweredRelation = {
    val (size, rel): LoweredRelation = getRelationLowered(relation)
    val idxCol: Int = resultSchema.indexOf(field)
    val intrCol = rel(idxCol)
    val nbCol = resultSchema.size

    var nbRes = dsl"""
      var nbRes: Int = 0
      for{
        idx <- 0 until $intrCol.length
      } (if ($intrCol(idx) == $value) nbRes += 1)
      nbRes
    """
    val res = new Array[Rep[Column]](nbCol) 
    for (i <- 0 until nbCol) {
      res(i) = dsl"new Array[String]($nbRes)"
    }

    val keepLine: (Rep[Int], Rep[Int]) => Rep[Unit] = (i: Rep[Int],
        occ: Rep[Int]) => {
      for(j <- 0 until nbCol) {
        var currCol = rel(j)
        var output = res(j)
        dsl"""$output($occ) = $currCol($i)"""
        res(j) = output
      }
      dsl"unit()"
    }

    dsl"""
      var occ = 0
      for (i <- 0 until $size) {
        if($intrCol(i) == $value) {
          $keepLine(i, occ)
          occ += 1
        }
      }
    """

    (nbRes, res)
  }
  
  def relationJoin(leftRelation: Rep[Relation], rightRelation: Rep[Relation],
      leftKey: String, rightKey: String,
      resultSchema: Schema): LoweredRelation = {
    val lSchema = getRelationSchema(leftRelation)
    val rSchema = getRelationSchema(rightRelation)
    val (sizeL, relL) = getRelationLowered(leftRelation)
    val (sizeR, relR) = getRelationLowered(rightRelation)
    val lKeyIdx = lSchema.indexOf(leftKey)
    val rKeyIdx = rSchema.indexOf(rightKey)
    val lColJoin = relL(lKeyIdx)
    val rColJoin = relR(rKeyIdx)
    val lNbCol = lSchema.size
    val rNbCol = rSchema.size
    // right key not copied
    val resNbCol = lNbCol + rNbCol - 1
    require(resNbCol == resultSchema.size)

    val res = new Array[Rep[Column]](resNbCol)
    val sizeRes = dsl"""
      var nbMatch = 0
      for (l <- 0 until $sizeL) {
        val lKeyV = $lColJoin(l)
        for (r <- 0 until $sizeR) {
          val rKeyV = $rColJoin(r)
          if(rKeyV == lKeyV) {
            nbMatch += 1
          }
        }
      }
      nbMatch
    """
    for (i <- 0 until resNbCol) {
      res(i) = dsl"new Array[String]($sizeRes)"
    }

    val mergeLine: (Rep[Int], Rep[Int], Rep[Int]) => Rep[Unit] = (l: Rep[Int],
        r: Rep[Int], occ: Rep[Int]) => {
      for(i <- 0 until lNbCol) {
        var currCol = relL(i)
        var output = res(i)
        dsl"""$output($occ) = $currCol($l)"""
        res(i) = output
      }
      var offset = lNbCol
      for (i <- 0 until rNbCol) {
        if (i == rKeyIdx) {
          offset -= 1
        } else {
          var currCol = relR(i)
          val resColNo = i + offset
          var output = res(resColNo)
          dsl"""$output($occ) = $currCol($r)"""
          res(resColNo) = output
        }
      }
      dsl"unit()"
    }

    dsl"""
      var occ = 0
      for (l <- 0 until $sizeL) {
        val lKeyV = $lColJoin(l)
        for (r <- 0 until $sizeR) {
          val rKeyV = $rColJoin(r)
          if (rKeyV == lKeyV) {
            $mergeLine(l, r, occ)
            occ += 1
          }
        }
      }
    """

    (sizeRes, res)
  }
  
  def relationPrint(relation: Rep[Relation]): Unit = {
    val (size, rel): LoweredRelation = getRelationLowered(relation)
    val schema: Schema = getRelationSchema(relation)
    val nbCol = schema.size
    val printLine: (Rep[Int]) => Rep[String] = (i: Rep[Int]) => {
      var str: Rep[String] = dsl""" "" """
      for(j <- 0 until nbCol) {
        var currCol = rel(j)
        val sep = if(j != nbCol - 1) dsl""" "|" """ else dsl""" "" """
        str = dsl"""$str + $currCol($i) + $sep"""
      }
      str
    }

    dsl"""
      for (j <- 0 until $size) {
        println($printLine(j))
      }
    """
  }
}