HIVE-19897: Add more tests for parallel compilation (Yongzhi Chen, reviewed by Aihua Xu)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/596edd18 Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/596edd18 Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/596edd18 Branch: refs/heads/master-txnstats Commit: 596edd181624afb5175fb98eea2c672775c2f378 Parents: 8f0973b Author: Yongzhi Chen <[email protected]> Authored: Thu Jun 21 10:08:01 2018 -0400 Committer: Yongzhi Chen <[email protected]> Committed: Thu Jun 21 10:27:11 2018 -0400 ---------------------------------------------------------------------- .../apache/hive/jdbc/TestJdbcWithMiniHS2.java | 67 ++++++++++++++++++-- 1 file changed, 62 insertions(+), 5 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/596edd18/itests/hive-unit/src/test/java/org/apache/hive/jdbc/TestJdbcWithMiniHS2.java ---------------------------------------------------------------------- diff --git a/itests/hive-unit/src/test/java/org/apache/hive/jdbc/TestJdbcWithMiniHS2.java b/itests/hive-unit/src/test/java/org/apache/hive/jdbc/TestJdbcWithMiniHS2.java index d7d7097..dba0721 100644 --- a/itests/hive-unit/src/test/java/org/apache/hive/jdbc/TestJdbcWithMiniHS2.java +++ b/itests/hive-unit/src/test/java/org/apache/hive/jdbc/TestJdbcWithMiniHS2.java @@ -281,21 +281,75 @@ public class TestJdbcWithMiniHS2 { } @Test + public void testParallelCompilation3() throws Exception { + Statement stmt = conTestDb.createStatement(); + stmt.execute("set hive.driver.parallel.compilation=true"); + stmt.execute("set hive.server2.async.exec.async.compile=true"); + stmt.close(); + Connection conn = getConnection(testDbName); + stmt = conn.createStatement(); + stmt.execute("set hive.driver.parallel.compilation=true"); + stmt.execute("set hive.server2.async.exec.async.compile=true"); + stmt.close(); + int poolSize = 100; + SynchronousQueue<Runnable> executorQueue1 = new SynchronousQueue<Runnable>(); + ExecutorService workers1 = + new ThreadPoolExecutor(1, poolSize, 20, TimeUnit.SECONDS, executorQueue1); + SynchronousQueue<Runnable> executorQueue2 = new SynchronousQueue<Runnable>(); + ExecutorService workers2 = + new ThreadPoolExecutor(1, poolSize, 20, TimeUnit.SECONDS, executorQueue2); + List<Future<Boolean>> list1 = startTasks(workers1, conTestDb, tableName, 10); + List<Future<Boolean>> list2 = startTasks(workers2, conn, tableName, 10); + finishTasks(list1, workers1); + finishTasks(list2, workers2); + conn.close(); + } + + @Test + public void testParallelCompilation4() throws Exception { + Statement stmt = conTestDb.createStatement(); + stmt.execute("set hive.driver.parallel.compilation=true"); + stmt.execute("set hive.server2.async.exec.async.compile=false"); + stmt.close(); + Connection conn = getConnection(testDbName); + stmt = conn.createStatement(); + stmt.execute("set hive.driver.parallel.compilation=true"); + stmt.execute("set hive.server2.async.exec.async.compile=false"); + stmt.close(); + int poolSize = 100; + SynchronousQueue<Runnable> executorQueue1 = new SynchronousQueue<Runnable>(); + ExecutorService workers1 = + new ThreadPoolExecutor(1, poolSize, 20, TimeUnit.SECONDS, executorQueue1); + SynchronousQueue<Runnable> executorQueue2 = new SynchronousQueue<Runnable>(); + ExecutorService workers2 = + new ThreadPoolExecutor(1, poolSize, 20, TimeUnit.SECONDS, executorQueue2); + List<Future<Boolean>> list1 = startTasks(workers1, conTestDb, tableName, 10); + List<Future<Boolean>> list2 = startTasks(workers2, conn, tableName, 10); + finishTasks(list1, workers1); + finishTasks(list2, workers2); + conn.close(); + } + + @Test public void testConcurrentStatements() throws Exception { startConcurrencyTest(conTestDb, tableName, 50); } private static void startConcurrencyTest(Connection conn, String tableName, int numTasks) { // Start concurrent testing - int POOL_SIZE = 100; - int TASK_COUNT = numTasks; - + int poolSize = 100; SynchronousQueue<Runnable> executorQueue = new SynchronousQueue<Runnable>(); ExecutorService workers = - new ThreadPoolExecutor(1, POOL_SIZE, 20, TimeUnit.SECONDS, executorQueue); + new ThreadPoolExecutor(1, poolSize, 20, TimeUnit.SECONDS, executorQueue); + List<Future<Boolean>> list = startTasks(workers, conn, tableName, numTasks); + finishTasks(list, workers); + } + + private static List<Future<Boolean>> startTasks(ExecutorService workers, Connection conn, + String tableName, int numTasks) { List<Future<Boolean>> list = new ArrayList<Future<Boolean>>(); int i = 0; - while (i < TASK_COUNT) { + while (i < numTasks) { try { Future<Boolean> future = workers.submit(new JDBCTask(conn, i, tableName)); list.add(future); @@ -308,7 +362,10 @@ public class TestJdbcWithMiniHS2 { } } } + return list; + } + private static void finishTasks(List<Future<Boolean>> list, ExecutorService workers) { for (Future<Boolean> future : list) { try { Boolean result = future.get(30, TimeUnit.SECONDS);
