Merge branch 'JdbcImport' into MongoImport

Author: gclaramunt

tools/import/jdbc/build.sbt

@@ -17,9 +17,13 @@ resolvers ++= Seq(
 
 scalacOptions ++= Seq("-unchecked", "-deprecation")
 
+seq(com.github.retronym.SbtOneJar.oneJarSettings: _*)
+
 libraryDependencies ++= Seq(
   "com.reportgrid" % "blueeyes-core_2.9.2" % "1.0.0-M6",
   "com.reportgrid" % "blueeyes-json_2.9.2" % "1.0.0-M6",
+  "org.scalaz"             % "scalaz-core_2.9.2"        % "7.0.0-M3" ,
   "org.specs2" %% "specs2" % "1.12.2" ,
+  //sbtVersion(v => "com.github.siasia" %% "xsbt-proguard-plugin" % (v+"-0.1.1")),
   "com.h2database" % "h2" % "1.2.134" % "test"
 )

tools/import/jdbc/src/main/scala/com/precog/tools/importers/jdbc/DbAccess.scala

@@ -2,6 +2,7 @@ package com.precog.tools.importers.jdbc
 
 import java.sql._
 import Datatypes._
+import scalaz.{StreamT, Id}
 
 /**
  * User: gabriel
@@ -26,11 +27,14 @@ object DbAccess {
     for ( i <- 1 to count) yield Column(tblMetaData.getColumnName(i),Table(tblMetaData.getTableName(i)))
   }
 
-  def rsIterator[T](rs:ResultSet)(f:ResultSet => T) = new Iterator[T] {
+  //don't use!
+  private def rsIterator[T](rs:ResultSet)(f:ResultSet => T) = new Iterator[T] {
     def hasNext = rs.next()
     def next():T = f(rs)
   }
 
+  def rsStreamT[T](rs:ResultSet)(f:ResultSet => T)=StreamT.unfold(rs)( (rs:ResultSet) => if (rs.next()) { Some(f(rs),rs)} else None )
+
   def oneColumnRs(rs:ResultSet) = rsIterator(rs)(rs=> rs.getString(1))
   def tables(rs:ResultSet) = rsIterator(rs)(rs=> Table(rs.getString("TABLE_NAME")))
   def columns(rs:ResultSet) = rsIterator(rs)(rs=> Column(rs.getString("COLUMN_NAME"), Table(rs.getString("TABLE_NAME"))))

tools/import/jdbc/src/main/scala/com/precog/tools/importers/jdbc/DbAnalysis.scala

@@ -18,7 +18,7 @@ object DbAnalysis{
 
 
   def findTables(metadata: DatabaseMetaData, oCat: Option[String], tableName: => Option[String]): Array[Table] = {
-    val cat= toNullUppercase(oCat)
+    val cat= null//toNullUppercase(oCat)
     val tableNm= tableName.map(_.toUpperCase).getOrElse(null)
     tables(metadata.getTables(cat, null, tableNm, Array("TABLE"))).toArray
   }

tools/import/jdbc/src/main/scala/com/precog/tools/importers/jdbc/ImportJdbc.scala

@@ -8,6 +8,17 @@ import blueeyes.bkka.AkkaDefaults.defaultFutureDispatch
 import scala.Some
 import blueeyes.core.service.engines.HttpClientXLightWeb
 import Datatypes._
+import blueeyes.bkka.FutureMonad
+import scalaz.{Hoist, StreamT, ~>}
+import akka.dispatch.{Await, ExecutionContext, Future}
+import java.nio.ByteBuffer
+import scalaz.Id._
+import annotation.tailrec
+import akka.util.Duration
+import blueeyes.core.http.HttpResponse
+import blueeyes.core.data.ByteChunk
+import akka.dispatch.Future
+
 
 /**
  * User: gabriel
@@ -17,6 +28,8 @@ object ImportJdbc {
 
   import DbAccess._
 
+  val httpClient=new HttpClientXLightWeb()(defaultFutureDispatch)
+
   case class ImportTable(name:String, columns:Seq[String], baseOrJoin:Either[Table,Join]){ val isCollection = baseOrJoin.right.toOption.map(_.exported).getOrElse(false) }
   case class IngestInfo(tables:Seq[ImportTable])
 
@@ -34,53 +47,63 @@ object ImportJdbc {
 
   def buildSort(ingestInfo:IngestInfo) =ingestInfo.tables.flatMap( t => t.columns.map("%s.%s".format(t.name,_)) )
 
+  def getElements(o:Option[JValue]):List[JValue]= o match {
+    case Some(l:JArray) => l.elements
+    case _ => Nil
+  }
+
+  def toJObject(o:JValue):JObject= o match {
+    case j:JObject => j
+    case _ => sys.error("base value is not jobject!")
+  }
 
-  def mkPartialJson(baseName:String, ingestInfo:IngestInfo, s: Seq[String], prevMap:Map[String,JValue]=Map())= {
+  def buildField( nm: (String,String)) =Option(nm._2).map( s=>JField(nm._1,JString(s)))
 
-    def getElements(o:Option[JValue]):List[JValue]= o match {
-      case Some(l:JArray) => l.elements
-      case _ => Nil
-    }
-    def toJObject(o:JValue):JObject= o match {
-      case j:JObject => j
-      case _ => sys.error("base value is not jobject!")
-    }
+  type StrJVMap= Map[String,JValue]
 
-    def buildJValues( ms:(Map[String,JValue],Seq[String]), tblDesc: ImportTable ):(Option[(String,JValue)],Seq[String])={
-      val (m,s)=ms
-      val (tblColValues,rest)=s.splitAt(tblDesc.columns.length)
-      val objValues =(tblDesc.columns.zip(tblColValues)).flatMap(buildField(_) ).toList
-      val tblName = tblDesc.name
-      val keyValue=if (objValues.isEmpty) if (tblDesc.isCollection) Some(tblName->JArray.empty) else None
+  def buildJValues( map:StrJVMap, s:Seq[String], tblDesc: ImportTable ):(Option[(String,JValue)],Seq[String])={
+    val (tblColValues,rest)=s.splitAt(tblDesc.columns.length)
+    val objValues =(tblDesc.columns.zip(tblColValues)).flatMap(buildField(_) ).toList
+    val tblName = tblDesc.name.toUpperCase
+    val keyValue=
+      if (objValues.isEmpty) if (tblDesc.isCollection) Some(tblName->JArray.empty) else None
       else {
         val data=JObject(objValues)
-        val obj= if (tblDesc.isCollection) JArray(data:: getElements(m.get(tblName)) ) else data
+        val obj= if (tblDesc.isCollection) JArray(getElements(map.get(tblName)):+data ) else data
         Some(tblName->obj)
       }
-      (keyValue,rest)
+    (keyValue,rest)
+  }
+
+  @tailrec
+  def mkPartialJson(baseName:String, ingestInfo:IngestInfo, dataStream: StreamT[Id,Seq[String]], prevMap:StrJVMap=Map()):Option[(JValue,StreamT[Id,Seq[String]])] =
+    if (dataStream.isEmpty) None
+    else {
+      val s=dataStream.head
+      val tail=dataStream.tail
+      val jsonMap=buildJsonObjMap(ingestInfo, prevMap, s)
+      val baseNameUC=baseName.toUpperCase
+      //peek into the stream
+      val nextJsonMap:StrJVMap=if (tail.isEmpty) Map() else buildJsonObjMap(ingestInfo, Map(), tail.head)
+      if ( !nextJsonMap.isEmpty && (jsonMap.get(baseNameUC) == nextJsonMap.get(baseNameUC)) ) {
+        //if next row is the same object, keep building
+        mkPartialJson(baseNameUC,ingestInfo,tail,jsonMap)
+      } else {
+        val base= toJObject(jsonMap(baseNameUC))
+        val values = (jsonMap-baseNameUC).map(nv => JField(nv._1, nv._2)).toList
+        Some(JObject(base.fields ++ values),tail)
+      }
     }
 
-    def buildField( nm: (String,String)) =Option(nm._2).map( s=>JField(nm._1,JString(s)))
-
-    val jsonMap:Map[String,JValue]=ingestInfo.tables.foldLeft( (prevMap,s) )(
-      (ms,v) =>{
-        val (opt,r)= buildJValues(ms,v)
-        val (m,_)=ms
-        opt.map( (kobj)=>{
-          val (k,obj) =kobj
-          if (k!=baseName)
-            (m+(kobj),r)
-          else if (prevMap.isEmpty || prevMap(k)!= obj)
-            (Map(kobj),r)
-          else (m,r)
-        }).getOrElse((m,r))
-      } )._1
-
-    val base:JObject = toJObject(jsonMap(baseName))
-    val values:List[JField] = (jsonMap-baseName).map(nv => JField(nv._1, nv._2)).toList
-    (JObject(base.fields ++ values),jsonMap)
-  }
 
+  def buildJsonObjMap(ingestInfo: ImportJdbc.IngestInfo, prevMap: ImportJdbc.StrJVMap, s: Seq[String]): StrJVMap = {
+    ingestInfo.tables.foldLeft((prevMap, s))(
+      (ms, v) => {
+        val (m,seq)=ms
+        val (opt, r): (Option[(String, JValue)], Seq[String]) = buildJValues(m, seq, v) //build a json object from the seq values
+        opt.map(kv => (m + kv, r)).getOrElse((m, r))
+      })._1
+  }
 
   def names(cs:Seq[Column])=cs.map(_.name)
 
@@ -92,33 +115,39 @@ object ImportJdbc {
     "select %s from %s order by %s".format(colSelect,join,sort)
   }
 
-  def executeQuery(connDb: Connection, query: String ): (Iterator[IndexedSeq[String]],IndexedSeq[Column]) = {
+  def executeQuery(connDb: Connection, query: String ): (StreamT[Id,IndexedSeq[String]],IndexedSeq[Column]) = {
     val stmt = connDb.prepareStatement(query)
     val columns = getColumns(stmt)
     val rs = stmt.executeQuery()
-    (rsIterator(rs)(row => for (i <- 1 to columns.size) yield row.getString(i)),columns)
+    (rsStreamT(rs)(row => for (i <- 1 to columns.size) yield row.getString(i)),columns)
   }
 
-  def getConnection(dbUrl: String, user: String, password: String): Connection = {
-    DriverManager.getConnection(dbUrl, user, password)
+  def getConnection(dbUrl: String, user: String, password: String, database:Option[String]): Connection = {
+    val uri= database.map( dbName=>if (dbUrl.endsWith(dbName)) dbUrl else "%s%s".format(dbUrl,dbName)).getOrElse(dbUrl)
+    DriverManager.getConnection(uri, user, password)
   }
 
-  def ingest(connDb: Connection, objName:String, query: String, oTblDesc:Option[IngestInfo], ingestPath: =>String, host: =>String, apiKey: =>String) = {
+  def ingest(connDb: Connection, objName:String, query: String, oTblDesc:Option[IngestInfo], ingestPath: =>String, host: =>String, apiKey: =>String)(implicit executor: ExecutionContext):Future[HttpResponse[ByteChunk]] = {
+    implicit val M = new FutureMonad(executor)
     val (data,columns) = executeQuery(connDb, query)
     val tblDesc= oTblDesc.getOrElse(IngestInfo(Seq(ImportTable(objName,names(columns),Left(Table("base"))))))
-    val body = buildBody(data, objName, tblDesc)
+
+    val dataStream:StreamT[Future,ByteBuffer] =buildBody(data, objName, tblDesc)
+      .map(jv=>ByteBuffer.wrap({val js="%s\n".format(jv.renderCompact); print(js); js}.getBytes("UTF-8")))
+
+    val body:ByteChunk= Right(dataStream)
     val fullPath = "%s/ingest/v1/sync/fs%s/%s".format(host, ingestPath,objName)
-    val httpClient=new HttpClientXLightWeb()(defaultFutureDispatch)
     //TODO add owner account id
-    httpClient.parameters('apiKey -> apiKey).post(fullPath)(jvalueToChunk(body))
+    println("sending to ingest: path=%s query=%s".format(fullPath,query))
+    httpClient.parameters('apiKey -> apiKey).post(fullPath)(body)
   }
 
-  def buildBody(data: Iterator[IndexedSeq[String]], baseTable: String, i: IngestInfo): JArray =
-    JArray(data.foldLeft((List[JValue](), Map[String, JValue]()))((lm, r) => {
-      val (l, m) = lm
-      val (values, map) = mkPartialJson(baseTable, i, r, m)
-      (values :: l, map)
-    })._1)
+  def buildBody(data: StreamT[Id,Seq[String]], baseTable: String, i: IngestInfo)(implicit executor: ExecutionContext, m:FutureMonad): StreamT[Future,JValue] =
+    StreamT.unfoldM[Future,JValue,StreamT[Id,Seq[String]]](data)(ds=>
+        if (ds.isEmpty) Future(None)
+        else Future(mkPartialJson(baseTable,i,ds)))
+
+
 }