HIVE-28822: Concurrent INSERTs can silently lose rows or fail with FileAlreadyExistsException on S3 (non-ACID)

ql/src/java/org/apache/hadoop/hive/ql/QueryPlan.java

@@ -311,6 +311,18 @@ public static String makeQueryId() {
         + UUID.randomUUID().toString();
   }
 
+  /**
+   * Extract a compact per-query uniqueness tag (16 lowercase hex chars) from a query id emitted
+   * by {@link #makeQueryId()}. The tag is the {@code getMostSignificantBits()} half of the UUID
+   * at the tail of the id, rendered as 16-char lowercase hex. Used by callers that need a stable
+   * per-query filename component — e.g. {@code Hive.mvFile}'s destination naming on filesystems
+   * whose {@code rename} is not atomic-if-absent.
+   */
+  public static String extractUniquenessTag(String queryId) {
+    UUID uuid = UUID.fromString(queryId.substring(queryId.lastIndexOf('_') + 1));
+    return String.format("%016x", uuid.getMostSignificantBits());
+  }
+
   /**
    * generate the operator graph and operator list for the given task based on
    * the operators corresponding to that task.

ql/src/java/org/apache/hadoop/hive/ql/exec/ParsedOutputFileName.java

@@ -31,8 +31,11 @@
  * 00001_02
  * 00001_02.gz
  * 00001_02.zlib.gz
- * 00001_02_copy_1
+ * 00001_02_copy_1                            (numeric copy suffix, HDFS-style)
  * 00001_02_copy_1.gz
+ * 00001_02_copy_abcd1234deadbeef             (per-query uniqueness tag as copy suffix,
+ *                                             used on non-atomic-rename filesystems)
+ * 00001_02_copy_abcd1234deadbeef.gz
  * <p>
  * All the components are here:
  * tmp_(taskPrefix)00001_02_copy_1.zlib.gz
@@ -41,9 +44,9 @@ public class ParsedOutputFileName {
   private static final Pattern COPY_FILE_NAME_TO_TASK_ID_REGEX = Pattern.compile(
       "^(.*?)?" + // any prefix
       "(\\(.*\\))?" + // taskId prefix
-      "([0-9]+)" + // taskId
-      "(?:_([0-9]{1,6}))?" + // _<attemptId> (limited to 6 digits)
-      "(?:_copy_([0-9]{1,6}))?" + // copy file index
+      "(\\d+)" + // taskId
+      "(?:_(\\d{1,6}))?" + // _<attemptId> (limited to 6 digits)
+      "(?:_copy_(\\d{1,6}|[\\da-fA-F]{16}))?" + // copy suffix: numeric counter, or 16-hex uniqueness tag
       "(\\..*)?$"); // any suffix/file extension
 
   public static ParsedOutputFileName parse(String fileName) {
@@ -108,6 +111,11 @@ public boolean isCopyFile() {
     return copyIndex != null;
   }
 
+  /**
+   * @return the copy suffix: either a numeric counter (HDFS-style) or an 8-hex per-query
+   *         uniqueness tag (used on non-atomic-rename filesystems), or {@code null} when the
+   *         filename has no copy suffix.
+   */
   public String getCopyIndex() {
     return copyIndex;
   }

ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java

