You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by sa...@apache.org on 2018/09/29 07:50:18 UTC
hive git commit: Revert "HIVE-20531: Repl load on cloud storage file
system can skip redundant move or add partition tasks (Mahesh Kumar Behera,
reviewed by Sankar Hariappan)"
Repository: hive
Updated Branches:
refs/heads/master b13eba000 -> 05ddd21c7
Revert "HIVE-20531: Repl load on cloud storage file system can skip redundant move or add partition tasks (Mahesh Kumar Behera, reviewed by Sankar Hariappan)"
This reverts commit b13eba000e1cce39a7449b289c659c2f236ce9c9.
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/05ddd21c
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/05ddd21c
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/05ddd21c
Branch: refs/heads/master
Commit: 05ddd21c7e7980e3a1b06d3faa0ebbd358ed472d
Parents: b13eba0
Author: Sankar Hariappan <sa...@apache.org>
Authored: Sat Sep 29 13:19:58 2018 +0530
Committer: Sankar Hariappan <sa...@apache.org>
Committed: Sat Sep 29 13:19:58 2018 +0530
----------------------------------------------------------------------
.../org/apache/hadoop/hive/conf/HiveConf.java | 6 --
.../hive/ql/parse/TestReplicationScenarios.java | 104 -------------------
...TestReplicationScenariosAcrossInstances.java | 3 +-
.../bootstrap/load/table/LoadPartitions.java | 17 +--
.../repl/bootstrap/load/table/LoadTable.java | 2 +-
.../hive/ql/parse/ImportSemanticAnalyzer.java | 13 ---
.../ql/parse/ReplicationSemanticAnalyzer.java | 18 +---
.../hadoop/hive/ql/plan/LoadTableDesc.java | 4 +-
8 files changed, 12 insertions(+), 155 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/05ddd21c/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
----------------------------------------------------------------------
diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
index b365f5e..0cecae5 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -495,12 +495,6 @@ public class HiveConf extends Configuration {
+ " the REPL LOAD on object data stores such as S3 or WASB where creating a directory and move \n"
+ " files are costly operations. In file system like HDFS where move operation is atomic, this \n"
+ " optimization should not be enabled as it may lead to inconsistent data read for non acid tables."),
- REPL_MOVE_OPTIMIZED_FILE_SCHEMES("hive.repl.move.optimized.scheme", "s3a, wasb",
- "Comma separated list of schemes for which move optimization will be enabled during repl load. \n"
- + "This configuration overrides the value set using REPL_ENABLE_MOVE_OPTIMIZATION for the given schemes. \n"
- + " Schemes of the file system which does not support atomic move (rename) can be specified here to \n "
- + " speed up the repl load operation. In file system like HDFS where move operation is atomic, this \n"
- + " optimization should not be enabled as it may lead to inconsistent data read for non acid tables."),
LOCALSCRATCHDIR("hive.exec.local.scratchdir",
"${system:java.io.tmpdir}" + File.separator + "${system:user.name}",
"Local scratch space for Hive jobs"),
http://git-wip-us.apache.org/repos/asf/hive/blob/05ddd21c/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenarios.java
----------------------------------------------------------------------
diff --git a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenarios.java b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenarios.java
index 0d6ad99..a9783ab 100644
--- a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenarios.java
+++ b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenarios.java
@@ -52,17 +52,9 @@ import org.apache.hadoop.hive.metastore.messaging.event.filters.AndFilter;
import org.apache.hadoop.hive.metastore.messaging.event.filters.DatabaseAndTableFilter;
import org.apache.hadoop.hive.metastore.messaging.event.filters.EventBoundaryFilter;
import org.apache.hadoop.hive.metastore.messaging.event.filters.MessageFormatFilter;
-import org.apache.hadoop.hive.ql.DriverContext;
import org.apache.hadoop.hive.ql.DriverFactory;
import org.apache.hadoop.hive.ql.IDriver;
-import org.apache.hadoop.hive.ql.exec.DDLTask;
-import org.apache.hadoop.hive.ql.exec.MoveTask;
-import org.apache.hadoop.hive.ql.exec.Task;
-import org.apache.hadoop.hive.ql.exec.TaskFactory;
import org.apache.hadoop.hive.ql.exec.repl.ReplDumpWork;
-import org.apache.hadoop.hive.ql.exec.repl.ReplLoadWork;
-import org.apache.hadoop.hive.ql.metadata.Hive;
-import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.parse.repl.load.EventDumpDirComparator;
import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
import org.apache.hadoop.hive.ql.session.SessionState;
@@ -88,7 +80,6 @@ import javax.annotation.Nullable;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
-import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -324,101 +315,6 @@ public class TestReplicationScenarios {
verifyRun("SELECT * from " + replicatedDbName + ".unptned_empty", empty, driverMirror);
}
- private abstract class checkTaskPresent {
- public boolean hasTask(Task rootTask) {
- if (rootTask == null) {
- return false;
- }
- if (validate(rootTask)) {
- return true;
- }
- List<Task<? extends Serializable>> childTasks = rootTask.getChildTasks();
- if (childTasks == null) {
- return false;
- }
- for (Task<? extends Serializable> childTask : childTasks) {
- if (hasTask(childTask)) {
- return true;
- }
- }
- return false;
- }
-
- public abstract boolean validate(Task task);
- }
-
- private boolean hasMoveTask(Task rootTask) {
- checkTaskPresent validator = new checkTaskPresent() {
- public boolean validate(Task task) {
- return (task instanceof MoveTask);
- }
- };
- return validator.hasTask(rootTask);
- }
-
- private boolean hasPartitionTask(Task rootTask) {
- checkTaskPresent validator = new checkTaskPresent() {
- public boolean validate(Task task) {
- if (task instanceof DDLTask) {
- DDLTask ddlTask = (DDLTask)task;
- if (ddlTask.getWork().getAddPartitionDesc() != null) {
- return true;
- }
- }
- return false;
- }
- };
- return validator.hasTask(rootTask);
- }
-
- private Task getReplLoadRootTask(String replicadb, boolean isIncrementalDump, Tuple tuple) throws Throwable {
- HiveConf confTemp = new HiveConf();
- confTemp.set("hive.repl.enable.move.optimization", "true");
- ReplLoadWork replLoadWork = new ReplLoadWork(confTemp, tuple.dumpLocation, replicadb,
- null, null, isIncrementalDump);
- Task replLoadTask = TaskFactory.get(replLoadWork, confTemp);
- replLoadTask.initialize(null, null, new DriverContext(driver.getContext()), null);
- replLoadTask.executeTask(null);
- Hive.getThreadLocal().closeCurrent();
- return replLoadWork.getRootTask();
- }
-
- @Test
- public void testTaskCreationOptimization() throws Throwable {
- String name = testName.getMethodName();
- String dbName = createDB(name, driver);
- String dbNameReplica = dbName + "_replica";
- run("create table " + dbName + ".t2 (place string) partitioned by (country string)", driver);
- run("insert into table " + dbName + ".t2 partition(country='india') values ('bangalore')", driver);
-
- Tuple dump = replDumpDb(dbName, null, null, null);
-
- //bootstrap load should not have move task
- Task task = getReplLoadRootTask(dbNameReplica, false, dump);
- assertEquals(false, hasMoveTask(task));
- assertEquals(true, hasPartitionTask(task));
-
- loadAndVerify(dbNameReplica, dump.dumpLocation, dump.lastReplId);
-
- run("insert into table " + dbName + ".t2 partition(country='india') values ('delhi')", driver);
- dump = replDumpDb(dbName, dump.lastReplId, null, null);
-
- //no partition task should be added as the operation is inserting into an existing partition
- task = getReplLoadRootTask(dbNameReplica, true, dump);
- assertEquals(true, hasMoveTask(task));
- assertEquals(false, hasPartitionTask(task));
-
- loadAndVerify(dbNameReplica, dump.dumpLocation, dump.lastReplId);
-
- run("insert into table " + dbName + ".t2 partition(country='us') values ('sf')", driver);
- dump = replDumpDb(dbName, dump.lastReplId, null, null);
-
- //no move task should be added as the operation is adding a dynamic partition
- task = getReplLoadRootTask(dbNameReplica, true, dump);
- assertEquals(false, hasMoveTask(task));
- assertEquals(true, hasPartitionTask(task));
- }
-
@Test
public void testBasicWithCM() throws Exception {
String name = testName.getMethodName();
http://git-wip-us.apache.org/repos/asf/hive/blob/05ddd21c/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcrossInstances.java
----------------------------------------------------------------------
diff --git a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcrossInstances.java b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcrossInstances.java
index 6cd1ac0..fb797ff 100644
--- a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcrossInstances.java
+++ b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcrossInstances.java
@@ -50,7 +50,6 @@ import org.apache.hadoop.hive.ql.exec.repl.incremental.IncrementalLoadTasksBuild
import org.junit.Assert;
import java.io.IOException;
-import java.io.Serializable;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
@@ -1394,7 +1393,7 @@ public class TestReplicationScenariosAcrossInstances {
.run("insert overwrite table t1 select * from t2")
.dump(primaryDbName, tuple.lastReplicationId);
- testMoveOptimization(primaryDbName, replicatedDbName, replicatedDbName_CM, "t1", "ADD_PARTITION", tuple);
+ testMoveOptimization(primaryDbName, replicatedDbName, replicatedDbName_CM, "t1", "INSERT", tuple);
}
@Test
http://git-wip-us.apache.org/repos/asf/hive/blob/05ddd21c/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadPartitions.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadPartitions.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadPartitions.java
index 172b4ac..45b674e 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadPartitions.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadPartitions.java
@@ -244,12 +244,7 @@ public class LoadPartitions {
tmpPath,
context.hiveConf, false, false
);
-
- Task<?> movePartitionTask = null;
- if (loadFileType != LoadFileType.IGNORE) {
- // no need to create move task, if file is moved directly to target location.
- movePartitionTask = movePartitionTask(table, partSpec, tmpPath, loadFileType);
- }
+ Task<?> movePartitionTask = movePartitionTask(table, partSpec, tmpPath, loadFileType);
// Set Checkpoint task as dependant to add partition tasks. So, if same dump is retried for
// bootstrap, we skip current partition update.
@@ -265,14 +260,10 @@ public class LoadPartitions {
} else {
ptnRootTask.addDependentTask(copyTask);
}
-
copyTask.addDependentTask(addPartTask);
- if (movePartitionTask != null) {
- addPartTask.addDependentTask(movePartitionTask);
- movePartitionTask.addDependentTask(ckptTask);
- } else {
- addPartTask.addDependentTask(ckptTask);
- }
+ addPartTask.addDependentTask(movePartitionTask);
+ movePartitionTask.addDependentTask(ckptTask);
+
return ptnRootTask;
}
http://git-wip-us.apache.org/repos/asf/hive/blob/05ddd21c/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadTable.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadTable.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadTable.java
index 8538463..82f687b 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadTable.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadTable.java
@@ -234,7 +234,7 @@ public class LoadTable {
LOG.debug("adding dependent CopyWork/AddPart/MoveWork for table "
+ table.getCompleteName() + " with source location: "
- + dataPath.toString() + " and target location " + tgtPath.toString());
+ + dataPath.toString() + " and target location " + tmpPath.toString());
Task<?> copyTask = ReplCopyTask.getLoadCopyTask(replicationSpec, dataPath, tmpPath, context.hiveConf,
false, false);
http://git-wip-us.apache.org/repos/asf/hive/blob/05ddd21c/ql/src/java/org/apache/hadoop/hive/ql/parse/ImportSemanticAnalyzer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/ImportSemanticAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/ImportSemanticAnalyzer.java
index 16ce5d5..c99d9c1 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/ImportSemanticAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/ImportSemanticAnalyzer.java
@@ -609,19 +609,6 @@ public class ImportSemanticAnalyzer extends BaseSemanticAnalyzer {
moveWork.setLoadTableWork(loadTableWork);
}
- if (loadFileType == LoadFileType.IGNORE) {
- // if file is coped directly to the target location, then no need of move task in case the operation getting
- // replayed is add partition. As add partition will add the event for create partition. Even the statics are
- // updated properly in create partition flow as the copy is done directly to the partition location. For insert
- // operations, add partition task is anyways a no-op as alter partition operation does just some statistics
- // update which is again done in load operations as part of move task.
- if (x.getEventType() == DumpType.EVENT_INSERT) {
- copyTask.addDependentTask(TaskFactory.get(moveWork, x.getConf()));
- } else {
- copyTask.addDependentTask(addPartTask);
- }
- return copyTask;
- }
Task<?> loadPartTask = TaskFactory.get(moveWork, x.getConf());
copyTask.addDependentTask(loadPartTask);
addPartTask.addDependentTask(loadPartTask);
http://git-wip-us.apache.org/repos/asf/hive/blob/05ddd21c/ql/src/java/org/apache/hadoop/hive/ql/parse/ReplicationSemanticAnalyzer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/ReplicationSemanticAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/ReplicationSemanticAnalyzer.java
index 87c69cf..cfeb31a 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/ReplicationSemanticAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/ReplicationSemanticAnalyzer.java
@@ -50,7 +50,6 @@ import java.util.Map;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVEQUERYID;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.REPL_DUMP_METADATA_ONLY;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.REPL_ENABLE_MOVE_OPTIMIZATION;
-import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.REPL_MOVE_OPTIMIZED_FILE_SCHEMES;
import static org.apache.hadoop.hive.ql.parse.HiveParser.TOK_DBNAME;
import static org.apache.hadoop.hive.ql.parse.HiveParser.TOK_LIMIT;
import static org.apache.hadoop.hive.ql.parse.HiveParser.TOK_REPL_CONFIG;
@@ -85,6 +84,8 @@ public class ReplicationSemanticAnalyzer extends BaseSemanticAnalyzer {
public static final String FUNCTIONS_ROOT_DIR_NAME = "_functions";
public static final String CONSTRAINTS_ROOT_DIR_NAME = "_constraints";
+ private static final List<String> CLOUD_SCHEME_PREFIXES = Arrays.asList("s3a", "wasb");
+
ReplicationSemanticAnalyzer(QueryState queryState) throws SemanticException {
super(queryState);
this.db = super.db;
@@ -221,7 +222,7 @@ public class ReplicationSemanticAnalyzer extends BaseSemanticAnalyzer {
}
}
- private boolean ifEnableMoveOptimization(Path filePath, org.apache.hadoop.conf.Configuration conf) throws Exception {
+ private boolean isCloudFS(Path filePath, org.apache.hadoop.conf.Configuration conf) throws Exception {
if (filePath == null) {
throw new HiveException("filePath cannot be null");
}
@@ -232,16 +233,7 @@ public class ReplicationSemanticAnalyzer extends BaseSemanticAnalyzer {
if (StringUtils.isBlank(scheme)) {
throw new HiveException("Cannot get valid scheme for " + filePath);
}
-
- LOG.info("scheme is " + scheme);
-
- String[] schmeList = conf.get(REPL_MOVE_OPTIMIZED_FILE_SCHEMES.varname).toLowerCase().split(",");
- for (String schemeIter : schmeList) {
- if (schemeIter.trim().equalsIgnoreCase(scheme.trim())) {
- return true;
- }
- }
- return false;
+ return CLOUD_SCHEME_PREFIXES.contains(scheme.toLowerCase().trim());
}
// REPL LOAD
@@ -334,7 +326,7 @@ public class ReplicationSemanticAnalyzer extends BaseSemanticAnalyzer {
try {
Warehouse wh = new Warehouse(conf);
Path filePath = wh.getWhRoot();
- if (ifEnableMoveOptimization(filePath, conf)) {
+ if (isCloudFS(filePath, conf)) {
conf.setBoolVar(REPL_ENABLE_MOVE_OPTIMIZATION, true);
LOG.info(" Set move optimization to true for warehouse " + filePath.toString());
}
http://git-wip-us.apache.org/repos/asf/hive/blob/05ddd21c/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadTableDesc.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadTableDesc.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadTableDesc.java
index d9333b5..2267df9 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadTableDesc.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadTableDesc.java
@@ -67,9 +67,7 @@ public class LoadTableDesc extends LoadDesc implements Serializable {
*/
OVERWRITE_EXISTING,
/**
- * No need to move the file, used in case of replication to s3. If load type is set to ignore, then only the file
- * operations(move/rename) is ignored at load table/partition method. Other operations like statistics update,
- * event notification happens as usual.
+ * No need to move the file, used in case of replication to s3
*/
IGNORE
}