diff --git metastore/src/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java metastore/src/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java index d4d0162..df7541a 100644 --- metastore/src/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java +++ metastore/src/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java @@ -81,6 +81,11 @@ * If we ever decide to run remote Derby server, according to * https://db.apache.org/derby/docs/10.0/manuals/develop/develop78.html all transactions will be * seriazlied, so that would also work though has not been tested. + * + * General design note: + * It's imperative that any operation on a txn (e.g. commit), ensure (atomically) that this txn is + * still valid and active. In the code this is usually achieved at the same time the txn record + * is locked for some operation. */ @InterfaceAudience.Private @InterfaceStability.Evolving diff --git ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java index e8ebe55..3aec8eb 100644 --- ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java +++ ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java @@ -19,6 +19,7 @@ import com.google.common.annotations.VisibleForTesting; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.ql.io.AcidUtils; import org.apache.hive.common.util.ShutdownHookManager; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -213,6 +214,17 @@ LockState acquireLocks(QueryPlan plan, Context ctx, String username, boolean isB break; case INSERT: + t = output.getTable(); + if(t == null) { + throw new IllegalStateException("No table info for " + output); + } + if(AcidUtils.isAcidTable(t)) { + compBuilder.setShared(); + } + else { + compBuilder.setExclusive(); + } + break; case DDL_SHARED: compBuilder.setShared(); break; diff --git ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java index f4debfe..02d0ae6 100644 --- ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java +++ ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java @@ -445,6 +445,28 @@ public void testDeleteIn() throws Exception { } /** + * Test update that hits multiple partitions (i.e. requries dynamic partition insert to process) + * @throws Exception + */ + @Test + public void updateDeletePartitioned() throws Exception { + int[][] tableData = {{1,2},{3,4},{5,6}}; + runStatementOnDriver("insert into " + Table.ACIDTBLPART + " partition(p=1) (a,b) " + makeValuesClause(tableData)); + runStatementOnDriver("insert into " + Table.ACIDTBLPART + " partition(p=2) (a,b) " + makeValuesClause(tableData)); + TxnStore txnHandler = TxnUtils.getTxnStore(hiveConf); + txnHandler.compact(new CompactionRequest("default", Table.ACIDTBLPART.name(), CompactionType.MAJOR)); + runWorker(hiveConf); + runCleaner(hiveConf); + runStatementOnDriver("update " + Table.ACIDTBLPART + " set b = b + 1 where a = 3"); + txnHandler.compact(new CompactionRequest("default", Table.ACIDTBLPART.toString(), CompactionType.MAJOR)); + runWorker(hiveConf); + runCleaner(hiveConf); + List rs = runStatementOnDriver("select p,a,b from " + Table.ACIDTBLPART + " order by p, a, b"); + int[][] expectedData = {{1,1,2},{1,3,5},{1,5,6},{2,1,2},{2,3,5},{2,5,6}}; + Assert.assertEquals("Update " + Table.ACIDTBLPART + " didn't match:", stringifyValues(expectedData), rs); + } + + /** * https://issues.apache.org/jira/browse/HIVE-10151 */ @Test diff --git ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager2.java ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager2.java index 1b07d4b..d4ce12e 100644 --- ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager2.java +++ ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager2.java @@ -276,6 +276,87 @@ public void testDummyTxnManagerOnAcidTable() throws Exception { Assert.assertTrue(cpr.getErrorMessage().contains("Attempt to do update or delete using transaction manager that does not support these operations.")); } + /** + * collection of queries where we ensure that we get the locks that are expected + * @throws Exception + */ + @Test + public void checkExpectedLocks() throws Exception { + CommandProcessorResponse cpr = null; + cpr = driver.run("create table acidPart(a int, b int) partitioned by (p string) clustered by (a) into 2 buckets stored as orc TBLPROPERTIES ('transactional'='true')"); + checkCmdOnDriver(cpr); + cpr = driver.run("create table nonAcidPart(a int, b int) partitioned by (p string) stored as orc"); + checkCmdOnDriver(cpr); + + cpr = driver.compileAndRespond("insert into nonAcidPart partition(p) values(1,2,3)"); + checkCmdOnDriver(cpr); + LockState lockState = ((DbTxnManager) txnMgr).acquireLocks(driver.getPlan(), ctx, "Practical", false); + List locks = getLocks(); + Assert.assertEquals("Unexpected lock count", 2, locks.size()); + checkLock(LockType.SHARED_READ, LockState.ACQUIRED, "default", "values__tmp__table__1", null, locks.get(0)); + checkLock(LockType.EXCLUSIVE, LockState.ACQUIRED, "default", "nonAcidPart", null, locks.get(1)); + List relLocks = new ArrayList(2); + relLocks.add(new DbLockManager.DbHiveLock(locks.get(0).getLockid())); + relLocks.add(new DbLockManager.DbHiveLock(locks.get(1).getLockid())); + txnMgr.getLockManager().releaseLocks(relLocks); + + cpr = driver.compileAndRespond("insert into nonAcidPart partition(p=1) values(5,6)"); + checkCmdOnDriver(cpr); + lockState = ((DbTxnManager) txnMgr).acquireLocks(driver.getPlan(), ctx, "Practical", false); + locks = getLocks(); + Assert.assertEquals("Unexpected lock count", 2, locks.size()); + checkLock(LockType.SHARED_READ, LockState.ACQUIRED, "default", "values__tmp__table__2", null, locks.get(0)); + checkLock(LockType.EXCLUSIVE, LockState.ACQUIRED, "default", "nonAcidPart", "p=1", locks.get(1)); + relLocks = new ArrayList(2); + relLocks.add(new DbLockManager.DbHiveLock(locks.get(0).getLockid())); + relLocks.add(new DbLockManager.DbHiveLock(locks.get(1).getLockid())); + txnMgr.getLockManager().releaseLocks(relLocks); + + cpr = driver.compileAndRespond("insert into acidPart partition(p) values(1,2,3)"); + checkCmdOnDriver(cpr); + lockState = ((DbTxnManager) txnMgr).acquireLocks(driver.getPlan(), ctx, "Practical", false); + locks = getLocks(); + Assert.assertEquals("Unexpected lock count", 2, locks.size()); + checkLock(LockType.SHARED_READ, LockState.ACQUIRED, "default", "values__tmp__table__3", null, locks.get(0)); + checkLock(LockType.SHARED_READ, LockState.ACQUIRED, "default", "acidPart", null, locks.get(1)); + relLocks = new ArrayList(2); + relLocks.add(new DbLockManager.DbHiveLock(locks.get(0).getLockid())); + relLocks.add(new DbLockManager.DbHiveLock(locks.get(1).getLockid())); + txnMgr.getLockManager().releaseLocks(relLocks); + + cpr = driver.compileAndRespond("insert into acidPart partition(p=1) values(5,6)"); + checkCmdOnDriver(cpr); + lockState = ((DbTxnManager) txnMgr).acquireLocks(driver.getPlan(), ctx, "Practical", false); + locks = getLocks(); + Assert.assertEquals("Unexpected lock count", 2, locks.size()); + checkLock(LockType.SHARED_READ, LockState.ACQUIRED, "default", "values__tmp__table__4", null, locks.get(0)); + checkLock(LockType.SHARED_READ, LockState.ACQUIRED, "default", "acidPart", "p=1", locks.get(1)); + relLocks = new ArrayList(2); + relLocks.add(new DbLockManager.DbHiveLock(locks.get(0).getLockid())); + relLocks.add(new DbLockManager.DbHiveLock(locks.get(1).getLockid())); + txnMgr.getLockManager().releaseLocks(relLocks); + + cpr = driver.compileAndRespond("update acidPart set b = 17 where a = 1"); + checkCmdOnDriver(cpr); + lockState = ((DbTxnManager) txnMgr).acquireLocks(driver.getPlan(), ctx, "Practical", false); + locks = getLocks(); + Assert.assertEquals("Unexpected lock count", 1, locks.size()); + checkLock(LockType.SHARED_WRITE, LockState.ACQUIRED, "default", "acidPart", null, locks.get(0)); + relLocks = new ArrayList(2); + relLocks.add(new DbLockManager.DbHiveLock(locks.get(0).getLockid())); + txnMgr.getLockManager().releaseLocks(relLocks); + + cpr = driver.compileAndRespond("update acidPart set b = 17 where p = 1"); + checkCmdOnDriver(cpr); + lockState = ((DbTxnManager) txnMgr).acquireLocks(driver.getPlan(), ctx, "Practical", false); + locks = getLocks(); + Assert.assertEquals("Unexpected lock count", 1, locks.size()); + checkLock(LockType.SHARED_WRITE, LockState.ACQUIRED, "default", "acidPart", null, locks.get(0));//https://issues.apache.org/jira/browse/HIVE-13212 + relLocks = new ArrayList(2); + relLocks.add(new DbLockManager.DbHiveLock(locks.get(0).getLockid())); + txnMgr.getLockManager().releaseLocks(relLocks); + } + private void checkLock(LockType type, LockState state, String db, String table, String partition, ShowLocksResponseElement l) { Assert.assertEquals(l.toString(),l.getType(), type); Assert.assertEquals(l.toString(),l.getState(), state);