HIVE-28822: Concurrent INSERTs into a new dynamic partition can silently lose rows or fail with FileAlreadyExistsException on S3 (non-ACID) - #6642
Conversation
|
@abstractdog recent fix in this area: 5ae5a70 cc @difin |
yeah, I confirmed that it didn't solve the problem I was investigating completely so that's why I would really appreciate a review on this patch from you guys :) |
c4ab838 to
977579c
Compare
asf-ci-hive
added
tests pending
tests unstable
and removed
tests unstable
tests pending
labels
977579c to
4f81258
Compare
…tly lose rows or fail with FileAlreadyExistsException on S3 (non-ACID)
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_<queryTag1>
basename_copy_<queryTag2>
— 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_<hex> keys in S3, no FAEE.
Co-Authored-By: Claude <noreply@anthropic.com>
4f81258 to
71f6fe8
Compare
none of the issues was introduced by this patch, this also fixed brain method problem by refactoring logic to a new method |
| * the set of unsafe filesystems is a property of the filesystem implementation, not something an | ||
| * operator should override. | ||
| */ | ||
| public enum UnstableRenameFileSystem { |
There was a problem hiding this comment.
Anything named *FileSystem in this codebase is a FileSystem subclass.
Why not simply use set
private static final Set<String> NON_ATOMIC_RENAME_SCHEMES = ImmutableSet.of("s3a", "s3n", "s3", "gs");
maybe introduce instead allowlist - the known-safe (hdfs, file, viewfs) ?
There was a problem hiding this comment.
ack, and this is over-engineered too, fixed in aab3c3b
| 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"); |
There was a problem hiding this comment.
did you consider abfs? please check BlobStorageUtils
There was a problem hiding this comment.
cannot make sure, depends on whether hierarchical namespaces are enabled
The atomic rename feature is not supported by the ABFS scheme ; however, rename, create and delete operations are atomic if Namespace is enabled for your Azure Storage account.
https://hadoop.apache.org/docs/stable/hadoop-azure/abfs.html#Rename_Options
https://learn.microsoft.com/en-us/azure/storage/blobs/data-lake-storage-namespace
given the new implementation has no performance implications, I'm simply applying this for abfs too, instead of hacking further to decide whether it's hiearchical or not
| if (qid == null || qid.isEmpty()) { | ||
| return ""; | ||
| } | ||
| return String.format("%08x", qid.hashCode()); |
There was a problem hiding this comment.
HIVE_QUERY_ID = hive_<ts>_<uuid>, can we extract uuid from there?
There was a problem hiding this comment.
maybe move to QueryPlan next to makeQueryId
public static String extractUniquenessTag(String queryId) {
UUID uuid = UUID.fromString(queryId.substring(queryId.lastIndexOf('_') + 1));
return String.format("%016x", uuid.getMostSignificantBits());
}
| */ | ||
| static String computeUniquenessTag(HiveConf conf) { | ||
| String qid = HiveConf.getVar(conf, ConfVars.HIVE_QUERY_ID); | ||
| if (qid == null || qid.isEmpty()) { |
There was a problem hiding this comment.
if (StringUtils.isEmpty(qid)) {
throw new IllegalStateException("hive.query.id is required to derive a unique destination name");
}
|
do we need to fix Utilities.moveFile as well ? |
the same pattern, yes, created follow-up ticket about that: https://issues.apache.org/jira/browse/HIVE-29775 |
aab3c3b to
7a10a00
Compare
|
The tag is per-query, but a single query can move multiple files with the same basename into the same destination directory, isn't it? FS On master, the exists-probe loop resolves it (000000_0 + 000000_0_copy_1); under the PR both legs compute the same 000000_0_copy_ and collide HIVE-21100 seems to add branch index, so we might be sorted |
| final String fullName = sourcePath.getName(); | ||
|
|
||
| final String name; | ||
| if (taskId == -1) { // non-acid |
There was a problem hiding this comment.
are you sure it also covers Insert-only (MM) transactional tables? otherwise we might fail in AcidUtils
| "(?:_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 |
There was a problem hiding this comment.
won't we fail on convertion non-ACID managed table to ACID ? AcidUtils.ORIGINAL_PATTERN_COPY won't match
There was a problem hiding this comment.
good catch, need to check
|
ack, this has to be sorted now, because "the exists-probe loop resolves it" is just true to a certain extent, which is still subject to the reported problem, which is the race in multiple places in the copy++ loop: I'm going to address this as well and let you know regarding HIVE-21100 that's another area that might be investigated, because it claims: |
3 participants
What changes were proposed in this pull request?
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.
Why are the changes needed?
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.Does this PR introduce any user-facing change?
It depends on whether the actual user is interested in the underlying file structure (not directory, but files).
Pre-patch, the unstable result of a highly concurrent INSERT INTO scenario was something like below: be mindful of 30 insert operations vs. 24 result files, this is just one of the symptoms I referred to as "Fail-silent" above:
After the patch, it becomes:
How was this patch tested?
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.
Disclaimer: end-to-end testing was done by Claude after I made a following small testing infra available for it:
post-patch: