diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/ParsedOutputFileName.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/ParsedOutputFileName.java index 713c7e56f848..ee0f278229e3 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/ParsedOutputFileName.java +++ b/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 *

* 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}))?" + // _ (limited to 6 digits) - "(?:_copy_([0-9]{1,6}))?" + // copy file index + "(\\d+)" + // taskId + "(?:_(\\d{1,6}))?" + // _ (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; } 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 d7eb7281eccc..67daf80e125f 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 @@ -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. + *

+ * 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 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_}) in place of the numeric {@code _copy_N} counter. + *

+ * 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 __}), 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). + *

+ * 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. + * + *

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. + * + *

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_` 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; + } + /** *

* 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,7 +5338,7 @@ 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."); } @@ -5249,10 +5346,10 @@ private static Path mvFile(HiveConf conf, FileSystem sourceFs, Path sourcePath, // 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; diff --git a/ql/src/test/org/apache/hadoop/hive/ql/exec/ParsedOutputFileNameTest.java b/ql/src/test/org/apache/hadoop/hive/ql/exec/ParsedOutputFileNameTest.java index e09a5ecc3c33..6b3d9b9bc8d4 100644 --- a/ql/src/test/org/apache/hadoop/hive/ql/exec/ParsedOutputFileNameTest.java +++ b/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"); diff --git a/ql/src/test/org/apache/hadoop/hive/ql/metadata/TestHiveCopyFiles.java b/ql/src/test/org/apache/hadoop/hive/ql/metadata/TestHiveCopyFiles.java index 2ef7bfcbccdd..21e8181c1813 100644 --- a/ql/src/test/org/apache/hadoop/hive/ql/metadata/TestHiveCopyFiles.java +++ b/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_} names — no plain {@code 000000_0}, no + * numeric {@code _copy_N}. + * + *

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: + *

    + *
  1. {@link Hive#isNonAtomicRenameFs(FileSystem)} recognizes S3-family schemes on the URI + * and rejects HDFS / local schemes.
    2. + *
  2. 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.
    3. + *
+ */ + @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 → "__"). 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")); + } + } }