You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by go...@apache.org on 2018/09/04 22:07:24 UTC
hive git commit: HIVE-20395. Parallelize files move in 'replaceFiles'
method. (Igor Dvorzhak, reviewed by Gopal Vijayaraghavan)
Repository: hive
Updated Branches:
refs/heads/master df1a02be3 -> 33fa62f18
HIVE-20395. Parallelize files move in 'replaceFiles' method. (Igor Dvorzhak, reviewed by Gopal Vijayaraghavan)
Signed-off-by: Gopal V <go...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/33fa62f1
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/33fa62f1
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/33fa62f1
Branch: refs/heads/master
Commit: 33fa62f18849f130fb14727d390888e7b0e5d09d
Parents: df1a02b
Author: Igor Dvorzhak <id...@google.com>
Authored: Tue Sep 4 15:01:45 2018 -0700
Committer: Gopal V <go...@apache.org>
Committed: Tue Sep 4 15:01:58 2018 -0700
----------------------------------------------------------------------
.../apache/hadoop/hive/ql/metadata/Hive.java | 41 +++++++++++++++++++-
1 file changed, 39 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/33fa62f1/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
----------------------------------------------------------------------
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 9cb5717..0b1048c 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
@@ -24,6 +24,7 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import static org.apache.hadoop.hive.conf.Constants.MATERIALIZED_VIEW_REWRITING_TIME_WINDOW;
@@ -4362,16 +4363,52 @@ private void constructOneLBLocationMap(FileStatus fSta,
listNewFilesRecursively(destFs, destf, newFiles);
}
} else {
+ final Map<Future<Boolean>, Path> moveFutures = Maps.newLinkedHashMapWithExpectedSize(srcs.length);
+ final int moveFilesThreadCount = HiveConf.getIntVar(conf, ConfVars.HIVE_MOVE_FILES_THREAD_COUNT);
+ final ExecutorService pool = moveFilesThreadCount > 0
+ ? Executors.newFixedThreadPool(
+ moveFilesThreadCount,
+ new ThreadFactoryBuilder().setDaemon(true).setNameFormat("Replace-Thread-%d").build())
+ : MoreExecutors.newDirectExecutorService();
+ final SessionState parentSession = SessionState.get();
// its either a file or glob
for (FileStatus src : srcs) {
Path destFile = new Path(destf, src.getPath().getName());
- if (!moveFile(conf, src.getPath(), destFile, true, isSrcLocal, isManaged)) {
+ moveFutures.put(
+ pool.submit(
+ new Callable<Boolean>() {
+ @Override
+ public Boolean call() throws Exception {
+ SessionState.setCurrentSessionState(parentSession);
+ return moveFile(
+ conf, src.getPath(), destFile, true, isSrcLocal, isManaged);
+ }
+ }),
+ destFile);
+ }
+
+ pool.shutdown();
+ for (Map.Entry<Future<Boolean>, Path> moveFuture : moveFutures.entrySet()) {
+ boolean moveFailed;
+ try {
+ moveFailed = !moveFuture.getKey().get();
+ } catch (InterruptedException | ExecutionException e) {
+ pool.shutdownNow();
+ if (e.getCause() instanceof IOException) {
+ throw (IOException) e.getCause();
+ }
+ if (e.getCause() instanceof HiveException) {
+ throw (HiveException) e.getCause();
+ }
+ throw handlePoolException(pool, e);
+ }
+ if (moveFailed) {
throw new IOException("Error moving: " + srcf + " into: " + destf);
}
// Add file paths of the files that will be moved to the destination if the caller needs it
if (null != newFiles) {
- newFiles.add(destFile);
+ newFiles.add(moveFuture.getValue());
}
}
}