From 71f6fe873fc93fcc08486890893918ce16afd681 Mon Sep 17 00:00:00 2001 From: Laszlo Bodor Date: Mon, 27 Jul 2026 17:45:19 +0200 Subject: [PATCH 1/2] HIVE-28822: Concurrent INSERTs into a new dynamic partition can silently lose rows or fail with FileAlreadyExistsException on S3 (non-ACID) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On file systems whose rename is not atomic-if-absent (S3A and other object stores), two concurrent non-ACID INSERTs that create the same new dynamic partition race in Hive.mvFile between the exists()-driven _copy_N picker and the destFs.rename() call. Depending on the timing this shows up as either: * Fail-loud — S3AFileSystem.initiateRename throws FileAlreadyExistsException, surfacing to the client as "MoveTask return code 40000". This matches the customer report: [load-dynamic-partitionsToAdd-0] Failed to move: ... Caused by: FileAlreadyExistsException: Failed to rename .../000001_N to .../000001_N_copy_M; destination file exists at S3AFileSystem.initiateRename at Hive.mvFile at Hive.copyFiles at Hive.loadPartitionInternal at Hive.lambda$loadDynamicPartitions * Fail-silent — both writers' internal exists() probes see the target as not-yet-present, both PUTs go to the same key, and the second silently overwrites the first (last writer wins, no error surfaces). Reproduced with 30 concurrent `insert into p_test values (i,2)` against an S3-backed external Parquet table: 2 sessions fail with MoveTask, 6 rows silently missing, and the final S3 listing shows the same _copy_N slot claimed by multiple writers. Fix: on filesystems in UnstableRenameFileSystem (S3A/S3N/S3/GS today), the copy suffix in Hive.mvFile carries a per-query 8-hex uniqueness tag (derived from hive.query.id) *in place of* the numeric counter. Two concurrent writers land at distinct destinations — basename_copy_ basename_copy_ — so there is no picker loop and no rename race. On stable-rename filesystems (HDFS, local) the historical numeric _copy_N picker is preserved unchanged. UnstableRenameFileSystem is an in-code enum rather than a configuration knob: the set of unsafe filesystems is a property of the filesystem impl, not something an operator should override. ParsedOutputFileName's copy-index regex group is widened from `[0-9]{1,6}` to `[0-9]{1,6}|[0-9a-fA-F]{8}` so both shapes parse. getCopyIndex returns either the numeric counter or the 8-hex tag verbatim; downstream taskId / attemptId extraction is unaffected. The ACID branch (taskId != -1) and the isOverwrite branch are unchanged — ACID writers already own unique taskIds, and overwrite explicitly clears the target first. If a future unstable-rename filesystem is ever missed by the enum, the failure mode is the same loud FileAlreadyExistsException → MoveTask return code 40000 that we surface today — a correct, actionable signal rather than a silent loss. Verification: * unit: TestHiveCopyFiles.testUniquenessTagAndUnstableFsGating covers the enum recognition (matches on s3a/s3n/s3/gs, rejects hdfs/file) and the per-query tag shape (distinct queryIds → distinct 8-hex tags, empty queryId → empty tag). ParsedOutputFileNameTest gains 3 cases: a copy suffix that is an 8-hex tag (plain and with extension), and a strict-shape check that rejects 7-char / non-hex forms. All 31 tests green (20 in TestHiveCopyFiles under 4 parameterizations + 11 in ParsedOutputFileNameTest). * end-to-end: 30-way concurrent burst against s3a://... table: Before: 24 rows persisted, 2 MoveTask failures, many _copy_N. After: 30 rows persisted, 0 MoveTask failures, 30 distinct 000001_N_copy_ keys in S3, no FAEE. Co-Authored-By: Claude --- .../hive/ql/exec/ParsedOutputFileName.java | 22 ++- .../apache/hadoop/hive/ql/metadata/Hive.java | 128 +++++++++++++----- .../ql/metadata/UnstableRenameFileSystem.java | 84 ++++++++++++ .../ql/exec/ParsedOutputFileNameTest.java | 42 ++++++ .../hive/ql/metadata/TestHiveCopyFiles.java | 52 +++++++ 5 files changed, 287 insertions(+), 41 deletions(-) create mode 100644 ql/src/java/org/apache/hadoop/hive/ql/metadata/UnstableRenameFileSystem.java 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..0f6f69bef519 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,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 *

