solve filenotexsitexception when table has an empty partition

amplab · Jun 13, 2014 · ff56841 · ff56841
1 parent 062a6f2
commit ff56841
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 12 deletions.
diff --git a/src/main/scala/shark/execution/MemoryStoreSinkOperator.scala b/src/main/scala/shark/execution/MemoryStoreSinkOperator.scala
@@ -130,7 +130,7 @@ class MemoryStoreSinkOperator extends TerminalOperator {
       op.logInfo("Putting RDD for %s.%s in off-heap storage".format(databaseName, tableName))
       offHeapWriter.createTable()
       outputRDD.context.runJob(
-          outputRDD, MemoryStoreSinkOperator.processOffHeapSinkPartition(op, offHeapWriter))
+          outputRDD, MemoryStoreSinkOperator.processOffHeapSinkPartition(offHeapWriter))
       offHeapWriter.cleanTmpPath()
     } else {
       // Run a job on the RDD that contains the query output to force the data into the memory
@@ -203,10 +203,8 @@ class MemoryStoreSinkOperator extends TerminalOperator {
 }
 
 object MemoryStoreSinkOperator {
-  def processOffHeapSinkPartition(op: OperatorSerializationWrapper[MemoryStoreSinkOperator], 
-      offHeapWriter: OffHeapTableWriter) = {
+  def processOffHeapSinkPartition(offHeapWriter: OffHeapTableWriter) = {
     def writeFiles(context: TaskContext, iter: Iterator[_]): Long = {
-      op.logDebug("Started executing writeFiles for operator: " + op)
       val partId = context.partitionId
       val partition = iter.next().asInstanceOf[TablePartition]
       val taskTmpDir = context.stageId + "_" + context.partitionId + "_" + context.attemptId
@@ -215,8 +213,8 @@ object MemoryStoreSinkOperator {
         offHeapWriter.writePartitionColumn(partId, column, buf, taskTmpDir)
         writeBytes += buf.limit 
       }
-      offHeapWriter.commitPartition(partId, taskTmpDir)
-      op.logDebug("Finished executing writeFiles for operator: " + op)
+      val numColumns = partition.columns.size + 1
+      offHeapWriter.commitPartition(partId, numColumns, taskTmpDir)
       writeBytes
     }
     writeFiles _

diff --git a/src/main/scala/shark/memstore2/OffHeapStorageClient.scala b/src/main/scala/shark/memstore2/OffHeapStorageClient.scala
@@ -79,7 +79,7 @@ abstract class OffHeapTableWriter extends Serializable {
 
   def writePartitionColumn(part: Int, column: Int, data: ByteBuffer, tempDir: String)
 
-  def commitPartition(part: Int, tempDir: String)
+  def commitPartition(part: Int, numColumns: Int, tempDir: String)
 
   def cleanTmpPath()
 }

diff --git a/src/tachyon_enabled/scala/shark/tachyon/TachyonOffHeapTableWriter.scala b/src/tachyon_enabled/scala/shark/tachyon/TachyonOffHeapTableWriter.scala
@@ -67,12 +67,14 @@ class TachyonOffHeapTableWriter(@transient path: String, @transient numColumns:
     outStream.close()
   }
 
-  override def commitPartition(part: Int, tempDir: String) {
+  override def commitPartition(part: Int, numColumns: Int, tempDir: String) {
     val tmpPath = rawTable.getPath() + Constants.PATH_SEPARATOR + TEMP
-    (0 until rawTable.getColumns()).reverse.foreach { column =>
-      tfs.rename(tmpPath + Constants.PATH_SEPARATOR + tempDir + Constants.PATH_SEPARATOR
-          + column + Constants.PATH_SEPARATOR + part, rawTable.getPath() + Constants.PATH_SEPARATOR 
-          + MasterInfo.COL + column + Constants.PATH_SEPARATOR + part)
+    (0 until numColumns).reverse.foreach { column =>
+      val srcPath = tmpPath + Constants.PATH_SEPARATOR + tempDir + Constants.PATH_SEPARATOR +
+          column + Constants.PATH_SEPARATOR + part
+      val destPath = rawTable.getPath() + Constants.PATH_SEPARATOR +
+          MasterInfo.COL + column + Constants.PATH_SEPARATOR + part
+      tfs.rename(srcPath, destPath)
     }
     tfs.delete(tmpPath + Constants.PATH_SEPARATOR + tempDir, true)
   }