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

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_abcd1234             (per-query uniqueness tag as copy suffix,
+ *                                     used on unstable-rename filesystems)
+ * 00001_02_copy_abcd1234.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]{8}))?" + // copy suffix: numeric counter, or 8-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 unstable-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/metadata/Hive.java

@@ -21,8 +21,10 @@
 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.ImmutableSet;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
@@ -222,6 +224,7 @@
 import java.util.LinkedHashSet;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map.Entry;
 import java.util.Map;
 import java.util.Optional;
@@ -272,6 +275,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.
+   */
+  private static final Set<String> NON_ATOMIC_RENAME_SCHEMES =
+      ImmutableSet.of("s3a", "s3n", "s3", "gs", "abfs", "abfss", "wasb", "wasbs");
+
   private HiveConf conf = null;
   private IMetaStoreClient metaStoreClient;
   private UserGroupInformation owner;
@@ -5170,6 +5190,113 @@ private static String getPathName(int taskId) {
     return Utilities.replaceTaskId("000000", taskId) + "_0";
   }
 
+  /**
+   * Compute a compact per-query uniqueness tag (8 lowercase hex chars) 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}, extracts the UUID at the tail
+   * (see {@code QueryPlan.makeQueryId}, which assembles the id as
+   * {@code <user>_<timestamp>_<uuid>}), and returns its leftmost 8 hex chars. 32 bits of UUID
+   * randomness keeps S3 listings readable and is collision-resistant for realistic
+   * per-partition concurrency (collides only at ~65k concurrent writers to the same
+   * partition).
+   * <p>
+   * 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");
+    }
+    int uuidStart = qid.lastIndexOf('_') + 1;
+    // hive_20240429111756_d39b59fb-31e2-4e89-853e-fac2844530e9 -> d39b59fb
+    return qid.substring(uuidStart, uuidStart + 8);
+  }
+
+  /**
+   * @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, 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 =
+        (taskId == -1 && isRenameAllowed && !isOverwrite && 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 +5326,7 @@ 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, sourcePath, destFs, destDirPath, taskId, isOverwrite, isRenameAllowed);
 
     if (isRenameAllowed) {
       destFs.rename(sourcePath, destFilePath);
@@ -5241,18 +5338,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 an 8-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_abcd1234");
+    Assert.assertTrue(p.matches());
+    Assert.assertEquals("000001", p.getTaskId());
+    Assert.assertEquals("0", p.getAttemptId());
+    Assert.assertEquals("abcd1234", 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_abcd1234.snappy.orc");
+    Assert.assertTrue(p.matches());
+    Assert.assertEquals("000001", p.getTaskId());
+    Assert.assertEquals("0", p.getAttemptId());
+    Assert.assertEquals("abcd1234", 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-8-hex tag (e.g. non-hex characters, or a numeric tag longer than 6 digits).
+   */
+  @Test
+  public void testUniquenessTagShapeIsStrict() {
+    // 7 chars — matches neither branch.
+    Assert.assertNull(ParsedOutputFileName.parse("000001_0_copy_abc1234").getCopyIndex());
+    // Non-hex character in an 8-char position.
+    Assert.assertNull(ParsedOutputFileName.parse("000001_0_copy_abcd123z").getCopyIndex());
+  }
+
   @Test
   public void testNoMatch() {
     ParsedOutputFileName p = ParsedOutputFileName.parse("ZfsLke");

ql/src/test/org/apache/hadoop/hive/ql/metadata/TestHiveCopyFiles.java

@@ -35,7 +35,11 @@
 import java.util.Arrays;
 import java.util.List;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 
 @RunWith(Parameterized.class)
@@ -227,4 +231,72 @@ public void testCopyExistingFilesOnDifferentFileSystem() throws IOException {
     assertTrue(spyTargetFs.exists(new Path(targetPath, "000000_0_copy_1.gz")));
     assertTrue(spyTargetFs.exists(new Path(targetPath, "000001_0_copy_1.gz")));
   }
+
+  /**
+   * When two concurrent writers stage a file with the same inner filename (e.g. {@code 000000_0})
+   * into the same destination directory on an S3-like filesystem, mvFile must pick distinct
+   * destination keys so the second writer does not silently overwrite the first. Both files
+   * must land under distinct {@code 000000_0_copy_<hex>} names — no plain {@code 000000_0}, no
+   * numeric {@code _copy_N}.
+   *
+   * <p>Covers the two moving parts individually since the full rename-branch path in
+   * {@link Hive#copyFiles} requires src and dest FileSystems to compare equal AND the dest
+   * scheme to be flagged non-atomic-rename, which is not easily synthesizable with
+   * LocalFileSystem in a JUnit environment:
+   * <ol>
+   *   <li>{@link Hive#isNonAtomicRenameFs(FileSystem)} recognizes S3-family schemes on the URI
+   *       and rejects HDFS / local schemes.</li>
+   *   <li>Two distinct {@code hive.query.id} values map to two distinct 8-hex uniqueness tags
+   *       — the compact per-query identifier that mvFile appends when the destination
+   *       filesystem is a non-atomic-rename one. Confirms the tag is stable for a given
+   *       queryId, and that the tag's shape (8 hex chars) matches the copy-suffix group in
+   *       {@link org.apache.hadoop.hive.ql.exec.ParsedOutputFileName}'s regex.</li>
+   * </ol>
+   */
+  @Test
+  public void testUniquenessTagAndUnstableFsGating() throws IOException {
+    // (1) non-atomic-rename filesystem detection via URI scheme
+    FileSystem localFs = new Path(targetFolder.getRoot().getAbsolutePath()).getFileSystem(hiveConf);
+    assertFalse("local FS is atomic-rename", Hive.isNonAtomicRenameFs(localFs));
+    assertFalse("null fs is not flagged", Hive.isNonAtomicRenameFs((FileSystem) null));
+
+    for (String scheme : new String[] {"s3a", "s3n", "s3", "gs", "abfs", "abfss", "wasb", "wasbs"}) {
+      FileSystem spy = Mockito.spy(localFs);
+      Mockito.when(spy.getUri()).thenReturn(URI.create(scheme + ":///bucket/path"));
+      assertTrue(scheme + " must be flagged non-atomic-rename",
+          Hive.isNonAtomicRenameFs(spy));
+    }
+    for (String scheme : new String[] {"hdfs", "file", "ofs", "adl"}) {
+      FileSystem spy = Mockito.spy(localFs);
+      Mockito.when(spy.getUri()).thenReturn(URI.create(scheme + ":///whatever"));
+      assertFalse(scheme + " must not be flagged non-atomic-rename",
+          Hive.isNonAtomicRenameFs(spy));
+    }
+
+    // (2) uniqueness tag: take the first 8 hex chars of the UUID at the tail of queryId
+    // (QueryPlan.makeQueryId → "<user>_<timestamp>_<uuid>"). Distinct UUIDs → distinct tags.
+    hiveConf.setVar(HiveConf.ConfVars.HIVE_QUERY_ID,
+        "lbodor_20260101120000_f47ac10b-58cc-4372-a567-0e02b2c3d479");
+    String tag1 = Hive.computeUniquenessTag(hiveConf);
+    hiveConf.setVar(HiveConf.ConfVars.HIVE_QUERY_ID,
+        "lbodor_20260101120001_9c8a44f1-e2b3-4a1c-9d3e-000000000000");
+    String tag2 = Hive.computeUniquenessTag(hiveConf);
+
+    assertEquals("first 8 chars of the UUID at the tail", "f47ac10b", tag1);
+    assertEquals("first 8 chars of the UUID at the tail", "9c8a44f1", tag2);
+    assertTrue("tag1 must match <8-hex>: " + tag1, tag1.matches("[0-9a-f]{8}"));
+    assertTrue("tag2 must match <8-hex>: " + tag2, tag2.matches("[0-9a-f]{8}"));
+    assertNotEquals("distinct queryIds must produce distinct tags", tag1, tag2);
+
+    // Missing queryId → hard failure (mvFile's non-atomic-rename branch must not silently
+    // fall back to a shared filename when the query state is absent).
+    hiveConf.unset(HiveConf.ConfVars.HIVE_QUERY_ID.varname);
+    try {
+      Hive.computeUniquenessTag(hiveConf);
+      fail("computeUniquenessTag must throw when hive.query.id is unset");
+    } catch (IllegalStateException expected) {
+      assertTrue("exception message must mention hive.query.id: " + expected.getMessage(),
+          expected.getMessage() != null && expected.getMessage().contains("hive.query.id"));
+    }
+  }
 }