You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by se...@apache.org on 2017/09/29 21:01:45 UTC
[04/17] hive git commit: HIVE-17482: External LLAP client: acquire
locks for tables queried directly by LLAP (Jason Dere,
reviewed by Eugene Koifman)
HIVE-17482: External LLAP client: acquire locks for tables queried directly by LLAP (Jason Dere, reviewed by Eugene Koifman)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/76d6e8a0
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/76d6e8a0
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/76d6e8a0
Branch: refs/heads/hive-14535
Commit: 76d6e8a092ed2c73d98f4e128d9620843d00c799
Parents: 2216dad
Author: Jason Dere <jd...@hortonworks.com>
Authored: Fri Sep 29 11:08:10 2017 -0700
Committer: Jason Dere <jd...@hortonworks.com>
Committed: Fri Sep 29 11:08:10 2017 -0700
----------------------------------------------------------------------
.../apache/hadoop/hive/ql/TestAcidOnTez.java | 87 +++++++++++++++
.../java/org/apache/hadoop/hive/ql/Driver.java | 111 +++++++++++++------
.../org/apache/hadoop/hive/ql/QueryState.java | 14 +++
.../hive/ql/parse/BaseSemanticAnalyzer.java | 13 ++-
.../hadoop/hive/ql/parse/SemanticAnalyzer.java | 4 +-
.../ql/parse/UpdateDeleteSemanticAnalyzer.java | 2 +-
.../hadoop/hive/ql/session/SessionState.java | 15 +++
.../ql/udf/generic/GenericUDTFGetSplits.java | 90 +++++++++++++--
8 files changed, 292 insertions(+), 44 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/76d6e8a0/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/TestAcidOnTez.java
----------------------------------------------------------------------
diff --git a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/TestAcidOnTez.java b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/TestAcidOnTez.java
index 8b4b21f..ea0aadf 100644
--- a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/TestAcidOnTez.java
+++ b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/TestAcidOnTez.java
@@ -18,9 +18,13 @@
package org.apache.hadoop.hive.ql;
+import static org.junit.Assert.assertEquals;
+
import java.io.File;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
@@ -28,14 +32,21 @@ import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.FileUtils;
import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.metastore.api.LockState;
+import org.apache.hadoop.hive.metastore.api.LockType;
import org.apache.hadoop.hive.metastore.api.ShowCompactRequest;
import org.apache.hadoop.hive.metastore.api.ShowCompactResponse;
+import org.apache.hadoop.hive.metastore.api.ShowLocksRequest;
+import org.apache.hadoop.hive.metastore.api.ShowLocksResponse;
+import org.apache.hadoop.hive.metastore.api.ShowLocksResponseElement;
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.ql.exec.AbstractFileMergeOperator;
import org.apache.hadoop.hive.ql.io.BucketCodec;
import org.apache.hadoop.hive.ql.io.HiveInputFormat;
+import org.apache.hadoop.hive.ql.lockmgr.TestDbTxnManager2;
import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.junit.After;
@@ -704,6 +715,82 @@ ekoifman:apache-hive-3.0.0-SNAPSHOT-bin ekoifman$ tree ~/dev/hiverwgit/itests/h
Assert.assertTrue("Actual line(file) " + i + " ac: " + rs.get(i), rs.get(i).endsWith(expected2[i][1]));
}
}
+
+ @Test
+ public void testGetSplitsLocks() throws Exception {
+ // Need to test this with LLAP settings, which requires some additional configurations set.
+ HiveConf modConf = new HiveConf(hiveConf);
+ setupTez(modConf);
+ modConf.setVar(ConfVars.HIVE_EXECUTION_ENGINE, "tez");
+ modConf.setVar(ConfVars.HIVEFETCHTASKCONVERSION, "more");
+ modConf.setVar(HiveConf.ConfVars.LLAP_DAEMON_SERVICE_HOSTS, "localhost");
+
+ // SessionState/Driver needs to be restarted with the Tez conf settings.
+ restartSessionAndDriver(modConf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(modConf);
+
+ try {
+ // Request LLAP splits for a table.
+ String queryParam = "select * from " + Table.ACIDTBL;
+ runStatementOnDriver("select get_splits(\"" + queryParam + "\", 1)");
+
+ // The get_splits call should have resulted in a lock on ACIDTBL
+ ShowLocksResponse slr = txnHandler.showLocks(new ShowLocksRequest());
+ TestDbTxnManager2.checkLock(LockType.SHARED_READ, LockState.ACQUIRED,
+ "default", Table.ACIDTBL.name, null, slr.getLocks());
+ assertEquals(1, slr.getLocksSize());
+
+ // Try another table.
+ queryParam = "select * from " + Table.ACIDTBLPART;
+ runStatementOnDriver("select get_splits(\"" + queryParam + "\", 1)");
+
+ // Should now have new lock on ACIDTBLPART
+ slr = txnHandler.showLocks(new ShowLocksRequest());
+ TestDbTxnManager2.checkLock(LockType.SHARED_READ, LockState.ACQUIRED,
+ "default", Table.ACIDTBLPART.name, null, slr.getLocks());
+ assertEquals(2, slr.getLocksSize());
+
+ // There should be different txn IDs associated with each lock.
+ Set<Long> txnSet = new HashSet<Long>();
+ for (ShowLocksResponseElement lockResponseElem : slr.getLocks()) {
+ txnSet.add(lockResponseElem.getTxnid());
+ }
+ assertEquals(2, txnSet.size());
+
+ List<String> rows = runStatementOnDriver("show transactions");
+ // Header row + 2 transactions = 3 rows
+ assertEquals(3, rows.size());
+ } finally {
+ // Close the session which should free up the TxnHandler/locks held by the session.
+ // Done in the finally block to make sure we free up the locks; otherwise
+ // the cleanup in tearDown() will get stuck waiting on the lock held here on ACIDTBL.
+ restartSessionAndDriver(hiveConf);
+ }
+
+ // Lock should be freed up now.
+ ShowLocksResponse slr = txnHandler.showLocks(new ShowLocksRequest());
+ assertEquals(0, slr.getLocksSize());
+
+ List<String> rows = runStatementOnDriver("show transactions");
+ // Transactions should be committed.
+ // No transactions - just the header row
+ assertEquals(1, rows.size());
+ }
+
+ private void restartSessionAndDriver(HiveConf conf) throws Exception {
+ SessionState ss = SessionState.get();
+ if (ss != null) {
+ ss.close();
+ }
+ if (d != null) {
+ d.destroy();
+ d.close();
+ }
+
+ SessionState.start(conf);
+ d = new Driver(conf);
+ }
+
// Ideally test like this should be a qfile test. However, the explain output from qfile is always
// slightly different depending on where the test is run, specifically due to file size estimation
private void testJoin(String engine, String joinType) throws Exception {
http://git-wip-us.apache.org/repos/asf/hive/blob/76d6e8a0/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
index ec228b4..f01edf8 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
@@ -175,6 +175,19 @@ public class Driver implements CommandProcessor {
private QueryLifeTimeHookRunner queryLifeTimeHookRunner;
private final HooksLoader hooksLoader;
+ // Transaction manager the Driver has been initialized with (can be null).
+ // If this is set then this Transaction manager will be used during query
+ // compilation/execution rather than using the current session's transaction manager.
+ // This might be needed in a situation where a Driver is nested within an already
+ // running Driver/query - the nested Driver requires a separate transaction manager
+ // so as not to conflict with the outer Driver/query which is using the session
+ // transaction manager.
+ private final HiveTxnManager initTxnMgr;
+
+ // Transaction manager used for the query. This will be set at compile time based on
+ // either initTxnMgr or from the SessionState, in that order.
+ private HiveTxnManager queryTxnMgr;
+
public enum DriverState {
INITIALIZED,
COMPILING,
@@ -352,8 +365,12 @@ public class Driver implements CommandProcessor {
this(getNewQueryState(conf), null);
}
+ public Driver(HiveConf conf, HiveTxnManager txnMgr) {
+ this(getNewQueryState(conf), null, null, txnMgr);
+ }
+
public Driver(HiveConf conf, Context ctx) {
- this(getNewQueryState(conf), null);
+ this(getNewQueryState(conf), null, null);
this.ctx = ctx;
}
@@ -362,18 +379,22 @@ public class Driver implements CommandProcessor {
}
public Driver(QueryState queryState, String userName) {
- this(queryState, userName, new HooksLoader(queryState.getConf()), null);
+ this(queryState, userName, new HooksLoader(queryState.getConf()), null, null);
}
public Driver(HiveConf conf, HooksLoader hooksLoader) {
- this(getNewQueryState(conf), null, hooksLoader, null);
+ this(getNewQueryState(conf), null, hooksLoader, null, null);
}
public Driver(QueryState queryState, String userName, QueryInfo queryInfo) {
- this(queryState, userName, new HooksLoader(queryState.getConf()), queryInfo);
+ this(queryState, userName, new HooksLoader(queryState.getConf()), queryInfo, null);
+ }
+
+ public Driver(QueryState queryState, String userName, QueryInfo queryInfo, HiveTxnManager txnMgr) {
+ this(queryState, userName, new HooksLoader(queryState.getConf()), queryInfo, txnMgr);
}
- public Driver(QueryState queryState, String userName, HooksLoader hooksLoader, QueryInfo queryInfo) {
+ public Driver(QueryState queryState, String userName, HooksLoader hooksLoader, QueryInfo queryInfo, HiveTxnManager txnMgr) {
this.queryState = queryState;
this.conf = queryState.getConf();
isParallelEnabled = (conf != null)
@@ -382,6 +403,7 @@ public class Driver implements CommandProcessor {
this.hooksLoader = hooksLoader;
this.queryLifeTimeHookRunner = new QueryLifeTimeHookRunner(conf, hooksLoader, console);
this.queryInfo = queryInfo;
+ this.initTxnMgr = txnMgr;
}
/**
@@ -477,16 +499,22 @@ public class Driver implements CommandProcessor {
try {
// Initialize the transaction manager. This must be done before analyze is called.
- final HiveTxnManager txnManager = SessionState.get().initTxnMgr(conf);
- // In case when user Ctrl-C twice to kill Hive CLI JVM, we want to release locks
+ if (initTxnMgr != null) {
+ queryTxnMgr = initTxnMgr;
+ } else {
+ queryTxnMgr = SessionState.get().initTxnMgr(conf);
+ }
+ queryState.setTxnManager(queryTxnMgr);
+ // In case when user Ctrl-C twice to kill Hive CLI JVM, we want to release locks
// if compile is being called multiple times, clear the old shutdownhook
ShutdownHookManager.removeShutdownHook(shutdownRunner);
+ final HiveTxnManager txnMgr = queryTxnMgr;
shutdownRunner = new Runnable() {
@Override
public void run() {
try {
- releaseLocksAndCommitOrRollback(false, txnManager);
+ releaseLocksAndCommitOrRollback(false, txnMgr);
} catch (LockException e) {
LOG.warn("Exception when releasing locks in ShutdownHook for Driver: " +
e.getMessage());
@@ -535,13 +563,13 @@ public class Driver implements CommandProcessor {
// because at that point we need access to the objects.
Hive.get().getMSC().flushCache();
- if(checkConcurrency() && startImplicitTxn(txnManager)) {
+ if(checkConcurrency() && startImplicitTxn(queryTxnMgr)) {
String userFromUGI = getUserFromUGI();
- if (!txnManager.isTxnOpen()) {
+ if (!queryTxnMgr.isTxnOpen()) {
if(userFromUGI == null) {
return 10;
}
- long txnid = txnManager.openTxn(ctx, userFromUGI);
+ long txnid = queryTxnMgr.openTxn(ctx, userFromUGI);
}
}
// Do semantic analysis and plan generation
@@ -1133,13 +1161,12 @@ public class Driver implements CommandProcessor {
// Write the current set of valid transactions into the conf file so that it can be read by
// the input format.
- private void recordValidTxns() throws LockException {
+ private void recordValidTxns(HiveTxnManager txnMgr) throws LockException {
ValidTxnList oldList = null;
String s = conf.get(ValidTxnList.VALID_TXNS_KEY);
if(s != null && s.length() > 0) {
oldList = new ValidReadTxnList(s);
}
- HiveTxnManager txnMgr = SessionState.get().getTxnMgr();
ValidTxnList txns = txnMgr.getValidTxns();
if(oldList != null) {
throw new IllegalStateException("calling recordValidTxn() more than once in the same " +
@@ -1172,6 +1199,7 @@ public class Driver implements CommandProcessor {
}
return null;
}
+
/**
* Acquire read and write locks needed by the statement. The list of objects to be locked are
* obtained from the inputs and outputs populated by the compiler. Locking strategy depends on
@@ -1184,9 +1212,7 @@ public class Driver implements CommandProcessor {
PerfLogger perfLogger = SessionState.getPerfLogger();
perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
- SessionState ss = SessionState.get();
- HiveTxnManager txnMgr = ss.getTxnMgr();
- if(!txnMgr.isTxnOpen() && txnMgr.supportsAcid()) {
+ if(!queryTxnMgr.isTxnOpen() && queryTxnMgr.supportsAcid()) {
/*non acid txn managers don't support txns but fwd lock requests to lock managers
acid txn manager requires all locks to be associated with a txn so if we
end up here w/o an open txn it's because we are processing something like "use <database>
@@ -1205,17 +1231,17 @@ public class Driver implements CommandProcessor {
//so this makes (file name -> data) mapping stable
acidSinks.sort((FileSinkDesc fsd1, FileSinkDesc fsd2) -> fsd1.getDirName().compareTo(fsd2.getDirName()));
for (FileSinkDesc desc : acidSinks) {
- desc.setTransactionId(txnMgr.getCurrentTxnId());
+ desc.setTransactionId(queryTxnMgr.getCurrentTxnId());
//it's possible to have > 1 FileSink writing to the same table/partition
//e.g. Merge stmt, multi-insert stmt when mixing DP and SP writes
- desc.setStatementId(txnMgr.getWriteIdAndIncrement());
+ desc.setStatementId(queryTxnMgr.getWriteIdAndIncrement());
}
}
/*It's imperative that {@code acquireLocks()} is called for all commands so that
HiveTxnManager can transition its state machine correctly*/
- txnMgr.acquireLocks(plan, ctx, userFromUGI, lDrvState);
- if(txnMgr.recordSnapshot(plan)) {
- recordValidTxns();
+ queryTxnMgr.acquireLocks(plan, ctx, userFromUGI, lDrvState);
+ if(queryTxnMgr.recordSnapshot(plan)) {
+ recordValidTxns(queryTxnMgr);
}
return 0;
} catch (Exception e) {
@@ -1233,6 +1259,11 @@ public class Driver implements CommandProcessor {
private boolean haveAcidWrite() {
return !plan.getAcidSinks().isEmpty();
}
+
+ public void releaseLocksAndCommitOrRollback(boolean commit) throws LockException {
+ releaseLocksAndCommitOrRollback(commit, queryTxnMgr);
+ }
+
/**
* @param commit if there is an open transaction and if true, commit,
* if false rollback. If there is no open transaction this parameter is ignored.
@@ -1246,8 +1277,8 @@ public class Driver implements CommandProcessor {
perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.RELEASE_LOCKS);
HiveTxnManager txnMgr;
if (txnManager == null) {
- SessionState ss = SessionState.get();
- txnMgr = ss.getTxnMgr();
+ // Default to driver's txn manager if no txn manager specified
+ txnMgr = queryTxnMgr;
} else {
txnMgr = txnManager;
}
@@ -1366,6 +1397,23 @@ public class Driver implements CommandProcessor {
return createProcessorResponse(compileInternal(command, false));
}
+ public CommandProcessorResponse lockAndRespond() {
+ // Assumes the query has already been compiled
+ if (plan == null) {
+ throw new IllegalStateException(
+ "No previously compiled query for driver - queryId=" + queryState.getQueryId());
+ }
+
+ int ret = 0;
+ if (requiresLock()) {
+ ret = acquireLocks();
+ }
+ if (ret != 0) {
+ return rollback(createProcessorResponse(ret));
+ }
+ return createProcessorResponse(ret);
+ }
+
private static final ReentrantLock globalCompileLock = new ReentrantLock();
private int compileInternal(String command, boolean deferClose) {
int ret;
@@ -1396,7 +1444,7 @@ public class Driver implements CommandProcessor {
if (ret != 0) {
try {
- releaseLocksAndCommitOrRollback(false, null);
+ releaseLocksAndCommitOrRollback(false);
} catch (LockException e) {
LOG.warn("Exception in releasing locks. "
+ org.apache.hadoop.util.StringUtils.stringifyException(e));
@@ -1536,8 +1584,7 @@ public class Driver implements CommandProcessor {
// the reason that we set the txn manager for the cxt here is because each
// query has its own ctx object. The txn mgr is shared across the
// same instance of Driver, which can run multiple queries.
- HiveTxnManager txnManager = SessionState.get().getTxnMgr();
- ctx.setHiveTxnManager(txnManager);
+ ctx.setHiveTxnManager(queryTxnMgr);
if (requiresLock()) {
// a checkpoint to see if the thread is interrupted or not before an expensive operation
@@ -1559,11 +1606,11 @@ public class Driver implements CommandProcessor {
//if needRequireLock is false, the release here will do nothing because there is no lock
try {
//since set autocommit starts an implicit txn, close it
- if(txnManager.isImplicitTransactionOpen() || plan.getOperation() == HiveOperation.COMMIT) {
- releaseLocksAndCommitOrRollback(true, null);
+ if(queryTxnMgr.isImplicitTransactionOpen() || plan.getOperation() == HiveOperation.COMMIT) {
+ releaseLocksAndCommitOrRollback(true);
}
else if(plan.getOperation() == HiveOperation.ROLLBACK) {
- releaseLocksAndCommitOrRollback(false, null);
+ releaseLocksAndCommitOrRollback(false);
}
else {
//txn (if there is one started) is not finished
@@ -1614,7 +1661,7 @@ public class Driver implements CommandProcessor {
private CommandProcessorResponse rollback(CommandProcessorResponse cpr) {
//console.printError(cpr.toString());
try {
- releaseLocksAndCommitOrRollback(false, null);
+ releaseLocksAndCommitOrRollback(false);
}
catch (LockException e) {
LOG.error("rollback() FAILED: " + cpr);//make sure not to loose
@@ -2373,7 +2420,7 @@ public class Driver implements CommandProcessor {
if(destroyed) {
if (!hiveLocks.isEmpty()) {
try {
- releaseLocksAndCommitOrRollback(false, null);
+ releaseLocksAndCommitOrRollback(false);
} catch (LockException e) {
LOG.warn("Exception when releasing locking in destroy: " +
e.getMessage());
@@ -2428,7 +2475,7 @@ public class Driver implements CommandProcessor {
}
if (!hiveLocks.isEmpty()) {
try {
- releaseLocksAndCommitOrRollback(false, null);
+ releaseLocksAndCommitOrRollback(false);
} catch (LockException e) {
LOG.warn("Exception when releasing locking in destroy: " +
e.getMessage());
http://git-wip-us.apache.org/repos/asf/hive/blob/76d6e8a0/ql/src/java/org/apache/hadoop/hive/ql/QueryState.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/QueryState.java b/ql/src/java/org/apache/hadoop/hive/ql/QueryState.java
index fa7c323..7d5aa8b 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/QueryState.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/QueryState.java
@@ -21,6 +21,7 @@ package org.apache.hadoop.hive.ql;
import java.util.Map;
import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.lockmgr.HiveTxnManager;
import org.apache.hadoop.hive.ql.plan.HiveOperation;
/**
@@ -39,6 +40,11 @@ public class QueryState {
private HiveOperation commandType;
/**
+ * transaction manager used in the query.
+ */
+ private HiveTxnManager txnManager;
+
+ /**
* Private constructor, use QueryState.Builder instead
* @param conf The query specific configuration object
*/
@@ -73,6 +79,14 @@ public class QueryState {
return queryConf;
}
+ public HiveTxnManager getTxnManager() {
+ return txnManager;
+ }
+
+ public void setTxnManager(HiveTxnManager txnManager) {
+ this.txnManager = txnManager;
+ }
+
/**
* Builder to instantiate the QueryState object.
*/
http://git-wip-us.apache.org/repos/asf/hive/blob/76d6e8a0/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java
index 3ad30c4..06e00d7 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java
@@ -64,6 +64,7 @@ import org.apache.hadoop.hive.ql.hooks.ReadEntity;
import org.apache.hadoop.hive.ql.hooks.WriteEntity;
import org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat;
import org.apache.hadoop.hive.ql.lib.Node;
+import org.apache.hadoop.hive.ql.lockmgr.HiveTxnManager;
import org.apache.hadoop.hive.ql.metadata.Hive;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.metadata.InvalidTableException;
@@ -78,6 +79,7 @@ import org.apache.hadoop.hive.ql.plan.FileSinkDesc;
import org.apache.hadoop.hive.ql.plan.ListBucketingCtx;
import org.apache.hadoop.hive.ql.plan.PlanUtils;
import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.hive.ql.session.SessionState.LogHelper;
import org.apache.hadoop.hive.serde.serdeConstants;
import org.apache.hadoop.hive.serde2.io.DateWritable;
@@ -123,6 +125,8 @@ public abstract class BaseSemanticAnalyzer {
// whether any ACID table is involved in a query
protected boolean acidInQuery;
+ protected HiveTxnManager txnManager;
+
public static final int HIVE_COLUMN_ORDER_ASC = 1;
public static final int HIVE_COLUMN_ORDER_DESC = 0;
public static final int HIVE_COLUMN_NULLS_FIRST = 0;
@@ -160,7 +164,6 @@ public abstract class BaseSemanticAnalyzer {
autoCommitValue = autoCommit;
}
-
public boolean skipAuthorization() {
return false;
}
@@ -230,6 +233,7 @@ public abstract class BaseSemanticAnalyzer {
idToTableNameMap = new HashMap<String, String>();
inputs = new LinkedHashSet<ReadEntity>();
outputs = new LinkedHashSet<WriteEntity>();
+ txnManager = queryState.getTxnManager();
} catch (Exception e) {
throw new SemanticException(e);
}
@@ -1918,4 +1922,11 @@ public abstract class BaseSemanticAnalyzer {
fetch.setSerializationNullFormat(" ");
return (FetchTask) TaskFactory.get(fetch, conf);
}
+
+ protected HiveTxnManager getTxnMgr() {
+ if (txnManager != null) {
+ return txnManager;
+ }
+ return SessionState.get().getTxnMgr();
+ }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/76d6e8a0/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 551aafc..68ddcf0 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
@@ -13657,7 +13657,7 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer {
}
private AcidUtils.Operation getAcidType(Class<? extends OutputFormat> of, String dest) {
- if (SessionState.get() == null || !SessionState.get().getTxnMgr().supportsAcid()) {
+ if (SessionState.get() == null || !getTxnMgr().supportsAcid()) {
return AcidUtils.Operation.NOT_ACID;
} else if (isAcidOutputFormat(of)) {
return getAcidType(dest);
@@ -13676,7 +13676,7 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer {
// Make sure the proper transaction manager that supports ACID is being used
protected void checkAcidTxnManager(Table table) throws SemanticException {
- if (SessionState.get() != null && !SessionState.get().getTxnMgr().supportsAcid()) {
+ if (SessionState.get() != null && !getTxnMgr().supportsAcid()) {
throw new SemanticException(ErrorMsg.TXNMGR_NOT_ACID, table.getDbName(), table.getTableName());
}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/76d6e8a0/ql/src/java/org/apache/hadoop/hive/ql/parse/UpdateDeleteSemanticAnalyzer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/UpdateDeleteSemanticAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/UpdateDeleteSemanticAnalyzer.java
index 702be2e..b3193d1 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/UpdateDeleteSemanticAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/UpdateDeleteSemanticAnalyzer.java
@@ -70,7 +70,7 @@ public class UpdateDeleteSemanticAnalyzer extends SemanticAnalyzer {
super.analyzeInternal(tree);
} else {
- if (!SessionState.get().getTxnMgr().supportsAcid()) {
+ if (!getTxnMgr().supportsAcid()) {
throw new SemanticException(ErrorMsg.ACID_OP_ON_NONACID_TXNMGR.getMsg());
}
switch (tree.getToken().getType()) {
http://git-wip-us.apache.org/repos/asf/hive/blob/76d6e8a0/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 cceeec0..6dece05 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
@@ -19,6 +19,7 @@
package org.apache.hadoop.hive.ql.session;
import static org.apache.hadoop.hive.metastore.Warehouse.DEFAULT_DATABASE_NAME;
+import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
@@ -290,6 +291,8 @@ public class SessionState {
private String atsDomainId;
+ private List<Closeable> cleanupItems = new LinkedList<Closeable>();
+
/**
* Get the lineage state stored in this session.
*
@@ -1651,6 +1654,14 @@ public class SessionState {
}
public void close() throws IOException {
+ for (Closeable cleanupItem : cleanupItems) {
+ try {
+ cleanupItem.close();
+ } catch (Exception err) {
+ LOG.error("Error processing SessionState cleanup item " + cleanupItem.toString(), err);
+ }
+ }
+
registry.clear();
if (txnMgr != null) txnMgr.closeTxnManager();
JavaUtils.closeClassLoadersTo(sessionConf.getClassLoader(), parentLoader);
@@ -1927,6 +1938,10 @@ public class SessionState {
public KillQuery getKillQuery() {
return killQuery;
}
+
+ public void addCleanupItem(Closeable item) {
+ cleanupItems.add(item);
+ }
}
class ResourceMaps {
http://git-wip-us.apache.org/repos/asf/hive/blob/76d6e8a0/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDTFGetSplits.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDTFGetSplits.java b/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDTFGetSplits.java
index 5003f42..610f36f 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDTFGetSplits.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDTFGetSplits.java
@@ -19,6 +19,7 @@
package org.apache.hadoop.hive.ql.udf.generic;
import java.io.ByteArrayOutputStream;
+import java.io.Closeable;
import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.FileNotFoundException;
@@ -69,6 +70,8 @@ import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.exec.tez.DagUtils;
import org.apache.hadoop.hive.ql.exec.tez.HiveSplitGenerator;
import org.apache.hadoop.hive.ql.exec.tez.TezTask;
+import org.apache.hadoop.hive.ql.lockmgr.HiveTxnManager;
+import org.apache.hadoop.hive.ql.lockmgr.TxnManagerFactory;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.plan.MapWork;
import org.apache.hadoop.hive.ql.plan.PlanUtils;
@@ -184,12 +187,21 @@ public class GenericUDTFGetSplits extends GenericUDTF {
String query = stringOI.getPrimitiveJavaObject(arguments[0]);
int num = intOI.get(arguments[1]);
- PlanFragment fragment = createPlanFragment(query, num);
+ // Generate applicationId for the LLAP splits
+ LlapCoordinator coordinator = LlapCoordinator.getInstance();
+ if (coordinator == null) {
+ throw new HiveException("LLAP coordinator is not initialized; must be running in HS2 with "
+ + ConfVars.LLAP_HS2_ENABLE_COORDINATOR.varname + " enabled");
+ }
+ ApplicationId applicationId = coordinator.createExtClientAppId();
+ LOG.info("Generated appID {} for LLAP splits", applicationId.toString());
+
+ PlanFragment fragment = createPlanFragment(query, num, applicationId);
TezWork tezWork = fragment.work;
Schema schema = fragment.schema;
try {
- for (InputSplit s : getSplits(jc, num, tezWork, schema)) {
+ for (InputSplit s : getSplits(jc, num, tezWork, schema, applicationId)) {
Object[] os = new Object[1];
bos.reset();
s.write(dos);
@@ -202,7 +214,7 @@ public class GenericUDTFGetSplits extends GenericUDTF {
}
}
- public PlanFragment createPlanFragment(String query, int num)
+ public PlanFragment createPlanFragment(String query, int num, ApplicationId splitsAppId)
throws HiveException {
HiveConf conf = new HiveConf(SessionState.get().getConf());
@@ -224,7 +236,17 @@ public class GenericUDTFGetSplits extends GenericUDTF {
throw new HiveException(e);
}
- Driver driver = new Driver(conf);
+ // Instantiate Driver to compile the query passed in.
+ // This UDF is running as part of an existing query, which may already be using the
+ // SessionState TxnManager. If this new Driver also tries to use the same TxnManager
+ // then this may mess up the existing state of the TxnManager.
+ // So initialize the new Driver with a new TxnManager so that it does not use the
+ // Session TxnManager that is already in use.
+ HiveTxnManager txnManager = TxnManagerFactory.getTxnManagerFactory().getTxnManager(conf);
+ Driver driver = new Driver(conf, txnManager);
+ driver.init();
+ DriverCleanup driverCleanup = new DriverCleanup(driver, txnManager, splitsAppId.toString());
+ boolean needsCleanup = true;
try {
CommandProcessorResponse cpr = driver.compileAndRespond(query);
if (cpr.getResponseCode() != 0) {
@@ -276,16 +298,38 @@ public class GenericUDTFGetSplits extends GenericUDTF {
}
tezWork = ((TezTask)roots.get(0)).getWork();
+ } else {
+ // Table will be queried directly by LLAP
+ // Acquire locks if necessary - they will be released during session cleanup.
+ // The read will have READ_COMMITTED level semantics.
+ cpr = driver.lockAndRespond();
+ if (cpr.getResponseCode() != 0) {
+ throw new HiveException("Failed to acquire locks: " + cpr.getException());
+ }
+
+ // Attach the resources to the session cleanup.
+ SessionState.get().addCleanupItem(driverCleanup);
+ needsCleanup = false;
}
return new PlanFragment(tezWork, schema, jc);
} finally {
- driver.close();
- driver.destroy();
+ if (needsCleanup) {
+ if (driverCleanup != null) {
+ try {
+ driverCleanup.close();
+ } catch (IOException err) {
+ throw new HiveException(err);
+ }
+ } else if (driver != null) {
+ driver.close();
+ driver.destroy();
+ }
+ }
}
}
- public InputSplit[] getSplits(JobConf job, int numSplits, TezWork work, Schema schema)
+ public InputSplit[] getSplits(JobConf job, int numSplits, TezWork work, Schema schema, ApplicationId applicationId)
throws IOException {
DAG dag = DAG.create(work.getName());
@@ -311,7 +355,6 @@ public class GenericUDTFGetSplits extends GenericUDTF {
// Update the queryId to use the generated applicationId. See comment below about
// why this is done.
- ApplicationId applicationId = coordinator.createExtClientAppId();
HiveConf.setVar(wxConf, HiveConf.ConfVars.HIVEQUERYID, applicationId.toString());
Vertex wx = utils.createVertex(wxConf, mapWork, scratchDir, appJarLr,
new ArrayList<LocalResource>(), fs, ctx, false, work,
@@ -412,6 +455,37 @@ public class GenericUDTFGetSplits extends GenericUDTF {
}
}
+ private static class DriverCleanup implements Closeable {
+ private final Driver driver;
+ private final HiveTxnManager txnManager;
+ private final String applicationId;
+
+ public DriverCleanup(Driver driver, HiveTxnManager txnManager, String applicationId) {
+ this.driver = driver;
+ this.txnManager = txnManager;
+ this.applicationId = applicationId;
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ LOG.info("DriverCleanup for LLAP splits: {}", applicationId);
+ driver.releaseLocksAndCommitOrRollback(true);
+ driver.close();
+ driver.destroy();
+ txnManager.closeTxnManager();
+ } catch (Exception err) {
+ LOG.error("Error closing driver resources", err);
+ throw new IOException(err);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "DriverCleanup for LLAP splits: " + applicationId;
+ }
+ }
+
private static class JobTokenCreator {
private static Token<JobTokenIdentifier> createJobToken(ApplicationId applicationId) {
String tokenIdentifier = applicationId.toString();