[jira] [Work logged] (HIVE-26076) Non blocking ADD PARTITION if not exists

["ASF GitHub Bot (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/HIVE-26076?focusedWorklogId=765452&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-765452 ]

ASF GitHub Bot logged work on HIVE-26076:
-----------------------------------------

                Author: ASF GitHub Bot
            Created on: 03/May/22 13:42
            Start Date: 03/May/22 13:42
    Worklog Time Spent: 10m 
      Work Description: asinkovits commented on code in PR #3122:
URL: https://github.com/apache/hive/pull/3122#discussion_r863754314


##########
ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager2.java:
##########
@@ -4028,4 +4005,88 @@ public void testAddDropConstraintNonBlocking() throws Exception {
     checkLock(LockType.EXCL_WRITE,
       LockState.ACQUIRED, "default", "tab_acid", null, locks);
   }
+
+  @Test
+  public void testAddPartitionIfNotExistsNonBlocking() throws Exception {
+    testAddPartitionIfNotExists(false);
+  }
+  @Test
+  public void testAddPartitionIfNotExistsBlocking() throws Exception {
+    testAddPartitionIfNotExists(true);
+  }
+
+  private void testAddPartitionIfNotExists(boolean blocking) throws Exception {
+    dropTable(new String[] {"T", "Tstage"});
+    
+    HiveConf.setIntVar(conf, HiveConf.ConfVars.HIVE_LOCKS_PARTITION_THRESHOLD, 1);
+    driver = Mockito.spy(driver);
+
+    HiveConf.setBoolVar(driver2.getConf(), HiveConf.ConfVars.HIVE_ACID_LOCKLESS_READS_ENABLED, !blocking);
+    driver2 = Mockito.spy(driver2);
+
+    driver.run("create table if not exists T (a int, b int) partitioned by (p string) " +
+      "stored as orc TBLPROPERTIES ('transactional'='true')");
+    //bucketed just so that we get 2 files
+    driver.run("create table if not exists Tstage (a int, b int) clustered by (a) into 2 " +
+      "buckets stored as orc TBLPROPERTIES ('transactional'='false')");
+    driver.run("insert into Tstage values(1,2),(3,4)");
+    String exportLoc = exportFolder.newFolder("1").toString();
+    driver.run("export table Tstage to '" + exportLoc + "'");
+
+    driver.compileAndRespond("select * from T");
+    List<String> res = new ArrayList<>();
+    
+    driver.lockAndRespond();
+    List<ShowLocksResponseElement> locks = getLocks();
+    Assert.assertEquals("Unexpected lock count", 1, locks.size());
+
+    checkLock(LockType.SHARED_READ,
+      LockState.ACQUIRED, "default", "T", null, locks);
+
+    DbTxnManager txnMgr2 = (DbTxnManager) TxnManagerFactory.getTxnManagerFactory().getTxnManager(conf);
+    swapTxnManager(txnMgr2);
+    driver2.compileAndRespond("alter table T add if not exists partition (p='foo') location '" + exportLoc + "/data'");
+
+    if (blocking) {
+      txnMgr2.acquireLocks(driver2.getPlan(), ctx, null, false);
+      locks = getLocks();
+
+      ShowLocksResponseElement checkLock = checkLock(LockType.EXCLUSIVE,
+        LockState.WAITING, "default", "T", null, locks);
+
+      swapTxnManager(txnMgr);
+      Mockito.doNothing().when(driver).lockAndRespond();
+      driver.run();
+      
+      driver.getFetchTask().fetch(res);
+      swapTxnManager(txnMgr2);
+
+      FieldSetter.setField(txnMgr2, txnMgr2.getClass().getDeclaredField("numStatements"), 0);
+      txnMgr2.getMS().unlock(checkLock.getLockid());
+    }
+    driver2.lockAndRespond();
+    locks = getLocks();
+    Assert.assertEquals("Unexpected lock count", blocking ? 1 : 2, locks.size());
+
+    checkLock(blocking ? LockType.EXCLUSIVE : LockType.EXCL_WRITE,
+      LockState.ACQUIRED, "default", "T", null, locks);
+
+    Mockito.doNothing().when(driver2).lockAndRespond();
+    driver2.run();
+
+    if (!blocking) {
+      swapTxnManager(txnMgr);
+      Mockito.doNothing().when(driver).lockAndRespond();
+      driver.run();
+    }
+    Mockito.reset(driver, driver2);
+    
+    driver.getFetchTask().fetch(res);
+    Assert.assertEquals("Expecting 2 rows and found " + res.size(), 0, res.size());

Review Comment:
   I think we're expecting 0 rows here, message says 2.



##########
ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager2.java:
##########
@@ -4028,4 +4005,88 @@ public void testAddDropConstraintNonBlocking() throws Exception {
     checkLock(LockType.EXCL_WRITE,
       LockState.ACQUIRED, "default", "tab_acid", null, locks);
   }
+
+  @Test
+  public void testAddPartitionIfNotExistsNonBlocking() throws Exception {
+    testAddPartitionIfNotExists(false);
+  }
+  @Test
+  public void testAddPartitionIfNotExistsBlocking() throws Exception {
+    testAddPartitionIfNotExists(true);
+  }
+
+  private void testAddPartitionIfNotExists(boolean blocking) throws Exception {
+    dropTable(new String[] {"T", "Tstage"});
+    
+    HiveConf.setIntVar(conf, HiveConf.ConfVars.HIVE_LOCKS_PARTITION_THRESHOLD, 1);
+    driver = Mockito.spy(driver);
+
+    HiveConf.setBoolVar(driver2.getConf(), HiveConf.ConfVars.HIVE_ACID_LOCKLESS_READS_ENABLED, !blocking);
+    driver2 = Mockito.spy(driver2);
+
+    driver.run("create table if not exists T (a int, b int) partitioned by (p string) " +
+      "stored as orc TBLPROPERTIES ('transactional'='true')");
+    //bucketed just so that we get 2 files
+    driver.run("create table if not exists Tstage (a int, b int) clustered by (a) into 2 " +
+      "buckets stored as orc TBLPROPERTIES ('transactional'='false')");
+    driver.run("insert into Tstage values(1,2),(3,4)");
+    String exportLoc = exportFolder.newFolder("1").toString();
+    driver.run("export table Tstage to '" + exportLoc + "'");
+
+    driver.compileAndRespond("select * from T");
+    List<String> res = new ArrayList<>();
+    
+    driver.lockAndRespond();
+    List<ShowLocksResponseElement> locks = getLocks();
+    Assert.assertEquals("Unexpected lock count", 1, locks.size());
+
+    checkLock(LockType.SHARED_READ,
+      LockState.ACQUIRED, "default", "T", null, locks);
+
+    DbTxnManager txnMgr2 = (DbTxnManager) TxnManagerFactory.getTxnManagerFactory().getTxnManager(conf);
+    swapTxnManager(txnMgr2);
+    driver2.compileAndRespond("alter table T add if not exists partition (p='foo') location '" + exportLoc + "/data'");
+
+    if (blocking) {
+      txnMgr2.acquireLocks(driver2.getPlan(), ctx, null, false);
+      locks = getLocks();
+
+      ShowLocksResponseElement checkLock = checkLock(LockType.EXCLUSIVE,
+        LockState.WAITING, "default", "T", null, locks);
+
+      swapTxnManager(txnMgr);
+      Mockito.doNothing().when(driver).lockAndRespond();
+      driver.run();
+      
+      driver.getFetchTask().fetch(res);
+      swapTxnManager(txnMgr2);
+
+      FieldSetter.setField(txnMgr2, txnMgr2.getClass().getDeclaredField("numStatements"), 0);
+      txnMgr2.getMS().unlock(checkLock.getLockid());
+    }
+    driver2.lockAndRespond();
+    locks = getLocks();
+    Assert.assertEquals("Unexpected lock count", blocking ? 1 : 2, locks.size());
+
+    checkLock(blocking ? LockType.EXCLUSIVE : LockType.EXCL_WRITE,
+      LockState.ACQUIRED, "default", "T", null, locks);
+
+    Mockito.doNothing().when(driver2).lockAndRespond();
+    driver2.run();
+
+    if (!blocking) {
+      swapTxnManager(txnMgr);
+      Mockito.doNothing().when(driver).lockAndRespond();
+      driver.run();
+    }
+    Mockito.reset(driver, driver2);
+    
+    driver.getFetchTask().fetch(res);
+    Assert.assertEquals("Expecting 2 rows and found " + res.size(), 0, res.size());
+    
+    driver.run("select * from T where p='foo'");
+    res = new ArrayList<>();
+    driver.getFetchTask().fetch(res);
+    Assert.assertEquals("Expecting 0 rows and found " + res.size(), 2, res.size());

Review Comment:
   I think we're expecting 2 rows here, message says 0.



##########
ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager2.java:
##########
@@ -4028,4 +4005,88 @@ public void testAddDropConstraintNonBlocking() throws Exception {
     checkLock(LockType.EXCL_WRITE,
       LockState.ACQUIRED, "default", "tab_acid", null, locks);
   }
+
+  @Test
+  public void testAddPartitionIfNotExistsNonBlocking() throws Exception {
+    testAddPartitionIfNotExists(false);
+  }
+  @Test
+  public void testAddPartitionIfNotExistsBlocking() throws Exception {
+    testAddPartitionIfNotExists(true);
+  }
+
+  private void testAddPartitionIfNotExists(boolean blocking) throws Exception {
+    dropTable(new String[] {"T", "Tstage"});
+    
+    HiveConf.setIntVar(conf, HiveConf.ConfVars.HIVE_LOCKS_PARTITION_THRESHOLD, 1);
+    driver = Mockito.spy(driver);
+
+    HiveConf.setBoolVar(driver2.getConf(), HiveConf.ConfVars.HIVE_ACID_LOCKLESS_READS_ENABLED, !blocking);
+    driver2 = Mockito.spy(driver2);
+
+    driver.run("create table if not exists T (a int, b int) partitioned by (p string) " +
+      "stored as orc TBLPROPERTIES ('transactional'='true')");
+    //bucketed just so that we get 2 files
+    driver.run("create table if not exists Tstage (a int, b int) clustered by (a) into 2 " +
+      "buckets stored as orc TBLPROPERTIES ('transactional'='false')");
+    driver.run("insert into Tstage values(1,2),(3,4)");
+    String exportLoc = exportFolder.newFolder("1").toString();
+    driver.run("export table Tstage to '" + exportLoc + "'");
+
+    driver.compileAndRespond("select * from T");
+    List<String> res = new ArrayList<>();

Review Comment:
   nit: It would be nicer to define this further down the line closer to the place where it's actually used. 
   In 4060 and 4083 so it's the same like in line 4088.





Issue Time Tracking
-------------------

            Worklog Id:     (was: 765452)
    Remaining Estimate: 0h
            Time Spent: 10m

> Non blocking ADD PARTITION if not exists
> ----------------------------------------
>
>                 Key: HIVE-26076
>                 URL: https://issues.apache.org/jira/browse/HIVE-26076
>             Project: Hive
>          Issue Type: Task
>            Reporter: Denys Kuzmenko
>            Priority: Major
>          Time Spent: 10m
>  Remaining Estimate: 0h
>
> Originally X (EXCLUSIVE) lock was used to prevent races between concurrent add partition calls with IF NOT EXISTS, however, the same behavior could be achieved with a less restrictive X-write (EXCL_WRITE) lock. 



--
This message was sent by Atlassian Jira
(v8.20.7#820007)