added takeFirst

DEIB-GECO · May 10, 2019 · 7230b4d · 7230b4d
1 parent 3f08bc8
commit 7230b4d
Show file tree

Hide file tree

Showing 5 changed files with 22 additions and 6 deletions.
diff --git a/GMQL-PythonAPI/src/main/scala/it/polimi/genomics/pythonapi/PythonManager.scala b/GMQL-PythonAPI/src/main/scala/it/polimi/genomics/pythonapi/PythonManager.scala
@@ -336,12 +336,12 @@ object PythonManager {
     new CollectedResult(result)
   }
 
-//  def take(index: Int, n: Int): CollectedResult = {
-//    this.checkSparkContext()
-//    val variableToTake = this.getVariable(index)
-//    val result = this.server.TAKE(variableToTake, n)
-//    new CollectedResult(result)
-//  }
+  def take(index: Int, n: Int): CollectedResult = {
+    this.checkSparkContext()
+    val variableToTake = this.getVariable(index)
+    val result = this.server.TAKE_FIRST(variableToTake, n)
+    new CollectedResult( (result._1.toIterator, result._2.toIterator, result._3) )
+  }
 
   def serializeVariable(index: Int): String = {
     val variableToSerialize = this.getVariable(index)

diff --git a/GMQL-PythonAPI/src/main/scala/it/polimi/genomics/pythonapi/StubExecutor.scala b/GMQL-PythonAPI/src/main/scala/it/polimi/genomics/pythonapi/StubExecutor.scala
@@ -24,4 +24,6 @@ class StubExecutor extends Implementation {
   override def getParser(name: String, dataset: String): GMQLLoaderBase = ???
 
   override def collectIterator(iRVariable: IRVariable): (Iterator[(GRecordKey, Array[GValue])], Iterator[(Long, (String, String))], List[(String, PARSING_TYPE)]) = ???
+
+  override def takeFirst(iRVariable: IRVariable, n: Int): (Array[(GRecordKey, Array[GValue])], Array[(Long, (String, String))], List[(String, PARSING_TYPE)]) = ???
 }
diff --git a/GMQL-Server/src/main/scala/it/polimi/genomics/GMQLServer/GmqlServer.scala b/GMQL-Server/src/main/scala/it/polimi/genomics/GMQLServer/GmqlServer.scala
@@ -54,6 +54,11 @@ class GmqlServer(var implementation: Implementation, binning_size: Option[Long]
     implementation.collectIterator(iRVariable)
   }
 
+  def TAKE_FIRST(iRVariable: IRVariable, n: Int): (Array[(GRecordKey, Array[GValue])], Array[(Long, (String, String))], List[(String, PARSING_TYPE)]) = {
+    optimise()
+    implementation.takeFirst(iRVariable, n)
+  }
+
   def TAKE(iRVariable: IRVariable, n: Int): Any = {
     optimise()
     implementation.take(iRVariable, n)

diff --git a/GMQL-Server/src/main/scala/it/polimi/genomics/GMQLServer/Implementation.scala b/GMQL-Server/src/main/scala/it/polimi/genomics/GMQLServer/Implementation.scala
@@ -55,6 +55,8 @@ abstract class Implementation {
 
   def take(iRVariable: IRVariable, n:Int):Any
 
+  def takeFirst(iRVariable: IRVariable, n:Int): (Array[(GRecordKey, Array[GValue])], Array[(Long, (String, String))], List[(String, PARSING_TYPE)])
+
   /** stop GMQL implementation (kill a job)*/
   def stop()
 

diff --git a/GMQL-Spark/src/main/scala/it/polimi/genomics/spark/implementation/GMQLSparkExecutor.scala b/GMQL-Spark/src/main/scala/it/polimi/genomics/spark/implementation/GMQLSparkExecutor.scala
@@ -82,6 +82,13 @@ class GMQLSparkExecutor(val binSize : BinSize = BinSize(), val maxBinDistance :
     (regionRDD,metaRDD,iRVariable.schema)
   }
 
+  override def takeFirst(iRVariable: IRVariable, n: Int): (Array[(GRecordKey, Array[GValue])], Array[(Long, (String, String))], List[(String, PARSING_TYPE)]) = {
+    val metaRDD = implement_md(iRVariable.metaDag, sc).collect()
+    val regionRDD = implement_rd(iRVariable.regionDag, sc).take(n)
+
+    (regionRDD, metaRDD, iRVariable.schema)
+  }
+
   override def stop(): Unit = {
     sc.stop()
   }