28339 bug fixed (apache#128)

Author: dsergeevmapr

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala

@@ -241,6 +241,12 @@ object FileFormatWriter extends Logging {
     try {
       Utils.tryWithSafeFinallyAndFailureCallbacks(block = {
         // Execute the task to write rows out and commit the task.
+        val summary = writeTask.execute(iterator)
+        writeTask.releaseResources()
+        val waitingTimeForInit =
+          SparkEnv.get.conf.getLong("spark.mapr.commitDelay", defaultValue = 0)
+        Thread.sleep(waitingTimeForInit)
+        WriteTaskResult(committer.commitTask(taskAttemptContext), summary)
         while (iterator.hasNext) {
           dataWriter.write(iterator.next())
         }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala

@@ -221,8 +221,10 @@ case class InsertIntoHadoopFsRelationCommand(
     // first clear the path determined by the static partition keys (e.g. /table/foo=1)
     val staticPrefixPath = qualifiedOutputPath.suffix(staticPartitionPrefix)
     if (fs.exists(staticPrefixPath) && !committer.deleteWithJob(fs, staticPrefixPath, true)) {
-      throw new IOException(s"Unable to clear output " +
-        s"directory $staticPrefixPath prior to writing to it")
+      if (fs.exists(staticPrefixPath)) {
+        throw new IOException(s"Unable to clear output " +
+          s"directory $staticPrefixPath prior to writing to it")
+      }
     }
     // now clear all custom partition locations (e.g. /custom/dir/where/foo=2/bar=4)
     for ((spec, customLoc) <- customPartitionLocations) {