* 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}))?" + // _ (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}))?" + /* _ (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; } 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..1b2ea5675dba 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 @@ -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_}) 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). + *

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

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 && UnstableRenameFileSystem.matches(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 +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,7 +5301,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 +5309,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/java/org/apache/hadoop/hive/ql/metadata/UnstableRenameFileSystem.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/UnstableRenameFileSystem.java new file mode 100644 index 000000000000..9972e2833fd8 --- /dev/null +++ b/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. + *

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

+ * 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 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()); + } +} 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..7e798bc3188f 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)); } + /** + * 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"); 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..434ac1c8d11e 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,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_} suffixed names — no plain {@code 000000_0}, no {@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 match {@link UnstableRenameFileSystem}, which is not easily synthesizable with + * LocalFileSystem in a JUnit environment: + *

    + *
  1. {@link UnstableRenameFileSystem#matches(String)} recognizes S3-family schemes 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 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.
    3. + *
+ */ + @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); + } } From 7a10a008a29ed667d00cb0b940a303ca2161105f Mon Sep 17 00:00:00 2001 From: Laszlo Bodor Date: Tue, 28 Jul 2026 14:17:44 +0200 Subject: [PATCH 2/2] PR comments --- .../hive/ql/exec/ParsedOutputFileName.java | 18 ++-- .../apache/hadoop/hive/ql/metadata/Hive.java | 53 ++++++++++-- .../ql/metadata/UnstableRenameFileSystem.java | 84 ------------------- .../ql/exec/ParsedOutputFileNameTest.java | 6 +- .../hive/ql/metadata/TestHiveCopyFiles.java | 78 ++++++++++------- 5 files changed, 106 insertions(+), 133 deletions(-) delete mode 100644 ql/src/java/org/apache/hadoop/hive/ql/metadata/UnstableRenameFileSystem.java 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 0f6f69bef519..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 @@ -33,7 +33,7 @@ * 00001_02.zlib.gz * 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, + * 00001_02_copy_abcd1234 (per-query uniqueness tag as copy suffix, * used on unstable-rename filesystems) * 00001_02_copy_abcd1234.gz *

@@ -42,12 +42,12 @@ */ public class ParsedOutputFileName { private static final Pattern COPY_FILE_NAME_TO_TASK_ID_REGEX = Pattern.compile( - "^(.*?)?" + /* any prefix */ - "(\\(.*\\))?" + /* taskId prefix */ - "(\\d+)" + /* taskId */ - "(?:_(\\d{1,6}))?" + /* _ (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 */ + "^(.*?)?" + // any prefix + "(\\(.*\\))?" + // taskId prefix + "(\\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) { return new ParsedOutputFileName(fileName); @@ -113,8 +113,8 @@ public boolean isCopyFile() { /** * @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. + * 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 1b2ea5675dba..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; @@ -5175,20 +5195,37 @@ private static String getPathName(int taskId) { * 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}; 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). + *

+ * 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 (qid == null || qid.isEmpty()) { - return ""; + 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 String.format("%08x", qid.hashCode()); + return NON_ATOMIC_RENAME_SCHEMES.contains(fs.getUri().getScheme().toLowerCase()); } /** @@ -5229,7 +5266,7 @@ private static Path pickDestFilePath(HiveConf conf, Path sourcePath, FileSystem Path destFilePath = new Path(destDirPath, taskId == -1 ? fullName : name); final String uniqueCopySuffix = - (taskId == -1 && isRenameAllowed && !isOverwrite && UnstableRenameFileSystem.matches(destFs)) + (taskId == -1 && isRenameAllowed && !isOverwrite && isNonAtomicRenameFs(destFs)) ? computeUniquenessTag(conf) : null; diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/UnstableRenameFileSystem.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/UnstableRenameFileSystem.java deleted file mode 100644 index 9972e2833fd8..000000000000 --- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/UnstableRenameFileSystem.java +++ /dev/null @@ -1,84 +0,0 @@ -/* - * 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. - *

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

- * 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 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()); - } -} 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 7e798bc3188f..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 @@ -121,9 +121,9 @@ public void testCopyAllParts() throws Exception { } /** - * 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. + * 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 { 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 434ac1c8d11e..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 @@ -39,6 +39,7 @@ 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) @@ -232,51 +233,70 @@ public void testCopyExistingFilesOnDifferentFileSystem() throws IOException { } /** - * 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_} suffixed names — no plain {@code 000000_0}, no {@code _copy_N}. + * 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 match {@link UnstableRenameFileSystem}, which is not easily synthesizable with + * scheme to be flagged non-atomic-rename, which is not easily synthesizable with * LocalFileSystem in a JUnit environment: *

    - *
  1. {@link UnstableRenameFileSystem#matches(String)} recognizes S3-family schemes and - * rejects HDFS / local schemes.
    2. + *
  2. {@link Hive#isNonAtomicRenameFs(FileSystem)} recognizes S3-family schemes on the URI + * and rejects HDFS / local schemes.
    3. *
  3. 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 + * 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.
    4. *
*/ @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"); + 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, "q2_lbodor_20260101_bbbbbbbb"); + hiveConf.setVar(HiveConf.ConfVars.HIVE_QUERY_ID, + "lbodor_20260101120001_9c8a44f1-e2b3-4a1c-9d3e-000000000000"); String tag2 = Hive.computeUniquenessTag(hiveConf); - hiveConf.unset(HiveConf.ConfVars.HIVE_QUERY_ID.varname); - String tagEmpty = 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); - assertEquals("empty queryId → empty tag", "", tagEmpty); + + // 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")); + } } }