@@ -240,9 +240,13 @@ private AcidUtils() {
       Pattern.compile("[0-9]+_[0-9]+");
   /**
    * @see org.apache.hadoop.hive.ql.exec.Utilities#COPY_KEYWORD
+   *
+   * The copy suffix is either a numeric counter (HDFS/local: _copy_N) or a
+   * 16-hex per-query uniqueness tag (non-atomic-rename FS such as S3A:
+   * _copy_<queryTag>). See ParsedOutputFileName#REGEX.
    */
   public static final Pattern ORIGINAL_PATTERN_COPY =
-    Pattern.compile("[0-9]+_[0-9]+" + COPY_KEYWORD + "[0-9]+");
+      Pattern.compile("[0-9]+_[0-9]+" + COPY_KEYWORD + "(?:[0-9]{1,6}|[0-9a-fA-F]{16})");
 
   public static final PathFilter acidHiddenFileFilter = new PathFilter() {
     @Override
@@ -450,8 +454,11 @@ public static BucketMetaData parse(String bucketFileName) {
         return new BucketMetaData(bucketId, 0);
       }
       else if(ORIGINAL_PATTERN_COPY.matcher(bucketFileName).matches()) {
-        int copyNumber = Integer.parseInt(
-            bucketFileName.substring(bucketFileName.lastIndexOf('_') + 1));
+        String copySuffix = bucketFileName.substring(bucketFileName.lastIndexOf('_') + 1);
+        // Copy suffix is either a numeric counter or a 16-hex per-query uniqueness tag.
+        // Hex-tagged files are unordered peers from concurrent writers on an
+        // non-atomic-rename FS, so there is no meaningful copy number to assign — use 0.
+        int copyNumber = (copySuffix.length() == 16) ? 0 : Integer.parseInt(copySuffix);
         int bucketId = Integer
             .parseInt(bucketFileName.substring(0, bucketFileName.indexOf('_')));
         return new BucketMetaData(bucketId, copyNumber);

ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java

@@ -21,6 +21,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Splitter;
+import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
@@ -163,6 +164,7 @@
 import org.apache.hadoop.hive.metastore.utils.RetryUtilities;
 import org.apache.hadoop.hive.ql.Context;
 import org.apache.hadoop.hive.ql.ErrorMsg;
+import org.apache.hadoop.hive.ql.QueryPlan;
 import org.apache.hadoop.hive.ql.ddl.database.drop.DropDatabaseDesc;
 import org.apache.hadoop.hive.ql.ddl.table.AlterTableType;
 import org.apache.hadoop.hive.ql.exec.AbstractFileMergeOperator;
@@ -272,6 +274,23 @@ public class Hive implements AutoCloseable {
   static final private Logger LOG = LoggerFactory.getLogger("hive.ql.metadata.Hive");
   private final String CLASS_NAME = Hive.class.getName();
 
+  /**
+   * Schemes whose single-file {@link FileSystem#rename(Path, Path)} is not atomic-if-absent and
+   * can silently overwrite an existing destination when two concurrent writers race between an
+   * {@code exists()} probe and the rename call (object stores where rename is client-side
+   * copy+delete). Callers use this to decide whether to switch to a uniqueness-tag copy suffix
+   * in {@link #mvFile}. The list is in code because the set of unsafe filesystems is a property
+   * of the filesystem implementation, not something an operator should override.
+   * <p>
+   * Note on Azure: {@code abfs}/{@code abfss} only guarantee atomic rename when the ADLS Gen2
+   * account has hierarchical namespace enabled; without HNS they degrade to copy+delete like
+   * {@code wasb}. Since {@code mvFile} cannot cheaply tell the two apart at rename time, the
+   * Azure schemes are included unconditionally — a false positive costs only a slightly longer
+   * filename, whereas a false negative would be silent data loss.
+   */
+  public static final Set<String> NON_ATOMIC_RENAME_SCHEMES = new HashSet<>(
+      Arrays.asList("s3a", "s3n", "s3", "gs", "abfs", "abfss", "wasb", "wasbs"));
+
   private HiveConf conf = null;
   private IMetaStoreClient metaStoreClient;
   private UserGroupInformation owner;
@@ -5170,6 +5189,110 @@ private static String getPathName(int taskId) {
     return Utilities.replaceTaskId("000000", taskId) + "_0";
   }
 
+  /**
+   * Compute a compact per-query uniqueness tag used by the non-ACID rename branch of
+   * {@link #mvFile} to make each concurrent writer's destination key unique on filesystems
+   * whose {@code rename} is not atomic-if-absent. The tag becomes the copy suffix
+   * ({@code basename_copy_<tag>}) in place of the numeric {@code _copy_N} counter.
+   * <p>
+   * Reads {@code hive.query.id} from the passed {@link HiveConf} and delegates to
+   * {@link QueryPlan#extractUniquenessTag(String)} for the actual UUID → hex derivation.
+   * The shape matches {@link ParsedOutputFileName}'s copy-index group so downstream filename
+   * parsing (taskId, attemptId, copyIndex) keeps working.
+   */
+  static String computeUniquenessTag(HiveConf conf) {
+    String qid = HiveConf.getVar(conf, ConfVars.HIVE_QUERY_ID);
+    if (Strings.isNullOrEmpty(qid)) {
+      throw new IllegalStateException("hive.query.id is required to derive a unique destination name");
+    }
+    return QueryPlan.extractUniquenessTag(qid);
+  }
+
+  /**
+   * @return {@code true} when the filesystem's URI scheme is one of the known non-atomic-rename
+   *         schemes ({@link #NON_ATOMIC_RENAME_SCHEMES}); {@code false} otherwise (including a
+   *         {@code null} fs or missing scheme).
+   */
+  static boolean isNonAtomicRenameFs(FileSystem fs) {
+    if (fs == null || fs.getUri() == null || fs.getUri().getScheme() == null) {
+      return false;
+    }
+    return NON_ATOMIC_RENAME_SCHEMES.contains(fs.getUri().getScheme().toLowerCase());
+  }
+
+  /**
+   * Picks the destination {@link Path} for {@link #mvFile}, choosing between a per-query
+   * uniqueness-tagged name (on filesystems without atomic rename-if-absent semantics) and the
+   * legacy {@code _copy_N} counter-based picker.
+   *
+   * <p>On file systems without atomic rename-if-absent semantics (e.g. S3), two concurrent inserts
+   * targeting the same new dynamic partition race in the counter-based picker below: their
+   * {@code exists()} probes both fire before either PUT commits, both rename to the same final
+   * key, and the second PUT silently overwrites the first (last writer wins, no error surfaces).
+   * To eliminate the collision, on such filesystems we skip the counter-based {@code _copy_N}
+   * picker entirely and use a per-query uniqueness tag (8-hex derived from {@code hive.query.id})
+   * as the copy suffix, so two concurrent writers rename to distinct keys.
+   *
+   * <p>The uniqueness-tag path is only taken in the non-ACID rename branch
+   * ({@code taskId == -1 && isRenameAllowed && !isOverwrite}): ACID writers already own unique
+   * taskIds, copy/copyFromLocal do not race on the destination filename, and overwrite explicitly
+   * clears the target first.
+   */
+  private static Path pickDestFilePath(HiveConf conf, FileSystem sourceFs, Path sourcePath, FileSystem destFs,
+                                       Path destDirPath, int taskId, boolean isOverwrite, boolean isRenameAllowed)
+      throws IOException {
+
+    final String type = FilenameUtils.getExtension(sourcePath.getName());
+
+    // Strip off the file type, if any so we don't make:
+    // 000000_0.gz -> 000000_0.gz_copy_1
+    final String fullName = sourcePath.getName();
+
+    final String name;
+    if (taskId == -1) { // non-acid
+      name = FilenameUtils.getBaseName(sourcePath.getName());
+    } else { // acid
+      name = getPathName(taskId);
+    }
+
+    // In case of ACID, the file is ORC so the extension is not relevant and should not be inherited.
+    Path destFilePath = new Path(destDirPath, taskId == -1 ? fullName : name);
+
+    final String uniqueCopySuffix =
+        // Only apply the unique suffix in case of files, as it's supposed to handle file name collisions.
+        // When mvFile is called with a directory, we can fall back to the original logic.
+        (taskId == -1 && isRenameAllowed && !isOverwrite && sourceFs.getFileStatus(sourcePath).isFile()
+            && isNonAtomicRenameFs(destFs))
+            ? computeUniquenessTag(conf)
+            : null;
+
+    if (uniqueCopySuffix != null && !uniqueCopySuffix.isEmpty()) {
+      // Unstable-rename FS: use `name_copy_<tag>` unconditionally as the destination. No
+      // exists()-probe loop, no _copy_N counter — the per-query tag alone is enough to keep
+      // concurrent writers from colliding, and ParsedOutputFileName recognizes the shape.
+      return new Path(destDirPath, name + Utilities.COPY_KEYWORD + uniqueCopySuffix +
+          (!type.isEmpty() ? "." + type : ""));
+    }
+
+    /*
+     * The below loop may perform bad when the destination file already exists and it has too many _copy_
+     * files as well. A desired approach was to call listFiles() and get a complete list of files from
+     * the destination, and check whether the file exists or not on that list. However, millions of files
+     * could live on the destination directory, and on concurrent situations, this can cause OOM problems.
+     *
+     * I'll leave the below loop for now until a better approach is found.
+     */
+    for (int counter = 1; destFs.exists(destFilePath); counter++) {
+      if (isOverwrite) {
+        destFs.delete(destFilePath, false);
+        break;
+      }
+      destFilePath = new Path(destDirPath, name + (Utilities.COPY_KEYWORD + counter) +
+          ((taskId == -1 && !type.isEmpty()) ? "." + type : ""));
+    }
+    return destFilePath;
+  }
+
   /**
    * <p>
    *   Moves a file from one {@link Path} to another. If {@code isRenameAllowed} is true then the
@@ -5199,37 +5322,8 @@ private static String getPathName(int taskId) {
   private static Path mvFile(HiveConf conf, FileSystem sourceFs, Path sourcePath, FileSystem destFs, Path destDirPath,
                              boolean isSrcLocal, boolean isOverwrite, boolean isRenameAllowed,
                              int taskId) throws IOException {
-
-    // Strip off the file type, if any so we don't make:
-    // 000000_0.gz -> 000000_0.gz_copy_1
-    final String fullname = sourcePath.getName();
-    final String name;
-    if (taskId == -1) { // non-acid
-      name = FilenameUtils.getBaseName(sourcePath.getName());
-    } else { // acid
-      name = getPathName(taskId);
-    }
-    final String type = FilenameUtils.getExtension(sourcePath.getName());
-
-    // Incase of ACID, the file is ORC so the extension is not relevant and should not be inherited.
-    Path destFilePath = new Path(destDirPath, taskId == -1 ? fullname : name);
-
-    /*
-    * The below loop may perform bad when the destination file already exists and it has too many _copy_
-    * files as well. A desired approach was to call listFiles() and get a complete list of files from
-    * the destination, and check whether the file exists or not on that list. However, millions of files
-    * could live on the destination directory, and on concurrent situations, this can cause OOM problems.
-    *
-    * I'll leave the below loop for now until a better approach is found.
-    */
-    for (int counter = 1; destFs.exists(destFilePath); counter++) {
-      if (isOverwrite) {
-        destFs.delete(destFilePath, false);
-        break;
-      }
-      destFilePath =  new Path(destDirPath, name + (Utilities.COPY_KEYWORD + counter) +
-              ((taskId == -1 && !type.isEmpty()) ? "." + type : ""));
-    }
+    Path destFilePath = pickDestFilePath(conf, sourceFs, sourcePath, destFs, destDirPath, taskId, isOverwrite,
+        isRenameAllowed);
 
     if (isRenameAllowed) {
       destFs.rename(sourcePath, destFilePath);
@@ -5241,18 +5335,18 @@ private static Path mvFile(HiveConf conf, FileSystem sourceFs, Path sourcePath,
           false,  // overwrite destination
           conf,
           new DataCopyStatistics())) {
-        LOG.error("Copy failed for source: " + sourcePath + " to destination: " + destFilePath);
+        LOG.error("Copy failed for source: {} to destination: {}", sourcePath, destFilePath);
         throw new IOException("File copy failed.");
       }
 
       // Source file delete may fail because of permission issue as executing user might not
       // have permission to delete the files in the source path. Ignore this failure.
       try {
         if (!sourceFs.delete(sourcePath, true)) {
-          LOG.warn("Delete source failed for source: " + sourcePath + " during copy to destination: " + destFilePath);
+          LOG.warn("Delete source failed for source: {} during copy to destination: {}", sourcePath, destFilePath);
         }
       } catch (Exception e) {
-        LOG.warn("Delete source failed for source: " + sourcePath + " during copy to destination: " + destFilePath, e);
+        LOG.warn("Delete source failed for source: {} during copy to destination: {}", sourcePath, destFilePath, e);
       }
     }
     return destFilePath;

ql/src/test/org/apache/hadoop/hive/ql/exec/ParsedOutputFileNameTest.java

@@ -120,6 +120,48 @@ public void testCopyAllParts() throws Exception {
     Assert.assertEquals("tmp_(prefix)00001_02_copy_4", p.makeFilenameWithCopyIndex(4));
   }
 
+  /**
+   * On filesystems without atomic rename-if-absent semantics (S3 etc.), the copy suffix
+   * carries a 16-hex per-query uniqueness tag instead of the numeric counter, so concurrent
+   * writers rename to distinct destination keys.
+   */
+  @Test
+  public void testUniquenessTagAsCopySuffix() throws Exception {
+    ParsedOutputFileName p = ParsedOutputFileName.parse("000001_0_copy_abcd1234deadbeef");
+    Assert.assertTrue(p.matches());
+    Assert.assertEquals("000001", p.getTaskId());
+    Assert.assertEquals("0", p.getAttemptId());
+    Assert.assertEquals("abcd1234deadbeef", p.getCopyIndex());
+    Assert.assertTrue(p.isCopyFile());
+    Assert.assertNull(p.getSuffix());
+    // Numeric-index renaming (used by legacy code paths) still works and replaces the tag.
+    Assert.assertEquals("000001_0_copy_3", p.makeFilenameWithCopyIndex(3));
+  }
+
+  @Test
+  public void testUniquenessTagAsCopySuffixWithExtension() throws Exception {
+    ParsedOutputFileName p = ParsedOutputFileName.parse("000001_0_copy_abcd1234deadbeef.snappy.orc");
+    Assert.assertTrue(p.matches());
+    Assert.assertEquals("000001", p.getTaskId());
+    Assert.assertEquals("0", p.getAttemptId());
+    Assert.assertEquals("abcd1234deadbeef", p.getCopyIndex());
+    Assert.assertTrue(p.isCopyFile());
+    Assert.assertEquals(".snappy.orc", p.getSuffix());
+    Assert.assertEquals("000001_0_copy_3", p.makeFilenameWithCopyIndex(3));
+  }
+
+  /**
+   * The copy-index group must reject shapes that are neither a 1..6 digit counter nor an
+   * exactly-16-hex tag (e.g. non-hex characters, or a numeric tag longer than 6 digits).
+   */
+  @Test
+  public void testUniquenessTagShapeIsStrict() {
+    // 15 chars — matches neither branch.
+    Assert.assertNull(ParsedOutputFileName.parse("000001_0_copy_abcd1234deadbee").getCopyIndex());
+    // Non-hex character in a 16-char position.
+    Assert.assertNull(ParsedOutputFileName.parse("000001_0_copy_abcd1234deadbeez").getCopyIndex());
+  }
+
   @Test
   public void testNoMatch() {
     ParsedOutputFileName p = ParsedOutputFileName.parse("ZfsLke");