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,20 +31,23 @@
  * 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, HIVE-28822,
+ *                                     used on unstable-rename filesystems)
+ * 00001_02_copy_abcd1234.gz
  * <p>
  * All the components are here:
  * tmp_(taskPrefix)00001_02_copy_1.zlib.gz
  */
 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
-      "(\\..*)?$"); // any suffix/file extension
+      "^(.*?)?" + /* any prefix */
+      "(\\(.*\\))?" + /* taskId prefix */
+      "(\\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 tag (HIVE-28822) */
+      "(\\..*)?$"); /* any suffix/file extension */
 
   public static ParsedOutputFileName parse(String fileName) {
     return new ParsedOutputFileName(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 (HIVE-28822, 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

@@ -5170,6 +5170,96 @@ 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.
+   * Reads {@code hive.query.id} from the passed {@link HiveConf}; returns the empty string
+   * when the id is missing. The 8 hex chars come from {@code queryId.hashCode()}; that is
+   * short enough to keep S3 listings readable and collision-free for realistic per-partition
+   * concurrency (birthday-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 (qid == null || qid.isEmpty()) {
+      return "";
+    }
+    return String.format("%08x", qid.hashCode());
+  }
+
+  /**
+   * 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 && UnstableRenameFileSystem.matches(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 +5289,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 +5301,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/java/org/apache/hadoop/hive/ql/metadata/UnstableRenameFileSystem.java

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.metadata;
+
+import java.util.EnumSet;
+import java.util.Locale;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.hadoop.fs.FileSystem;
+
+/**
+ * File systems whose {@link FileSystem#rename(org.apache.hadoop.fs.Path, org.apache.hadoop.fs.Path)}
+ * implementation is not atomic-if-absent and can silently overwrite an existing destination when
+ * two concurrent writers race between an {@code exists()} probe and the subsequent rename call.
+ * <p>
+ * Object stores (S3, GCS, etc.) fall in this category: the S3A rename is a copy+delete on the
+ * client, with the "does the destination exist?" check performed on the client before the copy;
+ * two writers whose probes both fire before either PUT commits will both proceed and one will
+ * silently overwrite the other.
+ * <p>
+ * Callers use this enum to decide whether to apply defensive strategies such as suffixing the
+ * destination filename with a per-query tag so concurrent writers pick distinct keys — see
+ * {@code Hive#mvFile}. It is intentionally an in-code enum rather than a configuration knob:
+ * the set of unsafe filesystems is a property of the filesystem implementation, not something an
+ * operator should override.
+ */
+public enum UnstableRenameFileSystem {
+  S3A("s3a"),
+  S3N("s3n"),
+  S3("s3"),
+  // Google Cloud Storage exposes the same "rename is copy+delete" semantics through the Hadoop
+  // connector; keep here so multi-cloud deployments are covered without further edits.
+  GS("gs");
+
+  private final String scheme;
+
+  UnstableRenameFileSystem(String scheme) {
+    this.scheme = scheme;
+  }
+
+  public String scheme() {
+    return scheme;
+  }
+
+  private static final Set<String> SCHEMES = EnumSet.allOf(UnstableRenameFileSystem.class).stream()
+      .map(UnstableRenameFileSystem::scheme).collect(Collectors.toSet());
+
+  /**
+   * @return {@code true} when {@code scheme} matches one of the known unstable-rename
+   *         filesystems; {@code false} otherwise (including {@code null} / empty).
+   */
+  public static boolean matches(String scheme) {
+    return scheme != null && SCHEMES.contains(scheme.toLowerCase(Locale.ROOT));
+  }
+
+  /**
+   * Convenience overload that inspects a {@link FileSystem}'s URI scheme.
+   *
+   * @return {@code true} when the filesystem's scheme matches one of the known unstable-rename
+   *         filesystems.
+   */
+  public static boolean matches(FileSystem fs) {
+    if (fs == null || fs.getUri() == null) {
+      return false;
+    }
+    return matches(fs.getUri().getScheme());
+  }
+}

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));
   }
 
+  /**
+   * HIVE-28822: 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,6 +35,9 @@
 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;
 
 
@@ -227,4 +230,53 @@ 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")));
   }
+
+  /**
+   * HIVE-28822 (root-cause fix): 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_<hex>} suffixed names — no plain {@code 000000_0}, no {@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 match {@link UnstableRenameFileSystem}, which is not easily synthesizable with
+   * LocalFileSystem in a JUnit environment:
+   * <ol>
+   *   <li>{@link UnstableRenameFileSystem#matches(String)} recognizes S3-family schemes 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 an unstable-rename one. Confirms the tag is stable for a given
+   *       queryId, and that the tag's shape (8 hex chars) matches the extra group in
+   *       {@link org.apache.hadoop.hive.ql.exec.ParsedOutputFileName}'s regex.</li>
+   * </ol>
+   */
+  @Test
+  public void testUniquenessTagAndUnstableFsGating() {
+    // (1) enum gating
+    assertTrue(UnstableRenameFileSystem.matches("s3a"));
+    assertTrue(UnstableRenameFileSystem.matches("s3n"));
+    assertTrue(UnstableRenameFileSystem.matches("s3"));
+    assertTrue(UnstableRenameFileSystem.matches("gs"));
+    assertFalse("hdfs is atomic-rename",
+        UnstableRenameFileSystem.matches("hdfs"));
+    assertFalse("local FS is atomic-rename",
+        UnstableRenameFileSystem.matches("file"));
+    assertFalse(UnstableRenameFileSystem.matches((String) null));
+    assertFalse(UnstableRenameFileSystem.matches(""));
+
+    // (2) uniqueness tag: distinct queryIds → distinct 8-hex tags, empty queryId → empty tag
+    hiveConf.setVar(HiveConf.ConfVars.HIVE_QUERY_ID, "q1_lbodor_20260101_aaaaaaaa");
+    String tag1 = Hive.computeUniquenessTag(hiveConf);
+    hiveConf.setVar(HiveConf.ConfVars.HIVE_QUERY_ID, "q2_lbodor_20260101_bbbbbbbb");
+    String tag2 = Hive.computeUniquenessTag(hiveConf);
+    hiveConf.unset(HiveConf.ConfVars.HIVE_QUERY_ID.varname);
+    String tagEmpty = Hive.computeUniquenessTag(hiveConf);
+
+    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);
+    assertEquals("empty queryId → empty tag", "", tagEmpty);
+  }
 }