Copilot commented on code in PR #6625:
URL: https://github.com/apache/hive/pull/6625#discussion_r3651951226
##########
ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java:
##########
@@ -1434,100 +1434,38 @@ public static FileStatus[] listStatusIfExists(Path
path, FileSystem fs) throws I
}
public static void mvFileToFinalPath(Path specPath, String unionSuffix,
Configuration hconf,
- boolean success, Logger log,
DynamicPartitionCtx dpCtx, FileSinkDesc conf,
+ boolean success, DynamicPartitionCtx
dpCtx, FileSinkDesc conf,
Reporter reporter) throws IOException,
HiveException {
- // There are following two paths this could could take based on the value
of shouldAvoidRename
- // shouldAvoidRename indicate if tmpPath should be renamed/moved or now.
- // if false:
- // Skip renaming/moving the tmpPath
- // Deduplicate and keep a list of files
- // Pass on the list of files to conf (to be used later by fetch
operator)
- // if true:
- // 1) Rename tmpPath to a new directory name to prevent additional
files
- // from being added by runaway processes.
- // 2) Remove duplicates from the temp directory
- // 3) Rename/move the temp directory to specPath
+ // On success, behavior is controlled by whether the staging directory
should be renamed (see computeAvoidRename).
+ // For SELECT statements and CTAS / Create MV on blob storage, staging
directory is not renamed;
+ // files are deduplicated and either passed to the fetch operator (SELECT)
or moved in place (CTAS) via
+ // moveKeptFiles. For INSERT, LOAD and similar write statements:
renameTmpIfNeeded fences runaway processes
+ // by renaming tmpPath to *.moved, then removeDuplicatesAndFillBuckets +
moveKeptFiles move files to specPath.
+ // If the branch produced no output, then .moved staging dir is deleted.
FileSystem fs = specPath.getFileSystem(hconf);
Path tmpPath = Utilities.toTempPath(specPath);
Path taskTmpPath = Utilities.toTaskTempPath(specPath);
if (!StringUtils.isEmpty(unionSuffix)) {
specPath = specPath.getParent();
}
- PerfLogger perfLogger = SessionState.getPerfLogger();
- boolean isBlobStorage = BlobStorageUtils.isBlobStorageFileSystem(hconf,
fs);
- boolean avoidRename = false;
- boolean shouldAvoidRename = shouldAvoidRename(conf, hconf);
-
- if(isBlobStorage && (shouldAvoidRename|| ((conf != null) &&
conf.isCTASorCM()))
- || (!isBlobStorage && shouldAvoidRename)) {
- avoidRename = true;
- }
+ boolean avoidRename = computeAvoidRename(conf, hconf, fs);
if (success) {
- if (!avoidRename && fs.exists(tmpPath)) {
- // 1) Rename tmpPath to a new directory name to prevent additional
files
- // from being added by runaway processes.
- // this is only done for all statements except SELECT, CTAS and Create
MV
- Path tmpPathOriginal = tmpPath;
- tmpPath = new Path(tmpPath.getParent(), tmpPath.getName() + ".moved");
- LOG.debug("shouldAvoidRename is false therefore moving/renaming " +
tmpPathOriginal + " to " + tmpPath);
- perfLogger.perfLogBegin("FileSinkOperator", "rename");
- Utilities.rename(fs, tmpPathOriginal, tmpPath);
- perfLogger.perfLogEnd("FileSinkOperator", "rename");
- }
-
- // Remove duplicates from tmpPath
+ tmpPath = renameTmpIfNeeded(fs, tmpPath, avoidRename);
List<FileStatus> statusList = HiveStatsUtils.getFileStatusRecurse(
- tmpPath, ((dpCtx == null) ? 1 : dpCtx.getNumDPCols()), fs);
- FileStatus[] statuses = statusList.toArray(new
FileStatus[statusList.size()]);
- if(statuses != null && statuses.length > 0) {
- Set<FileStatus> filesKept = new HashSet<>();
- perfLogger.perfLogBegin("FileSinkOperator",
"RemoveTempOrDuplicateFiles");
- // remove any tmp file or double-committed output files
- int dpLevels = dpCtx == null ? 0 : dpCtx.getNumDPCols(),
- numBuckets = (conf != null && conf.getTable() != null) ?
conf.getTable().getNumBuckets() : 0;
- List<Path> emptyBuckets = removeTempOrDuplicateFiles(
- fs, statuses, unionSuffix, dpLevels, numBuckets, hconf, null, 0,
false, filesKept, false);
- perfLogger.perfLogEnd("FileSinkOperator",
"RemoveTempOrDuplicateFiles");
- // create empty buckets if necessary
- if (!emptyBuckets.isEmpty()) {
- perfLogger.perfLogBegin("FileSinkOperator", "CreateEmptyBuckets");
- createEmptyBuckets(
- hconf, emptyBuckets, conf.getCompressed(), conf.getTableInfo(),
reporter);
- for(Path p:emptyBuckets) {
- FileStatus[] items = fs.listStatus(p);
- filesKept.addAll(Arrays.asList(items));
- }
- perfLogger.perfLogEnd("FileSinkOperator", "CreateEmptyBuckets");
- }
-
- // move to the file destination
- Utilities.FILE_OP_LOGGER.trace("Moving tmp dir: {} to: {}", tmpPath,
specPath);
- if(shouldAvoidRename(conf, hconf)){
- // for SELECT statements
- LOG.debug("Skipping rename/move files. Files to be kept are: " +
filesKept.toString());
- conf.getFilesToFetch().addAll(filesKept);
- } else if (conf !=null && conf.isCTASorCM() && isBlobStorage) {
- // for CTAS or Create MV statements
- perfLogger.perfLogBegin("FileSinkOperator",
"moveSpecifiedFileStatus");
- LOG.debug("CTAS/Create MV: Files being renamed: " +
filesKept.toString());
- if (conf.getTable() != null &&
conf.getTable().getTableType().equals(TableType.EXTERNAL_TABLE)) {
- // Do this optimisation only for External tables.
- createFileList(filesKept, tmpPath, specPath, fs);
- } else {
- Set<String> filesKeptPaths = filesKept.stream().map(x ->
x.getPath().toString()).collect(Collectors.toSet());
- moveSpecifiedFilesInParallel(hconf, fs, tmpPath, specPath,
filesKeptPaths);
- }
- perfLogger.perfLogEnd("FileSinkOperator", "moveSpecifiedFileStatus");
- } else {
- // for rest of the statement e.g. INSERT, LOAD etc
- perfLogger.perfLogBegin("FileSinkOperator", "RenameOrMoveFiles");
- LOG.debug("Final renaming/moving. Source: " + tmpPath + "
.Destination: " + specPath);
- renameOrMoveFilesInParallel(hconf, fs, tmpPath, specPath);
- perfLogger.perfLogEnd("FileSinkOperator", "RenameOrMoveFiles");
- }
+ tmpPath, (dpCtx == null ? 1 : dpCtx.getNumDPCols()), fs);
+ FileStatus[] statuses = statusList.toArray(new FileStatus[0]);
+ if (statuses.length > 0) {
+ Set<FileStatus> filesKept = removeDuplicatesAndFillBuckets(
+ hconf, fs, statuses, unionSuffix, dpCtx, conf, reporter);
+ moveKeptFiles(hconf, fs, tmpPath, specPath, conf, filesKept);
+ } else if (!avoidRename) {
+ // Empty UNION branch: .moved staging dir exists but contains no
output files - delete it
+ // (e.g. UNION ALL branch produced 0 rows)
+ Utilities.FILE_OP_LOGGER.debug("Deleting empty staging directory after
commit: {}", tmpPath);
+ fs.delete(tmpPath, false);
}
Review Comment:
`statuses.length == 0` is based on `HiveStatsUtils.getFileStatusRecurse`,
which uses `FileUtils.HIDDEN_FILES_PATH_FILTER` and may therefore miss hidden
children (and it also doesn’t prove the directory has no subdirectories). In
that case, `fs.delete(tmpPath, false)` will return false and leave the `.moved`
staging dir behind, undermining the cleanup this change is trying to ensure.
Prefer a recursive delete here (or at least handle the boolean return) so
cleanup is reliable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
