hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ser...@apache.org
Subject [57/57] [abbrv] hive git commit: HIVE-14640 : handle hive.merge.*files in select queries (Sergey Shelukhin)
Date Fri, 14 Oct 2016 00:20:18 GMT
HIVE-14640 : handle hive.merge.*files in select queries (Sergey Shelukhin)


Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/eacf9f9b
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/eacf9f9b
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/eacf9f9b

Branch: refs/heads/hive-14535
Commit: eacf9f9b6d7405b68def88ffc5fd755222375efc
Parents: bd78d66
Author: Sergey Shelukhin <sershe@apache.org>
Authored: Thu Oct 13 17:18:46 2016 -0700
Committer: Sergey Shelukhin <sershe@apache.org>
Committed: Thu Oct 13 17:18:46 2016 -0700

----------------------------------------------------------------------
 .../org/apache/hadoop/hive/conf/HiveConf.java   |   2 +-
 .../hive/ql/exec/AbstractFileMergeOperator.java | 181 +++++++++------
 .../hadoop/hive/ql/exec/FileSinkOperator.java   | 187 ++-------------
 .../apache/hadoop/hive/ql/exec/MoveTask.java    |   8 +-
 .../hive/ql/exec/OrcFileMergeOperator.java      |  11 +-
 .../hive/ql/exec/RCFileMergeOperator.java       |   3 +-
 .../apache/hadoop/hive/ql/exec/Utilities.java   | 225 ++++++++++++++++--
 .../rcfile/truncate/ColumnTruncateMapper.java   |   1 +
 .../apache/hadoop/hive/ql/metadata/Hive.java    |   1 +
 .../hive/ql/optimizer/GenMapRedUtils.java       | 214 +++++++++--------
 .../hive/ql/parse/DDLSemanticAnalyzer.java      |   9 +-
 .../hadoop/hive/ql/parse/GenTezUtils.java       |   4 +-
 .../hadoop/hive/ql/parse/SemanticAnalyzer.java  |   4 +-
 .../ql/plan/ConditionalResolverMergeFiles.java  |  17 +-
 .../hadoop/hive/ql/plan/FileMergeDesc.java      |   9 +
 .../hadoop/hive/ql/plan/FileSinkDesc.java       |  14 +-
 .../apache/hadoop/hive/ql/plan/MoveWork.java    |  10 +-
 ql/src/test/queries/clientpositive/mm_all.q     |  57 +++--
 ql/src/test/queries/clientpositive/mm_current.q |  40 +---
 .../results/clientpositive/llap/mm_all.q.out    | 232 +++++++++++++++----
 .../clientpositive/llap/mm_current.q.out        | 165 +------------
 21 files changed, 758 insertions(+), 636 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
----------------------------------------------------------------------
diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
index c89142c..6201c04 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -3122,7 +3122,7 @@ public class HiveConf extends Configuration {
 
     HIVE_METASTORE_MM_HEARTBEAT_TIMEOUT("hive.metastore.mm.heartbeat.timeout", "1800s",
         new TimeValidator(TimeUnit.SECONDS),
-        "MM write ID times out after this long if a heartbeat is not send. Currently disabled."), // TODO# heartbeating not implemented
+        "MM write ID times out after this long if a heartbeat is not send. Currently disabled."),
 
     HIVE_METASTORE_MM_ABSOLUTE_TIMEOUT("hive.metastore.mm.absolute.timeout", "7d",
         new TimeValidator(TimeUnit.SECONDS),

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/exec/AbstractFileMergeOperator.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/AbstractFileMergeOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/AbstractFileMergeOperator.java
index 40c784b..dedbb78 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/AbstractFileMergeOperator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/AbstractFileMergeOperator.java
@@ -34,6 +34,8 @@ import org.apache.hadoop.mapred.JobConf;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.collect.Lists;
+
 /**
  * Fast file merge operator for ORC and RCfile. This is an abstract class which
  * does not process any rows. Refer {@link org.apache.hadoop.hive.ql.exec.OrcFileMergeOperator}
@@ -47,20 +49,21 @@ public abstract class AbstractFileMergeOperator<T extends FileMergeDesc>
 
   protected JobConf jc;
   protected FileSystem fs;
-  protected boolean autoDelete;
-  protected boolean exception;
-  protected Path outPath;
-  protected Path finalPath;
-  protected Path dpPath;
-  protected Path tmpPath;
-  protected Path taskTmpPath;
-  protected int listBucketingDepth;
-  protected boolean hasDynamicPartitions;
-  protected boolean isListBucketingAlterTableConcatenate;
-  protected boolean tmpPathFixedConcatenate;
-  protected boolean tmpPathFixed;
-  protected Set<Path> incompatFileSet;
-  protected transient DynamicPartitionCtx dpCtx;
+  private boolean autoDelete;
+  private Path outPath; // The output path used by the subclasses.
+  private Path finalPath; // Used as a final destination; same as outPath for MM tables.
+  private Path dpPath;
+  private Path tmpPath; // Only stored to update based on the original in fixTmpPath.
+  private Path taskTmpPath; // Only stored to update based on the original in fixTmpPath.
+  private int listBucketingDepth;
+  private boolean hasDynamicPartitions;
+  private boolean isListBucketingAlterTableConcatenate;
+  private boolean tmpPathFixedConcatenate;
+  private boolean tmpPathFixed;
+  private Set<Path> incompatFileSet;
+  private transient DynamicPartitionCtx dpCtx;
+  private boolean isMmTable;
+  private String taskId;
 
   /** Kryo ctor. */
   protected AbstractFileMergeOperator() {
@@ -77,39 +80,50 @@ public abstract class AbstractFileMergeOperator<T extends FileMergeDesc>
     this.jc = new JobConf(hconf);
     incompatFileSet = new HashSet<Path>();
     autoDelete = false;
-    exception = false;
     tmpPathFixed = false;
     tmpPathFixedConcatenate = false;
-    outPath = null;
-    finalPath = null;
     dpPath = null;
-    tmpPath = null;
-    taskTmpPath = null;
     dpCtx = conf.getDpCtx();
     hasDynamicPartitions = conf.hasDynamicPartitions();
     isListBucketingAlterTableConcatenate = conf
         .isListBucketingAlterTableConcatenate();
     listBucketingDepth = conf.getListBucketingDepth();
     Path specPath = conf.getOutputPath();
-    updatePaths(Utilities.toTempPath(specPath),
-        Utilities.toTaskTempPath(specPath));
+    isMmTable = conf.getMmWriteId() != null;
+    if (isMmTable) {
+      updatePaths(specPath, null);
+    } else {
+      updatePaths(Utilities.toTempPath(specPath), Utilities.toTaskTempPath(specPath));
+    }
     try {
       fs = specPath.getFileSystem(hconf);
-      autoDelete = fs.deleteOnExit(outPath);
+      if (!isMmTable) {
+        // Do not delete for MM tables. We either want the file if we succeed, or we must
+        // delete is explicitly before proceeding if the merge fails.
+        autoDelete = fs.deleteOnExit(outPath);
+      }
     } catch (IOException e) {
-      this.exception = true;
-      throw new HiveException("Failed to initialize AbstractFileMergeOperator",
-          e);
+      throw new HiveException("Failed to initialize AbstractFileMergeOperator", e);
     }
   }
 
   // sets up temp and task temp path
   private void updatePaths(Path tp, Path ttp) {
-    String taskId = Utilities.getTaskId(jc);
+    if (taskId == null) {
+      taskId = Utilities.getTaskId(jc);
+    }
     tmpPath = tp;
-    taskTmpPath = ttp;
-    finalPath = new Path(tp, taskId);
-    outPath = new Path(ttp, Utilities.toTempPath(taskId));
+    if (isMmTable) {
+      taskTmpPath = null;
+      // Make sure we don't collide with the source.
+      outPath = finalPath = new Path(tmpPath, taskId + ".merged");
+    } else {
+      taskTmpPath = ttp;
+      finalPath = new Path(tp, taskId);
+      outPath = new Path(ttp, Utilities.toTempPath(taskId));
+    }
+    Utilities.LOG14535.info("Paths for merge " + taskId + ": tmp " + tmpPath + ", task "
+        + taskTmpPath + ", final " + finalPath + ", out " + outPath, new Exception());
   }
 
   /**
@@ -142,7 +156,7 @@ public abstract class AbstractFileMergeOperator<T extends FileMergeDesc>
   protected void fixTmpPath(Path inputPath, int depthDiff) throws IOException {
 
     // don't need to update tmp paths when there is no depth difference in paths
-    if (depthDiff <=0) {
+    if (depthDiff <= 0) {
       return;
     }
 
@@ -157,10 +171,12 @@ public abstract class AbstractFileMergeOperator<T extends FileMergeDesc>
     }
 
     Path newTmpPath = new Path(tmpPath, newPath);
-    Path newTaskTmpPath = new Path(taskTmpPath, newPath);
     if (!fs.exists(newTmpPath)) {
+      Utilities.LOG14535.info("Creating " + newTmpPath);
       fs.mkdirs(newTmpPath);
     }
+
+    Path newTaskTmpPath = (taskTmpPath != null) ? new Path(taskTmpPath, newPath) : null;
     updatePaths(newTmpPath, newTaskTmpPath);
   }
 
@@ -182,7 +198,7 @@ public abstract class AbstractFileMergeOperator<T extends FileMergeDesc>
   }
 
   protected void fixTmpPath(Path path) throws IOException {
-
+    Utilities.LOG14535.info("Calling fixTmpPath with " + path);
     // Fix temp path for alter table ... concatenate
     if (isListBucketingAlterTableConcatenate) {
       if (this.tmpPathFixedConcatenate) {
@@ -208,38 +224,49 @@ public abstract class AbstractFileMergeOperator<T extends FileMergeDesc>
   @Override
   public void closeOp(boolean abort) throws HiveException {
     try {
-      if (!abort) {
-        // if outPath does not exist, then it means all paths within combine split are skipped as
-        // they are incompatible for merge (for example: files without stripe stats).
-        // Those files will be added to incompatFileSet
-        if (fs.exists(outPath)) {
-          FileStatus fss = fs.getFileStatus(outPath);
+      if (abort) {
+        if (!autoDelete || isMmTable) {
+          fs.delete(outPath, true);
+        }
+        return;
+      }
+      // if outPath does not exist, then it means all paths within combine split are skipped as
+      // they are incompatible for merge (for example: files without stripe stats).
+      // Those files will be added to incompatFileSet
+      if (fs.exists(outPath)) {
+        FileStatus fss = fs.getFileStatus(outPath);
+        if (!isMmTable) {
           if (!fs.rename(outPath, finalPath)) {
-            throw new IOException(
-                "Unable to rename " + outPath + " to " + finalPath);
+            throw new IOException("Unable to rename " + outPath + " to " + finalPath);
           }
-          LOG.info("renamed path " + outPath + " to " + finalPath + " . File" +
-              " size is "
-              + fss.getLen());
+          LOG.info("Renamed path " + outPath + " to " + finalPath
+              + "(" + fss.getLen() + " bytes).");
+        } else {
+          assert finalPath.equals(outPath);
+          // There's always just one file that we have merged.
+          // The union/DP/etc. should already be account for in the path.
+          Utilities.writeMmCommitManifest(Lists.newArrayList(outPath),
+              tmpPath.getParent(), fs, taskId, conf.getMmWriteId(), null);
+          LOG.info("Merged into " + finalPath + "(" + fss.getLen() + " bytes).");
         }
+      }
 
-        // move any incompatible files to final path
-        if (incompatFileSet != null && !incompatFileSet.isEmpty()) {
-          for (Path incompatFile : incompatFileSet) {
-            Path destDir = finalPath.getParent();
-            try {
-              Utilities.renameOrMoveFiles(fs, incompatFile, destDir);
-              LOG.info("Moved incompatible file " + incompatFile + " to " +
-                  destDir);
-            } catch (HiveException e) {
-              LOG.error("Unable to move " + incompatFile + " to " + destDir);
-              throw new IOException(e);
-            }
-          }
+      // move any incompatible files to final path
+      if (incompatFileSet != null && !incompatFileSet.isEmpty()) {
+        if (isMmTable) {
+          // We only support query-time merge for MM tables, so don't handle this.
+          throw new HiveException("Incompatible files should not happen in MM tables.");
         }
-      } else {
-        if (!autoDelete) {
-          fs.delete(outPath, true);
+        for (Path incompatFile : incompatFileSet) {
+          Path destDir = finalPath.getParent();
+          try {
+            Utilities.renameOrMoveFiles(fs, incompatFile, destDir);
+            LOG.info("Moved incompatible file " + incompatFile + " to " +
+                destDir);
+          } catch (HiveException e) {
+            LOG.error("Unable to move " + incompatFile + " to " + destDir);
+            throw new IOException(e);
+          }
         }
       }
     } catch (IOException e) {
@@ -253,16 +280,26 @@ public abstract class AbstractFileMergeOperator<T extends FileMergeDesc>
     try {
       Path outputDir = conf.getOutputPath();
       FileSystem fs = outputDir.getFileSystem(hconf);
-      Path backupPath = backupOutputPath(fs, outputDir);
-      // TODO# merge-related move
-      Utilities.mvFileToFinalPath(outputDir, hconf, success, LOG, conf.getDpCtx(),
-              null, reporter);
-      if (success) {
-        LOG.info("jobCloseOp moved merged files to output dir: " + outputDir);
-      }
-      if (backupPath != null) {
-        fs.delete(backupPath, true);
+      Long mmWriteId = conf.getMmWriteId();
+      if (mmWriteId == null) {
+        Path backupPath = backupOutputPath(fs, outputDir);
+        Utilities.mvFileToFinalPath(
+            outputDir, hconf, success, LOG, conf.getDpCtx(), null, reporter);
+        if (success) {
+          LOG.info("jobCloseOp moved merged files to output dir: " + outputDir);
+        }
+        if (backupPath != null) {
+          fs.delete(backupPath, true);
+        }
+      } else {
+        int dpLevels = dpCtx == null ? 0 : dpCtx.getNumDPCols(),
+            lbLevels = conf.getListBucketingDepth();
+        // We don't expect missing buckets from mere (actually there should be no buckets),
+        // so just pass null as bucketing context. Union suffix should also be accounted for.
+        Utilities.handleMmTableFinalPath(outputDir.getParent(), null, hconf, success,
+            dpLevels, lbLevels, null, mmWriteId, reporter);
       }
+
     } catch (IOException e) {
       throw new HiveException("Failed jobCloseOp for AbstractFileMergeOperator",
           e);
@@ -290,4 +327,12 @@ public abstract class AbstractFileMergeOperator<T extends FileMergeDesc>
   public static String getOperatorName() {
     return "MERGE";
   }
+
+  protected final Path getOutPath() {
+    return outPath;
+  }
+
+  protected final void addIncompatibleFile(Path path) {
+    incompatFileSet.add(path);
+  }
 }

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
index 5902036..dda4b51 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
@@ -46,6 +46,7 @@ import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
 import org.apache.hadoop.hive.ql.CompilationOpContext;
 import org.apache.hadoop.hive.ql.ErrorMsg;
+import org.apache.hadoop.hive.ql.exec.Utilities.MissingBucketsContext;
 import org.apache.hadoop.hive.ql.io.AcidUtils;
 import org.apache.hadoop.hive.ql.io.HiveFileFormatUtils;
 import org.apache.hadoop.hive.ql.io.HiveKey;
@@ -98,7 +99,6 @@ import com.google.common.collect.Lists;
 public class FileSinkOperator extends TerminalOperator<FileSinkDesc> implements
     Serializable {
 
-  private static final String MANIFEST_EXTENSION = ".manifest";
   public static final Logger LOG = LoggerFactory.getLogger(FileSinkOperator.class);
   private static final boolean isInfoEnabled = LOG.isInfoEnabled();
   private static final boolean isDebugEnabled = LOG.isDebugEnabled();
@@ -1128,26 +1128,9 @@ public class FileSinkOperator extends TerminalOperator<FileSinkDesc> implements
           fsp.commit(fs, commitPaths);
         }
       }
-      if (!commitPaths.isEmpty()) {
-        Path manifestPath = getManifestDir(specPath, childSpecPathDynLinkedPartitions);
-        manifestPath = new Path(manifestPath, "_tmp." + ValidWriteIds.getMmFilePrefix(
-            conf.getMmWriteId()) + "_" + taskId + MANIFEST_EXTENSION);
-        Utilities.LOG14535.info("Writing manifest to " + manifestPath + " with " + commitPaths);
-        try {
-          // Don't overwrite the manifest... should fail if we have collisions.
-          // We assume one FSOP per task (per specPath), so we create it in specPath.
-          try (FSDataOutputStream out = fs.create(manifestPath, false)) {
-            if (out == null) {
-              throw new HiveException("Failed to create manifest at " + manifestPath);
-            }
-            out.writeInt(commitPaths.size());
-            for (Path path : commitPaths) {
-              out.writeUTF(path.toString());
-            }
-          }
-        } catch (IOException e) {
-          throw new HiveException(e);
-        }
+      if (conf.getMmWriteId() != null) {
+        Utilities.writeMmCommitManifest(commitPaths, specPath, fs, taskId, conf.getMmWriteId(),
+            childSpecPathDynLinkedPartitions);
       }
       // Only publish stats if this operator's flag was set to gather stats
       if (conf.isGatherStats()) {
@@ -1165,9 +1148,6 @@ public class FileSinkOperator extends TerminalOperator<FileSinkDesc> implements
     super.closeOp(abort);
   }
 
-  private static Path getManifestDir(Path specPath, String unionSuffix) {
-    return (unionSuffix == null) ? specPath : new Path(specPath, unionSuffix);
-  }
 
   /**
    * @return the name of the operator
@@ -1196,9 +1176,17 @@ public class FileSinkOperator extends TerminalOperator<FileSinkDesc> implements
           unionSuffix = conf.getDirName().getName();
         }
         if (!conf.isMmTable()) {
-          Utilities.mvFileToFinalPath(specPath, hconf, success, LOG, dpCtx, conf, reporter); // TODO# other callers
+          Utilities.mvFileToFinalPath(specPath, hconf, success, LOG, dpCtx, conf, reporter);
         } else {
-          handleMmTable(specPath, unionSuffix, hconf, success, dpCtx, lbCtx, conf, reporter);
+          int dpLevels = dpCtx == null ? 0 : dpCtx.getNumDPCols(),
+              lbLevels = lbCtx.calculateListBucketingLevel();
+          // TODO: why is it stored in both?
+          int numBuckets = (conf.getTable() != null) ? conf.getTable().getNumBuckets()
+              : (dpCtx != null ? dpCtx.getNumBuckets() : 0);
+          MissingBucketsContext mbc = new MissingBucketsContext(
+              conf.getTableInfo(), numBuckets, conf.getCompressed());
+          Utilities.handleMmTableFinalPath(specPath, unionSuffix, hconf, success,
+              dpLevels, lbLevels, mbc, conf.getMmWriteId(), reporter);
         }
       }
     } catch (IOException e) {
@@ -1207,152 +1195,6 @@ public class FileSinkOperator extends TerminalOperator<FileSinkDesc> implements
     super.jobCloseOp(hconf, success);
   }
 
-  private static FileStatus[] getMmDirectoryCandidates(FileSystem fs, Path path,
-      DynamicPartitionCtx dpCtx, ListBucketingCtx lbCtx, String unionSuffix, PathFilter filter)
-          throws IOException {
-    StringBuilder sb = new StringBuilder(path.toUri().getPath());
-    int dpLevels = dpCtx == null ? 0 : dpCtx.getNumDPCols(),
-        lbLevels = lbCtx == null ? 0 : lbCtx.getSkewedColNames().size();
-    for (int i = 0; i < dpLevels + lbLevels; i++) {
-      sb.append(Path.SEPARATOR).append("*");
-    }
-    if (unionSuffix != null) {
-      sb.append(Path.SEPARATOR).append(unionSuffix);
-    }
-    sb.append(Path.SEPARATOR).append("*"); // TODO: we could add exact mm prefix here
-    Utilities.LOG14535.info("Looking for files via: " + sb.toString());
-    Path pathPattern = new Path(path, sb.toString());
-    return fs.globStatus(pathPattern, filter);
-  }
-
-  private void handleMmTable(Path specPath, String unionSuffix, Configuration hconf,
-      boolean success, DynamicPartitionCtx dpCtx, ListBucketingCtx lbCtx, FileSinkDesc conf,
-      Reporter reporter) throws IOException, HiveException {
-    FileSystem fs = specPath.getFileSystem(hconf);
-    // Manifests would be at the root level, but the results at target level.
-    // TODO# special case - doesn't take bucketing into account
-    Path manifestDir = getManifestDir(specPath, unionSuffix);
-
-    ValidWriteIds.IdPathFilter filter = new ValidWriteIds.IdPathFilter(conf.getMmWriteId(), true);
-    if (!success) {
-      tryDeleteAllMmFiles(fs, specPath, manifestDir, dpCtx, lbCtx, unionSuffix, filter);
-      return;
-    }
-    FileStatus[] files = HiveStatsUtils.getFileStatusRecurse(manifestDir, 1, fs, filter);
-    Utilities.LOG14535.info("Looking for manifests in: " + manifestDir);
-    List<Path> manifests = new ArrayList<>();
-    if (files != null) {
-      for (FileStatus status : files) {
-        Path path = status.getPath();
-        if (path.getName().endsWith(MANIFEST_EXTENSION)) {
-          Utilities.LOG14535.info("Reading manifest " + path);
-          manifests.add(path);
-        }
-      }
-    }
-
-    Utilities.LOG14535.info("Looking for files in: " + specPath);
-    files = getMmDirectoryCandidates(fs, specPath, dpCtx, lbCtx, unionSuffix, filter);
-    ArrayList<FileStatus> results = new ArrayList<>();
-    if (files != null) {
-      for (FileStatus status : files) {
-        Path path = status.getPath();
-        Utilities.LOG14535.info("Looking at path: " + path + " from " + System.identityHashCode(this));
-        if (!status.isDirectory()) {
-          if (!path.getName().endsWith(MANIFEST_EXTENSION)) {
-            Utilities.LOG14535.warn("Unknown file found, deleting: " + path);
-            tryDelete(fs, path);
-          }
-        } else {
-          results.add(status);
-        }
-      }
-    }
-
-    HashSet<String> committed = new HashSet<>();
-    for (Path mfp : manifests) {
-      try (FSDataInputStream mdis = fs.open(mfp)) {
-        int fileCount = mdis.readInt();
-        for (int i = 0; i < fileCount; ++i) {
-          String nextFile = mdis.readUTF();
-          if (!committed.add(nextFile)) {
-            throw new HiveException(nextFile + " was specified in multiple manifests");
-          }
-        }
-      }
-    }
-
-    for (FileStatus status : results) {
-      for (FileStatus child : fs.listStatus(status.getPath())) {
-        Path childPath = child.getPath();
-        if (committed.remove(childPath.toString())) continue; // A good file.
-        Utilities.LOG14535.info("Deleting " + childPath + " that was not committed");
-        // We should actually succeed here - if we fail, don't commit the query.
-        if (!fs.delete(childPath, true)) {
-          throw new HiveException("Failed to delete an uncommitted path " + childPath);
-        }
-      }
-    }
-
-    if (!committed.isEmpty()) {
-      throw new HiveException("The following files were committed but not found: " + committed);
-    }
-    for (Path mfp : manifests) {
-      Utilities.LOG14535.info("Deleting manifest " + mfp);
-      tryDelete(fs, mfp);
-    }
-    // Delete the manifest directory if we only created it for manifests; otherwise the
-    // dynamic partition loader will find it and try to load it as a partition... what a mess.
-    if (manifestDir != specPath) {
-      FileStatus[] remainingFiles = fs.listStatus(manifestDir);
-      if (remainingFiles == null || remainingFiles.length == 0) {
-        Utilities.LOG14535.info("Deleting directory " + manifestDir);
-        tryDelete(fs, manifestDir);
-      }
-    }
-
-    if (results.isEmpty()) return;
-
-    // TODO: see HIVE-14886 - removeTempOrDuplicateFiles is broken for list bucketing,
-    //       so maintain parity here by not calling it at all.
-    if (lbCtx != null) return;
-    FileStatus[] finalResults = results.toArray(new FileStatus[results.size()]);
-    List<Path> emptyBuckets = Utilities.removeTempOrDuplicateFiles(
-        fs, finalResults, dpCtx, conf, hconf);
-    // create empty buckets if necessary
-    if (emptyBuckets.size() > 0) {
-      Utilities.createEmptyBuckets(hconf, emptyBuckets, conf, reporter);
-    }
-  }
-
-  private void tryDeleteAllMmFiles(FileSystem fs, Path specPath, Path manifestDir,
-      DynamicPartitionCtx dpCtx, ListBucketingCtx lbCtx, String unionSuffix,
-      ValidWriteIds.IdPathFilter filter) throws IOException {
-    FileStatus[] files = getMmDirectoryCandidates(fs, specPath, dpCtx, lbCtx, unionSuffix, filter);
-    if (files != null) {
-      for (FileStatus status : files) {
-        Utilities.LOG14535.info("Deleting " + status.getPath() + " on failure");
-        tryDelete(fs, status.getPath());
-      }
-    }
-    files = HiveStatsUtils.getFileStatusRecurse(manifestDir, 1, fs, filter);
-    if (files != null) {
-      for (FileStatus status : files) {
-        Utilities.LOG14535.info("Deleting " + status.getPath() + " on failure");
-        tryDelete(fs, status.getPath());
-      }
-    }
-  }
-
-
-  private void tryDelete(FileSystem fs, Path path) {
-    try {
-      fs.delete(path, true);
-    } catch (IOException ex) {
-      LOG.error("Failed to delete " + path, ex);
-    }
-  }
-
   @Override
   public OperatorType getType() {
     return OperatorType.FILESINK;
@@ -1427,7 +1269,6 @@ public class FileSinkOperator extends TerminalOperator<FileSinkDesc> implements
     for (Map.Entry<String, FSPaths> entry : valToPaths.entrySet()) {
       String fspKey = entry.getKey();     // DP/LB
       FSPaths fspValue = entry.getValue();
-      // TODO# useful code as reference, as it takes apart the crazy paths
       // for bucketed tables, hive.optimize.sort.dynamic.partition optimization
       // adds the taskId to the fspKey.
       if (conf.getDpSortState().equals(DPSortState.PARTITION_BUCKET_SORTED)) {

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java
index 9bc4836..f2b8ca3 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java
@@ -256,6 +256,9 @@ public class MoveTask extends Task<MoveWork> implements Serializable {
 
   @Override
   public int execute(DriverContext driverContext) {
+    Utilities.LOG14535.info("Executing MoveWork " + System.identityHashCode(work)
+        + " with " + work.getLoadFileWork() + "; " + work.getLoadTableWork() + "; "
+        + work.getLoadMultiFilesWork(), new Exception());
 
     try {
       if (driverContext.getCtx().getExplainAnalyze() == AnalyzeState.RUNNING) {
@@ -315,15 +318,14 @@ public class MoveTask extends Task<MoveWork> implements Serializable {
 
         boolean isAcid = work.getLoadTableWork().getWriteType() != AcidUtils.Operation.NOT_ACID;
         if (tbd.isMmTable() && isAcid) {
-          // TODO# need to make sure ACID writes to final directories. Otherwise, might need to move.
-          throw new HiveException("ACID and MM are not supported");
+           throw new HiveException("ACID and MM are not supported");
         }
 
         // Create a data container
         DataContainer dc = null;
         if (tbd.getPartitionSpec().size() == 0) {
           dc = new DataContainer(table.getTTable());
-          Utilities.LOG14535.info("loadTable called from " + tbd.getSourcePath() + " into " + tbd.getTable());
+          Utilities.LOG14535.info("loadTable called from " + tbd.getSourcePath() + " into " + tbd.getTable().getTableName(), new Exception());
           db.loadTable(tbd.getSourcePath(), tbd.getTable().getTableName(), tbd.getReplace(),
               work.isSrcLocal(), isSkewedStoredAsDirs(tbd), isAcid, hasFollowingStatsTask(),
               tbd.getMmWriteId());

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/exec/OrcFileMergeOperator.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/OrcFileMergeOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/OrcFileMergeOperator.java
index e3cb765..835791b 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/OrcFileMergeOperator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/OrcFileMergeOperator.java
@@ -75,6 +75,7 @@ public class OrcFileMergeOperator extends
   private void processKeyValuePairs(Object key, Object value)
       throws HiveException {
     String filePath = "";
+    boolean exception = false;
     try {
       OrcFileValueWrapper v;
       OrcFileKeyWrapper k;
@@ -87,12 +88,15 @@ public class OrcFileMergeOperator extends
       // skip incompatible file, files that are missing stripe statistics are set to incompatible
       if (k.isIncompatFile()) {
         LOG.warn("Incompatible ORC file merge! Stripe statistics is missing. " + k.getInputPath());
-        incompatFileSet.add(k.getInputPath());
+        addIncompatibleFile(k.getInputPath());
         return;
       }
 
       filePath = k.getInputPath().toUri().getPath();
 
+      Utilities.LOG14535.info("OrcFileMergeOperator processing " + filePath, new Exception());
+
+
       fixTmpPath(k.getInputPath().getParent());
 
       v = (OrcFileValueWrapper) value;
@@ -126,6 +130,7 @@ public class OrcFileMergeOperator extends
           options.bufferSize(compressBuffSize).enforceBufferSize();
         }
 
+        Path outPath = getOutPath();
         outWriter = OrcFile.createWriter(outPath, options);
         if (isLogDebugEnabled) {
           LOG.info("ORC merge file output path: " + outPath);
@@ -133,7 +138,7 @@ public class OrcFileMergeOperator extends
       }
 
       if (!checkCompatibility(k)) {
-        incompatFileSet.add(k.getInputPath());
+        addIncompatibleFile(k.getInputPath());
         return;
       }
 
@@ -164,7 +169,7 @@ public class OrcFileMergeOperator extends
         outWriter.appendUserMetadata(v.getUserMetadata());
       }
     } catch (Throwable e) {
-      this.exception = true;
+      exception = true;
       LOG.error("Closing operator..Exception: " + ExceptionUtils.getStackTrace(e));
       throw new HiveException(e);
     } finally {

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/exec/RCFileMergeOperator.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/RCFileMergeOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/RCFileMergeOperator.java
index 4dea1d2..349b459 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/RCFileMergeOperator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/RCFileMergeOperator.java
@@ -77,7 +77,7 @@ public class RCFileMergeOperator
         codec = key.getCodec();
         columnNumber = key.getKeyBuffer().getColumnNumber();
         RCFileOutputFormat.setColumnNumber(jc, columnNumber);
-        outWriter = new RCFile.Writer(fs, jc, outPath, null, codec);
+        outWriter = new RCFile.Writer(fs, jc, getOutPath(), null, codec);
       }
 
       boolean sameCodec = ((codec == key.getCodec()) || codec.getClass().equals(
@@ -94,7 +94,6 @@ public class RCFileMergeOperator
           key.getRecordLength(), key.getKeyLength(),
           key.getCompressedKeyLength());
     } catch (Throwable e) {
-      this.exception = true;
       closeOp(true);
       throw new HiveException(e);
     }

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
index d343e32..49bdd84 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
@@ -83,6 +83,8 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.filecache.DistributedCache;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.fs.ContentSummary;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -94,6 +96,7 @@ import org.apache.hadoop.hive.common.HiveInterruptUtils;
 import org.apache.hadoop.hive.common.HiveStatsUtils;
 import org.apache.hadoop.hive.common.JavaUtils;
 import org.apache.hadoop.hive.common.StatsSetupConst;
+import org.apache.hadoop.hive.common.ValidWriteIds;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
 import org.apache.hadoop.hive.metastore.MetaStoreUtils;
@@ -146,6 +149,7 @@ import org.apache.hadoop.hive.ql.parse.SemanticException;
 import org.apache.hadoop.hive.ql.plan.BaseWork;
 import org.apache.hadoop.hive.ql.plan.DynamicPartitionCtx;
 import org.apache.hadoop.hive.ql.plan.FileSinkDesc;
+import org.apache.hadoop.hive.ql.plan.ListBucketingCtx;
 import org.apache.hadoop.hive.ql.plan.MapWork;
 import org.apache.hadoop.hive.ql.plan.MapredWork;
 import org.apache.hadoop.hive.ql.plan.MergeJoinWork;
@@ -1411,7 +1415,7 @@ public final class Utilities {
       boolean success, Logger log, DynamicPartitionCtx dpCtx, FileSinkDesc conf,
       Reporter reporter) throws IOException,
       HiveException {
-
+    
     FileSystem fs = specPath.getFileSystem(hconf);
     Path tmpPath = Utilities.toTempPath(specPath);
     Path taskTmpPath = Utilities.toTaskTempPath(specPath);
@@ -1422,12 +1426,14 @@ public final class Utilities {
         PerfLogger perfLogger = SessionState.getPerfLogger();
         perfLogger.PerfLogBegin("FileSinkOperator", "RemoveTempOrDuplicateFiles");
         // remove any tmp file or double-committed output files
-        List<Path> emptyBuckets = Utilities.removeTempOrDuplicateFiles(fs, statuses, dpCtx, conf, hconf);
+        List<Path> emptyBuckets = Utilities.removeTempOrDuplicateFiles(
+            fs, statuses, dpCtx, conf, hconf);
         perfLogger.PerfLogEnd("FileSinkOperator", "RemoveTempOrDuplicateFiles");
         // create empty buckets if necessary
         if (emptyBuckets.size() > 0) {
           perfLogger.PerfLogBegin("FileSinkOperator", "CreateEmptyBuckets");
-          createEmptyBuckets(hconf, emptyBuckets, conf, reporter);
+          createEmptyBuckets(
+              hconf, emptyBuckets, conf.getCompressed(), conf.getTableInfo(), reporter);
           perfLogger.PerfLogEnd("FileSinkOperator", "CreateEmptyBuckets");
         }
         // move to the file destination
@@ -1457,7 +1463,7 @@ public final class Utilities {
    * @throws IOException
    */
   static void createEmptyBuckets(Configuration hconf, List<Path> paths,
-      FileSinkDesc conf, Reporter reporter)
+      boolean isCompressed, TableDesc tableInfo, Reporter reporter)
       throws HiveException, IOException {
 
     JobConf jc;
@@ -1469,13 +1475,11 @@ public final class Utilities {
     }
     HiveOutputFormat<?, ?> hiveOutputFormat = null;
     Class<? extends Writable> outputClass = null;
-    boolean isCompressed = conf.getCompressed();
-    TableDesc tableInfo = conf.getTableInfo();
     try {
       Serializer serializer = (Serializer) tableInfo.getDeserializerClass().newInstance();
       serializer.initialize(null, tableInfo.getProperties());
       outputClass = serializer.getSerializedClass();
-      hiveOutputFormat = HiveFileFormatUtils.getHiveOutputFormat(hconf, conf.getTableInfo());
+      hiveOutputFormat = HiveFileFormatUtils.getHiveOutputFormat(hconf, tableInfo);
     } catch (SerDeException e) {
       throw new HiveException(e);
     } catch (InstantiationException e) {
@@ -1518,13 +1522,21 @@ public final class Utilities {
    */
   public static List<Path> removeTempOrDuplicateFiles(FileSystem fs, FileStatus[] fileStats,
       DynamicPartitionCtx dpCtx, FileSinkDesc conf, Configuration hconf) throws IOException {
+    int dpLevels = dpCtx == null ? 0 : dpCtx.getNumDPCols(),
+        numBuckets = (conf != null && conf.getTable() != null)
+          ? conf.getTable().getNumBuckets() : 0;
+    return removeTempOrDuplicateFiles(fs, fileStats, dpLevels, numBuckets, hconf);
+  }
+
+  public static List<Path> removeTempOrDuplicateFiles(FileSystem fs, FileStatus[] fileStats,
+      int dpLevels, int numBuckets, Configuration hconf) throws IOException {
     if (fileStats == null) {
       return null;
     }
 
     List<Path> result = new ArrayList<Path>();
     HashMap<String, FileStatus> taskIDToFile = null;
-    if (dpCtx != null) {
+    if (dpLevels > 0) {
       FileStatus parts[] = fileStats;
 
       for (int i = 0; i < parts.length; ++i) {
@@ -1543,14 +1555,14 @@ public final class Utilities {
 
         taskIDToFile = removeTempOrDuplicateFiles(items, fs);
         // if the table is bucketed and enforce bucketing, we should check and generate all buckets
-        if (dpCtx.getNumBuckets() > 0 && taskIDToFile != null && !"tez".equalsIgnoreCase(hconf.get(ConfVars.HIVE_EXECUTION_ENGINE.varname))) {
+        if (numBuckets > 0 && taskIDToFile != null && !"tez".equalsIgnoreCase(hconf.get(ConfVars.HIVE_EXECUTION_ENGINE.varname))) {
           // refresh the file list
           items = fs.listStatus(parts[i].getPath());
           // get the missing buckets and generate empty buckets
           String taskID1 = taskIDToFile.keySet().iterator().next();
           Path bucketPath = taskIDToFile.values().iterator().next().getPath();
           Utilities.LOG14535.info("Bucket path " + bucketPath);
-          for (int j = 0; j < dpCtx.getNumBuckets(); ++j) {
+          for (int j = 0; j < numBuckets; ++j) {
             addBucketFileIfMissing(result, taskIDToFile, taskID1, bucketPath, j);
           }
         }
@@ -1561,13 +1573,13 @@ public final class Utilities {
         return result;
       }
       taskIDToFile = removeTempOrDuplicateFiles(items, fs);
-      if(taskIDToFile != null && taskIDToFile.size() > 0 && conf != null && conf.getTable() != null
-          && (conf.getTable().getNumBuckets() > taskIDToFile.size()) && !"tez".equalsIgnoreCase(hconf.get(ConfVars.HIVE_EXECUTION_ENGINE.varname))) {
+      if(taskIDToFile != null && taskIDToFile.size() > 0 && (numBuckets > taskIDToFile.size())
+          && !"tez".equalsIgnoreCase(hconf.get(ConfVars.HIVE_EXECUTION_ENGINE.varname))) {
           // get the missing buckets and generate empty buckets for non-dynamic partition
         String taskID1 = taskIDToFile.keySet().iterator().next();
         Path bucketPath = taskIDToFile.values().iterator().next().getPath();
         Utilities.LOG14535.info("Bucket path " + bucketPath);
-        for (int j = 0; j < conf.getTable().getNumBuckets(); ++j) {
+        for (int j = 0; j < numBuckets; ++j) {
           addBucketFileIfMissing(result, taskIDToFile, taskID1, bucketPath, j);
         }
       }
@@ -3746,4 +3758,191 @@ public final class Utilities {
     String suffix = "KMGTPE".charAt(exp-1) + "";
     return String.format("%.2f%sB", bytes / Math.pow(unit, exp), suffix);
   }
+
+  private static final String MANIFEST_EXTENSION = ".manifest";
+
+  private static Path getManifestDir(Path specPath, String unionSuffix) {
+    return (unionSuffix == null) ? specPath : new Path(specPath, unionSuffix);
+  }
+
+  private static void tryDelete(FileSystem fs, Path path) {
+    try {
+      fs.delete(path, true);
+    } catch (IOException ex) {
+      LOG.error("Failed to delete " + path, ex);
+    }
+  }
+
+  private static FileStatus[] getMmDirectoryCandidates(FileSystem fs, Path path,
+      int dpLevels, int lbLevels, String unionSuffix, PathFilter filter) throws IOException {
+    StringBuilder sb = new StringBuilder(path.toUri().getPath());
+    for (int i = 0; i < dpLevels + lbLevels; i++) {
+      sb.append(Path.SEPARATOR).append("*");
+    }
+    if (unionSuffix != null) {
+      sb.append(Path.SEPARATOR).append(unionSuffix);
+    }
+    sb.append(Path.SEPARATOR).append("*"); // TODO: we could add exact mm prefix here
+    Utilities.LOG14535.info("Looking for files via: " + sb.toString());
+    Path pathPattern = new Path(path, sb.toString());
+    return fs.globStatus(pathPattern, filter);
+  }
+
+  private static void tryDeleteAllMmFiles(FileSystem fs, Path specPath, Path manifestDir,
+      int dpLevels, int lbLevels, String unionSuffix, ValidWriteIds.IdPathFilter filter)
+          throws IOException {
+    FileStatus[] files = getMmDirectoryCandidates(
+        fs, specPath, dpLevels, lbLevels, unionSuffix, filter);
+    if (files != null) {
+      for (FileStatus status : files) {
+        Utilities.LOG14535.info("Deleting " + status.getPath() + " on failure");
+        tryDelete(fs, status.getPath());
+      }
+    }
+    files = HiveStatsUtils.getFileStatusRecurse(manifestDir, 1, fs, filter);
+    if (files != null) {
+      for (FileStatus status : files) {
+        Utilities.LOG14535.info("Deleting " + status.getPath() + " on failure");
+        tryDelete(fs, status.getPath());
+      }
+    }
+  }
+
+
+  public static void writeMmCommitManifest(List<Path> commitPaths, Path specPath, FileSystem fs,
+      String taskId, Long mmWriteId, String unionSuffix) throws HiveException {
+    if (commitPaths.isEmpty()) return;
+    Path manifestPath = getManifestDir(specPath, unionSuffix);
+    manifestPath = new Path(manifestPath, "_tmp." + ValidWriteIds.getMmFilePrefix(
+        mmWriteId) + "_" + taskId + MANIFEST_EXTENSION);
+    Utilities.LOG14535.info("Writing manifest to " + manifestPath + " with " + commitPaths);
+    try {
+      // Don't overwrite the manifest... should fail if we have collisions.
+      // We assume one FSOP per task (per specPath), so we create it in specPath.
+      try (FSDataOutputStream out = fs.create(manifestPath, false)) {
+        if (out == null) {
+          throw new HiveException("Failed to create manifest at " + manifestPath);
+        }
+        out.writeInt(commitPaths.size());
+        for (Path path : commitPaths) {
+          out.writeUTF(path.toString());
+        }
+      }
+    } catch (IOException e) {
+      throw new HiveException(e);
+    }
+  }
+
+  public static final class MissingBucketsContext {
+    public final TableDesc tableInfo;
+    public final int numBuckets;
+    public final boolean isCompressed;
+    public MissingBucketsContext(TableDesc tableInfo, int numBuckets, boolean isCompressed) {
+      this.tableInfo = tableInfo;
+      this.numBuckets = numBuckets;
+      this.isCompressed = isCompressed;
+    }
+  }
+
+  public static void handleMmTableFinalPath(Path specPath, String unionSuffix, Configuration hconf,
+      boolean success, int dpLevels, int lbLevels, MissingBucketsContext mbc, long mmWriteId,
+      Reporter reporter) throws IOException, HiveException {
+    FileSystem fs = specPath.getFileSystem(hconf);
+    // Manifests would be at the root level, but the results at target level.
+    // TODO# special case - doesn't take bucketing into account
+    Path manifestDir = getManifestDir(specPath, unionSuffix);
+
+    ValidWriteIds.IdPathFilter filter = new ValidWriteIds.IdPathFilter(mmWriteId, true);
+    if (!success) {
+      tryDeleteAllMmFiles(fs, specPath, manifestDir, dpLevels, lbLevels, unionSuffix, filter);
+      return;
+    }
+    FileStatus[] files = HiveStatsUtils.getFileStatusRecurse(manifestDir, 1, fs, filter);
+    Utilities.LOG14535.info("Looking for manifests in: " + manifestDir + " (" + mmWriteId + ")");
+    List<Path> manifests = new ArrayList<>();
+    if (files != null) {
+      for (FileStatus status : files) {
+        Path path = status.getPath();
+        if (path.getName().endsWith(MANIFEST_EXTENSION)) {
+          Utilities.LOG14535.info("Reading manifest " + path);
+          manifests.add(path);
+        }
+      }
+    }
+
+    Utilities.LOG14535.info("Looking for files in: " + specPath);
+    files = getMmDirectoryCandidates(fs, specPath, dpLevels, lbLevels, unionSuffix, filter);
+    ArrayList<FileStatus> results = new ArrayList<>();
+    if (files != null) {
+      for (FileStatus status : files) {
+        Path path = status.getPath();
+        Utilities.LOG14535.info("Looking at path: " + path);
+        if (!status.isDirectory()) {
+          if (!path.getName().endsWith(MANIFEST_EXTENSION)) {
+            Utilities.LOG14535.warn("Unknown file found, deleting: " + path);
+            tryDelete(fs, path);
+          }
+        } else {
+          results.add(status);
+        }
+      }
+    }
+
+    HashSet<String> committed = new HashSet<>();
+    for (Path mfp : manifests) {
+      try (FSDataInputStream mdis = fs.open(mfp)) {
+        int fileCount = mdis.readInt();
+        for (int i = 0; i < fileCount; ++i) {
+          String nextFile = mdis.readUTF();
+          if (!committed.add(nextFile)) {
+            throw new HiveException(nextFile + " was specified in multiple manifests");
+          }
+        }
+      }
+    }
+
+    for (FileStatus status : results) {
+      for (FileStatus child : fs.listStatus(status.getPath())) {
+        Path childPath = child.getPath();
+        if (committed.remove(childPath.toString())) continue; // A good file.
+        Utilities.LOG14535.info("Deleting " + childPath + " that was not committed");
+        // We should actually succeed here - if we fail, don't commit the query.
+        if (!fs.delete(childPath, true)) {
+          throw new HiveException("Failed to delete an uncommitted path " + childPath);
+        }
+      }
+    }
+
+    if (!committed.isEmpty()) {
+      throw new HiveException("The following files were committed but not found: " + committed);
+    }
+    for (Path mfp : manifests) {
+      Utilities.LOG14535.info("Deleting manifest " + mfp);
+      tryDelete(fs, mfp);
+    }
+    // Delete the manifest directory if we only created it for manifests; otherwise the
+    // dynamic partition loader will find it and try to load it as a partition... what a mess.
+    if (manifestDir != specPath) {
+      FileStatus[] remainingFiles = fs.listStatus(manifestDir);
+      if (remainingFiles == null || remainingFiles.length == 0) {
+        Utilities.LOG14535.info("Deleting directory " + manifestDir);
+        tryDelete(fs, manifestDir);
+      }
+    }
+
+    if (results.isEmpty()) return;
+
+    // TODO: see HIVE-14886 - removeTempOrDuplicateFiles is broken for list bucketing,
+    //       so maintain parity here by not calling it at all.
+    if (lbLevels != 0) return;
+    FileStatus[] finalResults = results.toArray(new FileStatus[results.size()]);
+    List<Path> emptyBuckets = Utilities.removeTempOrDuplicateFiles(
+        fs, finalResults, dpLevels, mbc == null ? 0 : mbc.numBuckets, hconf);
+    // create empty buckets if necessary
+    if (emptyBuckets.size() > 0) {
+      assert mbc != null;
+      Utilities.createEmptyBuckets(hconf, emptyBuckets, mbc.isCompressed, mbc.tableInfo, reporter);
+    }
+  }
+
 }

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateMapper.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateMapper.java b/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateMapper.java
index bd537cd..d013c6f 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateMapper.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateMapper.java
@@ -234,6 +234,7 @@ public class ColumnTruncateMapper extends MapReduceBase implements
       ) throws HiveException, IOException {
     FileSystem fs = outputPath.getFileSystem(job);
     Path backupPath = backupOutputPath(fs, outputPath, job);
+    // TODO# special case - what is this about?
     Utilities.mvFileToFinalPath(outputPath, job, success, LOG, dynPartCtx, null,
       reporter);
     fs.delete(backupPath, true);

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
index e66948f..9a1c1fa 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
@@ -1842,6 +1842,7 @@ private void constructOneLBLocationMap(FileStatus fSta,
         if (!s.isDirectory()) {
           throw new HiveException("partition " + s.getPath() + " is not a directory!");
         }
+        Utilities.LOG14535.info("Found DP " + s.getPath());
         validPartitions.add(s.getPath());
       }
     } catch (IOException e) {

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java
index 675bfd0..79ef4d0 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java
@@ -93,6 +93,7 @@ import org.apache.hadoop.hive.ql.plan.FetchWork;
 import org.apache.hadoop.hive.ql.plan.FileMergeDesc;
 import org.apache.hadoop.hive.ql.plan.FileSinkDesc;
 import org.apache.hadoop.hive.ql.plan.FilterDesc.SampleDesc;
+import org.apache.hadoop.hive.ql.plan.DependencyCollectionWork;
 import org.apache.hadoop.hive.ql.plan.LoadFileDesc;
 import org.apache.hadoop.hive.ql.plan.MapWork;
 import org.apache.hadoop.hive.ql.plan.MapredLocalWork;
@@ -1256,23 +1257,28 @@ public final class GenMapRedUtils {
    List<Task<MoveWork>> mvTasks, HiveConf conf,
    Task<? extends Serializable> currTask) throws SemanticException {
 
+
     //
     // 1. create the operator tree
     //
     FileSinkDesc fsInputDesc = fsInput.getConf();
+    Utilities.LOG14535.info("Creating merge work from " + System.identityHashCode(fsInput)
+        + " with write ID " + (fsInputDesc.isMmTable() ? fsInputDesc.getMmWriteId() : null) + " into " + finalName);
 
     // Create a TableScan operator
     RowSchema inputRS = fsInput.getSchema();
     TableScanOperator tsMerge = GenMapRedUtils.createTemporaryTableScanOperator(
         fsInput.getCompilationOpContext(), inputRS);
 
+    Long srcMmWriteId = fsInputDesc.isMmTable() ? fsInputDesc.getMmWriteId() : null;
+
     // Create a FileSink operator
     TableDesc ts = (TableDesc) fsInputDesc.getTableInfo().clone();
-    // TODO# special case #N - merge FS is created here
-    FileSinkDesc fsOutputDesc = new FileSinkDesc(finalName, ts,
-      conf.getBoolVar(ConfVars.COMPRESSRESULT));
-    FileSinkOperator fsOutput = (FileSinkOperator) OperatorFactory.getAndMakeChild(
-      fsOutputDesc, inputRS, tsMerge);
+    FileSinkDesc fsOutputDesc = new FileSinkDesc(
+        finalName, ts, conf.getBoolVar(ConfVars.COMPRESSRESULT));
+    fsOutputDesc.setMmWriteId(srcMmWriteId);
+    // Create and attach the filesink for the merge. We don't actually need it for anything here.
+    OperatorFactory.getAndMakeChild(fsOutputDesc, inputRS, tsMerge);
 
     // If the input FileSinkOperator is a dynamic partition enabled, the tsMerge input schema
     // needs to include the partition column, and the fsOutput should have
@@ -1305,9 +1311,7 @@ public final class GenMapRedUtils {
     //
     // 2. Constructing a conditional task consisting of a move task and a map reduce task
     //
-    // TODO# movetask is created here; handle MM tables
-    MoveWork dummyMv = new MoveWork(null, null, null,
-         new LoadFileDesc(fsInputDesc.getFinalDirName(), finalName, true, null, null), false);
+    Path inputDirName = fsInputDesc.getMergeInputDirName();
     MapWork cplan;
     Serializable work;
 
@@ -1348,8 +1352,15 @@ public final class GenMapRedUtils {
     cplan.setInputformat("org.apache.hadoop.hive.ql.io.CombineHiveInputFormat");
     // NOTE: we should gather stats in MR1 rather than MR2 at merge job since we don't
     // know if merge MR2 will be triggered at execution time
+    MoveWork dummyMv = null;
+    if (srcMmWriteId == null) {
+      // Only create the movework for non-MM table. No action needed for a MM table.
+      Utilities.LOG14535.info("creating dummy movetask for merge (with lfd)");
+      dummyMv = new MoveWork(null, null, null,
+         new LoadFileDesc(inputDirName, finalName, true, null, null), false);
+    }
     ConditionalTask cndTsk = GenMapRedUtils.createCondTask(conf, currTask, dummyMv, work,
-        fsInputDesc.getFinalDirName().toString());
+        fsInputDesc.getMergeInputDirName().toString());
 
     // keep the dynamic partition context in conditional task resolver context
     ConditionalResolverMergeFilesCtx mrCtx =
@@ -1360,7 +1371,13 @@ public final class GenMapRedUtils {
     //
     // 3. add the moveTask as the children of the conditional task
     //
-    linkMoveTask(fsOutput, cndTsk, mvTasks, conf, dependencyTask);
+    // Use the original fsOp path here in case of MM - while the new FSOP merges files inside the
+    // MM directory, the original MoveTask still commits based on the parent. Note that this path
+    // can only be triggered for a merge that's part of insert for now; MM tables do not support
+    // concatenate. Keeping the old logic for non-MM tables with temp directories and stuff.
+    Path fsopPath = srcMmWriteId != null
+        ? fsInputDesc.getFinalDirName() : fsOutputDesc.getFinalDirName();
+    linkMoveTask(fsopPath, cndTsk, mvTasks, conf, dependencyTask);
   }
 
   /**
@@ -1373,11 +1390,11 @@ public final class GenMapRedUtils {
    * @param hconf
    * @param dependencyTask
    */
-  public static void linkMoveTask(FileSinkOperator newOutput,
+  private static void linkMoveTask(Path fsopPath,
       ConditionalTask cndTsk, List<Task<MoveWork>> mvTasks, HiveConf hconf,
       DependencyCollectionTask dependencyTask) {
 
-    Task<MoveWork> mvTask = GenMapRedUtils.findMoveTask(mvTasks, newOutput);
+    Task<MoveWork> mvTask = GenMapRedUtils.findMoveTaskForFsopOutput(mvTasks, fsopPath);
 
     for (Task<? extends Serializable> tsk : cndTsk.getListTasks()) {
       linkMoveTask(mvTask, tsk, hconf, dependencyTask);
@@ -1392,7 +1409,7 @@ public final class GenMapRedUtils {
    * @param hconf
    * @param dependencyTask
    */
-  public static void linkMoveTask(Task<MoveWork> mvTask,
+  private static void linkMoveTask(Task<MoveWork> mvTask,
       Task<? extends Serializable> task, HiveConf hconf,
       DependencyCollectionTask dependencyTask) {
 
@@ -1527,10 +1544,11 @@ public final class GenMapRedUtils {
     TableScanOperator topOp,  FileSinkDesc fsDesc) {
 
     ArrayList<String> aliases = new ArrayList<String>();
-    Path inputDir = fsDesc.getFinalDirName();
+    Path inputDir = fsDesc.getMergeInputDirName();
     TableDesc tblDesc = fsDesc.getTableInfo();
     aliases.add(inputDir.toString()); // dummy alias: just use the input path
 
+    Utilities.LOG14535.info("createMRWorkForMergingFiles for " + inputDir);
     // constructing the default MapredWork
     MapredWork cMrPlan = GenMapRedUtils.getMapRedWorkFromConf(conf);
     MapWork cplan = cMrPlan.getMapWork();
@@ -1555,8 +1573,9 @@ public final class GenMapRedUtils {
    */
   public static MapWork createMergeTask(FileSinkDesc fsInputDesc, Path finalName,
       boolean hasDynamicPartitions, CompilationOpContext ctx) throws SemanticException {
+    
+    Path inputDir = fsInputDesc.getMergeInputDirName();
 
-    Path inputDir = fsInputDesc.getFinalDirName();
     TableDesc tblDesc = fsInputDesc.getTableInfo();
 
     List<Path> inputDirs = new ArrayList<Path>(1);
@@ -1580,6 +1599,7 @@ public final class GenMapRedUtils {
           + " format other than RCFile or ORCFile");
     }
 
+    Utilities.LOG14535.info("creating mergefilework from " + inputDirs + " to " + finalName);
     // create the merge file work
     MergeFileWork work = new MergeFileWork(inputDirs, finalName,
         hasDynamicPartitions, tblDesc.getInputFileFormatClass().getName());
@@ -1602,6 +1622,7 @@ public final class GenMapRedUtils {
     } else {
       fmd = new OrcFileMergeDesc();
     }
+    fmd.setMmWriteId(fsInputDesc.getMmWriteId());
     fmd.setDpCtx(fsInputDesc.getDynPartCtx());
     fmd.setOutputPath(finalName);
     fmd.setHasDynamicPartitions(work.hasDynamicPartitions());
@@ -1635,6 +1656,7 @@ public final class GenMapRedUtils {
   public static ConditionalTask createCondTask(HiveConf conf,
       Task<? extends Serializable> currTask, MoveWork mvWork,
       Serializable mergeWork, String inputPath) {
+    Utilities.LOG14535.info("Creating conditional merge task for " + inputPath);
 
     // There are 3 options for this ConditionalTask:
     // 1) Merge the partitions
@@ -1642,10 +1664,14 @@ public final class GenMapRedUtils {
     // 3) Merge some partitions and move other partitions (i.e. merge some partitions and don't
     // merge others) in this case the merge is done first followed by the move to prevent
     // conflicts.
+    // TODO: if we are not dealing with concatenate DDL, we should not create a merge+move path
+    //       because it should be impossible to get incompatible outputs.
+    // Create a dummy task if no move is needed.
+    Serializable moveWork = mvWork != null ? mvWork : new DependencyCollectionWork();
     Task<? extends Serializable> mergeOnlyMergeTask = TaskFactory.get(mergeWork, conf);
-    Task<? extends Serializable> moveOnlyMoveTask = TaskFactory.get(mvWork, conf);
+    Task<? extends Serializable> moveOnlyMoveTask = TaskFactory.get(moveWork, conf);
     Task<? extends Serializable> mergeAndMoveMergeTask = TaskFactory.get(mergeWork, conf);
-    Task<? extends Serializable> mergeAndMoveMoveTask = TaskFactory.get(mvWork, conf);
+    Task<? extends Serializable> mergeAndMoveMoveTask = TaskFactory.get(moveWork, conf);
 
     // NOTE! It is necessary merge task is the parent of the move task, and not
     // the other way around, for the proper execution of the execute method of
@@ -1653,7 +1679,7 @@ public final class GenMapRedUtils {
     mergeAndMoveMergeTask.addDependentTask(mergeAndMoveMoveTask);
 
     List<Serializable> listWorks = new ArrayList<Serializable>();
-    listWorks.add(mvWork);
+    listWorks.add(moveWork);
     listWorks.add(mergeWork);
 
     ConditionalWork cndWork = new ConditionalWork(listWorks);
@@ -1689,8 +1715,8 @@ public final class GenMapRedUtils {
         .isSkewedStoredAsDir();
   }
 
-  public static Task<MoveWork> findMoveTask(
-      List<Task<MoveWork>> mvTasks, FileSinkOperator fsOp) {
+  public static Task<MoveWork> findMoveTaskForFsopOutput(
+      List<Task<MoveWork>> mvTasks, Path fsopFinalDir) {
     // find the move task
     for (Task<MoveWork> mvTsk : mvTasks) {
       MoveWork mvWork = mvTsk.getWork();
@@ -1700,9 +1726,10 @@ public final class GenMapRedUtils {
       } else if (mvWork.getLoadTableWork() != null) {
         srcDir = mvWork.getLoadTableWork().getSourcePath();
       }
+      Utilities.LOG14535.info("Observing MoveWork " + System.identityHashCode(mvWork)
+          + " with " + srcDir + " while looking for " + fsopFinalDir);
 
-      if ((srcDir != null)
-          && (srcDir.equals(fsOp.getConf().getFinalDirName()))) {
+      if ((srcDir != null) && srcDir.equals(fsopFinalDir)) {
         return mvTsk;
       }
     }
@@ -1722,59 +1749,58 @@ public final class GenMapRedUtils {
       Task<? extends Serializable> currTask, boolean isInsertTable) {
 
     // Has the user enabled merging of files for map-only jobs or for all jobs
-    if ((mvTasks != null) && (!mvTasks.isEmpty())) {
-
-      // no need of merging if the move is to a local file system
-      MoveTask mvTask = (MoveTask) GenMapRedUtils.findMoveTask(mvTasks, fsOp);
-
-      if (mvTask != null && isInsertTable && hconf.getBoolVar(ConfVars.HIVESTATSAUTOGATHER)
-          && !fsOp.getConf().isMaterialization()) {
-        // mark the MapredWork and FileSinkOperator for gathering stats
-        fsOp.getConf().setGatherStats(true);
-        fsOp.getConf().setStatsReliable(hconf.getBoolVar(ConfVars.HIVE_STATS_RELIABLE));
-        if (!mvTask.hasFollowingStatsTask()) {
-          GenMapRedUtils.addStatsTask(fsOp, mvTask, currTask, hconf);
-        }
+    if (mvTasks == null  || mvTasks.isEmpty()) return false;
+
+    // no need of merging if the move is to a local file system
+    // We are looking based on the original FSOP, so use the original path as is.
+    MoveTask mvTask = (MoveTask) GenMapRedUtils.findMoveTaskForFsopOutput(
+        mvTasks, fsOp.getConf().getFinalDirName());
+
+    // TODO: wtf? wtf?!! why is this in this method?
+    if (mvTask != null && isInsertTable && hconf.getBoolVar(ConfVars.HIVESTATSAUTOGATHER)
+        && !fsOp.getConf().isMaterialization()) {
+      // mark the MapredWork and FileSinkOperator for gathering stats
+      fsOp.getConf().setGatherStats(true);
+      fsOp.getConf().setStatsReliable(hconf.getBoolVar(ConfVars.HIVE_STATS_RELIABLE));
+      if (!mvTask.hasFollowingStatsTask()) {
+        GenMapRedUtils.addStatsTask(fsOp, mvTask, currTask, hconf);
       }
+    }
 
-      if ((mvTask != null) && !mvTask.isLocal() && fsOp.getConf().canBeMerged()) {
+    if (mvTask == null || mvTask.isLocal() || !fsOp.getConf().canBeMerged()) return false;
 
-        if (currTask.getWork() instanceof TezWork) {
-          // tez blurs the boundary between map and reduce, thus it has it's own
-          // config
-          return hconf.getBoolVar(ConfVars.HIVEMERGETEZFILES);
-        } else if (currTask.getWork() instanceof SparkWork) {
-          // spark has its own config for merging
-          return hconf.getBoolVar(ConfVars.HIVEMERGESPARKFILES);
-        }
+    if (currTask.getWork() instanceof TezWork) {
+      // tez blurs the boundary between map and reduce, thus it has it's own config
+      return hconf.getBoolVar(ConfVars.HIVEMERGETEZFILES);
+    } else if (currTask.getWork() instanceof SparkWork) {
+      // spark has its own config for merging
+      return hconf.getBoolVar(ConfVars.HIVEMERGESPARKFILES);
+    }
+    return isMergeRequiredForMr(hconf, fsOp, currTask);
+  }
 
-        if (fsOp.getConf().isLinkedFileSink()) {
-          // If the user has HIVEMERGEMAPREDFILES set to false, the idea was the
-          // number of reducers are few, so the number of files anyway are small.
-          // However, with this optimization, we are increasing the number of files
-          // possibly by a big margin. So, merge aggresively.
-          if (hconf.getBoolVar(ConfVars.HIVEMERGEMAPFILES) ||
-              hconf.getBoolVar(ConfVars.HIVEMERGEMAPREDFILES)) {
-            return true;
-          }
-        } else {
-          // There are separate configuration parameters to control whether to
-          // merge for a map-only job
-          // or for a map-reduce job
-          if (currTask.getWork() instanceof MapredWork) {
-            ReduceWork reduceWork = ((MapredWork) currTask.getWork()).getReduceWork();
-            boolean mergeMapOnly =
-              hconf.getBoolVar(ConfVars.HIVEMERGEMAPFILES) && reduceWork == null;
-            boolean mergeMapRed =
-              hconf.getBoolVar(ConfVars.HIVEMERGEMAPREDFILES) &&
-              reduceWork != null;
-            if (mergeMapOnly || mergeMapRed) {
-              return true;
-            }
-          } else {
-            return false;
-          }
-        }
+  private static boolean isMergeRequiredForMr(HiveConf hconf,
+      FileSinkOperator fsOp, Task<? extends Serializable> currTask) {
+    if (fsOp.getConf().isLinkedFileSink()) {
+      // If the user has HIVEMERGEMAPREDFILES set to false, the idea was the
+      // number of reducers are few, so the number of files anyway are small.
+      // However, with this optimization, we are increasing the number of files
+      // possibly by a big margin. So, merge aggresively.
+      return (hconf.getBoolVar(ConfVars.HIVEMERGEMAPFILES) ||
+          hconf.getBoolVar(ConfVars.HIVEMERGEMAPREDFILES));
+    }
+    // There are separate configuration parameters to control whether to
+    // merge for a map-only job
+    // or for a map-reduce job
+    if (currTask.getWork() instanceof MapredWork) {
+      ReduceWork reduceWork = ((MapredWork) currTask.getWork()).getReduceWork();
+      boolean mergeMapOnly =
+        hconf.getBoolVar(ConfVars.HIVEMERGEMAPFILES) && reduceWork == null;
+      boolean mergeMapRed =
+        hconf.getBoolVar(ConfVars.HIVEMERGEMAPREDFILES) &&
+        reduceWork != null;
+      if (mergeMapOnly || mergeMapRed) {
+        return true;
       }
     }
     return false;
@@ -1798,36 +1824,38 @@ public final class GenMapRedUtils {
 
     Path dest = null;
 
+    FileSinkDesc fileSinkDesc = fsOp.getConf();
+    boolean isMmTable = fileSinkDesc.isMmTable();
     if (chDir) {
-      FileSinkDesc fileSinkDesc = fsOp.getConf();
-      dest = fileSinkDesc.getFinalDirName();
-
-      // generate the temporary file
-      // it must be on the same file system as the current destination
-      Context baseCtx = parseCtx.getContext();
-
-      // Create the required temporary file in the HDFS location if the destination
-      // path of the FileSinkOperator table is a blobstore path.
-      // TODO# special case #N - linked FDs (unions?)
-      Path tmpDir = baseCtx.getTempDirForPath(fileSinkDesc.getDestPath());
-
-      // Change all the linked file sink descriptors
-      if (fileSinkDesc.isLinkedFileSink()) {
-        for (FileSinkDesc fsConf:fileSinkDesc.getLinkedFileSinkDesc()) {
-          fsConf.setParentDir(tmpDir);
-          fsConf.setDirName(new Path(tmpDir, fsConf.getDirName().getName()));
-          Utilities.LOG14535.info("createMoveTask setting tmpDir for LinkedFileSink chDir " + fsConf.getDirName() + "; new parent " + tmpDir + ", dest was " + fileSinkDesc.getDestPath());
+
+      dest = fileSinkDesc.getMergeInputDirName();
+      if (!isMmTable) {
+        // generate the temporary file
+        // it must be on the same file system as the current destination
+        Context baseCtx = parseCtx.getContext();
+
+        // Create the required temporary file in the HDFS location if the destination
+        // path of the FileSinkOperator table is a blobstore path.
+        Path tmpDir = baseCtx.getTempDirForPath(fileSinkDesc.getDestPath());
+
+        // Change all the linked file sink descriptors
+        if (fileSinkDesc.isLinkedFileSink()) {
+          for (FileSinkDesc fsConf:fileSinkDesc.getLinkedFileSinkDesc()) {
+            fsConf.setParentDir(tmpDir);
+            fsConf.setDirName(new Path(tmpDir, fsConf.getDirName().getName()));
+            Utilities.LOG14535.info("createMoveTask setting tmpDir for LinkedFileSink chDir " + fsConf.getDirName() + "; new parent " + tmpDir + ", dest was " + fileSinkDesc.getDestPath());
+          }
+        } else {
+          fileSinkDesc.setDirName(tmpDir);
+          Utilities.LOG14535.info("createMoveTask setting tmpDir  chDir " + tmpDir + "; dest was " + fileSinkDesc.getDestPath());
         }
-      } else {
-        fileSinkDesc.setDirName(tmpDir);
-        Utilities.LOG14535.info("createMoveTask setting tmpDir for LinkedFileSink chDir " + tmpDir + "; dest was " + fileSinkDesc.getDestPath());
       }
     }
 
     Task<MoveWork> mvTask = null;
 
     if (!chDir) {
-      mvTask = GenMapRedUtils.findMoveTask(mvTasks, fsOp);
+      mvTask = GenMapRedUtils.findMoveTaskForFsopOutput(mvTasks, fsOp.getConf().getFinalDirName());
     }
 
     // Set the move task to be dependent on the current task

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java
index 5348500..03c2e79 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java
@@ -1586,6 +1586,10 @@ public class DDLSemanticAnalyzer extends BaseSemanticAnalyzer {
 
     try {
       tblObj = getTable(tableName);
+      // TODO: we should probably block all ACID tables here.
+      if (MetaStoreUtils.isMmTable(tblObj.getParameters())) {
+        throw new SemanticException("Merge is not supported for MM tables");
+      }
 
       List<String> bucketCols = null;
       Class<? extends InputFormat> inputFormatClass = null;
@@ -1676,9 +1680,8 @@ public class DDLSemanticAnalyzer extends BaseSemanticAnalyzer {
       LoadTableDesc ltd = new LoadTableDesc(queryTmpdir, tblDesc,
           partSpec == null ? new HashMap<String, String>() : partSpec);
       ltd.setLbCtx(lbCtx);
-      // TODO# movetask is created here; handle MM tables
-      Task<MoveWork> moveTsk = TaskFactory.get(new MoveWork(null, null, ltd, null, false),
-          conf);
+      // No need to handle MM tables - unsupported path.
+      Task<MoveWork> moveTsk = TaskFactory.get(new MoveWork(null, null, ltd, null, false), conf);
       mergeTask.addDependentTask(moveTsk);
 
       if (conf.getBoolVar(HiveConf.ConfVars.HIVESTATSAUTOGATHER)) {

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java
index 73cc95a..5c67fe2 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java
@@ -304,7 +304,6 @@ public class GenTezUtils {
         linked = context.linkedFileSinks.get(path);
         linked.add(desc);
 
-        // TODO# special case #N - unions (tez)
         desc.setDirName(new Path(path, "" + linked.size()));
         Utilities.LOG14535.info("removing union - new desc with " + desc.getDirName() + "; parent " + path);
         desc.setLinkedFileSink(true);
@@ -374,8 +373,7 @@ public class GenTezUtils {
       // If underlying data is RCFile or OrcFile, RCFileBlockMerge task or
       // OrcFileStripeMerge task would be created.
       LOG.info("using CombineHiveInputformat for the merge job");
-      Utilities.LOG14535.info("merging files from " + fileSink.getConf().getDirName() + " to " + finalName);
-      // TODO# special case #N - merge
+      Utilities.LOG14535.info("will generate MR work for merging files from " + fileSink.getConf().getDirName() + " to " + finalName);
       GenMapRedUtils.createMRWorkForMergingFiles(fileSink, finalName,
           context.dependencyTask, context.moveTask,
           hconf, context.currentTask);

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
index ede1bda..66e2d27 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
@@ -6575,7 +6575,7 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer {
       } else {
         queryTmpdir = ctx.getTempDirForPath(dest_path);
       }
-      Utilities.LOG14535.info("createFS for table specifying " + queryTmpdir + " from " + dest_path);
+      Utilities.LOG14535.info("create filesink w/DEST_TABLE specifying " + queryTmpdir + " from " + dest_path);
       if (dpCtx != null) {
         // set the root of the temporary path where dynamic partition columns will populate
         dpCtx.setRootPath(queryTmpdir);
@@ -6644,7 +6644,7 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer {
 
       isMmTable = MetaStoreUtils.isMmTable(dest_tab.getParameters());
       queryTmpdir = isMmTable ? dest_path : ctx.getTempDirForPath(dest_path);
-      Utilities.LOG14535.info("createFS for partition specifying " + queryTmpdir + " from " + dest_path);
+      Utilities.LOG14535.info("create filesink w/DEST_PARTITION specifying " + queryTmpdir + " from " + dest_path);
       table_desc = Utilities.getTableDesc(dest_tab);
 
       // Add sorting/bucketing if needed

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/plan/ConditionalResolverMergeFiles.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/ConditionalResolverMergeFiles.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/ConditionalResolverMergeFiles.java
index ffc9c3e..4635f18 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/plan/ConditionalResolverMergeFiles.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/ConditionalResolverMergeFiles.java
@@ -76,14 +76,6 @@ public class ConditionalResolverMergeFiles implements ConditionalResolver,
     }
 
     /**
-     * @param dir
-     *          the dir to set
-     */
-    public void setDir(String dir) {
-      this.dir = dir;
-    }
-
-    /**
      * @return the listTasks
      */
     public List<Task<? extends Serializable>> getListTasks() {
@@ -121,8 +113,7 @@ public class ConditionalResolverMergeFiles implements ConditionalResolver,
     }
   }
 
-  public List<Task<? extends Serializable>> getTasks(HiveConf conf,
-      Object objCtx) {
+  public List<Task<? extends Serializable>> getTasks(HiveConf conf, Object objCtx) {
     ConditionalResolverMergeFilesCtx ctx = (ConditionalResolverMergeFilesCtx) objCtx;
     String dirName = ctx.getDir();
 
@@ -179,6 +170,8 @@ public class ConditionalResolverMergeFiles implements ConditionalResolver,
           if(lbLevel == 0) {
             // static partition without list bucketing
             long totalSz = getMergeSize(inpFs, dirPath, avgConditionSize);
+            Utilities.LOG14535.info("merge resolve simple case - totalSz " + totalSz + " from " + dirPath);
+
             if (totalSz >= 0) { // add the merge job
               setupMapRedWork(conf, work, trgtSize, totalSz);
               resTsks.add(mrTask);
@@ -192,6 +185,7 @@ public class ConditionalResolverMergeFiles implements ConditionalResolver,
           }
         }
       } else {
+        Utilities.LOG14535.info("Resolver returning movetask for " + dirPath);
         resTsks.add(mvTask);
       }
     } catch (IOException e) {
@@ -234,6 +228,7 @@ public class ConditionalResolverMergeFiles implements ConditionalResolver,
       Task<? extends Serializable> mrTask, Task<? extends Serializable> mrAndMvTask, Path dirPath,
       FileSystem inpFs, ConditionalResolverMergeFilesCtx ctx, MapWork work, int dpLbLevel)
       throws IOException {
+    Utilities.LOG14535.info("generateActualTasks for " + dirPath);
     DynamicPartitionCtx dpCtx = ctx.getDPCtx();
     // get list of dynamic partitions
     FileStatus[] status = HiveStatsUtils.getFileStatusRecurse(dirPath, dpLbLevel, inpFs);
@@ -281,6 +276,7 @@ public class ConditionalResolverMergeFiles implements ConditionalResolver,
 
       // add the move task for those partitions that do not need merging
       if (toMove.size() > 0) {
+        // Note: this path should be specific to concatenate; never executed in a select query.
         // modify the existing move task as it is already in the candidate running tasks
 
         // running the MoveTask and MR task in parallel may
@@ -362,6 +358,7 @@ public class ConditionalResolverMergeFiles implements ConditionalResolver,
       long totalSz = 0;
       int numFiles = 0;
       for (FileStatus fStat : fStats) {
+        Utilities.LOG14535.info("Resolver looking at " + fStat.getPath());
         if (fStat.isDir()) {
           AverageSize avgSzDir = getAverageSize(inpFs, fStat.getPath());
           if (avgSzDir.getTotalSize() < 0) {

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/plan/FileMergeDesc.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/FileMergeDesc.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/FileMergeDesc.java
index 7ec1bdd..615c63d 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/plan/FileMergeDesc.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/FileMergeDesc.java
@@ -28,6 +28,7 @@ public class FileMergeDesc extends AbstractOperatorDesc {
   private int listBucketingDepth;
   private boolean hasDynamicPartitions;
   private boolean isListBucketingAlterTableConcatenate;
+  private Long mmWriteId;
 
   public FileMergeDesc(DynamicPartitionCtx dynPartCtx, Path outputDir) {
     this.dpCtx = dynPartCtx;
@@ -73,4 +74,12 @@ public class FileMergeDesc extends AbstractOperatorDesc {
   public void setListBucketingAlterTableConcatenate(boolean isListBucketingAlterTableConcatenate) {
     this.isListBucketingAlterTableConcatenate = isListBucketingAlterTableConcatenate;
   }
+
+  public Long getMmWriteId() {
+    return mmWriteId;
+  }
+
+  public void setMmWriteId(Long mmWriteId) {
+    this.mmWriteId = mmWriteId;
+  }
 }

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java
index def1c5f..8bef7a9 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java
@@ -22,6 +22,7 @@ import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.common.ValidWriteIds;
 import org.apache.hadoop.hive.ql.io.AcidUtils;
 import org.apache.hadoop.hive.ql.metadata.Table;
 import org.apache.hadoop.hive.ql.plan.Explain.Level;
@@ -182,6 +183,13 @@ public class FileSinkDesc extends AbstractOperatorDesc {
     return linkedFileSink ? parentDir : dirName;
   }
 
+  /** getFinalDirName that takes into account MM, but not DP, LB or buckets. */
+  public Path getMergeInputDirName() {
+    Path root = getFinalDirName();
+    if (mmWriteId == null) return root;
+    return new Path(root, ValidWriteIds.getMmFilePrefix(mmWriteId));
+  }
+
   @Explain(displayName = "table", explainLevels = { Level.USER, Level.DEFAULT, Level.EXTENDED })
   public TableDesc getTableInfo() {
     return tableInfo;
@@ -255,7 +263,7 @@ public class FileSinkDesc extends AbstractOperatorDesc {
     return mmWriteId != null;
   }
 
-  public long getMmWriteId() {
+  public Long getMmWriteId() {
     return mmWriteId;
   }
 
@@ -485,6 +493,10 @@ public class FileSinkDesc extends AbstractOperatorDesc {
     this.statsTmpDir = statsCollectionTempDir;
   }
 
+  public void setMmWriteId(Long mmWriteId) {
+    this.mmWriteId = mmWriteId;
+  }
+
   public class FileSinkOperatorExplainVectorization extends OperatorExplainVectorization {
 
     public FileSinkOperatorExplainVectorization(VectorDesc vectorDesc) {

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/java/org/apache/hadoop/hive/ql/plan/MoveWork.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/MoveWork.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/MoveWork.java
index 9f498c7..f0b2775 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/plan/MoveWork.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/MoveWork.java
@@ -23,6 +23,7 @@ import java.util.HashSet;
 import java.util.List;
 
 import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.ql.exec.Utilities;
 import org.apache.hadoop.hive.ql.hooks.ReadEntity;
 import org.apache.hadoop.hive.ql.hooks.WriteEntity;
 import org.apache.hadoop.hive.ql.plan.Explain.Level;
@@ -59,7 +60,7 @@ public class MoveWork implements Serializable {
   public MoveWork() {
   }
 
-  public MoveWork(HashSet<ReadEntity> inputs, HashSet<WriteEntity> outputs) {
+  private MoveWork(HashSet<ReadEntity> inputs, HashSet<WriteEntity> outputs) {
     this.inputs = inputs;
     this.outputs = outputs;
   }
@@ -68,6 +69,8 @@ public class MoveWork implements Serializable {
       final LoadTableDesc loadTableWork, final LoadFileDesc loadFileWork,
       boolean checkFileFormat, boolean srcLocal) {
     this(inputs, outputs);
+    Utilities.LOG14535.info("Creating MoveWork " + System.identityHashCode(this)
+        + " with " + loadTableWork + "; " + loadFileWork);
     this.loadTableWork = loadTableWork;
     this.loadFileWork = loadFileWork;
     this.checkFileFormat = checkFileFormat;
@@ -77,10 +80,7 @@ public class MoveWork implements Serializable {
   public MoveWork(HashSet<ReadEntity> inputs, HashSet<WriteEntity> outputs,
       final LoadTableDesc loadTableWork, final LoadFileDesc loadFileWork,
       boolean checkFileFormat) {
-    this(inputs, outputs);
-    this.loadTableWork = loadTableWork;
-    this.loadFileWork = loadFileWork;
-    this.checkFileFormat = checkFileFormat;
+    this(inputs, outputs, loadTableWork, loadFileWork, checkFileFormat, false);
   }
 
   @Explain(displayName = "tables", explainLevels = { Level.USER, Level.DEFAULT, Level.EXTENDED })

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/test/queries/clientpositive/mm_all.q
----------------------------------------------------------------------
diff --git a/ql/src/test/queries/clientpositive/mm_all.q b/ql/src/test/queries/clientpositive/mm_all.q
index 1f85c48..8ce42a2 100644
--- a/ql/src/test/queries/clientpositive/mm_all.q
+++ b/ql/src/test/queries/clientpositive/mm_all.q
@@ -131,6 +131,42 @@ drop table skew_dp_union_mm;
 
 
 
+set hive.merge.orcfile.stripe.level=true;
+set hive.merge.tezfiles=true;
+set hive.merge.mapfiles=true;
+set hive.merge.mapredfiles=true;
+
+
+create table merge0_mm (id int) stored as orc tblproperties('hivecommit'='true');
+
+insert into table merge0_mm select key from intermediate;
+select * from merge0_mm;
+
+set tez.grouping.split-count=1;
+insert into table merge0_mm select key from intermediate;
+set tez.grouping.split-count=0;
+select * from merge0_mm;
+
+drop table merge0_mm;
+
+
+create table merge1_mm (id int) partitioned by (key int) stored as orc tblproperties('hivecommit'='true');
+
+insert into table merge1_mm partition (key) select key, key from intermediate;
+select * from merge1_mm;
+
+set tez.grouping.split-count=1;
+insert into table merge1_mm partition (key) select key, key from intermediate;
+set tez.grouping.split-count=0;
+select * from merge1_mm;
+
+drop table merge1_mm;
+
+
+-- TODO: need to include merge+union, but it's broken for now
+
+
+
 
 
 
@@ -140,31 +176,14 @@ drop table skew_dp_union_mm;
 
 
 
---drop table merge_mm;
+
+
 --drop table ctas_mm;
 --
 --
 --create table ctas_mm tblproperties ('hivecommit'='true') as select * from src limit 3;
 --
 --
---set hive.merge.mapredfiles=true;
---set hive.merge.sparkfiles=true;
---set hive.merge.tezfiles=true;
---
---CREATE TABLE merge_mm (key INT, value STRING) 
---    PARTITIONED BY (ds STRING, part STRING) STORED AS ORC tblproperties ('hivecommit'='true');
---
---EXPLAIN
---INSERT OVERWRITE TABLE merge_mm PARTITION (ds='123', part)
---        SELECT key, value, PMOD(HASH(key), 2) as part
---        FROM src;
---
---INSERT OVERWRITE TABLE merge_mm PARTITION (ds='123', part)
---        SELECT key, value, PMOD(HASH(key), 2) as part
---        FROM src;
---
---
---
 ---- TODO load, multi-insert etc
 --
 --

http://git-wip-us.apache.org/repos/asf/hive/blob/eacf9f9b/ql/src/test/queries/clientpositive/mm_current.q
----------------------------------------------------------------------
diff --git a/ql/src/test/queries/clientpositive/mm_current.q b/ql/src/test/queries/clientpositive/mm_current.q
index ceb7a1a..f423b00 100644
--- a/ql/src/test/queries/clientpositive/mm_current.q
+++ b/ql/src/test/queries/clientpositive/mm_current.q
@@ -8,45 +8,17 @@ set hive.tez.auto.reducer.parallelism=false;
 
 drop table intermediate;
 create table intermediate(key int) partitioned by (p int) stored as orc;
-insert into table intermediate partition(p='455') select key from src limit 2;
-insert into table intermediate partition(p='456') select key from src limit 2;
+insert into table intermediate partition(p='455') select distinct key from src where key >= 0 order by key desc limit 2;
+insert into table intermediate partition(p='456') select distinct key from src where key is not null order by key asc limit 2;
 
 
-set hive.optimize.skewjoin.compiletime = true;
+set hive.merge.orcfile.stripe.level=true;
+set hive.merge.tezfiles=true;
+set hive.merge.mapfiles=true;
+set hive.merge.mapredfiles=true;
 
-create table skew_mm(k1 int, k2 int, k4 int) skewed by (k1, k4) on ((0,0),(1,1),(2,2),(3,3))
- stored as directories tblproperties ('hivecommit'='false');
 
-insert into table skew_mm 
-select key, key, key from intermediate;
 
-drop table skew_mm;
-
-
-create table skew_mm(k1 int, k2 int, k4 int) skewed by (k1, k4) on ((0,0),(1,1),(2,2),(3,3))
- stored as directories tblproperties ('hivecommit'='true');
-
-insert into table skew_mm 
-select key, key, key from intermediate;
-
-select * from skew_mm;
-drop table skew_mm;
-
-
-
-
-
-create table skew_mm(k1 int, k2 int, k4 int) partitioned by (k3 int) 
-skewed by (k1, k4) on ((0,0),(1,1),(2,2),(3,3)) stored as directories tblproperties ('hivecommit'='true');
-
-insert into table skew_mm partition (k3)
-select key as i, key as j, key as k, key as l from intermediate
-union all 
-select key +1 as i, key +2 as j, key +3 as k, key +4 as l from intermediate;
-
-
-select * from skew_mm;
-drop table skew_mm;
 
 
 


Mime
View raw message