You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by ha...@apache.org on 2014/03/05 01:20:57 UTC
svn commit: r1574266 [22/23] - in /hive/trunk:
common/src/java/org/apache/hadoop/hive/conf/
itests/hive-unit/src/test/java/org/apache/hadoop/hive/metastore/ metastore/
metastore/if/ metastore/scripts/upgrade/derby/
metastore/scripts/upgrade/mysql/ meta...
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/Driver.java?rev=1574266&r1=1574265&r2=1574266&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/Driver.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/Driver.java Wed Mar 5 00:20:53 2014
@@ -19,40 +19,18 @@
package org.apache.hadoop.hive.ql;
-import java.io.DataInput;
-import java.io.IOException;
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Queue;
-import java.util.Set;
-import java.util.concurrent.ConcurrentLinkedQueue;
-
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
import org.apache.hadoop.hive.metastore.MetaStoreUtils;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Schema;
-import org.apache.hadoop.hive.ql.exec.ConditionalTask;
-import org.apache.hadoop.hive.ql.exec.FetchTask;
-import org.apache.hadoop.hive.ql.exec.Operator;
-import org.apache.hadoop.hive.ql.exec.TableScanOperator;
-import org.apache.hadoop.hive.ql.exec.Task;
-import org.apache.hadoop.hive.ql.exec.TaskFactory;
-import org.apache.hadoop.hive.ql.exec.TaskResult;
-import org.apache.hadoop.hive.ql.exec.TaskRunner;
-import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.*;
import org.apache.hadoop.hive.ql.history.HiveHistory.Keys;
import org.apache.hadoop.hive.ql.hooks.Entity;
import org.apache.hadoop.hive.ql.hooks.ExecuteWithHookContext;
@@ -63,39 +41,15 @@ import org.apache.hadoop.hive.ql.hooks.P
import org.apache.hadoop.hive.ql.hooks.PreExecute;
import org.apache.hadoop.hive.ql.hooks.ReadEntity;
import org.apache.hadoop.hive.ql.hooks.WriteEntity;
-import org.apache.hadoop.hive.ql.lockmgr.HiveLock;
-import org.apache.hadoop.hive.ql.lockmgr.HiveLockManager;
-import org.apache.hadoop.hive.ql.lockmgr.HiveLockManagerCtx;
-import org.apache.hadoop.hive.ql.lockmgr.HiveLockMode;
-import org.apache.hadoop.hive.ql.lockmgr.HiveLockObj;
-import org.apache.hadoop.hive.ql.lockmgr.HiveLockObject;
+import org.apache.hadoop.hive.ql.lockmgr.*;
import org.apache.hadoop.hive.ql.lockmgr.HiveLockObject.HiveLockObjectData;
-import org.apache.hadoop.hive.ql.lockmgr.LockException;
import org.apache.hadoop.hive.ql.log.PerfLogger;
-import org.apache.hadoop.hive.ql.metadata.AuthorizationException;
-import org.apache.hadoop.hive.ql.metadata.DummyPartition;
-import org.apache.hadoop.hive.ql.metadata.Hive;
-import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.ql.metadata.Partition;
-import org.apache.hadoop.hive.ql.metadata.Table;
+import org.apache.hadoop.hive.ql.metadata.*;
import org.apache.hadoop.hive.ql.metadata.formatting.JsonMetaDataFormatter;
import org.apache.hadoop.hive.ql.metadata.formatting.MetaDataFormatUtils;
import org.apache.hadoop.hive.ql.metadata.formatting.MetaDataFormatter;
import org.apache.hadoop.hive.ql.optimizer.ppr.PartitionPruner;
-import org.apache.hadoop.hive.ql.parse.ASTNode;
-import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer;
-import org.apache.hadoop.hive.ql.parse.HiveSemanticAnalyzerHook;
-import org.apache.hadoop.hive.ql.parse.HiveSemanticAnalyzerHookContext;
-import org.apache.hadoop.hive.ql.parse.HiveSemanticAnalyzerHookContextImpl;
-import org.apache.hadoop.hive.ql.parse.ImportSemanticAnalyzer;
-import org.apache.hadoop.hive.ql.parse.ParseContext;
-import org.apache.hadoop.hive.ql.parse.ParseDriver;
-import org.apache.hadoop.hive.ql.parse.ParseUtils;
-import org.apache.hadoop.hive.ql.parse.PrunedPartitionList;
-import org.apache.hadoop.hive.ql.parse.SemanticAnalyzer;
-import org.apache.hadoop.hive.ql.parse.SemanticAnalyzerFactory;
-import org.apache.hadoop.hive.ql.parse.SemanticException;
-import org.apache.hadoop.hive.ql.parse.VariableSubstitution;
+import org.apache.hadoop.hive.ql.parse.*;
import org.apache.hadoop.hive.ql.plan.HiveOperation;
import org.apache.hadoop.hive.ql.plan.OperatorDesc;
import org.apache.hadoop.hive.ql.plan.TableDesc;
@@ -112,7 +66,13 @@ import org.apache.hadoop.hive.shims.Shim
import org.apache.hadoop.mapred.ClusterStatus;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.util.ReflectionUtils;
+
+import java.io.DataInput;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.*;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
public class Driver implements CommandProcessor {
@@ -130,11 +90,10 @@ public class Driver implements CommandPr
private Context ctx;
private QueryPlan plan;
private Schema schema;
- private HiveLockManager hiveLockMgr;
-
private String errorMessage;
private String SQLState;
private Throwable downstreamError;
+ private HiveTxnManager txnMgr;
// A limit on the number of threads that can be launched
private int maxthreads;
@@ -143,48 +102,29 @@ public class Driver implements CommandPr
private String userName;
+ private void createTxnManager() throws SemanticException {
+ if (txnMgr == null) {
+ try {
+ txnMgr = TxnManagerFactory.getTxnManagerFactory().getTxnManager(conf);
+ } catch (LockException e) {
+ throw new SemanticException(e.getMessage(), e);
+ }
+ }
+ // 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.
+ ctx.setHiveTxnManager(txnMgr);
+ }
+
private boolean checkConcurrency() throws SemanticException {
boolean supportConcurrency = conf.getBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY);
if (!supportConcurrency) {
LOG.info("Concurrency mode is disabled, not creating a lock manager");
return false;
}
- createLockManager();
- // the reason that we set the lock manager for the cxt here is because each
- // query has its own ctx object. The hiveLockMgr is shared accross the
- // same instance of Driver, which can run multiple queries.
- ctx.setHiveLockMgr(hiveLockMgr);
return true;
}
- private void createLockManager() throws SemanticException {
- if (hiveLockMgr != null) {
- return;
- }
- String lockMgr = conf.getVar(HiveConf.ConfVars.HIVE_LOCK_MANAGER);
- LOG.info("Creating lock manager of type " + lockMgr);
- if ((lockMgr == null) || (lockMgr.isEmpty())) {
- throw new SemanticException(ErrorMsg.LOCKMGR_NOT_SPECIFIED.getMsg());
- }
- try {
- hiveLockMgr = (HiveLockManager) ReflectionUtils.newInstance(conf.getClassByName(lockMgr),
- conf);
- hiveLockMgr.setContext(new HiveLockManagerCtx(conf));
- } catch (Exception e1) {
- // set hiveLockMgr to null just in case this invalid manager got set to
- // next query's ctx.
- if (hiveLockMgr != null) {
- try {
- hiveLockMgr.close();
- } catch (LockException e2) {
- //nothing can do here
- }
- hiveLockMgr = null;
- }
- throw new SemanticException(ErrorMsg.LOCKMGR_NOT_INITIALIZED.getMsg() + e1.getMessage(), e1);
- }
- }
-
@Override
public void init() {
Operator.resetId();
@@ -858,25 +798,20 @@ public class Driver implements CommandPr
return locks;
}
- /**
- * Dedup the list of lock objects so that there is only one lock per table/partition.
- * If there is both a shared and exclusive lock for the same object, this will deduped
- * to just a single exclusive lock.
- * @param lockObjects
- */
- static void dedupLockObjects(List<HiveLockObj> lockObjects) {
- Map<String, HiveLockObj> lockMap = new HashMap<String, HiveLockObj>();
- for (HiveLockObj lockObj : lockObjects) {
- String lockName = lockObj.getName();
- HiveLockObj foundLock = lockMap.get(lockName);
- if (foundLock == null || lockObj.getMode() == HiveLockMode.EXCLUSIVE) {
- lockMap.put(lockName, lockObj);
- }
- }
- // copy set of deduped locks back to original list
- lockObjects.clear();
- for (HiveLockObj lockObj : lockMap.values()) {
- lockObjects.add(lockObj);
+ // Write the current set of valid transactions into the conf file so that it can be read by
+ // the input format.
+ private int recordValidTxns() {
+ try {
+ IMetaStoreClient.ValidTxnList txns = txnMgr.getValidTxns();
+ ctx.getConf().set(IMetaStoreClient.ValidTxnList.VALID_TXNS_KEY, txns.toString());
+ return 0;
+ } catch (LockException e) {
+ errorMessage = "FAILED: Error in determing valid transactions: " + e.getMessage();
+ SQLState = ErrorMsg.findSQLState(e.getMessage());
+ downstreamError = e;
+ console.printError(errorMessage, "\n"
+ + org.apache.hadoop.util.StringUtils.stringifyException(e));
+ return 10;
}
}
@@ -886,96 +821,21 @@ public class Driver implements CommandPr
* pretty simple. If all the locks cannot be obtained, error out. Deadlock is avoided by making
* sure that the locks are lexicographically sorted.
**/
- public int acquireReadWriteLocks() {
+ private int acquireReadWriteLocks() {
PerfLogger perfLogger = PerfLogger.getPerfLogger();
perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
- try {
- boolean supportConcurrency = conf.getBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY);
- if (!supportConcurrency) {
- return 0;
- }
-
- List<HiveLockObj> lockObjects = new ArrayList<HiveLockObj>();
-
- // Sort all the inputs, outputs.
- // If a lock needs to be acquired on any partition, a read lock needs to be acquired on all
- // its parents also
- for (ReadEntity input : plan.getInputs()) {
- if (input.getType() == ReadEntity.Type.DATABASE) {
- lockObjects.addAll(getLockObjects(input.getDatabase(), null, null, HiveLockMode.SHARED));
- } else if (input.getType() == ReadEntity.Type.TABLE) {
- lockObjects.addAll(getLockObjects(null, input.getTable(), null, HiveLockMode.SHARED));
- } else {
- lockObjects.addAll(getLockObjects(null, null, input.getPartition(), HiveLockMode.SHARED));
- }
- }
-
- for (WriteEntity output : plan.getOutputs()) {
- List<HiveLockObj> lockObj = null;
- if (output.getType() == WriteEntity.Type.DATABASE) {
- lockObjects.addAll(getLockObjects(output.getDatabase(), null, null,
- output.isComplete() ? HiveLockMode.EXCLUSIVE : HiveLockMode.SHARED));
- } else if (output.getTyp() == WriteEntity.Type.TABLE) {
- lockObj = getLockObjects(null, output.getTable(), null,
- output.isComplete() ? HiveLockMode.EXCLUSIVE : HiveLockMode.SHARED);
- } else if (output.getTyp() == WriteEntity.Type.PARTITION) {
- lockObj = getLockObjects(null, null, output.getPartition(), HiveLockMode.EXCLUSIVE);
- }
- // In case of dynamic queries, it is possible to have incomplete dummy partitions
- else if (output.getTyp() == WriteEntity.Type.DUMMYPARTITION) {
- lockObj = getLockObjects(null, null, output.getPartition(), HiveLockMode.SHARED);
- }
-
- if(lockObj != null) {
- lockObjects.addAll(lockObj);
- ctx.getOutputLockObjects().put(output, lockObj);
- }
- }
-
- if (lockObjects.isEmpty() && !ctx.isNeedLockMgr()) {
- return 0;
- }
-
- HiveLockObjectData lockData =
- new HiveLockObjectData(plan.getQueryId(),
- String.valueOf(System.currentTimeMillis()),
- "IMPLICIT",
- plan.getQueryStr());
-
- // Lock the database also
- String currentDb = SessionState.get().getCurrentDatabase();
- lockObjects.add(
- new HiveLockObj(
- new HiveLockObject(currentDb, lockData),
- HiveLockMode.SHARED
- )
- );
-
- dedupLockObjects(lockObjects);
- List<HiveLock> hiveLocks = ctx.getHiveLockMgr().lock(lockObjects, false);
- if (hiveLocks == null) {
- throw new SemanticException(ErrorMsg.LOCK_CANNOT_BE_ACQUIRED.getMsg());
- } else {
- ctx.setHiveLocks(hiveLocks);
- }
-
- return (0);
- } catch (SemanticException e) {
- errorMessage = "FAILED: Error in acquiring locks: " + e.getMessage();
- SQLState = ErrorMsg.findSQLState(e.getMessage());
- downstreamError = e;
- console.printError(errorMessage, "\n"
- + org.apache.hadoop.util.StringUtils.stringifyException(e));
- return (10);
+ try {
+ txnMgr.acquireLocks(plan, ctx, userName);
+ return 0;
} catch (LockException e) {
errorMessage = "FAILED: Error in acquiring locks: " + e.getMessage();
SQLState = ErrorMsg.findSQLState(e.getMessage());
downstreamError = e;
console.printError(errorMessage, "\n"
+ org.apache.hadoop.util.StringUtils.stringifyException(e));
- return (10);
+ return 10;
} finally {
perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
}
@@ -986,12 +846,12 @@ public class Driver implements CommandPr
* list of hive locks to be released Release all the locks specified. If some of the
* locks have already been released, ignore them
**/
- private void releaseLocks(List<HiveLock> hiveLocks) {
+ private void releaseLocks(List<HiveLock> hiveLocks) throws LockException {
PerfLogger perfLogger = PerfLogger.getPerfLogger();
perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.RELEASE_LOCKS);
if (hiveLocks != null) {
- ctx.getHiveLockMgr().releaseLocks(hiveLocks);
+ ctx.getHiveTxnManager().getLockManager().releaseLocks(hiveLocks);
}
ctx.setHiveLocks(null);
@@ -1078,7 +938,12 @@ public class Driver implements CommandPr
ret = compile(command);
}
if (ret != 0) {
- releaseLocks(ctx.getHiveLocks());
+ try {
+ releaseLocks(ctx.getHiveLocks());
+ } catch (LockException e) {
+ LOG.warn("Exception in releasing locks. "
+ + org.apache.hadoop.util.StringUtils.stringifyException(e));
+ }
}
return ret;
}
@@ -1128,6 +993,7 @@ public class Driver implements CommandPr
boolean ckLock = false;
try {
ckLock = checkConcurrency();
+ createTxnManager();
} catch (SemanticException e) {
errorMessage = "FAILED: Error in semantic analysis: " + e.getMessage();
SQLState = ErrorMsg.findSQLState(e.getMessage());
@@ -1163,10 +1029,17 @@ public class Driver implements CommandPr
}
}
+ ret = recordValidTxns();
+ if (ret != 0) return new CommandProcessorResponse(ret, errorMessage, SQLState);
+
if (requireLock) {
ret = acquireReadWriteLocks();
if (ret != 0) {
- releaseLocks(ctx.getHiveLocks());
+ try {
+ releaseLocks(ctx.getHiveLocks());
+ } catch (LockException e) {
+ // Not much to do here
+ }
return new CommandProcessorResponse(ret, errorMessage, SQLState);
}
}
@@ -1174,12 +1047,25 @@ public class Driver implements CommandPr
ret = execute();
if (ret != 0) {
//if needRequireLock is false, the release here will do nothing because there is no lock
- releaseLocks(ctx.getHiveLocks());
+ try {
+ releaseLocks(ctx.getHiveLocks());
+ } catch (LockException e) {
+ // Nothing to do here
+ }
return new CommandProcessorResponse(ret, errorMessage, SQLState);
}
//if needRequireLock is false, the release here will do nothing because there is no lock
- releaseLocks(ctx.getHiveLocks());
+ try {
+ releaseLocks(ctx.getHiveLocks());
+ } catch (LockException e) {
+ errorMessage = "FAILED: Hive Internal Error: " + Utilities.getNameMessage(e);
+ SQLState = ErrorMsg.findSQLState(e.getMessage());
+ downstreamError = e;
+ console.printError(errorMessage + "\n"
+ + org.apache.hadoop.util.StringUtils.stringifyException(e));
+ return new CommandProcessorResponse(12, errorMessage, SQLState);
+ }
perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.DRIVER_RUN);
perfLogger.close(LOG, plan);
@@ -1731,17 +1617,16 @@ public class Driver implements CommandPr
public void destroy() {
if (ctx != null) {
- releaseLocks(ctx.getHiveLocks());
- }
-
- if (hiveLockMgr != null) {
try {
- hiveLockMgr.close();
- } catch(LockException e) {
- LOG.warn("Exception in closing hive lock manager. "
- + org.apache.hadoop.util.StringUtils.stringifyException(e));
+ releaseLocks(ctx.getHiveLocks());
+ } catch (LockException e) {
+ LOG.warn("Exception when releasing locking in destroy: " +
+ e.getMessage());
}
}
+ if (txnMgr != null) {
+ txnMgr.closeTxnManager();
+ }
}
public org.apache.hadoop.hive.ql.plan.api.Query getQueryPlan() throws IOException {
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java?rev=1574266&r1=1574265&r2=1574266&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java Wed Mar 5 00:20:53 2014
@@ -18,6 +18,11 @@
package org.apache.hadoop.hive.ql;
+import org.antlr.runtime.tree.Tree;
+import org.apache.hadoop.hive.ql.metadata.HiveUtils;
+import org.apache.hadoop.hive.ql.parse.ASTNode;
+import org.apache.hadoop.hive.ql.parse.ASTNodeOrigin;
+
import java.text.MessageFormat;
import java.util.HashMap;
import java.util.Map;
@@ -371,12 +376,35 @@ public enum ErrorMsg {
INVALID_HDFS_URI(10251, "{0} is not a hdfs uri", true),
INVALID_DIR(10252, "{0} is not a directory", true),
NO_VALID_LOCATIONS(10253, "Could not find any valid location to place the jars. " +
- "Please update hive.jar.directory or hive.user.install.directory with a valid location", false),
+ "Please update hive.jar.directory or hive.user.install.directory with a valid location", false),
UNNSUPPORTED_AUTHORIZATION_PRINCIPAL_TYPE_GROUP(10254,
"Principal type GROUP is not supported in this authorization setting", "28000"),
INVALID_TABLE_NAME(10255, "Invalid table name {0}", true),
INSERT_INTO_IMMUTABLE_TABLE(10256, "Inserting into a non-empty immutable table is not allowed"),
+ TXNMGR_NOT_SPECIFIED(10260, "Transaction manager not specified correctly, " +
+ "set hive.txn.manager"),
+ TXNMGR_NOT_INSTANTIATED(10261, "Transaction manager could not be " +
+ "instantiated, check hive.txn.manager"),
+ TXN_NO_SUCH_TRANSACTION(10262, "No record of transaction could be found, " +
+ "may have timed out"),
+ TXN_ABORTED(10263, "Transaction manager has aborted the transaction."),
+
+ LOCK_NO_SUCH_LOCK(10270, "No record of lock could be found, " +
+ "may have timed out"),
+ LOCK_REQUEST_UNSUPPORTED(10271, "Current transaction manager does not " +
+ "support explicit lock requests. Transaction manager: "),
+
+ METASTORE_COMMUNICATION_FAILED(10280, "Error communicating with the " +
+ "metastore"),
+ METASTORE_COULD_NOT_INITIATE(10281, "Unable to initiate connection to the " +
+ "metastore."),
+ INVALID_COMPACTION_TYPE(10282, "Invalid compaction type, supported values are 'major' and " +
+ "'minor'"),
+ NO_COMPACTION_PARTITION(10283, "You must specify a partition to compact for partitioned tables"),
+ TOO_MANY_COMPACTION_PARTITIONS(10284, "Compaction can only be requested on one partition at a " +
+ "time."),
+
//========================== 20000 range starts here ========================//
SCRIPT_INIT_ERROR(20000, "Unable to initialize custom script."),
SCRIPT_IO_ERROR(20001, "An error occurred while reading or writing to your custom script. "
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java?rev=1574266&r1=1574265&r2=1574266&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java Wed Mar 5 00:20:53 2014
@@ -47,11 +47,8 @@ import org.apache.commons.lang.StringEsc
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.FsShell;
-import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.type.HiveDecimal;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
@@ -59,24 +56,7 @@ import org.apache.hadoop.hive.metastore.
import org.apache.hadoop.hive.metastore.ProtectMode;
import org.apache.hadoop.hive.metastore.TableType;
import org.apache.hadoop.hive.metastore.Warehouse;
-import org.apache.hadoop.hive.metastore.api.AlreadyExistsException;
-import org.apache.hadoop.hive.metastore.api.Database;
-import org.apache.hadoop.hive.metastore.api.FieldSchema;
-import org.apache.hadoop.hive.metastore.api.HiveObjectPrivilege;
-import org.apache.hadoop.hive.metastore.api.HiveObjectRef;
-import org.apache.hadoop.hive.metastore.api.HiveObjectType;
-import org.apache.hadoop.hive.metastore.api.Index;
-import org.apache.hadoop.hive.metastore.api.InvalidOperationException;
-import org.apache.hadoop.hive.metastore.api.MetaException;
-import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
-import org.apache.hadoop.hive.metastore.api.Order;
-import org.apache.hadoop.hive.metastore.api.PrincipalType;
-import org.apache.hadoop.hive.metastore.api.PrivilegeBag;
-import org.apache.hadoop.hive.metastore.api.PrivilegeGrantInfo;
-import org.apache.hadoop.hive.metastore.api.Role;
-import org.apache.hadoop.hive.metastore.api.SerDeInfo;
-import org.apache.hadoop.hive.metastore.api.SkewedInfo;
-import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.*;
import org.apache.hadoop.hive.ql.Context;
import org.apache.hadoop.hive.ql.DriverContext;
import org.apache.hadoop.hive.ql.ErrorMsg;
@@ -88,29 +68,16 @@ import org.apache.hadoop.hive.ql.io.rcfi
import org.apache.hadoop.hive.ql.io.rcfile.merge.MergeWork;
import org.apache.hadoop.hive.ql.io.rcfile.truncate.ColumnTruncateTask;
import org.apache.hadoop.hive.ql.io.rcfile.truncate.ColumnTruncateWork;
-import org.apache.hadoop.hive.ql.lockmgr.HiveLock;
-import org.apache.hadoop.hive.ql.lockmgr.HiveLockManager;
-import org.apache.hadoop.hive.ql.lockmgr.HiveLockMode;
-import org.apache.hadoop.hive.ql.lockmgr.HiveLockObject;
+import org.apache.hadoop.hive.ql.lockmgr.*;
import org.apache.hadoop.hive.ql.lockmgr.HiveLockObject.HiveLockObjectData;
-import org.apache.hadoop.hive.ql.metadata.CheckResult;
-import org.apache.hadoop.hive.ql.metadata.Hive;
-import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.ql.metadata.HiveMetaStoreChecker;
-import org.apache.hadoop.hive.ql.metadata.HiveStorageHandler;
-import org.apache.hadoop.hive.ql.metadata.HiveUtils;
-import org.apache.hadoop.hive.ql.metadata.InvalidTableException;
+import org.apache.hadoop.hive.ql.metadata.*;
import org.apache.hadoop.hive.ql.metadata.Partition;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.metadata.formatting.MetaDataFormatUtils;
import org.apache.hadoop.hive.ql.metadata.formatting.MetaDataFormatter;
import org.apache.hadoop.hive.ql.parse.AlterTablePartMergeFilesDesc;
import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer;
-import org.apache.hadoop.hive.ql.plan.AddPartitionDesc;
-import org.apache.hadoop.hive.ql.plan.AlterDatabaseDesc;
-import org.apache.hadoop.hive.ql.plan.AlterIndexDesc;
-import org.apache.hadoop.hive.ql.plan.AlterTableAlterPartDesc;
-import org.apache.hadoop.hive.ql.plan.AlterTableDesc;
+import org.apache.hadoop.hive.ql.plan.*;
import org.apache.hadoop.hive.ql.plan.AlterTableDesc.AlterTableTypes;
import org.apache.hadoop.hive.ql.plan.AlterTableExchangePartition;
import org.apache.hadoop.hive.ql.plan.AlterTableSimpleDesc;
@@ -1069,7 +1036,7 @@ public class DDLTask extends Task<DDLWor
MetaStoreUtils.getIndexTableName(SessionState.get().getCurrentDatabase(),
crtIndex.getTableName(), crtIndex.getIndexName());
Table indexTable = db.getTable(indexTableName);
- work.getOutputs().add(new WriteEntity(indexTable));
+ work.getOutputs().add(new WriteEntity(indexTable, WriteEntity.WriteType.DDL));
}
return 0;
}
@@ -1162,7 +1129,7 @@ public class DDLTask extends Task<DDLWor
private int addPartitions(Hive db, AddPartitionDesc addPartitionDesc) throws HiveException {
List<Partition> parts = db.createPartitions(addPartitionDesc);
for (Partition part : parts) {
- work.getOutputs().add(new WriteEntity(part));
+ work.getOutputs().add(new WriteEntity(part, WriteEntity.WriteType.INSERT));
}
return 0;
}
@@ -1188,7 +1155,7 @@ public class DDLTask extends Task<DDLWor
Partition newPart = db
.getPartition(tbl, renamePartitionDesc.getNewPartSpec(), false);
work.getInputs().add(new ReadEntity(oldPart));
- work.getOutputs().add(new WriteEntity(newPart));
+ work.getOutputs().add(new WriteEntity(newPart, WriteEntity.WriteType.DDL));
return 0;
}
@@ -1230,7 +1197,7 @@ public class DDLTask extends Task<DDLWor
}
work.getInputs().add(new ReadEntity(tbl));
- work.getOutputs().add(new WriteEntity(tbl));
+ work.getOutputs().add(new WriteEntity(tbl, WriteEntity.WriteType.DDL));
return 0;
}
@@ -1259,7 +1226,7 @@ public class DDLTask extends Task<DDLWor
throw new HiveException("Uable to update table");
}
work.getInputs().add(new ReadEntity(tbl));
- work.getOutputs().add(new WriteEntity(tbl));
+ work.getOutputs().add(new WriteEntity(tbl, WriteEntity.WriteType.DDL_METADATA_ONLY));
} else {
Partition part = db.getPartition(tbl, touchDesc.getPartSpec(), false);
if (part == null) {
@@ -1271,7 +1238,7 @@ public class DDLTask extends Task<DDLWor
throw new HiveException(e);
}
work.getInputs().add(new ReadEntity(part));
- work.getOutputs().add(new WriteEntity(part));
+ work.getOutputs().add(new WriteEntity(part, WriteEntity.WriteType.DDL_METADATA_ONLY));
}
return 0;
}
@@ -2514,7 +2481,11 @@ public class DDLTask extends Task<DDLWor
*/
private int showLocks(ShowLocksDesc showLocks) throws HiveException {
Context ctx = driverContext.getCtx();
- HiveLockManager lockMgr = ctx.getHiveLockMgr();
+ HiveTxnManager txnManager = ctx.getHiveTxnManager();
+ HiveLockManager lockMgr = txnManager.getLockManager();
+
+ if (txnManager.useNewShowLocksFormat()) return showLocksNewFormat(showLocks, lockMgr);
+
boolean isExt = showLocks.isExt();
if (lockMgr == null) {
throw new HiveException("show Locks LockManager not specified");
@@ -2529,9 +2500,12 @@ public class DDLTask extends Task<DDLWor
List<HiveLock> locks = null;
if (showLocks.getTableName() == null) {
+ // TODO should be doing security check here. Users should not be
+ // able to see each other's locks.
locks = lockMgr.getLocks(false, isExt);
}
else {
+ // TODO make this work
locks = lockMgr.getLocks(getHiveObject(showLocks.getTableName(),
showLocks.getPartSpec()),
true, isExt);
@@ -2595,7 +2569,95 @@ public class DDLTask extends Task<DDLWor
return 0;
}
- /**
+ private int showLocksNewFormat(ShowLocksDesc showLocks, HiveLockManager lm)
+ throws HiveException {
+
+ DbLockManager lockMgr;
+ if (!(lm instanceof DbLockManager)) {
+ throw new RuntimeException("New lock format only supported with db lock manager.");
+ }
+ lockMgr = (DbLockManager)lm;
+
+ ShowLocksResponse rsp = lockMgr.getLocks();
+
+ // write the results in the file
+ DataOutputStream os = null;
+ try {
+ Path resFile = new Path(showLocks.getResFile());
+ FileSystem fs = resFile.getFileSystem(conf);
+ os = fs.create(resFile);
+
+ // Write a header
+ os.writeBytes("Lock ID");
+ os.write(separator);
+ os.writeBytes("Database");
+ os.write(separator);
+ os.writeBytes("Table");
+ os.write(separator);
+ os.writeBytes("Partition");
+ os.write(separator);
+ os.writeBytes("State");
+ os.write(separator);
+ os.writeBytes("Type");
+ os.write(separator);
+ os.writeBytes("Transaction ID");
+ os.write(separator);
+ os.writeBytes("Last Hearbeat");
+ os.write(separator);
+ os.writeBytes("Acquired At");
+ os.write(separator);
+ os.writeBytes("User");
+ os.write(separator);
+ os.writeBytes("Hostname");
+ os.write(terminator);
+
+ List<ShowLocksResponseElement> locks = rsp.getLocks();
+ if (locks != null) {
+ for (ShowLocksResponseElement lock : locks) {
+ os.writeBytes(Long.toString(lock.getLockid()));
+ os.write(separator);
+ os.writeBytes(lock.getDbname());
+ os.write(separator);
+ os.writeBytes((lock.getTablename() == null) ? "NULL" : lock.getTablename());
+ os.write(separator);
+ os.writeBytes((lock.getPartname() == null) ? "NULL" : lock.getPartname());
+ os.write(separator);
+ os.writeBytes(lock.getState().toString());
+ os.write(separator);
+ os.writeBytes(lock.getType().toString());
+ os.write(separator);
+ os.writeBytes((lock.getTxnid() == 0) ? "NULL" : Long.toString(lock.getTxnid()));
+ os.write(separator);
+ os.writeBytes(Long.toString(lock.getLastheartbeat()));
+ os.write(separator);
+ os.writeBytes((lock.getAcquiredat() == 0) ? "NULL" : Long.toString(lock.getAcquiredat()));
+ os.write(separator);
+ os.writeBytes(lock.getUser());
+ os.write(separator);
+ os.writeBytes(lock.getHostname());
+ os.write(separator);
+ os.write(terminator);
+ }
+
+ }
+
+ os.close();
+ os = null;
+ } catch (FileNotFoundException e) {
+ LOG.warn("show function: " + stringifyException(e));
+ return 1;
+ } catch (IOException e) {
+ LOG.warn("show function: " + stringifyException(e));
+ return 1;
+ } catch (Exception e) {
+ throw new HiveException(e.toString());
+ } finally {
+ IOUtils.closeStream((FSDataOutputStream) os);
+ }
+ return 0;
+ }
+
+ /**
* Lock the table/partition specified
*
* @param lockTbl
@@ -2606,7 +2668,12 @@ public class DDLTask extends Task<DDLWor
*/
private int lockTable(LockTableDesc lockTbl) throws HiveException {
Context ctx = driverContext.getCtx();
- HiveLockManager lockMgr = ctx.getHiveLockMgr();
+ HiveTxnManager txnManager = ctx.getHiveTxnManager();
+ if (!txnManager.supportsExplicitLock()) {
+ throw new HiveException(ErrorMsg.LOCK_REQUEST_UNSUPPORTED,
+ conf.getVar(HiveConf.ConfVars.HIVE_TXN_MANAGER));
+ }
+ HiveLockManager lockMgr = txnManager.getLockManager();
if (lockMgr == null) {
throw new HiveException("lock Table LockManager not specified");
}
@@ -2655,7 +2722,12 @@ public class DDLTask extends Task<DDLWor
*/
private int lockDatabase(LockDatabaseDesc lockDb) throws HiveException {
Context ctx = driverContext.getCtx();
- HiveLockManager lockMgr = ctx.getHiveLockMgr();
+ HiveTxnManager txnManager = ctx.getHiveTxnManager();
+ if (!txnManager.supportsExplicitLock()) {
+ throw new HiveException(ErrorMsg.LOCK_REQUEST_UNSUPPORTED,
+ conf.getVar(HiveConf.ConfVars.HIVE_TXN_MANAGER));
+ }
+ HiveLockManager lockMgr = txnManager.getLockManager();
if (lockMgr == null) {
throw new HiveException("lock Database LockManager not specified");
}
@@ -2691,7 +2763,12 @@ public class DDLTask extends Task<DDLWor
*/
private int unlockDatabase(UnlockDatabaseDesc unlockDb) throws HiveException {
Context ctx = driverContext.getCtx();
- HiveLockManager lockMgr = ctx.getHiveLockMgr();
+ HiveTxnManager txnManager = ctx.getHiveTxnManager();
+ if (!txnManager.supportsExplicitLock()) {
+ throw new HiveException(ErrorMsg.LOCK_REQUEST_UNSUPPORTED,
+ conf.getVar(HiveConf.ConfVars.HIVE_TXN_MANAGER));
+ }
+ HiveLockManager lockMgr = txnManager.getLockManager();
if (lockMgr == null) {
throw new HiveException("unlock Database LockManager not specified");
}
@@ -2749,7 +2826,12 @@ public class DDLTask extends Task<DDLWor
*/
private int unlockTable(UnlockTableDesc unlockTbl) throws HiveException {
Context ctx = driverContext.getCtx();
- HiveLockManager lockMgr = ctx.getHiveLockMgr();
+ HiveTxnManager txnManager = ctx.getHiveTxnManager();
+ if (!txnManager.supportsExplicitLock()) {
+ throw new HiveException(ErrorMsg.LOCK_REQUEST_UNSUPPORTED,
+ conf.getVar(HiveConf.ConfVars.HIVE_TXN_MANAGER));
+ }
+ HiveLockManager lockMgr = txnManager.getLockManager();
if (lockMgr == null) {
throw new HiveException("unlock Table LockManager not specified");
}
@@ -3496,7 +3578,7 @@ public class DDLTask extends Task<DDLWor
}
tbl.setNumBuckets(alterTbl.getNumberBuckets());
}
- } else {
+ } else {
throw new HiveException(ErrorMsg.UNSUPPORTED_ALTER_TBL_OP, alterTbl.getOp().toString());
}
@@ -3533,17 +3615,17 @@ public class DDLTask extends Task<DDLWor
// passed
if(part != null) {
work.getInputs().add(new ReadEntity(part));
- work.getOutputs().add(new WriteEntity(part));
+ work.getOutputs().add(new WriteEntity(part, WriteEntity.WriteType.DDL));
}
else if (allPartitions != null ){
for (Partition tmpPart: allPartitions) {
work.getInputs().add(new ReadEntity(tmpPart));
- work.getOutputs().add(new WriteEntity(tmpPart));
+ work.getOutputs().add(new WriteEntity(tmpPart, WriteEntity.WriteType.DDL));
}
}
else {
work.getInputs().add(new ReadEntity(oldTbl));
- work.getOutputs().add(new WriteEntity(tbl));
+ work.getOutputs().add(new WriteEntity(tbl, WriteEntity.WriteType.DDL));
}
return 0;
}
@@ -3581,7 +3663,7 @@ public class DDLTask extends Task<DDLWor
dropTbl.getPartSpecs(), true, dropTbl.getIgnoreProtection(), true);
for (Partition partition : droppedParts) {
console.printInfo("Dropped the partition " + partition.getName());
- work.getOutputs().add(new WriteEntity(partition));
+ work.getOutputs().add(new WriteEntity(partition, WriteEntity.WriteType.DDL));
};
}
@@ -3635,7 +3717,7 @@ public class DDLTask extends Task<DDLWor
// drop the table
db.dropTable(dropTbl.getTableName());
if (tbl != null) {
- work.getOutputs().add(new WriteEntity(tbl));
+ work.getOutputs().add(new WriteEntity(tbl, WriteEntity.WriteType.DDL));
}
}
@@ -3910,7 +3992,7 @@ public class DDLTask extends Task<DDLWor
// create the table
db.createTable(tbl, crtTbl.getIfNotExists());
- work.getOutputs().add(new WriteEntity(tbl));
+ work.getOutputs().add(new WriteEntity(tbl, WriteEntity.WriteType.DDL));
return 0;
}
@@ -4018,7 +4100,7 @@ public class DDLTask extends Task<DDLWor
// create the table
db.createTable(tbl, crtTbl.getIfNotExists());
- work.getOutputs().add(new WriteEntity(tbl));
+ work.getOutputs().add(new WriteEntity(tbl, WriteEntity.WriteType.DDL));
return 0;
}
@@ -4054,7 +4136,7 @@ public class DDLTask extends Task<DDLWor
} catch (InvalidOperationException e) {
throw new HiveException(e);
}
- work.getOutputs().add(new WriteEntity(oldview));
+ work.getOutputs().add(new WriteEntity(oldview, WriteEntity.WriteType.DDL));
} else {
// create new view
Table tbl = db.newTable(crtView.getViewName());
@@ -4081,7 +4163,7 @@ public class DDLTask extends Task<DDLWor
}
db.createTable(tbl, crtView.getIfNotExists());
- work.getOutputs().add(new WriteEntity(tbl));
+ work.getOutputs().add(new WriteEntity(tbl, WriteEntity.WriteType.DDL));
}
return 0;
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java?rev=1574266&r1=1574265&r2=1574266&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java Wed Mar 5 00:20:53 2014
@@ -18,16 +18,6 @@
package org.apache.hadoop.hive.ql.exec;
-import java.io.IOException;
-import java.io.Serializable;
-import java.security.AccessControlException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileStatus;
@@ -56,17 +46,16 @@ import org.apache.hadoop.hive.ql.metadat
import org.apache.hadoop.hive.ql.optimizer.physical.BucketingSortingCtx.BucketCol;
import org.apache.hadoop.hive.ql.optimizer.physical.BucketingSortingCtx.SortCol;
import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer;
-import org.apache.hadoop.hive.ql.plan.DynamicPartitionCtx;
-import org.apache.hadoop.hive.ql.plan.LoadFileDesc;
-import org.apache.hadoop.hive.ql.plan.LoadMultiFilesDesc;
-import org.apache.hadoop.hive.ql.plan.LoadTableDesc;
-import org.apache.hadoop.hive.ql.plan.MapWork;
-import org.apache.hadoop.hive.ql.plan.MapredWork;
-import org.apache.hadoop.hive.ql.plan.MoveWork;
+import org.apache.hadoop.hive.ql.plan.*;
import org.apache.hadoop.hive.ql.plan.api.StageType;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.util.StringUtils;
+import java.io.IOException;
+import java.io.Serializable;
+import java.security.AccessControlException;
+import java.util.*;
+
/**
* MoveTask implementation.
**/
@@ -175,7 +164,7 @@ public class MoveTask extends Task<MoveW
}
Context ctx = driverContext.getCtx();
- HiveLockManager lockMgr = ctx.getHiveLockMgr();
+ HiveLockManager lockMgr = ctx.getHiveTxnManager().getLockManager();
WriteEntity output = ctx.getLoadTableOutputMap().get(ltd);
List<HiveLockObj> lockObjects = ctx.getOutputLockObjects().get(output);
if (lockObjects == null) {
@@ -284,7 +273,9 @@ public class MoveTask extends Task<MoveW
db.loadTable(tbd.getSourcePath(), tbd.getTable()
.getTableName(), tbd.getReplace(), tbd.getHoldDDLTime());
if (work.getOutputs() != null) {
- work.getOutputs().add(new WriteEntity(table));
+ work.getOutputs().add(new WriteEntity(table,
+ (tbd.getReplace() ? WriteEntity.WriteType.INSERT_OVERWRITE :
+ WriteEntity.WriteType.INSERT)));
}
} else {
LOG.info("Partition is: " + tbd.getPartitionSpec().toString());
@@ -376,7 +367,9 @@ public class MoveTask extends Task<MoveW
updatePartitionBucketSortColumns(table, partn, bucketCols, numBuckets, sortCols);
}
- WriteEntity enty = new WriteEntity(partn);
+ WriteEntity enty = new WriteEntity(partn,
+ (tbd.getReplace() ? WriteEntity.WriteType.INSERT_OVERWRITE :
+ WriteEntity.WriteType.INSERT));
if (work.getOutputs() != null) {
work.getOutputs().add(enty);
}
@@ -417,7 +410,9 @@ public class MoveTask extends Task<MoveW
dc = new DataContainer(table.getTTable(), partn.getTPartition());
// add this partition to post-execution hook
if (work.getOutputs() != null) {
- work.getOutputs().add(new WriteEntity(partn));
+ work.getOutputs().add(new WriteEntity(partn,
+ (tbd.getReplace() ? WriteEntity.WriteType.INSERT_OVERWRITE
+ : WriteEntity.WriteType.INSERT)));
}
}
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java?rev=1574266&r1=1574265&r2=1574266&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java Wed Mar 5 00:20:53 2014
@@ -2099,6 +2099,9 @@ public final class Utilities {
for (Map.Entry<String, String> entry : jobProperties.entrySet()) {
job.set(entry.getKey(), entry.getValue());
}
+ // copy the bucket count
+ job.set(hive_metastoreConstants.BUCKET_COUNT,
+ tbl.getProperties().getProperty(hive_metastoreConstants.BUCKET_COUNT));
}
private static final Object INPUT_SUMMARY_LOCK = new Object();
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/hooks/WriteEntity.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/hooks/WriteEntity.java?rev=1574266&r1=1574265&r2=1574266&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/hooks/WriteEntity.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/hooks/WriteEntity.java Wed Mar 5 00:20:53 2014
@@ -18,14 +18,14 @@
package org.apache.hadoop.hive.ql.hooks;
-import java.io.Serializable;
-
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.ql.metadata.DummyPartition;
import org.apache.hadoop.hive.ql.metadata.Partition;
import org.apache.hadoop.hive.ql.metadata.Table;
+import java.io.Serializable;
+
/**
* This class encapsulates an object that is being written to by the query. This
* object may be a table, partition, dfs directory or a local directory.
@@ -34,6 +34,18 @@ public class WriteEntity extends Entity
private boolean isTempURI = false;
+ public static enum WriteType {
+ DDL, // for use in DDL statements that will touch data,
+ // will result in an exclusive lock,
+ DDL_METADATA_ONLY, // for use in DDL statements that touch only
+ // metadata and don't need a lock
+ INSERT,
+ INSERT_OVERWRITE,
+ UPDATE,
+ DELETE};
+
+ private WriteType writeType;
+
/**
* Only used by serialization.
*/
@@ -41,8 +53,9 @@ public class WriteEntity extends Entity
super();
}
- public WriteEntity(Database database) {
+ public WriteEntity(Database database, WriteType type) {
super(database, true);
+ writeType = type;
}
/**
@@ -51,12 +64,14 @@ public class WriteEntity extends Entity
* @param t
* Table that is written to.
*/
- public WriteEntity(Table t) {
- this(t, true);
+ public WriteEntity(Table t, WriteType type) {
+ super(t, true);
+ writeType = type;
}
- public WriteEntity(Table t, boolean complete) {
+ public WriteEntity(Table t, WriteType type, boolean complete) {
super(t, complete);
+ writeType = type;
}
/**
@@ -65,12 +80,14 @@ public class WriteEntity extends Entity
* @param p
* Partition that is written to.
*/
- public WriteEntity(Partition p) {
+ public WriteEntity(Partition p, WriteType type) {
super(p, true);
+ writeType = type;
}
- public WriteEntity(DummyPartition p, boolean complete) {
+ public WriteEntity(DummyPartition p, WriteType type, boolean complete) {
super(p, complete);
+ writeType = type;
}
/**
@@ -101,6 +118,15 @@ public class WriteEntity extends Entity
}
/**
+ * Determine which type of write this is. This is needed by the lock
+ * manager so it can understand what kind of lock to acquire.
+ * @return write type
+ */
+ public WriteType getWriteType() {
+ return writeType;
+ }
+
+ /**
* Equals function.
*/
@Override
Added: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbLockManager.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbLockManager.java?rev=1574266&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbLockManager.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbLockManager.java Wed Mar 5 00:20:53 2014
@@ -0,0 +1,220 @@
+/**
+ * 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.lockmgr;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.*;
+import org.apache.hadoop.hive.ql.ErrorMsg;
+import org.apache.thrift.TException;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * An implementation of HiveLockManager for use with {@link org.apache.hadoop.hive.ql.lockmgr.DbTxnManager}.
+ * Note, this lock manager is not meant to stand alone. It cannot be used
+ * without the DbTxnManager.
+ */
+public class DbLockManager implements HiveLockManager{
+
+ static final private String CLASS_NAME = DbLockManager.class.getName();
+ static final private Log LOG = LogFactory.getLog(CLASS_NAME);
+
+ private static final long MAX_SLEEP = 15000;
+ private HiveLockManagerCtx context;
+ private Set<DbHiveLock> locks;
+ private HiveMetaStoreClient client;
+ private long nextSleep = 50;
+
+ DbLockManager(HiveMetaStoreClient client) {
+ locks = new HashSet<DbHiveLock>();
+ this.client = client;
+ }
+
+ @Override
+ public void setContext(HiveLockManagerCtx ctx) throws LockException {
+ context = ctx;
+ }
+
+ @Override
+ public HiveLock lock(HiveLockObject key, HiveLockMode mode,
+ boolean keepAlive) throws LockException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public List<HiveLock> lock(List<HiveLockObj> objs, boolean keepAlive) throws
+ LockException {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * Send a lock request to the metastore. This is intended for use by
+ * {@link DbTxnManager}.
+ * @param lock lock request
+ * @throws LockException
+ */
+ List<HiveLock> lock(LockRequest lock) throws LockException {
+ try {
+ LOG.debug("Requesting lock");
+ LockResponse res = client.lock(lock);
+ while (res.getState() == LockState.WAITING) {
+ backoff();
+ res = client.checkLock(res.getLockid());
+
+ }
+ DbHiveLock hl = new DbHiveLock(res.getLockid());
+ locks.add(hl);
+ if (res.getState() != LockState.ACQUIRED) {
+ throw new LockException(ErrorMsg.LOCK_CANNOT_BE_ACQUIRED.getMsg());
+ }
+ List<HiveLock> locks = new ArrayList<HiveLock>(1);
+ locks.add(hl);
+ return locks;
+ } catch (NoSuchTxnException e) {
+ LOG.error("Metastore could not find txnid " + lock.getTxnid());
+ throw new LockException(ErrorMsg.TXNMGR_NOT_INSTANTIATED.getMsg(), e);
+ } catch (TxnAbortedException e) {
+ LOG.error("Transaction " + lock.getTxnid() + " already aborted.");
+ throw new LockException(ErrorMsg.TXN_ABORTED.getMsg(), e);
+ } catch (TException e) {
+ throw new LockException(ErrorMsg.METASTORE_COMMUNICATION_FAILED.getMsg(),
+ e);
+ }
+ }
+
+ @Override
+ public void unlock(HiveLock hiveLock) throws LockException {
+ long lockId = ((DbHiveLock)hiveLock).lockId;
+ try {
+ LOG.debug("Unlocking id:" + lockId);
+ client.unlock(lockId);
+ boolean removed = locks.remove((DbHiveLock)hiveLock);
+ LOG.debug("Removed a lock " + removed);
+ } catch (NoSuchLockException e) {
+ LOG.error("Metastore could find no record of lock " + lockId);
+ throw new LockException(ErrorMsg.LOCK_NO_SUCH_LOCK.getMsg(), e);
+ } catch (TxnOpenException e) {
+ throw new RuntimeException("Attempt to unlock lock " + lockId +
+ "associated with an open transaction, " + e.getMessage(), e);
+ } catch (TException e) {
+ throw new LockException(ErrorMsg.METASTORE_COMMUNICATION_FAILED.getMsg(),
+ e);
+ }
+ }
+
+ @Override
+ public void releaseLocks(List<HiveLock> hiveLocks) {
+ for (HiveLock lock : hiveLocks) {
+ try {
+ unlock(lock);
+ } catch (LockException e) {
+ // Not sure why this method doesn't throw any exceptions,
+ // but since the interface doesn't allow it we'll just swallow them and
+ // move on.
+ }
+ }
+ }
+
+ @Override
+ public List<HiveLock> getLocks(boolean verifyTablePartitions,
+ boolean fetchData) throws LockException {
+ return new ArrayList<HiveLock>(locks);
+ }
+
+ @Override
+ public List<HiveLock> getLocks(HiveLockObject key,
+ boolean verifyTablePartitions,
+ boolean fetchData) throws LockException {
+ throw new UnsupportedOperationException();
+ }
+
+ public ShowLocksResponse getLocks() throws LockException {
+ try {
+ return client.showLocks();
+ } catch (TException e) {
+ throw new LockException(ErrorMsg.METASTORE_COMMUNICATION_FAILED.getMsg(), e);
+ }
+ }
+
+ @Override
+ public void close() throws LockException {
+ for (HiveLock lock : locks) {
+ unlock(lock);
+ }
+ locks.clear();
+ }
+
+ @Override
+ public void prepareRetry() throws LockException {
+ // NOP
+ }
+
+ @Override
+ public void refresh() {
+ // NOP
+ }
+
+ static class DbHiveLock extends HiveLock {
+
+ long lockId;
+
+ DbHiveLock(long id) {
+ lockId = id;
+ }
+
+ @Override
+ public HiveLockObject getHiveLockObject() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public HiveLockMode getHiveLockMode() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (other instanceof DbHiveLock) {
+ return lockId == ((DbHiveLock)other).lockId;
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public int hashCode() {
+ return (int)(lockId % Integer.MAX_VALUE);
+ }
+ }
+
+ // Sleep before we send checkLock again, but do it with a back off
+ // off so we don't sit and hammer the metastore in a tight loop
+ private void backoff() {
+ nextSleep *= 2;
+ if (nextSleep > MAX_SLEEP) nextSleep = MAX_SLEEP;
+ try {
+ Thread.sleep(nextSleep);
+ } catch (InterruptedException e) {
+ }
+ }
+}
Added: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java?rev=1574266&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java Wed Mar 5 00:20:53 2014
@@ -0,0 +1,317 @@
+/**
+ * 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.lockmgr;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.LockComponentBuilder;
+import org.apache.hadoop.hive.metastore.LockRequestBuilder;
+import org.apache.hadoop.hive.metastore.api.*;
+import org.apache.hadoop.hive.ql.Context;
+import org.apache.hadoop.hive.ql.ErrorMsg;
+import org.apache.hadoop.hive.ql.QueryPlan;
+import org.apache.hadoop.hive.ql.hooks.Entity;
+import org.apache.hadoop.hive.ql.hooks.ReadEntity;
+import org.apache.hadoop.hive.ql.hooks.WriteEntity;
+import org.apache.hadoop.hive.ql.metadata.Table;
+import org.apache.thrift.TException;
+
+import java.util.List;
+
+/**
+ * An implementation of HiveTxnManager that stores the transactions in the
+ * metastore database.
+ */
+public class DbTxnManager extends HiveTxnManagerImpl {
+
+ static final private String CLASS_NAME = DbTxnManager.class.getName();
+ static final private Log LOG = LogFactory.getLog(CLASS_NAME);
+
+ private DbLockManager lockMgr = null;
+ private HiveMetaStoreClient client = null;
+ private long txnId = 0;
+
+ DbTxnManager() {
+ }
+
+ @Override
+ public void openTxn(String user) throws LockException {
+ init();
+ try {
+ txnId = client.openTxn(user);
+ LOG.debug("Opened txn " + txnId);
+ } catch (TException e) {
+ throw new LockException(ErrorMsg.METASTORE_COMMUNICATION_FAILED.getMsg(),
+ e);
+ }
+ }
+
+ @Override
+ public HiveLockManager getLockManager() throws LockException {
+ init();
+ if (lockMgr == null) {
+ lockMgr = new DbLockManager(client);
+ }
+ return lockMgr;
+ }
+
+ @Override
+ public void acquireLocks(QueryPlan plan, Context ctx, String username) throws LockException {
+ init();
+ // Make sure we've built the lock manager
+ getLockManager();
+
+ boolean atLeastOneLock = false;
+
+ LockRequestBuilder rqstBuilder = new LockRequestBuilder();
+ LOG.debug("Setting lock request transaction to " + txnId);
+ rqstBuilder.setTransactionId(txnId)
+ .setUser(username);
+
+ // For each source to read, get a shared lock
+ for (ReadEntity input : plan.getInputs()) {
+ LockComponentBuilder compBuilder = new LockComponentBuilder();
+ compBuilder.setShared();
+
+ Table t = null;
+ switch (input.getType()) {
+ case DATABASE:
+ compBuilder.setDbName(input.getDatabase().getName());
+ break;
+
+ case TABLE:
+ t = input.getTable();
+ compBuilder.setDbName(t.getDbName());
+ compBuilder.setTableName(t.getTableName());
+ break;
+
+ case PARTITION:
+ case DUMMYPARTITION:
+ compBuilder.setPartitionName(input.getPartition().getName());
+ t = input.getPartition().getTable();
+ compBuilder.setDbName(t.getDbName());
+ compBuilder.setTableName(t.getTableName());
+ break;
+
+ default:
+ // This is a file or something we don't hold locks for.
+ continue;
+ }
+ LockComponent comp = compBuilder.build();
+ LOG.debug("Adding lock component to lock request " + comp.toString());
+ rqstBuilder.addLockComponent(comp);
+ atLeastOneLock = true;
+ }
+
+ // For each source to write to, get the appropriate lock type. If it's
+ // an OVERWRITE, we need to get an exclusive lock. If it's an insert (no
+ // overwrite) than we need a shared. If it's update or delete then we
+ // need a SEMI-SHARED.
+ for (WriteEntity output : plan.getOutputs()) {
+ if (output.getType() == Entity.Type.DFS_DIR || output.getType() ==
+ Entity.Type.LOCAL_DIR) {
+ // We don't lock files or directories.
+ continue;
+ }
+ LockComponentBuilder compBuilder = new LockComponentBuilder();
+ Table t = null;
+ LOG.debug("output is null " + (output == null));
+ switch (output.getWriteType()) {
+ case DDL:
+ case INSERT_OVERWRITE:
+ compBuilder.setExclusive();
+ break;
+
+ case INSERT:
+ case DDL_METADATA_ONLY:
+ compBuilder.setShared();
+ break;
+
+ case UPDATE:
+ case DELETE:
+ compBuilder.setSemiShared();
+ break;
+
+ default:
+ throw new RuntimeException("Unknown write type " +
+ output.getWriteType().toString());
+
+ }
+ switch (output.getType()) {
+ case DATABASE:
+ compBuilder.setDbName(output.getDatabase().getName());
+ break;
+
+ case TABLE:
+ t = output.getTable();
+ compBuilder.setDbName(t.getDbName());
+ compBuilder.setTableName(t.getTableName());
+ break;
+
+ case PARTITION:
+ case DUMMYPARTITION:
+ compBuilder.setPartitionName(output.getPartition().getName());
+ t = output.getPartition().getTable();
+ compBuilder.setDbName(t.getDbName());
+ compBuilder.setTableName(t.getTableName());
+ break;
+
+ default:
+ // This is a file or something we don't hold locks for.
+ continue;
+ }
+ LockComponent comp = compBuilder.build();
+ LOG.debug("Adding lock component to lock request " + comp.toString());
+ rqstBuilder.addLockComponent(comp);
+ atLeastOneLock = true;
+ }
+
+ // Make sure we need locks. It's possible there's nothing to lock in
+ // this operation.
+ if (!atLeastOneLock) return;
+
+ List<HiveLock> locks = lockMgr.lock(rqstBuilder.build());
+ ctx.setHiveLocks(locks);
+ }
+
+ @Override
+ public void commitTxn() throws LockException {
+ if (txnId == 0) {
+ throw new RuntimeException("Attempt to commit before opening a " +
+ "transaction");
+ }
+ try {
+ LOG.debug("Committing txn " + txnId);
+ client.commitTxn(txnId);
+ } catch (NoSuchTxnException e) {
+ LOG.error("Metastore could not find txn " + txnId);
+ throw new LockException(ErrorMsg.TXN_NO_SUCH_TRANSACTION.getMsg() , e);
+ } catch (TxnAbortedException e) {
+ LOG.error("Transaction " + txnId + " aborted");
+ throw new LockException(ErrorMsg.TXN_ABORTED.getMsg(), e);
+ } catch (TException e) {
+ throw new LockException(ErrorMsg.METASTORE_COMMUNICATION_FAILED.getMsg(),
+ e);
+ } finally {
+ txnId = 0;
+ }
+ }
+
+ @Override
+ public void rollbackTxn() throws LockException {
+ if (txnId == 0) {
+ throw new RuntimeException("Attempt to rollback before opening a " +
+ "transaction");
+ }
+ try {
+ LOG.debug("Rolling back txn " + txnId);
+ client.rollbackTxn(txnId);
+ } catch (NoSuchTxnException e) {
+ LOG.error("Metastore could not find txn " + txnId);
+ throw new LockException(ErrorMsg.TXN_NO_SUCH_TRANSACTION.getMsg() , e);
+ } catch (TException e) {
+ throw new LockException(ErrorMsg.METASTORE_COMMUNICATION_FAILED.getMsg(),
+ e);
+ } finally {
+ txnId = 0;
+ }
+ }
+
+ @Override
+ public void heartbeat() throws LockException {
+ LOG.debug("Heartbeating lock and transaction " + txnId);
+ List<HiveLock> locks = lockMgr.getLocks(false, false);
+ if (locks.size() == 0) {
+ if (txnId == 0) {
+ // No locks, no txn, we outta here.
+ return;
+ } else {
+ // Create one dummy lock so we can go through the loop below
+ DbLockManager.DbHiveLock dummyLock = new DbLockManager.DbHiveLock(0L);
+ locks.add(dummyLock);
+ }
+ }
+ for (HiveLock lock : locks) {
+ long lockId = ((DbLockManager.DbHiveLock)lock).lockId;
+ try {
+ client.heartbeat(txnId, lockId);
+ } catch (NoSuchLockException e) {
+ LOG.error("Unable to find lock " + lockId);
+ throw new LockException(ErrorMsg.LOCK_NO_SUCH_LOCK.getMsg(), e);
+ } catch (NoSuchTxnException e) {
+ LOG.error("Unable to find transaction " + txnId);
+ throw new LockException(ErrorMsg.TXN_NO_SUCH_TRANSACTION.getMsg(), e);
+ } catch (TxnAbortedException e) {
+ LOG.error("Transaction aborted " + txnId);
+ throw new LockException(ErrorMsg.TXN_ABORTED.getMsg(), e);
+ } catch (TException e) {
+ throw new LockException(
+ ErrorMsg.METASTORE_COMMUNICATION_FAILED.getMsg(), e);
+ }
+ }
+ }
+
+ @Override
+ public IMetaStoreClient.ValidTxnList getValidTxns() throws LockException {
+ init();
+ try {
+ return client.getValidTxns();
+ } catch (TException e) {
+ throw new LockException(ErrorMsg.METASTORE_COMMUNICATION_FAILED.getMsg(),
+ e);
+ }
+ }
+
+ @Override
+ public boolean supportsExplicitLock() {
+ return false;
+ }
+
+ @Override
+ public boolean useNewShowLocksFormat() {
+ return true;
+ }
+
+ @Override
+ protected void destruct() {
+ try {
+ if (txnId > 0) rollbackTxn();
+ if (lockMgr != null) lockMgr.close();
+ } catch (Exception e) {
+ // Not much we can do about it here.
+ }
+ }
+
+ private void init() throws LockException {
+ if (client == null) {
+ if (conf == null) {
+ throw new RuntimeException("Must call setHiveConf before any other " +
+ "methods.");
+ }
+ try {
+ client = new HiveMetaStoreClient(conf);
+ } catch (MetaException e) {
+ throw new LockException(ErrorMsg.METASTORE_COULD_NOT_INITIATE.getMsg(),
+ e);
+ }
+ }
+ }
+
+}
Added: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DummyTxnManager.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DummyTxnManager.java?rev=1574266&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DummyTxnManager.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DummyTxnManager.java Wed Mar 5 00:20:53 2014
@@ -0,0 +1,343 @@
+/**
+ * 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.lockmgr;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.GetOpenTxnsResponse;
+import org.apache.hadoop.hive.ql.Context;
+import org.apache.hadoop.hive.ql.ErrorMsg;
+import org.apache.hadoop.hive.ql.QueryPlan;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.hooks.ReadEntity;
+import org.apache.hadoop.hive.ql.hooks.WriteEntity;
+import org.apache.hadoop.hive.ql.metadata.DummyPartition;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.metadata.Partition;
+import org.apache.hadoop.hive.ql.metadata.Table;
+import org.apache.hadoop.hive.ql.session.SessionState;
+import org.apache.hadoop.util.ReflectionUtils;
+
+import java.util.*;
+
+/**
+ * An implementation of {@link HiveTxnManager} that does not support
+ * transactions. This provides default Hive behavior.
+ */
+class DummyTxnManager extends HiveTxnManagerImpl {
+ private static final int separator = Utilities.tabCode;
+ private static final int terminator = Utilities.newLineCode;
+ static final private Log LOG =
+ LogFactory.getLog(DummyTxnManager.class.getName());
+
+ private HiveLockManager lockMgr;
+
+ @Override
+ public void openTxn(String user) throws LockException {
+ // No-op
+ }
+
+ @Override
+ public HiveLockManager getLockManager() throws LockException {
+ if (lockMgr == null) {
+ boolean supportConcurrency =
+ conf.getBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY);
+ if (supportConcurrency) {
+ String lockMgrName =
+ conf.getVar(HiveConf.ConfVars.HIVE_LOCK_MANAGER);
+ if ((lockMgrName == null) || (lockMgrName.isEmpty())) {
+ throw new LockException(ErrorMsg.LOCKMGR_NOT_SPECIFIED.getMsg());
+ }
+
+ try {
+ LOG.info("Creating lock manager of type " + lockMgrName);
+ lockMgr = (HiveLockManager)ReflectionUtils.newInstance(
+ conf.getClassByName(lockMgrName), conf);
+ lockMgr.setContext(new HiveLockManagerCtx(conf));
+ } catch (Exception e) {
+ // set hiveLockMgr to null just in case this invalid manager got set to
+ // next query's ctx.
+ if (lockMgr != null) {
+ try {
+ lockMgr.close();
+ } catch (LockException e1) {
+ //nothing can do here
+ }
+ lockMgr = null;
+ }
+ throw new LockException(ErrorMsg.LOCKMGR_NOT_INITIALIZED.getMsg() +
+ e.getMessage());
+ }
+ } else {
+ LOG.info("Concurrency mode is disabled, not creating a lock manager");
+ return null;
+ }
+ }
+ // Force a re-read of the configuration file. This is done because
+ // different queries in the session may be using the same lock manager.
+ lockMgr.refresh();
+ return lockMgr;
+ }
+
+ @Override
+ public void acquireLocks(QueryPlan plan, Context ctx, String username) throws LockException {
+ // Make sure we've built the lock manager
+ getLockManager();
+
+ // If the lock manager is still null, then it means we aren't using a
+ // lock manager
+ if (lockMgr == null) return;
+
+ List<HiveLockObj> lockObjects = new ArrayList<HiveLockObj>();
+
+ // Sort all the inputs, outputs.
+ // If a lock needs to be acquired on any partition, a read lock needs to be acquired on all
+ // its parents also
+ for (ReadEntity input : plan.getInputs()) {
+ LOG.debug("Adding " + input.getName() + " to list of lock inputs");
+ if (input.getType() == ReadEntity.Type.DATABASE) {
+ lockObjects.addAll(getLockObjects(plan, input.getDatabase(), null,
+ null, HiveLockMode.SHARED));
+ } else if (input.getType() == ReadEntity.Type.TABLE) {
+ lockObjects.addAll(getLockObjects(plan, null, input.getTable(), null,
+ HiveLockMode.SHARED));
+ } else {
+ lockObjects.addAll(getLockObjects(plan, null, null,
+ input.getPartition(),
+ HiveLockMode.SHARED));
+ }
+ }
+
+ for (WriteEntity output : plan.getOutputs()) {
+ LOG.debug("Adding " + output.getName() + " to list of lock outputs");
+ List<HiveLockObj> lockObj = null;
+ if (output.getType() == WriteEntity.Type.DATABASE) {
+ lockObjects.addAll(getLockObjects(plan, output.getDatabase(), null,
+ null,
+ output.isComplete() ? HiveLockMode.EXCLUSIVE : HiveLockMode.SHARED));
+ } else if (output.getTyp() == WriteEntity.Type.TABLE) {
+ lockObj = getLockObjects(plan, null, output.getTable(), null,
+ output.isComplete() ? HiveLockMode.EXCLUSIVE : HiveLockMode.SHARED);
+ } else if (output.getTyp() == WriteEntity.Type.PARTITION) {
+ lockObj = getLockObjects(plan, null, null, output.getPartition(),
+ HiveLockMode.EXCLUSIVE);
+ }
+ // In case of dynamic queries, it is possible to have incomplete dummy partitions
+ else if (output.getTyp() == WriteEntity.Type.DUMMYPARTITION) {
+ lockObj = getLockObjects(plan, null, null, output.getPartition(),
+ HiveLockMode.SHARED);
+ }
+
+ if(lockObj != null) {
+ lockObjects.addAll(lockObj);
+ ctx.getOutputLockObjects().put(output, lockObj);
+ }
+ }
+
+ if (lockObjects.isEmpty() && !ctx.isNeedLockMgr()) {
+ return;
+ }
+
+ HiveLockObject.HiveLockObjectData lockData =
+ new HiveLockObject.HiveLockObjectData(plan.getQueryId(),
+ String.valueOf(System.currentTimeMillis()),
+ "IMPLICIT",
+ plan.getQueryStr());
+
+ // Lock the database also
+ String currentDb = SessionState.get().getCurrentDatabase();
+ lockObjects.add(
+ new HiveLockObj(
+ new HiveLockObject(currentDb, lockData),
+ HiveLockMode.SHARED
+ )
+ );
+
+ dedupLockObjects(lockObjects);
+ List<HiveLock> hiveLocks = lockMgr.lock(lockObjects, false);
+
+ if (hiveLocks == null) {
+ throw new LockException(ErrorMsg.LOCK_CANNOT_BE_ACQUIRED.getMsg());
+ } else {
+ ctx.setHiveLocks(hiveLocks);
+ }
+ }
+
+ @Override
+ public void commitTxn() throws LockException {
+ // No-op
+ }
+
+ @Override
+ public void rollbackTxn() throws LockException {
+ // No-op
+ }
+
+ @Override
+ public void heartbeat() throws LockException {
+ // No-op
+ }
+
+ @Override
+ public IMetaStoreClient.ValidTxnList getValidTxns() throws LockException {
+ return new IMetaStoreClient.ValidTxnList() {
+ @Override
+ public boolean isTxnCommitted(long txnid) {
+ return false;
+ }
+
+ @Override
+ public RangeResponse isTxnRangeCommitted(long min, long max) {
+ return RangeResponse.NONE;
+ }
+
+ @Override
+ public GetOpenTxnsResponse getOpenTxns() {
+ return null;
+ }
+
+ @Override
+ public void fromString(String src) {
+ }
+
+ @Override
+ public String toString() {
+ return "";
+ }
+
+ };
+ }
+
+ @Override
+ public boolean supportsExplicitLock() {
+ return true;
+ }
+
+ @Override
+ public boolean useNewShowLocksFormat() {
+ return false;
+ }
+
+
+ protected void destruct() {
+ if (lockMgr != null) {
+ try {
+ lockMgr.close();
+ } catch (LockException e) {
+ // Not much I can do about it.
+ LOG.warn("Got exception when closing lock manager " + e.getMessage());
+ }
+ }
+ }
+
+ /**
+ * Dedup the list of lock objects so that there is only one lock per table/partition.
+ * If there is both a shared and exclusive lock for the same object, this will deduped
+ * to just a single exclusive lock. Package level so that the unit tests
+ * can access it. Not intended for use outside this class.
+ * @param lockObjects
+ */
+ static void dedupLockObjects(List<HiveLockObj> lockObjects) {
+ Map<String, HiveLockObj> lockMap = new HashMap<String, HiveLockObj>();
+ for (HiveLockObj lockObj : lockObjects) {
+ String lockName = lockObj.getName();
+ HiveLockObj foundLock = lockMap.get(lockName);
+ if (foundLock == null || lockObj.getMode() == HiveLockMode.EXCLUSIVE) {
+ lockMap.put(lockName, lockObj);
+ }
+ }
+ // copy set of deduped locks back to original list
+ lockObjects.clear();
+ for (HiveLockObj lockObj : lockMap.values()) {
+ lockObjects.add(lockObj);
+ }
+ }
+
+ private List<HiveLockObj> getLockObjects(QueryPlan plan, Database db,
+ Table t, Partition p,
+ HiveLockMode mode)
+ throws LockException {
+ List<HiveLockObj> locks = new LinkedList<HiveLockObj>();
+
+ HiveLockObject.HiveLockObjectData lockData =
+ new HiveLockObject.HiveLockObjectData(plan.getQueryId(),
+ String.valueOf(System.currentTimeMillis()),
+ "IMPLICIT",
+ plan.getQueryStr());
+
+ if (db != null) {
+ locks.add(new HiveLockObj(new HiveLockObject(db.getName(), lockData),
+ mode));
+ return locks;
+ }
+
+ if (t != null) {
+ locks.add(new HiveLockObj(new HiveLockObject(t, lockData), mode));
+ mode = HiveLockMode.SHARED;
+ locks.add(new HiveLockObj(new HiveLockObject(t.getDbName(), lockData), mode));
+ return locks;
+ }
+
+ if (p != null) {
+ if (!(p instanceof DummyPartition)) {
+ locks.add(new HiveLockObj(new HiveLockObject(p, lockData), mode));
+ }
+
+ // All the parents are locked in shared mode
+ mode = HiveLockMode.SHARED;
+
+ // For dummy partitions, only partition name is needed
+ String name = p.getName();
+
+ if (p instanceof DummyPartition) {
+ name = p.getName().split("@")[2];
+ }
+
+ String partialName = "";
+ String[] partns = name.split("/");
+ int len = p instanceof DummyPartition ? partns.length : partns.length - 1;
+ Map<String, String> partialSpec = new LinkedHashMap<String, String>();
+ for (int idx = 0; idx < len; idx++) {
+ String partn = partns[idx];
+ partialName += partn;
+ String[] nameValue = partn.split("=");
+ assert(nameValue.length == 2);
+ partialSpec.put(nameValue[0], nameValue[1]);
+ try {
+ locks.add(new HiveLockObj(
+ new HiveLockObject(new DummyPartition(p.getTable(), p.getTable().getDbName()
+ + "/" + p.getTable().getTableName()
+ + "/" + partialName,
+ partialSpec), lockData), mode));
+ partialName += "/";
+ } catch (HiveException e) {
+ throw new LockException(e.getMessage());
+ }
+ }
+
+ locks.add(new HiveLockObj(new HiveLockObject(p.getTable(), lockData), mode));
+ locks.add(new HiveLockObj(new HiveLockObject(p.getTable().getDbName(), lockData), mode));
+ }
+ return locks;
+ }
+
+
+}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/EmbeddedLockManager.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/EmbeddedLockManager.java?rev=1574266&r1=1574265&r2=1574266&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/EmbeddedLockManager.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/EmbeddedLockManager.java Wed Mar 5 00:20:53 2014
@@ -18,24 +18,14 @@
package org.apache.hadoop.hive.ql.lockmgr;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Stack;
-import java.util.concurrent.locks.ReentrantLock;
-
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.lockmgr.HiveLockObject.HiveLockObjectData;
-import org.apache.hadoop.hive.ql.metadata.DummyPartition;
-import org.apache.hadoop.hive.ql.metadata.Hive;
-import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.ql.metadata.Partition;
-import org.apache.hadoop.hive.ql.metadata.Table;
+import org.apache.hadoop.hive.ql.metadata.*;
+
+import java.util.*;
+import java.util.concurrent.locks.ReentrantLock;
/**
* shared lock manager for dedicated hive server. all locks are managed in memory
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveLockManager.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveLockManager.java?rev=1574266&r1=1574265&r2=1574266&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveLockManager.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveLockManager.java Wed Mar 5 00:20:53 2014
@@ -20,6 +20,11 @@ package org.apache.hadoop.hive.ql.lockmg
import java.util.List;
+/**
+ * Manager for locks in Hive. Users should not instantiate a lock manager
+ * directly. Instead they should get an instance from their instance of
+ * {@link HiveTxnManager}.
+ */
public interface HiveLockManager {
public void setContext(HiveLockManagerCtx ctx) throws LockException;
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveLockMode.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveLockMode.java?rev=1574266&r1=1574265&r2=1574266&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveLockMode.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveLockMode.java Wed Mar 5 00:20:53 2014
@@ -19,6 +19,8 @@
package org.apache.hadoop.hive.ql.lockmgr;
public enum HiveLockMode {
- SHARED, EXCLUSIVE;
+ SHARED,
+ EXCLUSIVE,
+ SEMI_SHARED; // SEMI_SHARED can share with SHARED but not with itself
}
Added: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveTxnManager.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveTxnManager.java?rev=1574266&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveTxnManager.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveTxnManager.java Wed Mar 5 00:20:53 2014
@@ -0,0 +1,122 @@
+/**
+ * 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.lockmgr;
+
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.ql.Context;
+import org.apache.hadoop.hive.ql.QueryPlan;
+
+/**
+ * An interface that allows Hive to manage transactions. All classes
+ * implementing this should extend {@link HiveTxnManagerImpl} rather than
+ * implementing this directly.
+ */
+public interface HiveTxnManager {
+
+ /**
+ * Open a new transaction.
+ * @param user Hive user who is opening this transaction.
+ * @throws LockException if a transaction is already open.
+ */
+ void openTxn(String user) throws LockException;
+
+ /**
+ * Get the lock manager. This must be used rather than instantiating an
+ * instance of the lock manager directly as the transaction manager will
+ * choose which lock manager to instantiate.
+ * @return the instance of the lock manager
+ * @throws LockException if there is an issue obtaining the lock manager.
+ */
+ HiveLockManager getLockManager() throws LockException;
+
+ /**
+ * Acquire all of the locks needed by a query. If used with a query that
+ * requires transactions, this should be called after {@link #openTxn(String)}.
+ * A list of acquired locks will be stored in the
+ * {@link org.apache.hadoop.hive.ql.Context} object and can be retrieved
+ * via {@link org.apache.hadoop.hive.ql.Context#getHiveLocks}.
+ * @param plan query plan
+ * @param ctx Context for this query
+ * @param username name of the user for this query
+ * @throws LockException if there is an error getting the locks
+ */
+ void acquireLocks(QueryPlan plan, Context ctx, String username) throws LockException;
+
+ /**
+ * Commit the current transaction. This will release all locks obtained in
+ * {@link #acquireLocks(org.apache.hadoop.hive.ql.QueryPlan,
+ * org.apache.hadoop.hive.ql.Context, java.lang.String)}.
+ * @throws LockException if there is no current transaction or the
+ * transaction has already been committed or aborted.
+ */
+ void commitTxn() throws LockException;
+
+ /**
+ * Abort the current transaction. This will release all locks obtained in
+ * {@link #acquireLocks(org.apache.hadoop.hive.ql.QueryPlan,
+ * org.apache.hadoop.hive.ql.Context, java.lang.String)}.
+ * @throws LockException if there is no current transaction or the
+ * transaction has already been committed or aborted.
+ */
+ void rollbackTxn() throws LockException;
+
+ /**
+ * Send a heartbeat to the transaction management storage so other Hive
+ * clients know that the transaction and locks held by this client are
+ * still valid. For implementations that do not require heartbeats this
+ * can be a no-op.
+ * @throws LockException If current transaction exists or the transaction
+ * has already been committed or aborted.
+ */
+ void heartbeat() throws LockException;
+
+ /**
+ * Get the transactions that are currently valid. The resulting
+ * {@link IMetaStoreClient.ValidTxnList} object is a thrift object and can
+ * be passed to the processing
+ * tasks for use in the reading the data. This call should be made once up
+ * front by the planner and should never be called on the backend,
+ * as this will violate the isolation level semantics.
+ * @return list of valid transactions.
+ * @throws LockException
+ */
+ IMetaStoreClient.ValidTxnList getValidTxns() throws LockException;
+
+ /**
+ * This call closes down the transaction manager. All open transactions
+ * are aborted. If no transactions are open but locks are held those locks
+ * are released. This method should be called if processing of a session
+ * is being halted in an abnormal way. It avoids locks and transactions
+ * timing out.
+ */
+ void closeTxnManager();
+
+ /**
+ * Indicate whether this lock manager supports the use of <code>lock
+ * <i>database</i></code> or <code>lock <i>table</i></code>.
+ * @return
+ */
+ boolean supportsExplicitLock();
+
+ /**
+ * Indicate whether this transaction manager returns information about locks in the new format
+ * for show locks or the old one.
+ * @return true if the new format should be used.
+ */
+ boolean useNewShowLocksFormat();
+}
Added: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveTxnManagerImpl.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveTxnManagerImpl.java?rev=1574266&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveTxnManagerImpl.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/HiveTxnManagerImpl.java Wed Mar 5 00:20:53 2014
@@ -0,0 +1,47 @@
+/**
+ * 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.lockmgr;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+
+/**
+ * An implementation HiveTxnManager that includes internal methods that all
+ * transaction managers need to implement but that we don't want to expose to
+ * outside.
+ */
+abstract class HiveTxnManagerImpl implements HiveTxnManager {
+
+ protected HiveConf conf;
+
+ void setHiveConf(HiveConf c) {
+ conf = c;
+ }
+
+ abstract protected void destruct();
+
+ @Override
+ public void closeTxnManager() {
+ destruct();
+ }
+
+ @Override
+ protected void finalize() throws Throwable {
+ destruct();
+ }
+
+}