You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by ma...@apache.org on 2019/05/03 03:56:52 UTC
[hive] branch master updated: HIVE-21598 : CTAS on ACID table
during incremental does not replicate data (Ashutosh Bapat reviewed by
Mahesh Kumar Behera)
This is an automated email from the ASF dual-hosted git repository.
mahesh pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new df5a8f1 HIVE-21598 : CTAS on ACID table during incremental does not replicate data (Ashutosh Bapat reviewed by Mahesh Kumar Behera)
df5a8f1 is described below
commit df5a8f124174f580df51d6217822852fcc3fb1ee
Author: Ashutosh Bapat <ab...@cloudera.com>
AuthorDate: Fri May 3 09:25:44 2019 +0530
HIVE-21598 : CTAS on ACID table during incremental does not replicate data (Ashutosh Bapat reviewed by Mahesh Kumar Behera)
Signed-off-by: Mahesh Kumar Behera <ma...@apache.org>
---
.../parse/BaseReplicationScenariosAcidTables.java | 387 +++++++++++++
...estReplAcidTablesBootstrapWithJsonMessage.java} | 11 +-
.../parse/TestReplAcidTablesWithJsonMessage.java | 2 -
.../parse/TestReplicationScenariosAcidTables.java | 626 +--------------------
...estReplicationScenariosAcidTablesBootstrap.java | 334 +++++++++++
.../hive/ql/ddl/table/CreateTableOperation.java | 26 +
.../org/apache/hadoop/hive/ql/metadata/Hive.java | 4 +-
.../ptest2/conf/deployed/master-mr2.properties | 4 +-
8 files changed, 782 insertions(+), 612 deletions(-)
diff --git a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/BaseReplicationScenariosAcidTables.java b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/BaseReplicationScenariosAcidTables.java
new file mode 100644
index 0000000..bfc4abd
--- /dev/null
+++ b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/BaseReplicationScenariosAcidTables.java
@@ -0,0 +1,387 @@
+/*
+ * 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.parse;
+
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hive.cli.CliSessionState;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.api.AllocateTableWriteIdsRequest;
+import org.apache.hadoop.hive.metastore.api.OpenTxnRequest;
+import org.apache.hadoop.hive.metastore.api.OpenTxnsResponse;
+import org.apache.hadoop.hive.metastore.txn.TxnDbUtil;
+import org.apache.hadoop.hive.metastore.txn.TxnStore;
+import org.apache.hadoop.hive.ql.DriverFactory;
+import org.apache.hadoop.hive.ql.IDriver;
+import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
+import org.apache.hadoop.hive.ql.session.SessionState;
+import org.apache.hadoop.hive.shims.Utils;
+
+import org.junit.rules.TestName;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.AfterClass;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Collections;
+import java.util.Map;
+
+import static org.apache.hadoop.hive.metastore.ReplChangeManager.SOURCE_OF_REPLICATION;
+
+/**
+ * TestReplicationScenariosAcidTablesBase - base class for replication for ACID tables tests
+ */
+public class BaseReplicationScenariosAcidTables {
+
+ @Rule
+ public final TestName testName = new TestName();
+
+ protected static final Logger LOG = LoggerFactory.getLogger(TestReplicationScenarios.class);
+ static WarehouseInstance primary;
+ static WarehouseInstance replica, replicaNonAcid;
+ static HiveConf conf;
+ String primaryDbName, replicatedDbName;
+ List<String> acidTableNames = new LinkedList<>();
+ private List<String> nonAcidTableNames = new LinkedList<>();
+
+ static void internalBeforeClassSetup(Map<String, String> overrides, Class clazz)
+ throws Exception {
+ conf = new HiveConf(clazz);
+ conf.set("dfs.client.use.datanode.hostname", "true");
+ conf.set("hadoop.proxyuser." + Utils.getUGI().getShortUserName() + ".hosts", "*");
+ MiniDFSCluster miniDFSCluster =
+ new MiniDFSCluster.Builder(conf).numDataNodes(1).format(true).build();
+ HashMap<String, String> acidEnableConf = new HashMap<String, String>() {{
+ put("fs.defaultFS", miniDFSCluster.getFileSystem().getUri().toString());
+ put("hive.support.concurrency", "true");
+ put("hive.txn.manager", "org.apache.hadoop.hive.ql.lockmgr.DbTxnManager");
+ put("hive.metastore.client.capability.check", "false");
+ put("hive.repl.bootstrap.dump.open.txn.timeout", "1s");
+ put("hive.exec.dynamic.partition.mode", "nonstrict");
+ put("hive.strict.checks.bucketing", "false");
+ put("hive.mapred.mode", "nonstrict");
+ put("mapred.input.dir.recursive", "true");
+ put("hive.metastore.disallow.incompatible.col.type.changes", "false");
+ put("hive.in.repl.test", "true");
+ }};
+
+ acidEnableConf.putAll(overrides);
+
+ primary = new WarehouseInstance(LOG, miniDFSCluster, acidEnableConf);
+ replica = new WarehouseInstance(LOG, miniDFSCluster, acidEnableConf);
+ HashMap<String, String> overridesForHiveConf1 = new HashMap<String, String>() {{
+ put("fs.defaultFS", miniDFSCluster.getFileSystem().getUri().toString());
+ put("hive.support.concurrency", "false");
+ put("hive.txn.manager", "org.apache.hadoop.hive.ql.lockmgr.DummyTxnManager");
+ put("hive.metastore.client.capability.check", "false");
+ }};
+ replicaNonAcid = new WarehouseInstance(LOG, miniDFSCluster, overridesForHiveConf1);
+ }
+
+ @AfterClass
+ public static void classLevelTearDown() throws IOException {
+ primary.close();
+ replica.close();
+ }
+
+ @Before
+ public void setup() throws Throwable {
+ primaryDbName = testName.getMethodName() + "_" + +System.currentTimeMillis();
+ replicatedDbName = "replicated_" + primaryDbName;
+ primary.run("create database " + primaryDbName + " WITH DBPROPERTIES ( '" +
+ SOURCE_OF_REPLICATION + "' = '1,2,3')");
+ String primaryDbNameExtra = primaryDbName+"_extra";
+ primary.run("create database " + primaryDbNameExtra + " WITH DBPROPERTIES ( '" +
+ SOURCE_OF_REPLICATION + "' = '1,2,3')");
+ }
+
+ @After
+ public void tearDown() throws Throwable {
+ primary.run("drop database if exists " + primaryDbName + " cascade");
+ replica.run("drop database if exists " + replicatedDbName + " cascade");
+ replicaNonAcid.run("drop database if exists " + replicatedDbName + " cascade");
+ primary.run("drop database if exists " + primaryDbName + "_extra cascade");
+ }
+
+ private void prepareAcidData(String primaryDbName) throws Throwable {
+ primary.run("use " + primaryDbName)
+ .run("create table t1 (id int) clustered by(id) into 3 buckets stored as orc " +
+ "tblproperties (\"transactional\"=\"true\")")
+ .run("insert into t1 values(1)")
+ .run("insert into t1 values(2)")
+ .run("create table t2 (place string) partitioned by (country string) clustered by(place) " +
+ "into 3 buckets stored as orc tblproperties (\"transactional\"=\"true\")")
+ .run("insert into t2 partition(country='india') values ('bangalore')")
+ .run("insert into t2 partition(country='us') values ('austin')")
+ .run("insert into t2 partition(country='france') values ('paris')")
+ .run("alter table t2 add partition(country='italy')")
+ .run("create table t3 (rank int) tblproperties(\"transactional\"=\"true\", " +
+ "\"transactional_properties\"=\"insert_only\")")
+ .run("insert into t3 values(11)")
+ .run("insert into t3 values(22)")
+ .run("create table t5 (id int) stored as orc ")
+ .run("insert into t5 values(1111), (2222)")
+ .run("alter table t5 set tblproperties (\"transactional\"=\"true\")")
+ .run("insert into t5 values(3333)");
+ acidTableNames.add("t1");
+ acidTableNames.add("t2");
+ acidTableNames.add("t3");
+ acidTableNames.add("t5");
+ }
+
+ private void prepareNonAcidData(String primaryDbName) throws Throwable {
+ primary.run("use " + primaryDbName)
+ .run("create table t4 (id int)")
+ .run("insert into t4 values(111), (222)");
+ nonAcidTableNames.add("t4");
+ }
+
+ WarehouseInstance.Tuple prepareDataAndDump(String primaryDbName, String fromReplId,
+ List<String> withClause) throws Throwable {
+ prepareAcidData(primaryDbName);
+ prepareNonAcidData(primaryDbName);
+ return primary.run("use " + primaryDbName)
+ .dump(primaryDbName, fromReplId, withClause != null ?
+ withClause : Collections.emptyList());
+ }
+
+ private void verifyNonAcidTableLoad(String replicatedDbName) throws Throwable {
+ replica.run("use " + replicatedDbName)
+ .run("select id from t4 order by id")
+ .verifyResults(new String[] {"111", "222"});
+ }
+
+ private void verifyAcidTableLoad(String replicatedDbName) throws Throwable {
+ replica.run("use " + replicatedDbName)
+ .run("select id from t1 order by id")
+ .verifyResults(new String[]{"1", "2"})
+ .run("select country from t2 order by country")
+ .verifyResults(new String[] {"france", "india", "us"})
+ .run("select rank from t3 order by rank")
+ .verifyResults(new String[] {"11", "22"})
+ .run("select id from t5 order by id")
+ .verifyResults(new String[] {"1111", "2222", "3333"});
+ }
+
+ void verifyLoadExecution(String replicatedDbName, String lastReplId, boolean includeAcid)
+ throws Throwable {
+ List<String> tableNames = new LinkedList<>(nonAcidTableNames);
+ if (includeAcid) {
+ tableNames.addAll(acidTableNames);
+ }
+ replica.run("use " + replicatedDbName)
+ .run("show tables")
+ .verifyResults(tableNames)
+ .run("repl status " + replicatedDbName)
+ .verifyResult(lastReplId);
+ verifyNonAcidTableLoad(replicatedDbName);
+ if (includeAcid) {
+ verifyAcidTableLoad(replicatedDbName);
+ }
+ }
+
+ void prepareIncAcidData(String dbName) throws Throwable {
+ primary.run("use " + dbName)
+ .run("create table t6 stored as orc tblproperties (\"transactional\"=\"true\")" +
+ " as select * from t1")
+ .run("alter table t2 add columns (placetype string)")
+ .run("update t2 set placetype = 'city'");
+ acidTableNames.add("t6");
+ }
+
+ private void verifyIncAcidLoad(String dbName) throws Throwable {
+ replica.run("use " + dbName)
+ .run("select id from t6 order by id")
+ .verifyResults(new String[]{"1", "2"})
+ .run("select country from t2 order by country")
+ .verifyResults(new String[] {"france", "india", "us"})
+ .run("select distinct placetype from t2")
+ .verifyResult("city")
+ .run("select id from t1 order by id")
+ .verifyResults(new String[]{"1", "2"})
+ .run("select rank from t3 order by rank")
+ .verifyResults(new String[] {"11", "22"})
+ .run("select id from t5 order by id")
+ .verifyResults(new String[] {"1111", "2222", "3333"});
+ }
+
+ private void runUsingDriver(IDriver driver, String command) throws Throwable {
+ CommandProcessorResponse ret = driver.run(command);
+ if (ret.getException() != null) {
+ throw ret.getException();
+ }
+ }
+
+ void prepareInc2AcidData(String dbName, HiveConf hiveConf) throws Throwable {
+ IDriver driver = DriverFactory.newDriver(hiveConf);
+ SessionState.start(new CliSessionState(hiveConf));
+ runUsingDriver(driver, "use " + dbName);
+ runUsingDriver(driver, "insert into t1 values (3)");
+ runUsingDriver(driver, "insert into t5 values (4444)");
+ }
+
+ private void verifyInc2AcidLoad(String dbName) throws Throwable {
+ replica.run("use " + dbName)
+ .run("select id from t6 order by id")
+ .verifyResults(new String[]{"1", "2"})
+ .run("select country from t2 order by country")
+ .verifyResults(new String[] {"france", "india", "us"})
+ .run("select distinct placetype from t2")
+ .verifyResult("city")
+ .run("select id from t1 order by id")
+ .verifyResults(new String[]{"1", "2", "3"})
+ .run("select rank from t3 order by rank")
+ .verifyResults(new String[] {"11", "22"})
+ .run("select id from t5 order by id")
+ .verifyResults(new String[] {"1111", "2222", "3333", "4444"});
+ }
+
+ void prepareIncNonAcidData(String dbName) throws Throwable {
+ primary.run("use " + dbName)
+ .run("insert into t4 values (333)")
+ .run("create table t7 (str string)")
+ .run("insert into t7 values ('aaa')");
+ nonAcidTableNames.add("t7");
+ }
+
+ private void verifyIncNonAcidLoad(String dbName) throws Throwable {
+ replica.run("use " + dbName)
+ .run("select * from t4 order by id")
+ .verifyResults(new String[] {"111", "222", "333"})
+ .run("select * from t7")
+ .verifyResult("aaa");
+ }
+
+ void prepareInc2NonAcidData(String dbName, HiveConf hiveConf) throws Throwable {
+ IDriver driver = DriverFactory.newDriver(hiveConf);
+ SessionState.start(new CliSessionState(hiveConf));
+ runUsingDriver(driver, "use " + dbName);
+ runUsingDriver(driver, "insert into t4 values (444)");
+ runUsingDriver(driver, "insert into t7 values ('bbb')");
+ }
+
+ private void verifyInc2NonAcidLoad(String dbName) throws Throwable {
+ replica.run("use " + dbName)
+ .run("select * from t4 order by id")
+ .verifyResults(new String[] {"111", "222", "333", "444"})
+ .run("select * from t7")
+ .verifyResults(new String[] {"aaa", "bbb"});
+ }
+
+ void verifyIncLoad(String dbName, String lastReplId)
+ throws Throwable {
+ List<String> tableNames = new LinkedList<>(nonAcidTableNames);
+ tableNames.addAll(acidTableNames);
+ replica.run("use " + dbName)
+ .run("show tables")
+ .verifyResults(tableNames)
+ .run("repl status " + dbName)
+ .verifyResult(lastReplId);
+ verifyIncNonAcidLoad(dbName);
+ verifyIncAcidLoad(dbName);
+ }
+
+ void verifyInc2Load(String dbName, String lastReplId)
+ throws Throwable {
+ List<String> tableNames = new LinkedList<>(nonAcidTableNames);
+ tableNames.addAll(acidTableNames);
+ replica.run("use " + dbName)
+ .run("show tables")
+ .verifyResults(tableNames)
+ .run("repl status " + dbName)
+ .verifyResult(lastReplId);
+ verifyInc2NonAcidLoad(dbName);
+ verifyInc2AcidLoad(dbName);
+ }
+
+ List<Long> openTxns(int numTxns, TxnStore txnHandler, HiveConf primaryConf) throws Throwable {
+ OpenTxnsResponse otResp = txnHandler.openTxns(new OpenTxnRequest(numTxns, "u1", "localhost"));
+ List<Long> txns = otResp.getTxn_ids();
+ String txnIdRange = " txn_id >= " + txns.get(0) + " and txn_id <= " + txns.get(numTxns - 1);
+ Assert.assertEquals(TxnDbUtil.queryToString(primaryConf, "select * from TXNS"),
+ numTxns, TxnDbUtil.countQueryAgent(primaryConf,
+ "select count(*) from TXNS where txn_state = 'o' and " + txnIdRange));
+ return txns;
+ }
+
+ void allocateWriteIdsForTables(String primaryDbName, Map<String, Long> tables,
+ TxnStore txnHandler,
+ List<Long> txns, HiveConf primaryConf) throws Throwable {
+ AllocateTableWriteIdsRequest rqst = new AllocateTableWriteIdsRequest();
+ rqst.setDbName(primaryDbName);
+
+ for(Map.Entry<String, Long> entry : tables.entrySet()) {
+ rqst.setTableName(entry.getKey());
+ rqst.setTxnIds(txns);
+ txnHandler.allocateTableWriteIds(rqst);
+ }
+ verifyWriteIdsForTables(tables, primaryConf, primaryDbName);
+ }
+
+ void verifyWriteIdsForTables(Map<String, Long> tables, HiveConf conf, String dbName)
+ throws Throwable {
+ for(Map.Entry<String, Long> entry : tables.entrySet()) {
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_TO_WRITE_ID"),
+ entry.getValue().longValue(),
+ TxnDbUtil.countQueryAgent(conf,
+ "select count(*) from TXN_TO_WRITE_ID where t2w_database = '"
+ + dbName.toLowerCase()
+ + "' and t2w_table = '" + entry.getKey() + "'"));
+ }
+ }
+
+ void verifyAllOpenTxnsAborted(List<Long> txns, HiveConf primaryConf) throws Throwable {
+ int numTxns = txns.size();
+ String txnIdRange = " txn_id >= " + txns.get(0) + " and txn_id <= " + txns.get(numTxns - 1);
+ Assert.assertEquals(TxnDbUtil.queryToString(primaryConf, "select * from TXNS"),
+ 0, TxnDbUtil.countQueryAgent(primaryConf,
+ "select count(*) from TXNS where txn_state = 'o' and " + txnIdRange));
+ Assert.assertEquals(TxnDbUtil.queryToString(primaryConf, "select * from TXNS"),
+ numTxns, TxnDbUtil.countQueryAgent(primaryConf,
+ "select count(*) from TXNS where txn_state = 'a' and " + txnIdRange));
+ }
+
+ void verifyNextId(Map<String, Long> tables, String dbName, HiveConf conf) throws Throwable {
+ // Verify the next write id
+ for(Map.Entry<String, Long> entry : tables.entrySet()) {
+ String[] nextWriteId =
+ TxnDbUtil.queryToString(conf,
+ "select nwi_next from NEXT_WRITE_ID where nwi_database = '"
+ + dbName.toLowerCase() + "' and nwi_table = '"
+ + entry.getKey() + "'").split("\n");
+ Assert.assertEquals(Long.parseLong(nextWriteId[1].trim()), entry.getValue() + 1);
+ }
+ }
+
+ void verifyCompactionQueue(Map<String, Long> tables, String dbName, HiveConf conf)
+ throws Throwable {
+ for(Map.Entry<String, Long> entry : tables.entrySet()) {
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"),
+ entry.getValue().longValue(),
+ TxnDbUtil.countQueryAgent(conf,
+ "select count(*) from COMPACTION_QUEUE where cq_database = '" + dbName
+ + "' and cq_table = '" + entry.getKey() + "'"));
+ }
+ }
+}
diff --git a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplAcidTablesWithJsonMessage.java b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplAcidTablesBootstrapWithJsonMessage.java
similarity index 82%
copy from itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplAcidTablesWithJsonMessage.java
copy to itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplAcidTablesBootstrapWithJsonMessage.java
index 13dc362..00a9e13 100644
--- a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplAcidTablesWithJsonMessage.java
+++ b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplAcidTablesBootstrapWithJsonMessage.java
@@ -22,7 +22,6 @@ import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.hadoop.hive.metastore.messaging.json.JSONMessageEncoder;
import org.junit.Before;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Rule;
import org.junit.rules.TestRule;
@@ -30,8 +29,12 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
-@Ignore("See HIVE-21647, HIVE-21648")
-public class TestReplAcidTablesWithJsonMessage extends TestReplicationScenariosAcidTables {
+/**
+ * TestReplAcidTablesBootstrapWithJsonMessage - same as
+ * TestReplicationScenariosAcidTablesBootstrap but uses JSON messages.
+ */
+public class TestReplAcidTablesBootstrapWithJsonMessage
+ extends TestReplicationScenariosAcidTablesBootstrap {
@Rule
public TestRule replV1BackwardCompat;
@@ -41,7 +44,7 @@ public class TestReplAcidTablesWithJsonMessage extends TestReplicationScenariosA
Map<String, String> overrides = new HashMap<>();
overrides.put(MetastoreConf.ConfVars.EVENT_MESSAGE_FACTORY.getHiveName(),
JSONMessageEncoder.class.getCanonicalName());
- internalBeforeClassSetup(overrides, TestReplAcidTablesWithJsonMessage.class);
+ internalBeforeClassSetup(overrides, TestReplAcidTablesBootstrapWithJsonMessage.class);
}
@Before
diff --git a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplAcidTablesWithJsonMessage.java b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplAcidTablesWithJsonMessage.java
index 13dc362..747b7a8 100644
--- a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplAcidTablesWithJsonMessage.java
+++ b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplAcidTablesWithJsonMessage.java
@@ -22,7 +22,6 @@ import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.hadoop.hive.metastore.messaging.json.JSONMessageEncoder;
import org.junit.Before;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Rule;
import org.junit.rules.TestRule;
@@ -30,7 +29,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
-@Ignore("See HIVE-21647, HIVE-21648")
public class TestReplAcidTablesWithJsonMessage extends TestReplicationScenariosAcidTables {
@Rule
diff --git a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcidTables.java b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcidTables.java
index 20520ac..96b074d 100644
--- a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcidTables.java
+++ b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcidTables.java
@@ -22,13 +22,8 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hive.cli.CliSessionState;
import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.metastore.api.AllocateTableWriteIdsRequest;
-import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
-import org.apache.hadoop.hive.metastore.api.OpenTxnRequest;
-import org.apache.hadoop.hive.metastore.api.OpenTxnsResponse;
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.hadoop.hive.metastore.messaging.json.gzip.GzipJSONMessageEncoder;
-import org.apache.hadoop.hive.metastore.txn.TxnDbUtil;
import org.apache.hadoop.hive.metastore.txn.TxnStore;
import org.apache.hadoop.hive.metastore.txn.TxnUtils;
import org.apache.hadoop.hive.metastore.InjectableBehaviourObjectStore;
@@ -37,28 +32,19 @@ import org.apache.hadoop.hive.metastore.InjectableBehaviourObjectStore.Behaviour
import org.apache.hadoop.hive.ql.DriverFactory;
import org.apache.hadoop.hive.ql.ErrorMsg;
import org.apache.hadoop.hive.ql.IDriver;
-import org.apache.hadoop.hive.ql.exec.repl.util.ReplUtils;
import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.hive.shims.Utils;
-import org.junit.rules.TestName;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
-import org.junit.Ignore;
-import org.junit.Rule;
import org.junit.Test;
import org.junit.BeforeClass;
-import org.junit.AfterClass;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
-import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
-import java.util.LinkedList;
import java.util.List;
import java.util.Collections;
import java.util.Map;
@@ -67,36 +53,13 @@ import static org.junit.Assert.assertTrue;
import static org.apache.hadoop.hive.metastore.ReplChangeManager.SOURCE_OF_REPLICATION;
/**
- * TestReplicationScenariosAcidTables - test replication for ACID tables
+ * TestReplicationScenariosAcidTables - test replication for ACID tables.
*/
-@Ignore("See HIVE-21647, HIVE-21648")
-public class TestReplicationScenariosAcidTables {
-
- @Rule
- public final TestName testName = new TestName();
-
- protected static final Logger LOG = LoggerFactory.getLogger(TestReplicationScenarios.class);
- static WarehouseInstance primary;
- private static WarehouseInstance replica, replicaNonAcid;
- static HiveConf conf;
- private String primaryDbName, replicatedDbName, primaryDbNameExtra;
- private enum OperationType {
- REPL_TEST_ACID_INSERT, REPL_TEST_ACID_INSERT_SELECT, REPL_TEST_ACID_CTAS,
- REPL_TEST_ACID_INSERT_OVERWRITE, REPL_TEST_ACID_INSERT_IMPORT, REPL_TEST_ACID_INSERT_LOADLOCAL,
- REPL_TEST_ACID_INSERT_UNION
- }
- private static List<String> dumpWithoutAcidClause = Collections.singletonList(
- "'" + ReplUtils.REPL_DUMP_INCLUDE_ACID_TABLES + "'='false'");
- private static List<String> dumpWithAcidBootstrapClause = Arrays.asList(
- "'" + ReplUtils.REPL_DUMP_INCLUDE_ACID_TABLES + "'='true'",
- "'" + HiveConf.ConfVars.REPL_BOOTSTRAP_ACID_TABLES + "'='true'");
- private List<String> acidTableNames = new LinkedList<>();
- private List<String> nonAcidTableNames = new LinkedList<>();
-
+public class TestReplicationScenariosAcidTables extends BaseReplicationScenariosAcidTables {
@BeforeClass
public static void classLevelSetup() throws Exception {
- HashMap<String, String> overrides = new HashMap<>();
+ Map<String, String> overrides = new HashMap<>();
overrides.put(MetastoreConf.ConfVars.EVENT_MESSAGE_FACTORY.getHiveName(),
GzipJSONMessageEncoder.class.getCanonicalName());
@@ -111,48 +74,36 @@ public class TestReplicationScenariosAcidTables {
conf.set("hadoop.proxyuser." + Utils.getUGI().getShortUserName() + ".hosts", "*");
MiniDFSCluster miniDFSCluster =
new MiniDFSCluster.Builder(conf).numDataNodes(1).format(true).build();
- HashMap<String, String> acidEnableConf = new HashMap<String, String>() {{
- put("fs.defaultFS", miniDFSCluster.getFileSystem().getUri().toString());
- put("hive.support.concurrency", "true");
- put("hive.txn.manager", "org.apache.hadoop.hive.ql.lockmgr.DbTxnManager");
- put("hive.metastore.client.capability.check", "false");
- put("hive.repl.bootstrap.dump.open.txn.timeout", "1s");
- put("hive.exec.dynamic.partition.mode", "nonstrict");
- put("hive.strict.checks.bucketing", "false");
- put("hive.mapred.mode", "nonstrict");
- put("mapred.input.dir.recursive", "true");
- put("hive.metastore.disallow.incompatible.col.type.changes", "false");
- put("hive.in.repl.test", "true");
- }};
+ Map<String, String> acidEnableConf = new HashMap<String, String>() {{
+ put("fs.defaultFS", miniDFSCluster.getFileSystem().getUri().toString());
+ put("hive.support.concurrency", "true");
+ put("hive.txn.manager", "org.apache.hadoop.hive.ql.lockmgr.DbTxnManager");
+ put("hive.metastore.client.capability.check", "false");
+ put("hive.repl.bootstrap.dump.open.txn.timeout", "1s");
+ put("hive.exec.dynamic.partition.mode", "nonstrict");
+ put("hive.strict.checks.bucketing", "false");
+ put("hive.mapred.mode", "nonstrict");
+ put("mapred.input.dir.recursive", "true");
+ put("hive.metastore.disallow.incompatible.col.type.changes", "false");
+ put("hive.in.repl.test", "true");
+ }};
acidEnableConf.putAll(overrides);
primary = new WarehouseInstance(LOG, miniDFSCluster, acidEnableConf);
replica = new WarehouseInstance(LOG, miniDFSCluster, acidEnableConf);
- HashMap<String, String> overridesForHiveConf1 = new HashMap<String, String>() {{
- put("fs.defaultFS", miniDFSCluster.getFileSystem().getUri().toString());
- put("hive.support.concurrency", "false");
- put("hive.txn.manager", "org.apache.hadoop.hive.ql.lockmgr.DummyTxnManager");
- put("hive.metastore.client.capability.check", "false");
- }};
+ Map<String, String> overridesForHiveConf1 = new HashMap<String, String>() {{
+ put("fs.defaultFS", miniDFSCluster.getFileSystem().getUri().toString());
+ put("hive.support.concurrency", "false");
+ put("hive.txn.manager", "org.apache.hadoop.hive.ql.lockmgr.DummyTxnManager");
+ put("hive.metastore.client.capability.check", "false");
+ }};
replicaNonAcid = new WarehouseInstance(LOG, miniDFSCluster, overridesForHiveConf1);
}
- @AfterClass
- public static void classLevelTearDown() throws IOException {
- primary.close();
- replica.close();
- }
-
@Before
public void setup() throws Throwable {
- primaryDbName = testName.getMethodName() + "_" + +System.currentTimeMillis();
- replicatedDbName = "replicated_" + primaryDbName;
- primary.run("create database " + primaryDbName + " WITH DBPROPERTIES ( '" +
- SOURCE_OF_REPLICATION + "' = '1,2,3')");
- primaryDbNameExtra = primaryDbName+"_extra";
- primary.run("create database " + primaryDbNameExtra + " WITH DBPROPERTIES ( '" +
- SOURCE_OF_REPLICATION + "' = '1,2,3')");
+ super.setup();
}
@After
@@ -163,82 +114,6 @@ public class TestReplicationScenariosAcidTables {
primary.run("drop database if exists " + primaryDbName + "_extra cascade");
}
- private void prepareAcidData(String primaryDbName) throws Throwable {
- primary.run("use " + primaryDbName)
- .run("create table t1 (id int) clustered by(id) into 3 buckets stored as orc " +
- "tblproperties (\"transactional\"=\"true\")")
- .run("insert into t1 values(1)")
- .run("insert into t1 values(2)")
- .run("create table t2 (place string) partitioned by (country string) clustered by(place) " +
- "into 3 buckets stored as orc tblproperties (\"transactional\"=\"true\")")
- .run("insert into t2 partition(country='india') values ('bangalore')")
- .run("insert into t2 partition(country='us') values ('austin')")
- .run("insert into t2 partition(country='france') values ('paris')")
- .run("alter table t2 add partition(country='italy')")
- .run("create table t3 (rank int) tblproperties(\"transactional\"=\"true\", " +
- "\"transactional_properties\"=\"insert_only\")")
- .run("insert into t3 values(11)")
- .run("insert into t3 values(22)")
- .run("create table t5 (id int) stored as orc ")
- .run("insert into t5 values(1111), (2222)")
- .run("alter table t5 set tblproperties (\"transactional\"=\"true\")")
- .run("insert into t5 values(3333)");
- acidTableNames.add("t1");
- acidTableNames.add("t2");
- acidTableNames.add("t3");
- acidTableNames.add("t5");
- }
-
- private void prepareNonAcidData(String primaryDbName) throws Throwable {
- primary.run("use " + primaryDbName)
- .run("create table t4 (id int)")
- .run("insert into t4 values(111), (222)");
- nonAcidTableNames.add("t4");
- }
- private WarehouseInstance.Tuple prepareDataAndDump(String primaryDbName, String fromReplId,
- List<String> withClause) throws Throwable {
- prepareAcidData(primaryDbName);
- prepareNonAcidData(primaryDbName);
- return primary.run("use " + primaryDbName)
- .dump(primaryDbName, fromReplId, withClause != null ?
- withClause : Collections.emptyList());
- }
-
- private void verifyNonAcidTableLoad(String replicatedDbName) throws Throwable {
- replica.run("use " + replicatedDbName)
- .run("select id from t4 order by id")
- .verifyResults(new String[] {"111", "222"});
- }
-
- private void verifyAcidTableLoad(String replicatedDbName) throws Throwable {
- replica.run("use " + replicatedDbName)
- .run("select id from t1 order by id")
- .verifyResults(new String[]{"1", "2"})
- .run("select country from t2 order by country")
- .verifyResults(new String[] {"france", "india", "us"})
- .run("select rank from t3 order by rank")
- .verifyResults(new String[] {"11", "22"})
- .run("select id from t5 order by id")
- .verifyResults(new String[] {"1111", "2222", "3333"});
- }
-
- private void verifyLoadExecution(String replicatedDbName, String lastReplId, boolean includeAcid)
- throws Throwable {
- List<String> tableNames = new LinkedList<>(nonAcidTableNames);
- if (includeAcid) {
- tableNames.addAll(acidTableNames);
- }
- replica.run("use " + replicatedDbName)
- .run("show tables")
- .verifyResults(tableNames)
- .run("repl status " + replicatedDbName)
- .verifyResult(lastReplId);
- verifyNonAcidTableLoad(replicatedDbName);
- if (includeAcid) {
- verifyAcidTableLoad(replicatedDbName);
- }
- }
-
@Test
public void testAcidTablesBootstrap() throws Throwable {
// Bootstrap
@@ -273,252 +148,6 @@ public class TestReplicationScenariosAcidTables {
verifyLoadExecution(replicatedDbName, bootstrapDump.lastReplicationId, true);
}
- private void prepareIncAcidData(String dbName) throws Throwable {
- primary.run("use " + dbName)
- .run("create table t6 (str string) stored as orc tblproperties " +
- "(\"transactional\"=\"true\")")
- .run("insert into t6 values ('aaa'), ('bbb')")
- .run("alter table t2 add columns (placetype string)")
- .run("update t2 set placetype = 'city'");
- acidTableNames.add("t6");
- }
-
- private void verifyIncAcidLoad(String dbName) throws Throwable {
- replica.run("use " + dbName)
- .run("select str from t6 order by str")
- .verifyResults(new String[]{"aaa", "bbb"})
- .run("select country from t2 order by country")
- .verifyResults(new String[] {"france", "india", "us"})
- .run("select distinct placetype from t2")
- .verifyResult("city")
- .run("select id from t1 order by id")
- .verifyResults(new String[]{"1", "2"})
- .run("select rank from t3 order by rank")
- .verifyResults(new String[] {"11", "22"})
- .run("select id from t5 order by id")
- .verifyResults(new String[] {"1111", "2222", "3333"});
- }
-
- private void runUsingDriver(IDriver driver, String command) throws Throwable {
- CommandProcessorResponse ret = driver.run(command);
- if (ret.getException() != null) {
- throw ret.getException();
- }
- }
-
- private void prepareInc2AcidData(String dbName, HiveConf hiveConf) throws Throwable {
- IDriver driver = DriverFactory.newDriver(hiveConf);
- SessionState.start(new CliSessionState(hiveConf));
- runUsingDriver(driver, "use " + dbName);
- runUsingDriver(driver, "insert into t1 values (3)");
- runUsingDriver(driver, "insert into t5 values (4444)");
- }
-
- private void verifyInc2AcidLoad(String dbName) throws Throwable {
- replica.run("use " + dbName)
- .run("select str from t6 order by str")
- .verifyResults(new String[]{"aaa", "bbb"})
- .run("select country from t2 order by country")
- .verifyResults(new String[] {"france", "india", "us"})
- .run("select distinct placetype from t2")
- .verifyResult("city")
- .run("select id from t1 order by id")
- .verifyResults(new String[]{"1", "2", "3"})
- .run("select rank from t3 order by rank")
- .verifyResults(new String[] {"11", "22"})
- .run("select id from t5 order by id")
- .verifyResults(new String[] {"1111", "2222", "3333", "4444"});
- }
-
- private void prepareIncNonAcidData(String dbName) throws Throwable {
- primary.run("use " + dbName)
- .run("insert into t4 values (333)")
- .run("create table t7 (str string)")
- .run("insert into t7 values ('aaa')");
- nonAcidTableNames.add("t7");
- }
-
- private void verifyIncNonAcidLoad(String dbName) throws Throwable {
- replica.run("use " + dbName)
- .run("select * from t4 order by id")
- .verifyResults(new String[] {"111", "222", "333"})
- .run("select * from t7")
- .verifyResult("aaa");
- }
-
- private void prepareInc2NonAcidData(String dbName, HiveConf hiveConf) throws Throwable {
- IDriver driver = DriverFactory.newDriver(hiveConf);
- SessionState.start(new CliSessionState(hiveConf));
- runUsingDriver(driver, "use " + dbName);
- runUsingDriver(driver, "insert into t4 values (444)");
- runUsingDriver(driver, "insert into t7 values ('bbb')");
- }
-
- private void verifyInc2NonAcidLoad(String dbName) throws Throwable {
- replica.run("use " + dbName)
- .run("select * from t4 order by id")
- .verifyResults(new String[] {"111", "222", "333", "444"})
- .run("select * from t7")
- .verifyResults(new String[] {"aaa", "bbb"});
- }
-
- private void verifyIncLoad(String dbName, String lastReplId)
- throws Throwable {
- List<String> tableNames = new LinkedList<>(nonAcidTableNames);
- tableNames.addAll(acidTableNames);
- replica.run("use " + dbName)
- .run("show tables")
- .verifyResults(tableNames)
- .run("repl status " + dbName)
- .verifyResult(lastReplId);
- verifyIncNonAcidLoad(dbName);
- verifyIncAcidLoad(dbName);
- }
-
- private void verifyInc2Load(String dbName, String lastReplId)
- throws Throwable {
- List<String> tableNames = new LinkedList<>(nonAcidTableNames);
- tableNames.addAll(acidTableNames);
- replica.run("use " + dbName)
- .run("show tables")
- .verifyResults(tableNames)
- .run("repl status " + dbName)
- .verifyResult(lastReplId);
- verifyInc2NonAcidLoad(dbName);
- verifyInc2AcidLoad(dbName);
- }
-
- @Test
- public void testAcidTablesBootstrapDuringIncremental() throws Throwable {
- // Take a bootstrap dump without acid tables
- WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName, null,
- dumpWithoutAcidClause);
- LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
- replica.load(replicatedDbName, bootstrapDump.dumpLocation);
- verifyLoadExecution(replicatedDbName, bootstrapDump.lastReplicationId, false);
-
- // Take a incremental dump with acid table bootstrap
- prepareIncAcidData(primaryDbName);
- prepareIncNonAcidData(primaryDbName);
- LOG.info(testName.getMethodName() + ": incremental dump and load dump with acid table bootstrap.");
- WarehouseInstance.Tuple incrementalDump = primary.run("use " + primaryDbName)
- .dump(primaryDbName, bootstrapDump.lastReplicationId, dumpWithAcidBootstrapClause);
- replica.load(replicatedDbName, incrementalDump.dumpLocation);
- verifyIncLoad(replicatedDbName, incrementalDump.lastReplicationId);
- // Ckpt should be set on bootstrapped tables.
- replica.verifyIfCkptSetForTables(replicatedDbName, acidTableNames, incrementalDump.dumpLocation);
-
- // Take a second normal incremental dump after Acid table boostrap
- prepareInc2AcidData(primaryDbName, primary.hiveConf);
- prepareInc2NonAcidData(primaryDbName, primary.hiveConf);
- LOG.info(testName.getMethodName()
- + ": second incremental dump and load dump after incremental with acid table " +
- "bootstrap.");
- WarehouseInstance.Tuple inc2Dump = primary.run("use " + primaryDbName)
- .dump(primaryDbName, incrementalDump.lastReplicationId);
- replica.load(replicatedDbName, inc2Dump.dumpLocation);
- verifyInc2Load(replicatedDbName, inc2Dump.lastReplicationId);
- }
-
- @Test
- public void testRetryAcidTablesBootstrapFromDifferentDump() throws Throwable {
- WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName, null,
- dumpWithoutAcidClause);
- LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
- replica.load(replicatedDbName, bootstrapDump.dumpLocation);
- verifyLoadExecution(replicatedDbName, bootstrapDump.lastReplicationId, false);
-
- prepareIncAcidData(primaryDbName);
- prepareIncNonAcidData(primaryDbName);
- LOG.info(testName.getMethodName() + ": first incremental dump with acid table bootstrap.");
- WarehouseInstance.Tuple incDump = primary.run("use " + primaryDbName)
- .dump(primaryDbName, bootstrapDump.lastReplicationId, dumpWithAcidBootstrapClause);
-
- // Fail setting ckpt property for table t5 but success for earlier tables
- BehaviourInjection<CallerArguments, Boolean> callerVerifier
- = new BehaviourInjection<CallerArguments, Boolean>() {
- @Nullable
- @Override
- public Boolean apply(@Nullable CallerArguments args) {
- if (args.tblName.equalsIgnoreCase("t5") && args.dbName.equalsIgnoreCase(replicatedDbName)) {
- injectionPathCalled = true;
- LOG.warn("Verifier - DB : " + args.dbName + " TABLE : " + args.tblName);
- return false;
- }
- return true;
- }
- };
-
- // Fail repl load before the ckpt property is set for t4 and after it is set for t2.
- // In the retry, these half baked tables should be dropped and bootstrap should be successful.
- InjectableBehaviourObjectStore.setAlterTableModifier(callerVerifier);
- try {
- LOG.info(testName.getMethodName()
- + ": loading first incremental dump with acid table bootstrap (will fail)");
- replica.loadFailure(replicatedDbName, incDump.dumpLocation);
- callerVerifier.assertInjectionsPerformed(true, false);
- } finally {
- InjectableBehaviourObjectStore.resetAlterTableModifier();
- }
-
- prepareInc2AcidData(primaryDbName, primary.hiveConf);
- prepareInc2NonAcidData(primaryDbName, primary.hiveConf);
- LOG.info(testName.getMethodName() + ": second incremental dump with acid table bootstrap");
- WarehouseInstance.Tuple inc2Dump = primary.run("use " + primaryDbName)
- .dump(primaryDbName, bootstrapDump.lastReplicationId, dumpWithAcidBootstrapClause);
-
- // Set incorrect bootstrap dump to clean tables. Here, used the full bootstrap dump which is invalid.
- // So, REPL LOAD fails.
- List<String> loadWithClause = Collections.singletonList(
- "'" + ReplUtils.REPL_CLEAN_TABLES_FROM_BOOTSTRAP_CONFIG + "'='"
- + bootstrapDump.dumpLocation + "'");
- LOG.info(testName.getMethodName()
- + ": trying to load second incremental dump with wrong bootstrap dump "
- + " specified for cleaning ACID tables. Should fail.");
- replica.loadFailure(replicatedDbName, inc2Dump.dumpLocation, loadWithClause);
-
- // Set previously failed bootstrap dump to clean-up. Now, new bootstrap should overwrite the old one.
- loadWithClause = Collections.singletonList(
- "'" + ReplUtils.REPL_CLEAN_TABLES_FROM_BOOTSTRAP_CONFIG + "'='"
- + incDump.dumpLocation + "'");
-
- LOG.info(testName.getMethodName()
- + ": trying to load second incremental dump with correct bootstrap dump "
- + "specified for cleaning ACID tables. Should succeed.");
- replica.load(replicatedDbName, inc2Dump.dumpLocation, loadWithClause);
- verifyInc2Load(replicatedDbName, inc2Dump.lastReplicationId);
-
- // Once the REPL LOAD is successful, the this config should be unset or else, the subsequent REPL LOAD
- // will also drop those tables which will cause data loss.
- loadWithClause = Collections.emptyList();
-
- // Verify if bootstrapping with same dump is idempotent and return same result
- LOG.info(testName.getMethodName()
- + ": trying to load second incremental dump (with acid bootstrap) again."
- + " Should succeed.");
- replica.load(replicatedDbName, inc2Dump.dumpLocation, loadWithClause);
- verifyInc2Load(replicatedDbName, inc2Dump.lastReplicationId);
- }
-
- @Test
- public void retryIncBootstrapAcidFromDifferentDumpWithoutCleanTablesConfig() throws Throwable {
- WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName, null,
- dumpWithoutAcidClause);
- replica.load(replicatedDbName, bootstrapDump.dumpLocation);
-
- prepareIncAcidData(primaryDbName);
- prepareIncNonAcidData(primaryDbName);
- WarehouseInstance.Tuple incDump = primary.run("use " + primaryDbName)
- .dump(primaryDbName, bootstrapDump.lastReplicationId, dumpWithAcidBootstrapClause);
- WarehouseInstance.Tuple inc2Dump = primary.run("use " + primaryDbName)
- .dump(primaryDbName, bootstrapDump.lastReplicationId, dumpWithAcidBootstrapClause);
- replica.load(replicatedDbName, incDump.dumpLocation);
-
- // Re-bootstrapping from different bootstrap dump without clean tables config should fail.
- replica.loadFailure(replicatedDbName, inc2Dump.dumpLocation, Collections.emptyList(),
- ErrorMsg.REPL_BOOTSTRAP_LOAD_PATH_NOT_VALID.getErrorCode());
- }
-
@Test
public void testAcidTablesMoveOptimizationIncremental() throws Throwable {
WarehouseInstance.Tuple bootstrapDump = primary.dump(primaryDbName, null);
@@ -531,76 +160,6 @@ public class TestReplicationScenariosAcidTables {
verifyLoadExecution(replicatedDbName, incrDump.lastReplicationId, true);
}
- private List<Long> openTxns(int numTxns, TxnStore txnHandler, HiveConf primaryConf) throws Throwable {
- OpenTxnsResponse otResp = txnHandler.openTxns(new OpenTxnRequest(numTxns, "u1", "localhost"));
- List<Long> txns = otResp.getTxn_ids();
- String txnIdRange = " txn_id >= " + txns.get(0) + " and txn_id <= " + txns.get(numTxns - 1);
- Assert.assertEquals(TxnDbUtil.queryToString(primaryConf, "select * from TXNS"),
- numTxns, TxnDbUtil.countQueryAgent(primaryConf,
- "select count(*) from TXNS where txn_state = 'o' and " + txnIdRange));
- return txns;
- }
-
- private void allocateWriteIdsForTables(String primaryDbName, Map<String, Long> tables,
- TxnStore txnHandler,
- List<Long> txns, HiveConf primaryConf) throws Throwable {
- AllocateTableWriteIdsRequest rqst = new AllocateTableWriteIdsRequest();
- rqst.setDbName(primaryDbName);
-
- for(Map.Entry<String, Long> entry : tables.entrySet()) {
- rqst.setTableName(entry.getKey());
- rqst.setTxnIds(txns);
- txnHandler.allocateTableWriteIds(rqst);
- }
- verifyWriteIdsForTables(tables, primaryConf, primaryDbName);
- }
-
- private void verifyWriteIdsForTables(Map<String, Long> tables, HiveConf conf, String dbName)
- throws Throwable {
- for(Map.Entry<String, Long> entry : tables.entrySet()) {
- Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_TO_WRITE_ID"),
- entry.getValue().longValue(),
- TxnDbUtil.countQueryAgent(conf,
- "select count(*) from TXN_TO_WRITE_ID where t2w_database = '"
- + dbName.toLowerCase()
- + "' and t2w_table = '" + entry.getKey() + "'"));
- }
- }
-
- private void verifyAllOpenTxnsAborted(List<Long> txns, HiveConf primaryConf) throws Throwable {
- int numTxns = txns.size();
- String txnIdRange = " txn_id >= " + txns.get(0) + " and txn_id <= " + txns.get(numTxns - 1);
- Assert.assertEquals(TxnDbUtil.queryToString(primaryConf, "select * from TXNS"),
- 0, TxnDbUtil.countQueryAgent(primaryConf,
- "select count(*) from TXNS where txn_state = 'o' and " + txnIdRange));
- Assert.assertEquals(TxnDbUtil.queryToString(primaryConf, "select * from TXNS"),
- numTxns, TxnDbUtil.countQueryAgent(primaryConf,
- "select count(*) from TXNS where txn_state = 'a' and " + txnIdRange));
- }
-
- private void verifyNextId(Map<String, Long> tables, String dbName, HiveConf conf) throws Throwable {
- // Verify the next write id
- for(Map.Entry<String, Long> entry : tables.entrySet()) {
- String[] nextWriteId =
- TxnDbUtil.queryToString(conf,
- "select nwi_next from NEXT_WRITE_ID where nwi_database = '"
- + dbName.toLowerCase() + "' and nwi_table = '"
- + entry.getKey() + "'").split("\n");
- Assert.assertEquals(Long.parseLong(nextWriteId[1].trim()), entry.getValue() + 1);
- }
- }
-
- private void verifyCompactionQueue(Map<String, Long> tables, String dbName, HiveConf conf)
- throws Throwable {
- for(Map.Entry<String, Long> entry : tables.entrySet()) {
- Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"),
- entry.getValue().longValue(),
- TxnDbUtil.countQueryAgent(conf,
- "select count(*) from COMPACTION_QUEUE where cq_database = '" + dbName
- + "' and cq_table = '" + entry.getKey() + "'"));
- }
- }
-
@Test
public void testAcidTablesBootstrapWithOpenTxnsTimeout() throws Throwable {
int numTxns = 5;
@@ -668,68 +227,6 @@ public class TestReplicationScenariosAcidTables {
}
@Test
- public void testAcidTablesBootstrapDuringIncrementalWithOpenTxnsTimeout() throws Throwable {
- // Take a dump without ACID tables
- WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName, null,
- dumpWithoutAcidClause);
- LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
- replica.load(replicatedDbName, bootstrapDump.dumpLocation);
-
- // Open concurrent transactions, create data for incremental and take an incremental dump
- // with ACID table bootstrap.
- int numTxns = 5;
- HiveConf primaryConf = primary.getConf();
- TxnStore txnHandler = TxnUtils.getTxnStore(primary.getConf());
- // Open 5 txns
- List<Long> txns = openTxns(numTxns, txnHandler, primaryConf);
- prepareIncNonAcidData(primaryDbName);
- prepareIncAcidData(primaryDbName);
- // Allocate write ids for tables t1 and t2 for all txns
- // t1=5+2(insert) and t2=5+5(insert, alter add column)
- Map<String, Long> tables = new HashMap<>();
- tables.put("t1", numTxns+2L);
- tables.put("t2", numTxns+5L);
- allocateWriteIdsForTables(primaryDbName, tables, txnHandler, txns, primaryConf);
-
- // Bootstrap dump with open txn timeout as 1s.
- List<String> withConfigs = new LinkedList<>(dumpWithAcidBootstrapClause);
- withConfigs.add("'hive.repl.bootstrap.dump.open.txn.timeout'='1s'");
- WarehouseInstance.Tuple incDump = primary
- .run("use " + primaryDbName)
- .dump(primaryDbName, bootstrapDump.lastReplicationId, withConfigs);
-
- // After bootstrap dump, all the opened txns should be aborted. Verify it.
- verifyAllOpenTxnsAborted(txns, primaryConf);
- verifyNextId(tables, primaryDbName, primaryConf);
-
- // Incremental load with ACID bootstrap should also replicate the aborted write ids on
- // tables t1 and t2
- HiveConf replicaConf = replica.getConf();
- LOG.info(testName.getMethodName() + ": loading incremental dump with ACID bootstrap.");
- replica.load(replicatedDbName, incDump.dumpLocation);
- // During incremental dump with ACID bootstrap we do not dump ALLOC_WRITE_ID events. So the
- // two ALLOC_WRITE_ID events corresponding aborted transactions on t1 and t2 will not be
- // repliaced. Discount those.
- verifyIncLoad(replicatedDbName,
- (new Long(Long.valueOf(incDump.lastReplicationId) - 2)).toString());
- // Verify if HWM is properly set after REPL LOAD
- verifyNextId(tables, replicatedDbName, replicaConf);
-
- // Verify if all the aborted write ids are replicated to the replicated DB
- for(Map.Entry<String, Long> entry : tables.entrySet()) {
- entry.setValue((long) numTxns);
- }
- verifyWriteIdsForTables(tables, replicaConf, replicatedDbName);
-
- // Verify if entries added in COMPACTION_QUEUE for each table/partition
- // t1-> 1 entry and t2-> 2 entries (1 per partition)
- tables.clear();
- tables.put("t1", 1L);
- tables.put("t2", 4L);
- verifyCompactionQueue(tables, replicatedDbName, replicaConf);
- }
-
- @Test
public void testAcidTablesBootstrapWithConcurrentWrites() throws Throwable {
HiveConf primaryConf = primary.getConf();
primary.run("use " + primaryDbName)
@@ -801,83 +298,6 @@ public class TestReplicationScenariosAcidTables {
}
@Test
- public void testBootstrapAcidTablesDuringIncrementalWithConcurrentWrites() throws Throwable {
- // Dump and load bootstrap without ACID tables.
- WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName, null,
- dumpWithoutAcidClause);
- LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
- replica.load(replicatedDbName, bootstrapDump.dumpLocation);
-
- // Create incremental data for incremental load with bootstrap of ACID
- prepareIncNonAcidData(primaryDbName);
- prepareIncAcidData(primaryDbName);
- // Perform concurrent writes. Bootstrap won't see the written data but the subsequent
- // incremental repl should see it. We can not inject callerVerifier since an incremental dump
- // would not cause an ALTER DATABASE event. Instead we piggy back on
- // getCurrentNotificationEventId() which is anyway required for a bootstrap.
- BehaviourInjection<CurrentNotificationEventId, CurrentNotificationEventId> callerInjectedBehavior
- = new BehaviourInjection<CurrentNotificationEventId, CurrentNotificationEventId>() {
- @Nullable
- @Override
- public CurrentNotificationEventId apply(@Nullable CurrentNotificationEventId input) {
- if (injectionPathCalled) {
- nonInjectedPathCalled = true;
- } else {
- // Do some writes through concurrent thread
- injectionPathCalled = true;
- Thread t = new Thread(new Runnable() {
- @Override
- public void run() {
- LOG.info("Entered new thread");
- try {
- prepareInc2NonAcidData(primaryDbName, primary.hiveConf);
- prepareInc2AcidData(primaryDbName, primary.hiveConf);
- } catch (Throwable t) {
- Assert.assertNull(t);
- }
- LOG.info("Exit new thread success");
- }
- });
- t.start();
- LOG.info("Created new thread {}", t.getName());
- try {
- t.join();
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- }
- }
- return input;
- }
- };
-
- InjectableBehaviourObjectStore.setGetCurrentNotificationEventIdBehaviour(callerInjectedBehavior);
- WarehouseInstance.Tuple incDump = null;
- try {
- incDump = primary.dump(primaryDbName, bootstrapDump.lastReplicationId, dumpWithAcidBootstrapClause);
- callerInjectedBehavior.assertInjectionsPerformed(true, true);
- } finally {
- // reset the behaviour
- InjectableBehaviourObjectStore.resetGetCurrentNotificationEventIdBehaviour();
- }
-
- // While bootstrapping ACID tables it has taken snapshot before concurrent thread performed
- // write. So concurrent writes won't be dumped.
- LOG.info(testName.getMethodName() +
- ": loading incremental dump containing bootstrapped ACID tables.");
- replica.load(replicatedDbName, incDump.dumpLocation);
- verifyIncLoad(replicatedDbName, incDump.lastReplicationId);
-
- // Next Incremental should include the concurrent writes
- LOG.info(testName.getMethodName() +
- ": dumping second normal incremental dump from event id = " + incDump.lastReplicationId);
- WarehouseInstance.Tuple inc2Dump = primary.dump(primaryDbName, incDump.lastReplicationId);
- LOG.info(testName.getMethodName() +
- ": loading second normal incremental dump from event id = " + incDump.lastReplicationId);
- replica.load(replicatedDbName, inc2Dump.dumpLocation);
- verifyInc2Load(replicatedDbName, inc2Dump.lastReplicationId);
- }
-
- @Test
public void testAcidTablesBootstrapWithConcurrentDropTable() throws Throwable {
HiveConf primaryConf = primary.getConf();
primary.run("use " + primaryDbName)
diff --git a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcidTablesBootstrap.java b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcidTablesBootstrap.java
new file mode 100644
index 0000000..f475b1e
--- /dev/null
+++ b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcidTablesBootstrap.java
@@ -0,0 +1,334 @@
+/*
+ * 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.parse;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
+import org.apache.hadoop.hive.metastore.messaging.json.gzip.GzipJSONMessageEncoder;
+import org.apache.hadoop.hive.metastore.txn.TxnStore;
+import org.apache.hadoop.hive.metastore.txn.TxnUtils;
+import org.apache.hadoop.hive.metastore.InjectableBehaviourObjectStore;
+import org.apache.hadoop.hive.metastore.InjectableBehaviourObjectStore.CallerArguments;
+import org.apache.hadoop.hive.metastore.InjectableBehaviourObjectStore.BehaviourInjection;
+import org.apache.hadoop.hive.ql.ErrorMsg;
+import org.apache.hadoop.hive.ql.exec.repl.util.ReplUtils;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.BeforeClass;
+
+import javax.annotation.Nullable;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * TestReplicationScenariosAcidTables - test bootstrap of ACID tables during an incremental.
+ */
+public class TestReplicationScenariosAcidTablesBootstrap
+ extends BaseReplicationScenariosAcidTables {
+
+ private static List<String> dumpWithoutAcidClause = Collections.singletonList(
+ "'" + ReplUtils.REPL_DUMP_INCLUDE_ACID_TABLES + "'='false'");
+ private static List<String> dumpWithAcidBootstrapClause = Arrays.asList(
+ "'" + ReplUtils.REPL_DUMP_INCLUDE_ACID_TABLES + "'='true'",
+ "'" + HiveConf.ConfVars.REPL_BOOTSTRAP_ACID_TABLES + "'='true'");
+
+ @BeforeClass
+ public static void classLevelSetup() throws Exception {
+ Map<String, String> overrides = new HashMap<>();
+ overrides.put(MetastoreConf.ConfVars.EVENT_MESSAGE_FACTORY.getHiveName(),
+ GzipJSONMessageEncoder.class.getCanonicalName());
+
+ internalBeforeClassSetup(overrides, TestReplicationScenariosAcidTablesBootstrap.class);
+ }
+
+ @Test
+ public void testAcidTablesBootstrapDuringIncremental() throws Throwable {
+ // Take a bootstrap dump without acid tables
+ WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName, null,
+ dumpWithoutAcidClause);
+ LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
+ replica.load(replicatedDbName, bootstrapDump.dumpLocation);
+ verifyLoadExecution(replicatedDbName, bootstrapDump.lastReplicationId, false);
+
+ // Take a incremental dump with acid table bootstrap
+ prepareIncAcidData(primaryDbName);
+ prepareIncNonAcidData(primaryDbName);
+ LOG.info(testName.getMethodName() + ": incremental dump and load dump with acid table bootstrap.");
+ WarehouseInstance.Tuple incrementalDump = primary.run("use " + primaryDbName)
+ .dump(primaryDbName, bootstrapDump.lastReplicationId, dumpWithAcidBootstrapClause);
+ replica.load(replicatedDbName, incrementalDump.dumpLocation);
+ verifyIncLoad(replicatedDbName, incrementalDump.lastReplicationId);
+ // Ckpt should be set on bootstrapped tables.
+ replica.verifyIfCkptSetForTables(replicatedDbName, acidTableNames, incrementalDump.dumpLocation);
+
+ // Take a second normal incremental dump after Acid table boostrap
+ prepareInc2AcidData(primaryDbName, primary.hiveConf);
+ prepareInc2NonAcidData(primaryDbName, primary.hiveConf);
+ LOG.info(testName.getMethodName()
+ + ": second incremental dump and load dump after incremental with acid table " +
+ "bootstrap.");
+ WarehouseInstance.Tuple inc2Dump = primary.run("use " + primaryDbName)
+ .dump(primaryDbName, incrementalDump.lastReplicationId);
+ replica.load(replicatedDbName, inc2Dump.dumpLocation);
+ verifyInc2Load(replicatedDbName, inc2Dump.lastReplicationId);
+ }
+
+ @Test
+ public void testRetryAcidTablesBootstrapFromDifferentDump() throws Throwable {
+ WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName, null,
+ dumpWithoutAcidClause);
+ LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
+ replica.load(replicatedDbName, bootstrapDump.dumpLocation);
+ verifyLoadExecution(replicatedDbName, bootstrapDump.lastReplicationId, false);
+
+ prepareIncAcidData(primaryDbName);
+ prepareIncNonAcidData(primaryDbName);
+ LOG.info(testName.getMethodName() + ": first incremental dump with acid table bootstrap.");
+ WarehouseInstance.Tuple incDump = primary.run("use " + primaryDbName)
+ .dump(primaryDbName, bootstrapDump.lastReplicationId, dumpWithAcidBootstrapClause);
+
+ // Fail setting ckpt property for table t5 but success for earlier tables
+ BehaviourInjection<CallerArguments, Boolean> callerVerifier
+ = new BehaviourInjection<CallerArguments, Boolean>() {
+ @Nullable
+ @Override
+ public Boolean apply(@Nullable CallerArguments args) {
+ if (args.tblName.equalsIgnoreCase("t5") && args.dbName.equalsIgnoreCase(replicatedDbName)) {
+ injectionPathCalled = true;
+ LOG.warn("Verifier - DB : " + args.dbName + " TABLE : " + args.tblName);
+ return false;
+ }
+ return true;
+ }
+ };
+
+ // Fail repl load before the ckpt property is set for t4 and after it is set for t2.
+ // In the retry, these half baked tables should be dropped and bootstrap should be successful.
+ InjectableBehaviourObjectStore.setAlterTableModifier(callerVerifier);
+ try {
+ LOG.info(testName.getMethodName()
+ + ": loading first incremental dump with acid table bootstrap (will fail)");
+ replica.loadFailure(replicatedDbName, incDump.dumpLocation);
+ callerVerifier.assertInjectionsPerformed(true, false);
+ } finally {
+ InjectableBehaviourObjectStore.resetAlterTableModifier();
+ }
+
+ prepareInc2AcidData(primaryDbName, primary.hiveConf);
+ prepareInc2NonAcidData(primaryDbName, primary.hiveConf);
+ LOG.info(testName.getMethodName() + ": second incremental dump with acid table bootstrap");
+ WarehouseInstance.Tuple inc2Dump = primary.run("use " + primaryDbName)
+ .dump(primaryDbName, bootstrapDump.lastReplicationId, dumpWithAcidBootstrapClause);
+
+ // Set incorrect bootstrap dump to clean tables. Here, used the full bootstrap dump which is invalid.
+ // So, REPL LOAD fails.
+ List<String> loadWithClause = Collections.singletonList(
+ "'" + ReplUtils.REPL_CLEAN_TABLES_FROM_BOOTSTRAP_CONFIG + "'='"
+ + bootstrapDump.dumpLocation + "'");
+ LOG.info(testName.getMethodName()
+ + ": trying to load second incremental dump with wrong bootstrap dump "
+ + " specified for cleaning ACID tables. Should fail.");
+ replica.loadFailure(replicatedDbName, inc2Dump.dumpLocation, loadWithClause);
+
+ // Set previously failed bootstrap dump to clean-up. Now, new bootstrap should overwrite the old one.
+ loadWithClause = Collections.singletonList(
+ "'" + ReplUtils.REPL_CLEAN_TABLES_FROM_BOOTSTRAP_CONFIG + "'='"
+ + incDump.dumpLocation + "'");
+
+ LOG.info(testName.getMethodName()
+ + ": trying to load second incremental dump with correct bootstrap dump "
+ + "specified for cleaning ACID tables. Should succeed.");
+ replica.load(replicatedDbName, inc2Dump.dumpLocation, loadWithClause);
+ verifyInc2Load(replicatedDbName, inc2Dump.lastReplicationId);
+
+ // Once the REPL LOAD is successful, the this config should be unset or else, the subsequent REPL LOAD
+ // will also drop those tables which will cause data loss.
+ loadWithClause = Collections.emptyList();
+
+ // Verify if bootstrapping with same dump is idempotent and return same result
+ LOG.info(testName.getMethodName()
+ + ": trying to load second incremental dump (with acid bootstrap) again."
+ + " Should succeed.");
+ replica.load(replicatedDbName, inc2Dump.dumpLocation, loadWithClause);
+ verifyInc2Load(replicatedDbName, inc2Dump.lastReplicationId);
+ }
+
+ @Test
+ public void retryIncBootstrapAcidFromDifferentDumpWithoutCleanTablesConfig() throws Throwable {
+ WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName, null,
+ dumpWithoutAcidClause);
+ replica.load(replicatedDbName, bootstrapDump.dumpLocation);
+
+ prepareIncAcidData(primaryDbName);
+ prepareIncNonAcidData(primaryDbName);
+ WarehouseInstance.Tuple incDump = primary.run("use " + primaryDbName)
+ .dump(primaryDbName, bootstrapDump.lastReplicationId, dumpWithAcidBootstrapClause);
+ WarehouseInstance.Tuple inc2Dump = primary.run("use " + primaryDbName)
+ .dump(primaryDbName, bootstrapDump.lastReplicationId, dumpWithAcidBootstrapClause);
+ replica.load(replicatedDbName, incDump.dumpLocation);
+
+ // Re-bootstrapping from different bootstrap dump without clean tables config should fail.
+ replica.loadFailure(replicatedDbName, inc2Dump.dumpLocation, Collections.emptyList(),
+ ErrorMsg.REPL_BOOTSTRAP_LOAD_PATH_NOT_VALID.getErrorCode());
+ }
+
+ @Test
+ public void testAcidTablesBootstrapDuringIncrementalWithOpenTxnsTimeout() throws Throwable {
+ // Take a dump without ACID tables
+ WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName, null,
+ dumpWithoutAcidClause);
+ LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
+ replica.load(replicatedDbName, bootstrapDump.dumpLocation);
+
+ // Open concurrent transactions, create data for incremental and take an incremental dump
+ // with ACID table bootstrap.
+ int numTxns = 5;
+ HiveConf primaryConf = primary.getConf();
+ TxnStore txnHandler = TxnUtils.getTxnStore(primary.getConf());
+ // Open 5 txns
+ List<Long> txns = openTxns(numTxns, txnHandler, primaryConf);
+ prepareIncNonAcidData(primaryDbName);
+ prepareIncAcidData(primaryDbName);
+ // Allocate write ids for tables t1 and t2 for all txns
+ // t1=5+2(insert) and t2=5+5(insert, alter add column)
+ Map<String, Long> tables = new HashMap<>();
+ tables.put("t1", numTxns+2L);
+ tables.put("t2", numTxns+5L);
+ allocateWriteIdsForTables(primaryDbName, tables, txnHandler, txns, primaryConf);
+
+ // Bootstrap dump with open txn timeout as 1s.
+ List<String> withConfigs = new LinkedList<>(dumpWithAcidBootstrapClause);
+ withConfigs.add("'hive.repl.bootstrap.dump.open.txn.timeout'='1s'");
+ WarehouseInstance.Tuple incDump = primary
+ .run("use " + primaryDbName)
+ .dump(primaryDbName, bootstrapDump.lastReplicationId, withConfigs);
+
+ // After bootstrap dump, all the opened txns should be aborted. Verify it.
+ verifyAllOpenTxnsAborted(txns, primaryConf);
+ verifyNextId(tables, primaryDbName, primaryConf);
+
+ // Incremental load with ACID bootstrap should also replicate the aborted write ids on
+ // tables t1 and t2
+ HiveConf replicaConf = replica.getConf();
+ LOG.info(testName.getMethodName() + ": loading incremental dump with ACID bootstrap.");
+ replica.load(replicatedDbName, incDump.dumpLocation);
+ // During incremental dump with ACID bootstrap we do not dump ALLOC_WRITE_ID events. So the
+ // two ALLOC_WRITE_ID events corresponding aborted transactions on t1 and t2 will not be
+ // repliaced. Discount those.
+ verifyIncLoad(replicatedDbName,
+ (new Long(Long.valueOf(incDump.lastReplicationId) - 2)).toString());
+ // Verify if HWM is properly set after REPL LOAD
+ verifyNextId(tables, replicatedDbName, replicaConf);
+
+ // Verify if all the aborted write ids are replicated to the replicated DB
+ for(Map.Entry<String, Long> entry : tables.entrySet()) {
+ entry.setValue((long) numTxns);
+ }
+ verifyWriteIdsForTables(tables, replicaConf, replicatedDbName);
+
+ // Verify if entries added in COMPACTION_QUEUE for each table/partition
+ // t1-> 1 entry and t2-> 2 entries (1 per partition)
+ tables.clear();
+ tables.put("t1", 1L);
+ tables.put("t2", 4L);
+ verifyCompactionQueue(tables, replicatedDbName, replicaConf);
+ }
+
+ @Test
+ public void testBootstrapAcidTablesDuringIncrementalWithConcurrentWrites() throws Throwable {
+ // Dump and load bootstrap without ACID tables.
+ WarehouseInstance.Tuple bootstrapDump = prepareDataAndDump(primaryDbName, null,
+ dumpWithoutAcidClause);
+ LOG.info(testName.getMethodName() + ": loading dump without acid tables.");
+ replica.load(replicatedDbName, bootstrapDump.dumpLocation);
+
+ // Create incremental data for incremental load with bootstrap of ACID
+ prepareIncNonAcidData(primaryDbName);
+ prepareIncAcidData(primaryDbName);
+ // Perform concurrent writes. Bootstrap won't see the written data but the subsequent
+ // incremental repl should see it. We can not inject callerVerifier since an incremental dump
+ // would not cause an ALTER DATABASE event. Instead we piggy back on
+ // getCurrentNotificationEventId() which is anyway required for a bootstrap.
+ BehaviourInjection<CurrentNotificationEventId, CurrentNotificationEventId> callerInjectedBehavior
+ = new BehaviourInjection<CurrentNotificationEventId, CurrentNotificationEventId>() {
+ @Nullable
+ @Override
+ public CurrentNotificationEventId apply(@Nullable CurrentNotificationEventId input) {
+ if (injectionPathCalled) {
+ nonInjectedPathCalled = true;
+ } else {
+ // Do some writes through concurrent thread
+ injectionPathCalled = true;
+ Thread t = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ LOG.info("Entered new thread");
+ try {
+ prepareInc2NonAcidData(primaryDbName, primary.hiveConf);
+ prepareInc2AcidData(primaryDbName, primary.hiveConf);
+ } catch (Throwable t) {
+ Assert.assertNull(t);
+ }
+ LOG.info("Exit new thread success");
+ }
+ });
+ t.start();
+ LOG.info("Created new thread {}", t.getName());
+ try {
+ t.join();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ return input;
+ }
+ };
+
+ InjectableBehaviourObjectStore.setGetCurrentNotificationEventIdBehaviour(callerInjectedBehavior);
+ WarehouseInstance.Tuple incDump = null;
+ try {
+ incDump = primary.dump(primaryDbName, bootstrapDump.lastReplicationId, dumpWithAcidBootstrapClause);
+ callerInjectedBehavior.assertInjectionsPerformed(true, true);
+ } finally {
+ // reset the behaviour
+ InjectableBehaviourObjectStore.resetGetCurrentNotificationEventIdBehaviour();
+ }
+
+ // While bootstrapping ACID tables it has taken snapshot before concurrent thread performed
+ // write. So concurrent writes won't be dumped.
+ LOG.info(testName.getMethodName() +
+ ": loading incremental dump containing bootstrapped ACID tables.");
+ replica.load(replicatedDbName, incDump.dumpLocation);
+ verifyIncLoad(replicatedDbName, incDump.lastReplicationId);
+
+ // Next Incremental should include the concurrent writes
+ LOG.info(testName.getMethodName() +
+ ": dumping second normal incremental dump from event id = " + incDump.lastReplicationId);
+ WarehouseInstance.Tuple inc2Dump = primary.dump(primaryDbName, incDump.lastReplicationId);
+ LOG.info(testName.getMethodName() +
+ ": loading second normal incremental dump from event id = " + incDump.lastReplicationId);
+ replica.load(replicatedDbName, inc2Dump.dumpLocation);
+ verifyInc2Load(replicatedDbName, inc2Dump.lastReplicationId);
+ }
+}
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/ddl/table/CreateTableOperation.java b/ql/src/java/org/apache/hadoop/hive/ql/ddl/table/CreateTableOperation.java
index 7da3d26..2987cab 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/ddl/table/CreateTableOperation.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/ddl/table/CreateTableOperation.java
@@ -33,10 +33,15 @@ import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.exec.repl.util.ReplUtils;
import org.apache.hadoop.hive.ql.hooks.WriteEntity;
import org.apache.hadoop.hive.ql.hooks.LineageInfo.DataContainer;
+import org.apache.hadoop.hive.ql.io.AcidUtils;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.parse.ReplicationSpec;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
/**
* Operation process of creating a table.
*/
@@ -140,6 +145,27 @@ public class CreateTableOperation extends DDLOperation {
Table createdTable = context.getDb().getTable(tbl.getDbName(), tbl.getTableName());
DataContainer dc = new DataContainer(createdTable.getTTable());
context.getQueryState().getLineageState().setLineage(createdTable.getPath(), dc, createdTable.getCols());
+
+ // We did not create the table before moving the data files for a non-partitioned table i.e
+ // we used load file instead of load table (see SemanticAnalyzer#getFileSinkPlan() for
+ // more details). Thus could not add a write notification required for a transactional
+ // table. Do that here, after we have created the table. Since this is a newly created
+ // table, listing all the files in the directory and listing only the ones corresponding to
+ // the given id doesn't have much difference.
+ if (!createdTable.isPartitioned() && AcidUtils.isTransactionalTable(createdTable)) {
+ org.apache.hadoop.hive.metastore.api.Table tTable = createdTable.getTTable();
+ Path tabLocation = new Path(tTable.getSd().getLocation());
+ List<Path> newFilesList = new ArrayList<>();
+ try {
+ context.getDb().listFilesInsideAcidDirectory(tabLocation,
+ tabLocation.getFileSystem(context.getConf()), newFilesList);
+ } catch (IOException e) {
+ LOG.error("Error listing files", e);
+ throw new HiveException(e);
+ }
+ context.getDb().addWriteNotificationLog(createdTable, null, newFilesList,
+ tTable.getWriteId());
+ }
}
}
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 420270a..111cd1d 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
@@ -2362,7 +2362,7 @@ public class Hive {
((null != oldPart) || AcidUtils.isTransactionalTable(tbl));
}
- private void listFilesInsideAcidDirectory(Path acidDir, FileSystem srcFs, List<Path> newFiles) throws IOException {
+ public void listFilesInsideAcidDirectory(Path acidDir, FileSystem srcFs, List<Path> newFiles) throws IOException {
// list out all the files/directory in the path
FileStatus[] acidFiles;
acidFiles = srcFs.listStatus(acidDir);
@@ -3261,7 +3261,7 @@ private void constructOneLBLocationMap(FileStatus fSta,
tpart.getSd().setLocation(partPath);
}
- private void addWriteNotificationLog(Table tbl, Map<String, String> partitionSpec,
+ public void addWriteNotificationLog(Table tbl, Map<String, String> partitionSpec,
List<Path> newFiles, Long writeId) throws HiveException {
if (!conf.getBoolVar(ConfVars.FIRE_EVENTS_FOR_DML)) {
LOG.debug("write notification log is ignored as dml event logging is disabled");
diff --git a/testutils/ptest2/conf/deployed/master-mr2.properties b/testutils/ptest2/conf/deployed/master-mr2.properties
index 66c23ab..45980be 100644
--- a/testutils/ptest2/conf/deployed/master-mr2.properties
+++ b/testutils/ptest2/conf/deployed/master-mr2.properties
@@ -76,7 +76,9 @@ ut.itests.hive-unit.skipBatching=TestAcidOnTezWithSplitUpdate TestAcidOnTez Test
TestStatsReplicationScenariosACIDNoAutogather TestStatsReplicationScenariosMMNoAutogather \
TestStatsReplicationScenariosACID TestStatsReplicationScenariosMM \
TestStatsReplicationScenariosMigrationNoAutogather TestStatsReplicationScenariosMigration \
- TestStatsReplicationScenariosNoAutogather TestStatsReplicationScenarios
+ TestStatsReplicationScenariosNoAutogather TestStatsReplicationScenarios \
+ TestReplicationScenariosAcidTables TestReplAcidTablesBootstrapWithJsonMessage \
+ TestReplicationScenariosAcidTablesBootstrap TestReplAcidTablesWithJsonMessage
unitTests.module.itests.qtest=itests.qtest
ut.itests.qtest.batchSize=9