You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by ek...@apache.org on 2018/05/07 19:34:23 UTC
[1/3] hive git commit: HIVE-18131 - Truncate table for Acid tables
(Eugene Koifman, reviewed by Prasanth Jayachandran)
Repository: hive
Updated Branches:
refs/heads/branch-3 f93f76d18 -> 63abf9be0
HIVE-18131 - Truncate table for Acid tables (Eugene Koifman, reviewed by Prasanth Jayachandran)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/a7d2fb80
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/a7d2fb80
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/a7d2fb80
Branch: refs/heads/branch-3
Commit: a7d2fb805ce8725525651c54804511ffe5dfbaee
Parents: f93f76d
Author: Eugene Koifman <ek...@apache.org>
Authored: Mon May 7 12:05:19 2018 -0700
Committer: Eugene Koifman <ek...@apache.org>
Committed: Mon May 7 12:05:19 2018 -0700
----------------------------------------------------------------------
.../hadoop/hive/ql/lockmgr/TestDbTxnManager2.java | 15 +++++++++++++++
1 file changed, 15 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/a7d2fb80/ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager2.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager2.java b/ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager2.java
index 1d79242..e06f0a4 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager2.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager2.java
@@ -2512,4 +2512,19 @@ public class TestDbTxnManager2 {
Assert.assertEquals("Unexpected lock count", 1, locks.size());
checkLock(LockType.EXCLUSIVE, LockState.ACQUIRED, "default", "T", null, locks);
}
+ @Test
+ public void testTruncate() throws Exception {
+ dropTable(new String[] {"T"});
+ CommandProcessorResponse cpr = driver.run("create table T (a int, b int) stored as" +
+ " orc tblproperties('transactional'='true')");
+ checkCmdOnDriver(cpr);
+ checkCmdOnDriver(driver.run("insert into T values(0,2),(1,4)"));
+ checkCmdOnDriver(driver.run("truncate table T"));
+ cpr = driver.compileAndRespond("truncate table T");
+ checkCmdOnDriver(cpr);
+ txnMgr.acquireLocks(driver.getPlan(), ctx, "Fifer");//gets X lock on T
+ List<ShowLocksResponseElement> locks = getLocks();
+ Assert.assertEquals("Unexpected lock count", 1, locks.size());
+ checkLock(LockType.EXCLUSIVE, LockState.ACQUIRED, "default", "T", null, locks);
+ }
}
[2/3] hive git commit: HIVE-18288 - merge/concat not supported on
Acid table (Eugene Koifman, reviewed by Sergey Shelukhin)
Posted by ek...@apache.org.
HIVE-18288 - merge/concat not supported on Acid table (Eugene Koifman, reviewed by Sergey Shelukhin)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/cb61ed63
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/cb61ed63
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/cb61ed63
Branch: refs/heads/branch-3
Commit: cb61ed631c28811fc169e3a33280edd1f41f2186
Parents: a7d2fb8
Author: Eugene Koifman <ek...@apache.org>
Authored: Mon May 7 12:06:59 2018 -0700
Committer: Eugene Koifman <ek...@apache.org>
Committed: Mon May 7 12:06:59 2018 -0700
----------------------------------------------------------------------
.../org/apache/hadoop/hive/conf/HiveConf.java | 3 +
.../apache/hadoop/hive/ql/metadata/Hive.java | 4 +-
.../hive/ql/parse/DDLSemanticAnalyzer.java | 21 ++-
.../hadoop/hive/ql/TestTxnConcatenate.java | 159 +++++++++++++++++++
4 files changed, 177 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/cb61ed63/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 4b8dc19..090c255 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -2565,6 +2565,9 @@ public class HiveConf extends Configuration {
COMPACTOR_JOB_QUEUE("hive.compactor.job.queue", "", "Used to specify name of Hadoop queue to which\n" +
"Compaction jobs will be submitted. Set to empty string to let Hadoop choose the queue."),
+ TRANSACTIONAL_CONCATENATE_NOBLOCK("hive.transactional.concatenate.noblock", false,
+ "Will cause 'alter table T concatenate' to be non-blocking"),
+
HIVE_COMPACTOR_COMPACT_MM("hive.compactor.compact.insert.only", true,
"Whether the compactor should compact insert-only tables. A safety switch."),
/**
http://git-wip-us.apache.org/repos/asf/hive/blob/cb61ed63/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 5ab86b4..3218f96 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
@@ -4405,9 +4405,9 @@ private void constructOneLBLocationMap(FileStatus fSta,
throws HiveException {
try {
CompactionType cr = null;
- if ("major".equals(compactType)) {
+ if ("major".equalsIgnoreCase(compactType)) {
cr = CompactionType.MAJOR;
- } else if ("minor".equals(compactType)) {
+ } else if ("minor".equalsIgnoreCase(compactType)) {
cr = CompactionType.MINOR;
} else {
throw new RuntimeException("Unknown compaction type " + compactType);
http://git-wip-us.apache.org/repos/asf/hive/blob/cb61ed63/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java
index defb8be..f0b9eda 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java
@@ -1964,9 +1964,19 @@ public class DDLSemanticAnalyzer extends BaseSemanticAnalyzer {
try {
tblObj = getTable(tableName);
- // TODO: we should probably block all ACID tables here.
- if (AcidUtils.isInsertOnlyTable(tblObj.getParameters())) {
- throw new SemanticException("Merge is not supported for MM tables");
+ if(AcidUtils.isTransactionalTable(tblObj)) {
+ LinkedHashMap<String, String> newPartSpec = null;
+ if (partSpec != null) {
+ newPartSpec = new LinkedHashMap<>(partSpec);
+ }
+
+ boolean isBlocking = !HiveConf.getBoolVar(conf,
+ ConfVars.TRANSACTIONAL_CONCATENATE_NOBLOCK, false);
+ AlterTableSimpleDesc desc = new AlterTableSimpleDesc(
+ tableName, newPartSpec, "MAJOR", isBlocking);
+
+ rootTasks.add(TaskFactory.get(new DDLWork(getInputs(), getOutputs(), desc)));
+ return;
}
mergeDesc.setTableDesc(Utilities.getTableDesc(tblObj));
@@ -2039,11 +2049,6 @@ public class DDLSemanticAnalyzer extends BaseSemanticAnalyzer {
throw new SemanticException(ErrorMsg.CONCATENATE_UNSUPPORTED_TABLE_NOT_MANAGED.getMsg());
}
- // transactional tables are compacted and no longer needs to be bucketed, so not safe for merge/concatenation
- boolean isAcid = AcidUtils.isTransactionalTable(tblObj);
- if (isAcid) {
- throw new SemanticException(ErrorMsg.CONCATENATE_UNSUPPORTED_TABLE_TRANSACTIONAL.getMsg());
- }
inputDir.add(oldTblPartLoc);
mergeDesc.setInputDir(inputDir);
http://git-wip-us.apache.org/repos/asf/hive/blob/cb61ed63/ql/src/test/org/apache/hadoop/hive/ql/TestTxnConcatenate.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnConcatenate.java b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnConcatenate.java
new file mode 100644
index 0000000..92bcefe
--- /dev/null
+++ b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnConcatenate.java
@@ -0,0 +1,159 @@
+package org.apache.hadoop.hive.ql;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.api.ShowCompactRequest;
+import org.apache.hadoop.hive.metastore.api.ShowCompactResponse;
+import org.apache.hadoop.hive.metastore.txn.TxnStore;
+import org.apache.hadoop.hive.metastore.txn.TxnUtils;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+
+public class TestTxnConcatenate extends TxnCommandsBaseForTests {
+ static final private Logger LOG = LoggerFactory.getLogger(TestTxnConcatenate.class);
+ private static final String TEST_DATA_DIR =
+ new File(System.getProperty("java.io.tmpdir") +
+ File.separator + TestTxnLoadData.class.getCanonicalName()
+ + "-" + System.currentTimeMillis()
+ ).getPath().replaceAll("\\\\", "/");
+ @Rule
+ public TemporaryFolder folder = new TemporaryFolder();
+
+ @Override
+ String getTestDataDir() {
+ return TEST_DATA_DIR;
+ }
+
+ @Test
+ public void testConcatenate() throws Exception {
+ runStatementOnDriver("insert into " + Table.ACIDTBL + " values(1,2),(4,5)");
+ runStatementOnDriver("update " + Table.ACIDTBL + " set b = 4");
+ runStatementOnDriver("insert into " + Table.ACIDTBL + " values(5,6),(8,8)");
+ String testQuery = "select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.ACIDTBL + " order by a, b";
+ String[][] expected = new String[][] {
+ {"{\"writeid\":2,\"bucketid\":536936448,\"rowid\":1}\t1\t4",
+ "acidtbl/delta_0000002_0000002_0000/bucket_00001"},
+ {"{\"writeid\":2,\"bucketid\":536936448,\"rowid\":0}\t4\t4",
+ "acidtbl/delta_0000002_0000002_0000/bucket_00001"},
+ {"{\"writeid\":3,\"bucketid\":536936448,\"rowid\":1}\t5\t6",
+ "acidtbl/delta_0000003_0000003_0000/bucket_00001"},
+ {"{\"writeid\":3,\"bucketid\":536936448,\"rowid\":0}\t8\t8",
+ "acidtbl/delta_0000003_0000003_0000/bucket_00001"}};
+ checkResult(expected, testQuery, false, "check data", LOG);
+
+ /*in UTs, there is no standalone HMS running to kick off compaction so it's done via runWorker()
+ but in normal usage 'concatenate' is blocking, */
+ hiveConf.setBoolVar(HiveConf.ConfVars.TRANSACTIONAL_CONCATENATE_NOBLOCK, true);
+ runStatementOnDriver("alter table " + Table.ACIDTBL + " concatenate");
+
+ TxnStore txnStore = TxnUtils.getTxnStore(hiveConf);
+ ShowCompactResponse rsp = txnStore.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompactsSize());
+ Assert.assertEquals(TxnStore.INITIATED_RESPONSE, rsp.getCompacts().get(0).getState());
+ runWorker(hiveConf);
+ rsp = txnStore.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompactsSize());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());
+ String[][] expected2 = new String[][] {
+ {"{\"writeid\":2,\"bucketid\":536936448,\"rowid\":1}\t1\t4",
+ "acidtbl/base_0000003/bucket_00001"},
+ {"{\"writeid\":2,\"bucketid\":536936448,\"rowid\":0}\t4\t4",
+ "acidtbl/base_0000003/bucket_00001"},
+ {"{\"writeid\":3,\"bucketid\":536936448,\"rowid\":1}\t5\t6",
+ "acidtbl/base_0000003/bucket_00001"},
+ {"{\"writeid\":3,\"bucketid\":536936448,\"rowid\":0}\t8\t8",
+ "acidtbl/base_0000003/bucket_00001"}};
+ checkResult(expected2, testQuery, false, "check data after concatenate", LOG);
+ }
+ @Test
+ public void testConcatenatePart() throws Exception {
+ runStatementOnDriver("insert into " + Table.ACIDTBLPART + " values(1,2,'p1'),(4,5,'p2')");
+ runStatementOnDriver("update " + Table.ACIDTBLPART + " set b = 4 where p='p1'");
+ runStatementOnDriver("insert into " + Table.ACIDTBLPART + " values(5,6,'p1'),(8,8,'p2')");
+ String testQuery = "select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.ACIDTBLPART + " order by a, b";
+ String[][] expected = new String[][] {
+ {"{\"writeid\":2,\"bucketid\":536936448,\"rowid\":0}\t1\t4",
+ "acidtblpart/p=p1/delta_0000002_0000002_0000/bucket_00001"},
+ {"{\"writeid\":1,\"bucketid\":536936448,\"rowid\":0}\t4\t5",
+ "acidtblpart/p=p2/delta_0000001_0000001_0000/bucket_00001"},
+ {"{\"writeid\":3,\"bucketid\":536936448,\"rowid\":0}\t5\t6",
+ "acidtblpart/p=p1/delta_0000003_0000003_0000/bucket_00001"},
+ {"{\"writeid\":3,\"bucketid\":536936448,\"rowid\":0}\t8\t8",
+ "acidtblpart/p=p2/delta_0000003_0000003_0000/bucket_00001"}};
+ checkResult(expected, testQuery, false, "check data", LOG);
+
+ /*in UTs, there is no standalone HMS running to kick off compaction so it's done via runWorker()
+ but in normal usage 'concatenate' is blocking, */
+ hiveConf.setBoolVar(HiveConf.ConfVars.TRANSACTIONAL_CONCATENATE_NOBLOCK, true);
+ runStatementOnDriver("alter table " + Table.ACIDTBLPART + " PARTITION(p='p1') concatenate");
+
+ TxnStore txnStore = TxnUtils.getTxnStore(hiveConf);
+ ShowCompactResponse rsp = txnStore.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompactsSize());
+ Assert.assertEquals(TxnStore.INITIATED_RESPONSE, rsp.getCompacts().get(0).getState());
+ runWorker(hiveConf);
+ rsp = txnStore.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompactsSize());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());
+ String[][] expected2 = new String[][] {
+ {"{\"writeid\":2,\"bucketid\":536936448,\"rowid\":0}\t1\t4",
+ "acidtblpart/p=p1/base_0000003/bucket_00001"},
+ {"{\"writeid\":1,\"bucketid\":536936448,\"rowid\":0}\t4\t5",
+ "acidtblpart/p=p2/delta_0000001_0000001_0000/bucket_00001"},
+ {"{\"writeid\":3,\"bucketid\":536936448,\"rowid\":0}\t5\t6",
+ "acidtblpart/p=p1/base_0000003/bucket_00001"},
+ {"{\"writeid\":3,\"bucketid\":536936448,\"rowid\":0}\t8\t8",
+ "acidtblpart/p=p2/delta_0000003_0000003_0000/bucket_00001"}};
+
+ checkResult(expected2, testQuery, false, "check data after concatenate", LOG);
+ }
+
+ @Test
+ public void testConcatenateMM() throws Exception {
+ HiveConf.setBoolVar(hiveConf, HiveConf.ConfVars.HIVE_CREATE_TABLES_AS_INSERT_ONLY, true);
+ runStatementOnDriver("drop table if exists T");
+ runStatementOnDriver("create table T(a int, b int)");
+ runStatementOnDriver("insert into T values(1,2),(4,5)");
+ runStatementOnDriver("insert into T values(5,6),(8,8)");
+ String testQuery = "select a, b, INPUT__FILE__NAME from T order by a, b";
+ String[][] expected = new String[][] {
+ {"1\t2",
+ "t/delta_0000001_0000001_0000/000000_0"},
+ {"4\t5",
+ "t/delta_0000001_0000001_0000/000000_0"},
+ {"5\t6",
+ "t/delta_0000002_0000002_0000/000000_0"},
+ {"8\t8",
+ "t/delta_0000002_0000002_0000/000000_0"}};
+ checkResult(expected, testQuery, false, "check data", LOG);
+
+ /*in UTs, there is no standalone HMS running to kick off compaction so it's done via runWorker()
+ but in normal usage 'concatenate' is blocking, */
+ hiveConf.setBoolVar(HiveConf.ConfVars.TRANSACTIONAL_CONCATENATE_NOBLOCK, true);
+ runStatementOnDriver("alter table T concatenate");
+
+ TxnStore txnStore = TxnUtils.getTxnStore(hiveConf);
+ ShowCompactResponse rsp = txnStore.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompactsSize());
+ Assert.assertEquals(TxnStore.INITIATED_RESPONSE, rsp.getCompacts().get(0).getState());
+ runWorker(hiveConf);
+ rsp = txnStore.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompactsSize());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());
+ String[][] expected2 = new String[][] {
+ {"1\t2",
+ "t/base_0000002/000000_0"},
+ {"4\t5",
+ "t/base_0000002/000000_0"},
+ {"5\t6",
+ "t/base_0000002/000000_0"},
+ {"8\t8",
+ "t/base_0000002/000000_0"}};
+ checkResult(expected2, testQuery, false, "check data after concatenate", LOG);
+ }
+}
[3/3] hive git commit: HIVE-19322 - broken test:
TestNegativeMinimrCliDriver#testCliDriver[minimr_broken_pipe] (Eugene Koifman,
reviewed by Jason Dere)
Posted by ek...@apache.org.
HIVE-19322 - broken test: TestNegativeMinimrCliDriver#testCliDriver[minimr_broken_pipe] (Eugene Koifman, reviewed by Jason Dere)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/63abf9be
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/63abf9be
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/63abf9be
Branch: refs/heads/branch-3
Commit: 63abf9be053c2f3a4ff524b7ab2b1c17248decb5
Parents: cb61ed6
Author: Eugene Koifman <ek...@apache.org>
Authored: Mon May 7 12:08:35 2018 -0700
Committer: Eugene Koifman <ek...@apache.org>
Committed: Mon May 7 12:08:35 2018 -0700
----------------------------------------------------------------------
.../ql/cache/results/QueryResultsCache.java | 2 +-
.../ql/metadata/SessionHiveMetaStoreClient.java | 66 +++++++++++++-------
.../hadoop/hive/ql/parse/SemanticAnalyzer.java | 3 +-
.../ql/parse/repl/dump/PartitionExport.java | 9 ++-
.../hadoop/hive/ql/session/SessionState.java | 14 +++--
.../check_constraint_tbl_level.q.out | 1 -
.../check_constraint_violation.q.out | 1 -
.../merge_constraint_notnull.q.out | 1 -
8 files changed, 63 insertions(+), 34 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/63abf9be/ql/src/java/org/apache/hadoop/hive/ql/cache/results/QueryResultsCache.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/cache/results/QueryResultsCache.java b/ql/src/java/org/apache/hadoop/hive/ql/cache/results/QueryResultsCache.java
index 56a9faa..6734ac5 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/cache/results/QueryResultsCache.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/cache/results/QueryResultsCache.java
@@ -687,7 +687,7 @@ public final class QueryResultsCache {
if (readEntity.getType() == Type.TABLE) {
Table tableUsed = readEntity.getTable();
Map<String, Table> tempTables =
- SessionHiveMetaStoreClient.getTempTablesForDatabase(tableUsed.getDbName());
+ SessionHiveMetaStoreClient.getTempTablesForDatabase(tableUsed.getDbName(), tableUsed.getTableName());
if (tempTables != null && tempTables.containsKey(tableUsed.getTableName())) {
LOG.info("{} resolves to a temporary table in the current session. This query cannot use the cache.",
tableUsed.getTableName());
http://git-wip-us.apache.org/repos/asf/hive/blob/63abf9be/ql/src/java/org/apache/hadoop/hive/ql/metadata/SessionHiveMetaStoreClient.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/SessionHiveMetaStoreClient.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/SessionHiveMetaStoreClient.java
index d89df48..071756d 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/SessionHiveMetaStoreClient.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/SessionHiveMetaStoreClient.java
@@ -68,10 +68,22 @@ import org.apache.hadoop.hive.shims.HadoopShims;
import org.apache.hadoop.hive.shims.ShimLoader;
import org.apache.hadoop.hive.ql.stats.StatsUtils;
import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import static org.apache.hadoop.hive.metastore.Warehouse.DEFAULT_CATALOG_NAME;
+/**
+ * todo: This need review re: thread safety. Various places (see callsers of
+ * {@link SessionState#setCurrentSessionState(SessionState)}) pass SessionState to forked threads.
+ * Currently it looks like those threads only read metadata but this is fragile.
+ * Also, maps (in SessionState) where tempt table metadata is stored are concurrent and so
+ * any put/get crosses a memory barrier and so does using most {@code java.util.concurrent.*}
+ * so the readers of the objects in these maps should have the most recent view of the object.
+ * But again, could be fragile.
+ */
public class SessionHiveMetaStoreClient extends HiveMetaStoreClient implements IMetaStoreClient {
+ private static final Logger LOG = LoggerFactory.getLogger(SessionHiveMetaStoreClient.class);
SessionHiveMetaStoreClient(Configuration conf, Boolean allowEmbedded) throws MetaException {
super(conf, null, allowEmbedded);
@@ -174,7 +186,7 @@ public class SessionHiveMetaStoreClient extends HiveMetaStoreClient implements I
List<String> tableNames = super.getAllTables(dbName);
// May need to merge with list of temp tables
- Map<String, Table> tables = getTempTablesForDatabase(dbName);
+ Map<String, Table> tables = getTempTablesForDatabase(dbName, "?");
if (tables == null || tables.size() == 0) {
return tableNames;
}
@@ -198,7 +210,7 @@ public class SessionHiveMetaStoreClient extends HiveMetaStoreClient implements I
// May need to merge with list of temp tables
dbName = dbName.toLowerCase();
tablePattern = tablePattern.toLowerCase();
- Map<String, Table> tables = getTempTablesForDatabase(dbName);
+ Map<String, Table> tables = getTempTablesForDatabase(dbName, tablePattern);
if (tables == null || tables.size() == 0) {
return tableNames;
}
@@ -224,7 +236,8 @@ public class SessionHiveMetaStoreClient extends HiveMetaStoreClient implements I
public List<TableMeta> getTableMeta(String dbPatterns, String tablePatterns, List<String> tableTypes)
throws MetaException {
List<TableMeta> tableMetas = super.getTableMeta(dbPatterns, tablePatterns, tableTypes);
- Map<String, Map<String, Table>> tmpTables = getTempTables();
+ Map<String, Map<String, Table>> tmpTables = getTempTables("dbPatterns='" + dbPatterns +
+ "' tablePatterns='" + tablePatterns + "'");
if (tmpTables.isEmpty()) {
return tableMetas;
}
@@ -426,7 +439,7 @@ public class SessionHiveMetaStoreClient extends HiveMetaStoreClient implements I
SessionState ss = SessionState.get();
if (ss == null) {
- throw new MetaException("No current SessionState, cannot create temporary table"
+ throw new MetaException("No current SessionState, cannot create temporary table: "
+ Warehouse.getQualifiedName(tbl));
}
@@ -435,7 +448,7 @@ public class SessionHiveMetaStoreClient extends HiveMetaStoreClient implements I
String dbName = tbl.getDbName();
String tblName = tbl.getTableName();
- Map<String, Table> tables = getTempTablesForDatabase(dbName);
+ Map<String, Table> tables = getTempTablesForDatabase(dbName, tblName);
if (tables != null && tables.containsKey(tblName)) {
throw new MetaException(
"Temporary table " + StatsUtils.getFullyQualifiedTableName(dbName, tblName) + " already exists");
@@ -472,7 +485,8 @@ public class SessionHiveMetaStoreClient extends HiveMetaStoreClient implements I
}
private org.apache.hadoop.hive.metastore.api.Table getTempTable(String dbName, String tableName) {
- Map<String, Table> tables = getTempTablesForDatabase(dbName.toLowerCase());
+ Map<String, Table> tables = getTempTablesForDatabase(dbName.toLowerCase(),
+ tableName.toLowerCase());
if (tables != null) {
Table table = tables.get(tableName.toLowerCase());
if (table != null) {
@@ -510,13 +524,13 @@ public class SessionHiveMetaStoreClient extends HiveMetaStoreClient implements I
// Remove old temp table entry, and add new entry to list of temp tables.
// Note that for temp tables there is no need to rename directories
- Map<String, Table> tables = getTempTablesForDatabase(dbname);
+ Map<String, Table> tables = getTempTablesForDatabase(dbname, tbl_name);
if (tables == null || tables.remove(tbl_name) == null) {
throw new MetaException("Could not find temp table entry for " + dbname + "." + tbl_name);
}
shouldDeleteColStats = true;
- tables = getTempTablesForDatabase(newDbName);
+ tables = getTempTablesForDatabase(newDbName, tbl_name);
if (tables == null) {
tables = new HashMap<String, Table>();
SessionState.get().getTempTables().put(newDbName, tables);
@@ -526,7 +540,7 @@ public class SessionHiveMetaStoreClient extends HiveMetaStoreClient implements I
if (haveTableColumnsChanged(oldt, newt)) {
shouldDeleteColStats = true;
}
- getTempTablesForDatabase(dbname).put(tbl_name, newTable);
+ getTempTablesForDatabase(dbname, tbl_name).put(tbl_name, newTable);
}
if (shouldDeleteColStats) {
@@ -652,7 +666,7 @@ public class SessionHiveMetaStoreClient extends HiveMetaStoreClient implements I
}
// Remove table entry from SessionState
- Map<String, Table> tables = getTempTablesForDatabase(dbName);
+ Map<String, Table> tables = getTempTablesForDatabase(dbName, tableName);
if (tables == null || tables.remove(tableName) == null) {
throw new MetaException(
"Could not find temp table entry for " + StatsUtils.getFullyQualifiedTableName(dbName, tableName));
@@ -682,14 +696,20 @@ public class SessionHiveMetaStoreClient extends HiveMetaStoreClient implements I
return newCopy;
}
- public static Map<String, Table> getTempTablesForDatabase(String dbName) {
- return getTempTables().get(dbName);
+ /**
+ * @param dbName actual database name
+ * @param tblName actual table name or search pattern (for error message)
+ */
+ public static Map<String, Table> getTempTablesForDatabase(String dbName,
+ String tblName) {
+ return getTempTables(Warehouse.getQualifiedName(dbName, tblName)).
+ get(dbName);
}
- public static Map<String, Map<String, Table>> getTempTables() {
+ private static Map<String, Map<String, Table>> getTempTables(String msg) {
SessionState ss = SessionState.get();
if (ss == null) {
- LOG.debug("No current SessionState, skipping temp tables");
+ LOG.warn("No current SessionState, skipping temp tables for " + msg);
return Collections.emptyMap();
}
return ss.getTempTables();
@@ -699,7 +719,8 @@ public class SessionHiveMetaStoreClient extends HiveMetaStoreClient implements I
String tableName) {
SessionState ss = SessionState.get();
if (ss == null) {
- LOG.debug("No current SessionState, skipping temp tables");
+ LOG.debug("No current SessionState, skipping temp tables for " +
+ Warehouse.getQualifiedName(dbName, tableName));
return null;
}
String lookupName = StatsUtils.getFullyQualifiedTableName(dbName.toLowerCase(),
@@ -976,15 +997,17 @@ public class SessionHiveMetaStoreClient extends HiveMetaStoreClient implements I
getQualifiedName(t.getDbName().toLowerCase(), t.getTableName().toLowerCase());
SessionState ss = SessionState.get();
if (ss == null) {
- LOG.debug("No current SessionState, skipping temp partitions");
+ LOG.warn("No current SessionState, skipping temp partitions for " + qualifiedTableName);
return null;
}
return ss.getTempPartitions().get(qualifiedTableName);
}
private static void removeTempTable(org.apache.hadoop.hive.metastore.api.Table t) {
+ String qualifiedTableName = Warehouse.
+ getQualifiedName(t.getDbName().toLowerCase(), t.getTableName().toLowerCase());
SessionState ss = SessionState.get();
if (ss == null) {
- LOG.debug("No current SessionState, skipping temp partitions");
+ LOG.warn("No current SessionState, skipping temp partitions for " + qualifiedTableName);
return;
}
ss.getTempPartitions().remove(Warehouse.getQualifiedName(t));
@@ -994,15 +1017,16 @@ public class SessionHiveMetaStoreClient extends HiveMetaStoreClient implements I
//do nothing as it's not a partitioned table
return;
}
+ String qualifiedTableName = Warehouse.
+ getQualifiedName(t.getDbName().toLowerCase(), t.getTableName().toLowerCase());
SessionState ss = SessionState.get();
if (ss == null) {
- LOG.debug("No current SessionState, skipping temp partitions");
+ LOG.warn("No current SessionState, skipping temp partitions for " + qualifiedTableName);
return;
}
TempTable tt = new TempTable(t);
- String qualifiedName = Warehouse.getQualifiedName(t);
- if(ss.getTempPartitions().putIfAbsent(qualifiedName, tt) != null) {
- throw new IllegalStateException("TempTable for " + qualifiedName + " already exists");
+ if(ss.getTempPartitions().putIfAbsent(qualifiedTableName, tt) != null) {
+ throw new IllegalStateException("TempTable for " + qualifiedTableName + " already exists");
}
}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/63abf9be/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
index 6eb67f0..1ab7158 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
@@ -13126,7 +13126,8 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer {
throw new SemanticException("No current SessionState, cannot create temporary table "
+ dbName + "." + tblName);
}
- Map<String, Table> tables = SessionHiveMetaStoreClient.getTempTablesForDatabase(dbName);
+ Map<String, Table> tables = SessionHiveMetaStoreClient.
+ getTempTablesForDatabase(dbName, tblName);
if (tables != null && tables.containsKey(tblName)) {
throw new SemanticException("Temporary table " + dbName + "." + tblName
+ " already exists");
http://git-wip-us.apache.org/repos/asf/hive/blob/63abf9be/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/PartitionExport.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/PartitionExport.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/PartitionExport.java
index c960d2c..70eb750 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/PartitionExport.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/dump/PartitionExport.java
@@ -20,11 +20,11 @@ package org.apache.hadoop.hive.ql.parse.repl.dump;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.ql.hooks.ReadEntity;
import org.apache.hadoop.hive.ql.metadata.Partition;
import org.apache.hadoop.hive.ql.metadata.PartitionIterable;
import org.apache.hadoop.hive.ql.parse.ReplicationSpec;
import org.apache.hadoop.hive.ql.parse.repl.dump.io.FileOperations;
+import org.apache.hadoop.hive.ql.session.SessionState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,7 +36,6 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
-import static org.apache.hadoop.hive.ql.parse.repl.dump.TableExport.AuthEntities;
import static org.apache.hadoop.hive.ql.parse.repl.dump.TableExport.Paths;
/**
@@ -50,6 +49,7 @@ class PartitionExport {
private final String distCpDoAsUser;
private final HiveConf hiveConf;
private final int nThreads;
+ private final SessionState callersSession;
private static final Logger LOG = LoggerFactory.getLogger(PartitionExport.class);
private BlockingQueue<Partition> queue;
@@ -62,11 +62,14 @@ class PartitionExport {
this.hiveConf = hiveConf;
this.nThreads = hiveConf.getIntVar(HiveConf.ConfVars.REPL_PARTITIONS_DUMP_PARALLELISM);
this.queue = new ArrayBlockingQueue<>(2 * nThreads);
+ this.callersSession = SessionState.get();
}
void write(final ReplicationSpec forReplicationSpec) throws InterruptedException {
- ExecutorService producer = Executors.newFixedThreadPool(1);
+ ExecutorService producer = Executors.newFixedThreadPool(1,
+ new ThreadFactoryBuilder().setNameFormat("partition-submitter-thread-%d").build());
producer.submit(() -> {
+ SessionState.setCurrentSessionState(callersSession);
for (Partition partition : partitionIterable) {
try {
queue.put(partition);
http://git-wip-us.apache.org/repos/asf/hive/blob/63abf9be/ql/src/java/org/apache/hadoop/hive/ql/session/SessionState.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/session/SessionState.java b/ql/src/java/org/apache/hadoop/hive/ql/session/SessionState.java
index 5198574..6bb756c 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/session/SessionState.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/session/SessionState.java
@@ -41,6 +41,7 @@ import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CancellationException;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.commons.lang.StringUtils;
@@ -117,10 +118,14 @@ public class SessionState {
static final String LOCK_FILE_NAME = "inuse.lck";
static final String INFO_FILE_NAME = "inuse.info";
- private final Map<String, Map<String, Table>> tempTables = new HashMap<>();
+ /**
+ * Concurrent since SessionState is often propagated to workers in thread pools
+ */
+ private final Map<String, Map<String, Table>> tempTables = new ConcurrentHashMap<>();
private final Map<String, Map<String, ColumnStatisticsObj>> tempTableColStats =
- new HashMap<String, Map<String, ColumnStatisticsObj>>();
- private final Map<String, SessionHiveMetaStoreClient.TempTable> tempPartitions = new HashMap<>();
+ new ConcurrentHashMap<>();
+ private final Map<String, SessionHiveMetaStoreClient.TempTable> tempPartitions =
+ new ConcurrentHashMap<>();
protected ClassLoader parentLoader;
@@ -537,8 +542,7 @@ public class SessionState {
* Singleton Session object per thread.
*
**/
- private static InheritableThreadLocal<SessionStates> tss =
- new InheritableThreadLocal<SessionStates>() {
+ private static ThreadLocal<SessionStates> tss = new ThreadLocal<SessionStates>() {
@Override
protected SessionStates initialValue() {
return new SessionStates();
http://git-wip-us.apache.org/repos/asf/hive/blob/63abf9be/ql/src/test/results/clientnegative/check_constraint_tbl_level.q.out
----------------------------------------------------------------------
diff --git a/ql/src/test/results/clientnegative/check_constraint_tbl_level.q.out b/ql/src/test/results/clientnegative/check_constraint_tbl_level.q.out
index c18f733..3ed9c25 100644
--- a/ql/src/test/results/clientnegative/check_constraint_tbl_level.q.out
+++ b/ql/src/test/results/clientnegative/check_constraint_tbl_level.q.out
@@ -11,5 +11,4 @@ PREHOOK: type: QUERY
PREHOOK: Input: _dummy_database@_dummy_table
PREHOOK: Output: default@tti
#### A masked pattern was here ####
-Error during job, obtaining debugging information...
FAILED: Execution Error, return code 2 from org.apache.hadoop.hive.ql.exec.mr.MapRedTask
http://git-wip-us.apache.org/repos/asf/hive/blob/63abf9be/ql/src/test/results/clientnegative/check_constraint_violation.q.out
----------------------------------------------------------------------
diff --git a/ql/src/test/results/clientnegative/check_constraint_violation.q.out b/ql/src/test/results/clientnegative/check_constraint_violation.q.out
index 05e3ff5..82f367c 100644
--- a/ql/src/test/results/clientnegative/check_constraint_violation.q.out
+++ b/ql/src/test/results/clientnegative/check_constraint_violation.q.out
@@ -11,5 +11,4 @@ PREHOOK: type: QUERY
PREHOOK: Input: _dummy_database@_dummy_table
PREHOOK: Output: default@tti
#### A masked pattern was here ####
-Error during job, obtaining debugging information...
FAILED: Execution Error, return code 2 from org.apache.hadoop.hive.ql.exec.mr.MapRedTask
http://git-wip-us.apache.org/repos/asf/hive/blob/63abf9be/ql/src/test/results/clientnegative/merge_constraint_notnull.q.out
----------------------------------------------------------------------
diff --git a/ql/src/test/results/clientnegative/merge_constraint_notnull.q.out b/ql/src/test/results/clientnegative/merge_constraint_notnull.q.out
index 0fd26ff..088e249 100644
--- a/ql/src/test/results/clientnegative/merge_constraint_notnull.q.out
+++ b/ql/src/test/results/clientnegative/merge_constraint_notnull.q.out
@@ -52,5 +52,4 @@ PREHOOK: Output: default@testt
PREHOOK: Output: default@testt
PREHOOK: Output: default@testt
#### A masked pattern was here ####
-Error during job, obtaining debugging information...
FAILED: Execution Error, return code 2 from org.apache.hadoop.hive.ql.exec.mr.MapRedTask