Tue Nov 8 20:04:12 PST 2016

Author: ericl

core/src/main/scala/org/apache/spark/internal/io/FileCommitProtocol.scala

@@ -82,12 +82,18 @@ abstract class FileCommitProtocol {
    *
    * The "dir" parameter specifies 2, and "ext" parameter specifies both 4 and 5, and the rest
    * are left to the commit protocol implementation to decide.
+   *
+   * Important: it is the caller's responsibility to add uniquely identifying content to "ext"
+   * if a task is going to write out multiple files to the same dir.
    */
   def newTaskTempFile(taskContext: TaskAttemptContext, dir: Option[String], ext: String): String
 
   /**
    * Similar to newTaskTempFile(), but allows files to committed to an absolute output location.
    * Depending on the implementation, there may be weaker guarantees around adding files this way.
+   *
+   * Important: it is the caller's responsibility to add uniquely identifying content to "ext"
+   * if a task is going to write out multiple files at all (even to different directories).
    */
   def newTaskTempFileAbsPath(
       taskContext: TaskAttemptContext, absoluteDir: String, ext: String): String

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

@@ -321,8 +321,10 @@ object FileFormatWriter extends Logging {
           None
       }
       val path = if (customPath.isDefined) {
+        // We need to include a uuid here since the commit protocol does not guarantee that
+        // temp files requested by the same task for absolute placement do not collide.
         committer.newTaskTempFileAbsPath(
-          taskAttemptContext, customPath.get, java.util.UUID.randomUUID().toString + ext)
+          taskAttemptContext, customPath.get, UUID.randomUUID().toString + ext)
       } else {
         committer.newTaskTempFile(taskAttemptContext, partDir, ext)
       }

sql/hive/src/test/scala/org/apache/spark/sql/hive/PartitionProviderCompatibilitySuite.scala

@@ -191,7 +191,8 @@ class PartitionProviderCompatibilitySuite
 */
 
   /**
-   * Runs a test against a multi-level partitioned table, and validate some post-test invariants.
+   * Runs a test against a multi-level partitioned table, then validates that the custom locations
+   * were respected by the output writer.
    *
    * The initial partitioning structure is:
    *   /p1=0/p2=0  -- custom location a