pr comments

Author: 0aix

README.md

@@ -183,6 +183,12 @@ You may want queries, asserts or query metadata to use a specific date column, o
 
 Simply pass `--parameter KEY=VALUE` in the CLI and the `KEY` will be replaced with `VALUE` in all queries, query metadata and test assertions. For example, to query June 23rd 2015, you could use `--parameter DATE=2015-06-23`. If the query uses `${DATE}` in the query it will be replaced before execution with `2015-06-23`.
 
+### Query Parallelism
+
+You may want to run queries in parallel rather than sequentially especially if you have many time-consuming queries.
+
+To enable this feature, pass in `--query-parallel-enable true` when launching Validatar. By default, this will run all queries in parallel. If this number needs to be limited, pass in `--query-parallel-max VALUE` where `VALUE` is the max number of queries that should run concurrently.  
+
 ## Execution Engines
 
 ### Hive
@@ -330,6 +336,17 @@ Option                                 Description
   fully qualified classes to plug in>
 
 
+Engine Options:
+Option                              Description                           
+------                              -----------                           
+--query-parallel-enable <Boolean:   Whether or not queries should run in  
+  Query parallelism option>           parallel. (default: false)          
+--query-parallel-max <Integer: Max  The max number of queries that will   
+  query parallelism>                  run concurrently. If non-positive or
+                                      unspecified, all queries will run at
+                                      once. (default: 0)                  
+
+
 Hive engine options:
 Option (* = required)                   Description
 ---------------------                   -----------
@@ -488,10 +505,11 @@ Version | Notes
 0.5.4   | Added a flag ```--report-on-failure-only``` to only generate reports if there were failures in tests (including warnOnly) or queries
 0.5.5   | Shaded ```org.objectweb.asm``` to not clash with asm in Hadoop environments
 0.5.6   | Fixed a bug with pretty-printing results with nulls
-0.6.0   | Better reporting (show data with only the assertion columns with the assertion result column, columns now in sorted order, EMail Subject Prefix). Can now write multiple reports per invocation (specify more than one report formatter using --report-format)
+0.6.0   | Better reporting (show data with only the assertion columns with the assertion result column, columns now in sorted order, Email Subject Prefix). Can now write multiple reports per invocation (specify more than one report formatter using --report-format)
 0.6.1   | Added a flag to configure the Email reporter SMTP strategy. Use ```--email-smtp-strategy``` to pass in ```SMTP_PLAIN```, ```SMTP_TLS``` or ```SMTP_SSL```.
 0.6.2   | Bintray EOL. First rerelease of 0.6.1 on Maven Central instead
 0.6.3   | Screwdriver migration. First rerelease of 0.6.1 using Screwdriver instead of Travis.
+0.6.4   | Added support for running queries in parallel
 
 ## Members
 

src/main/java/com/yahoo/validatar/execution/EngineManager.java

@@ -30,31 +30,30 @@
 public class EngineManager extends Pluggable<Engine> implements Helpable {
     public static final String CUSTOM_ENGINE = "custom-engine";
     public static final String CUSTOM_ENGINE_DESCRIPTION = "Additional custom engine to load.";
+    public static final String QUERY_PARALLEL_ENABLE = "query-parallel-enable";
+    public static final String QUERY_PARALLEL_MAX = "query-parallel-max";
+    private static final int QUERY_PARALLEL_MIN = 1;
 
-
-    public static final String PARALLEL = "parallel";
-    public static final String THREAD_POOL_SIZE = "thread-pool-size";
-
-    protected boolean parallel;
-    protected int threadPoolSize;
+    protected boolean queryParallelEnable;
+    protected int queryParallelMax;
 
     private static final OptionParser PARSER = new OptionParser() {
         {
-            accepts(PARALLEL, "Whether or not queries should run in parallel.")
+            accepts(QUERY_PARALLEL_ENABLE, "Whether or not queries should run in parallel.")
                     .withRequiredArg()
-                    .describedAs("Parallel option")
+                    .describedAs("Query parallelism option")
                     .ofType(Boolean.class)
                     .defaultsTo(false);
-            accepts(THREAD_POOL_SIZE, "The number of queries that can run in parallel.")
+            accepts(QUERY_PARALLEL_MAX, "The max number of queries that will run concurrently. If non-positive or " +
+                                        "unspecified, all queries will run at once.")
                     .withRequiredArg()
-                    .describedAs("Thread pool size")
+                    .describedAs("Max query parallelism")
                     .ofType(Integer.class)
                     .defaultsTo(0);
             allowsUnrecognizedOptions();
         }
     };
 
-
     /**
      * The Engine classes to manage.
      */
@@ -114,8 +113,8 @@ public EngineManager(String[] arguments) {
         }
 
         OptionSet parser = PARSER.parse(arguments);
-        parallel = (Boolean) parser.valueOf(PARALLEL);
-        threadPoolSize = (Integer) parser.valueOf(THREAD_POOL_SIZE);
+        queryParallelEnable = (Boolean) parser.valueOf(QUERY_PARALLEL_ENABLE);
+        queryParallelMax = (Integer) parser.valueOf(QUERY_PARALLEL_MAX);
     }
 
     /**
@@ -138,8 +137,7 @@ protected boolean startEngines(List<Query> queries) {
         List<Query> all = queries == null ? Collections.emptyList() : queries;
         // Queries -> engine name Set -> start engine -> verify all started
         return all.stream().map(q -> q.engine).collect(Collectors.toSet())
-               .stream().map(this::startEngine)
-               .allMatch(b -> b);
+               .stream().allMatch(this::startEngine);
     }
 
     private boolean startEngine(String engine) {
@@ -163,6 +161,7 @@ private boolean startEngine(String engine) {
 
     @Override
     public void printHelp() {
+        Helpable.printHelp("Engine Options", PARSER);
         engines.values().stream().map(WorkingEngine::getEngine).forEach(Engine::printHelp);
         Helpable.printHelp("Advanced Engine Options", getPluginOptionsParser());
     }
@@ -185,12 +184,11 @@ public boolean run(List<Query> queries) {
         if (!startEngines(queries)) {
             return false;
         }
-
         // Run each query.
-        if (!parallel) {
+        if (!queryParallelEnable) {
             queries.forEach(this::run);
         } else {
-            int poolSize = threadPoolSize > 0 ? threadPoolSize : queries.size();
+            int poolSize = Math.max(queryParallelMax > 0 ? queryParallelMax : queries.size(), QUERY_PARALLEL_MIN);
             log.info("Creating a ForkJoinPool with size {}", poolSize);
             ForkJoinPool forkJoinPool = new ForkJoinPool(poolSize);
             try {

src/test/java/com/yahoo/validatar/execution/EngineManagerTest.java

@@ -261,7 +261,7 @@ public void testParallelRun() {
         query.engine = MockRunningEngine.ENGINE_NAME;
         query.name = "Foo";
 
-        String[] args = {"--parallel", "true"};
+        String[] args = {"--query-parallel-enable", "true"};
         manager = new EngineManager(args);
         manager.setEngines(engines);
 
@@ -291,9 +291,9 @@ public void testParallelRun() {
 
     @Test
     public void testConstructor() {
-        String[] args = {"--parallel", "true", "--thread-pool-size", "10"};
+        String[] args = {"--query-parallel-enable", "true", "--query-parallel-max", "10"};
         manager = new EngineManager(args);
-        Assert.assertTrue(manager.parallel);
-        Assert.assertEquals(manager.threadPoolSize, 10);
+        Assert.assertTrue(manager.queryParallelEnable);
+        Assert.assertEquals(manager.queryParallelMax, 10);
     }
 }