@Test public void testSlowPlanNode() { AtomicBoolean notInterrupted = new AtomicBoolean(); List<PlanNode> planNodes = new ArrayList<>(); for (int i = 0; i < 20; i++) { planNodes.add(() -> { try { Thread.sleep(10000); } catch (InterruptedException e) { throw new RuntimeException(e); } notInterrupted.set(true); return null; }); } CombinePlanNode combinePlanNode = new CombinePlanNode(planNodes, _queryContext, _executorService, System.currentTimeMillis() + 100, InstancePlanMakerImplV2.DEFAULT_NUM_GROUPS_LIMIT, null); try { combinePlanNode.run(); } catch (RuntimeException e) { Assert.assertTrue(e.getCause() instanceof TimeoutException); Assert.assertFalse(notInterrupted.get()); return; } Assert.fail(); }
@SuppressWarnings({"rawtypes", "unchecked"}) @Override public Operator<IntermediateResultsBlock> run() { int numPlanNodes = _planNodes.size(); List<Operator> operators = new ArrayList<>(numPlanNodes); if (numPlanNodes <= TARGET_NUM_PLANS_PER_THREAD) { for (PlanNode planNode : _planNodes) { operators.add(planNode.run()); } } else { int numThreads = Math.min((numPlanNodes + TARGET_NUM_PLANS_PER_THREAD - 1) / TARGET_NUM_PLANS_PER_THREAD, MAX_NUM_THREADS_PER_QUERY); Phaser phaser = new Phaser(1); Future[] futures = new Future[numThreads]; for (int i = 0; i < numThreads; i++) { int index = i; futures[i] = _executorService.submit(new TraceCallable<List<Operator>>() { @Override public List<Operator> callJob() { try { if (phaser.register() < 0) { return Collections.emptyList(); } List<Operator> operators = new ArrayList<>(); for (int i = index; i < numPlanNodes; i += numThreads) { operators.add(_planNodes.get(i).run()); } return operators; } finally { phaser.arriveAndDeregister(); } } }); } try { for (Future future : futures) { List<Operator> ops = (List<Operator>) future.get(_endTimeMs - System.currentTimeMillis(), TimeUnit.MILLISECONDS); operators.addAll(ops); } } catch (Exception e) { Throwable cause = e.getCause(); if (cause instanceof BadQueryRequestException) { throw (BadQueryRequestException) cause; } else { throw new RuntimeException("Caught exception while running CombinePlanNode.", e); } } finally { for (Future future : futures) { if (!future.isDone()) { future.cancel(true); } } phaser.awaitAdvance(phaser.arriveAndDeregister()); } } if (_streamObserver != null) { return new StreamingSelectionOnlyCombineOperator(operators, _queryContext, _executorService, _endTimeMs, _streamObserver); } if (QueryContextUtils.isAggregationQuery(_queryContext)) { if (_queryContext.getGroupByExpressions() == null) { return new AggregationOnlyCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } else { QueryOptions queryOptions = new QueryOptions(_queryContext.getQueryOptions()); if (queryOptions.isGroupByModeSQL()) { return new GroupByOrderByCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } return new GroupByCombineOperator(operators, _queryContext, _executorService, _endTimeMs, _numGroupsLimit); } } else { if (_queryContext.getLimit() == 0 || _queryContext.getOrderByExpressions() == null) { return new SelectionOnlyCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } else { return new SelectionOrderByCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } } }
CombinePlanNode implements PlanNode { @SuppressWarnings({"rawtypes", "unchecked"}) @Override public Operator<IntermediateResultsBlock> run() { int numPlanNodes = _planNodes.size(); List<Operator> operators = new ArrayList<>(numPlanNodes); if (numPlanNodes <= TARGET_NUM_PLANS_PER_THREAD) { for (PlanNode planNode : _planNodes) { operators.add(planNode.run()); } } else { int numThreads = Math.min((numPlanNodes + TARGET_NUM_PLANS_PER_THREAD - 1) / TARGET_NUM_PLANS_PER_THREAD, MAX_NUM_THREADS_PER_QUERY); Phaser phaser = new Phaser(1); Future[] futures = new Future[numThreads]; for (int i = 0; i < numThreads; i++) { int index = i; futures[i] = _executorService.submit(new TraceCallable<List<Operator>>() { @Override public List<Operator> callJob() { try { if (phaser.register() < 0) { return Collections.emptyList(); } List<Operator> operators = new ArrayList<>(); for (int i = index; i < numPlanNodes; i += numThreads) { operators.add(_planNodes.get(i).run()); } return operators; } finally { phaser.arriveAndDeregister(); } } }); } try { for (Future future : futures) { List<Operator> ops = (List<Operator>) future.get(_endTimeMs - System.currentTimeMillis(), TimeUnit.MILLISECONDS); operators.addAll(ops); } } catch (Exception e) { Throwable cause = e.getCause(); if (cause instanceof BadQueryRequestException) { throw (BadQueryRequestException) cause; } else { throw new RuntimeException("Caught exception while running CombinePlanNode.", e); } } finally { for (Future future : futures) { if (!future.isDone()) { future.cancel(true); } } phaser.awaitAdvance(phaser.arriveAndDeregister()); } } if (_streamObserver != null) { return new StreamingSelectionOnlyCombineOperator(operators, _queryContext, _executorService, _endTimeMs, _streamObserver); } if (QueryContextUtils.isAggregationQuery(_queryContext)) { if (_queryContext.getGroupByExpressions() == null) { return new AggregationOnlyCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } else { QueryOptions queryOptions = new QueryOptions(_queryContext.getQueryOptions()); if (queryOptions.isGroupByModeSQL()) { return new GroupByOrderByCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } return new GroupByCombineOperator(operators, _queryContext, _executorService, _endTimeMs, _numGroupsLimit); } } else { if (_queryContext.getLimit() == 0 || _queryContext.getOrderByExpressions() == null) { return new SelectionOnlyCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } else { return new SelectionOrderByCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } } } }
CombinePlanNode implements PlanNode { @SuppressWarnings({"rawtypes", "unchecked"}) @Override public Operator<IntermediateResultsBlock> run() { int numPlanNodes = _planNodes.size(); List<Operator> operators = new ArrayList<>(numPlanNodes); if (numPlanNodes <= TARGET_NUM_PLANS_PER_THREAD) { for (PlanNode planNode : _planNodes) { operators.add(planNode.run()); } } else { int numThreads = Math.min((numPlanNodes + TARGET_NUM_PLANS_PER_THREAD - 1) / TARGET_NUM_PLANS_PER_THREAD, MAX_NUM_THREADS_PER_QUERY); Phaser phaser = new Phaser(1); Future[] futures = new Future[numThreads]; for (int i = 0; i < numThreads; i++) { int index = i; futures[i] = _executorService.submit(new TraceCallable<List<Operator>>() { @Override public List<Operator> callJob() { try { if (phaser.register() < 0) { return Collections.emptyList(); } List<Operator> operators = new ArrayList<>(); for (int i = index; i < numPlanNodes; i += numThreads) { operators.add(_planNodes.get(i).run()); } return operators; } finally { phaser.arriveAndDeregister(); } } }); } try { for (Future future : futures) { List<Operator> ops = (List<Operator>) future.get(_endTimeMs - System.currentTimeMillis(), TimeUnit.MILLISECONDS); operators.addAll(ops); } } catch (Exception e) { Throwable cause = e.getCause(); if (cause instanceof BadQueryRequestException) { throw (BadQueryRequestException) cause; } else { throw new RuntimeException("Caught exception while running CombinePlanNode.", e); } } finally { for (Future future : futures) { if (!future.isDone()) { future.cancel(true); } } phaser.awaitAdvance(phaser.arriveAndDeregister()); } } if (_streamObserver != null) { return new StreamingSelectionOnlyCombineOperator(operators, _queryContext, _executorService, _endTimeMs, _streamObserver); } if (QueryContextUtils.isAggregationQuery(_queryContext)) { if (_queryContext.getGroupByExpressions() == null) { return new AggregationOnlyCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } else { QueryOptions queryOptions = new QueryOptions(_queryContext.getQueryOptions()); if (queryOptions.isGroupByModeSQL()) { return new GroupByOrderByCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } return new GroupByCombineOperator(operators, _queryContext, _executorService, _endTimeMs, _numGroupsLimit); } } else { if (_queryContext.getLimit() == 0 || _queryContext.getOrderByExpressions() == null) { return new SelectionOnlyCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } else { return new SelectionOrderByCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } } } CombinePlanNode(List<PlanNode> planNodes, QueryContext queryContext, ExecutorService executorService,
      long endTimeMs, int numGroupsLimit, @Nullable StreamObserver<Server.ServerResponse> streamObserver); }
CombinePlanNode implements PlanNode { @SuppressWarnings({"rawtypes", "unchecked"}) @Override public Operator<IntermediateResultsBlock> run() { int numPlanNodes = _planNodes.size(); List<Operator> operators = new ArrayList<>(numPlanNodes); if (numPlanNodes <= TARGET_NUM_PLANS_PER_THREAD) { for (PlanNode planNode : _planNodes) { operators.add(planNode.run()); } } else { int numThreads = Math.min((numPlanNodes + TARGET_NUM_PLANS_PER_THREAD - 1) / TARGET_NUM_PLANS_PER_THREAD, MAX_NUM_THREADS_PER_QUERY); Phaser phaser = new Phaser(1); Future[] futures = new Future[numThreads]; for (int i = 0; i < numThreads; i++) { int index = i; futures[i] = _executorService.submit(new TraceCallable<List<Operator>>() { @Override public List<Operator> callJob() { try { if (phaser.register() < 0) { return Collections.emptyList(); } List<Operator> operators = new ArrayList<>(); for (int i = index; i < numPlanNodes; i += numThreads) { operators.add(_planNodes.get(i).run()); } return operators; } finally { phaser.arriveAndDeregister(); } } }); } try { for (Future future : futures) { List<Operator> ops = (List<Operator>) future.get(_endTimeMs - System.currentTimeMillis(), TimeUnit.MILLISECONDS); operators.addAll(ops); } } catch (Exception e) { Throwable cause = e.getCause(); if (cause instanceof BadQueryRequestException) { throw (BadQueryRequestException) cause; } else { throw new RuntimeException("Caught exception while running CombinePlanNode.", e); } } finally { for (Future future : futures) { if (!future.isDone()) { future.cancel(true); } } phaser.awaitAdvance(phaser.arriveAndDeregister()); } } if (_streamObserver != null) { return new StreamingSelectionOnlyCombineOperator(operators, _queryContext, _executorService, _endTimeMs, _streamObserver); } if (QueryContextUtils.isAggregationQuery(_queryContext)) { if (_queryContext.getGroupByExpressions() == null) { return new AggregationOnlyCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } else { QueryOptions queryOptions = new QueryOptions(_queryContext.getQueryOptions()); if (queryOptions.isGroupByModeSQL()) { return new GroupByOrderByCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } return new GroupByCombineOperator(operators, _queryContext, _executorService, _endTimeMs, _numGroupsLimit); } } else { if (_queryContext.getLimit() == 0 || _queryContext.getOrderByExpressions() == null) { return new SelectionOnlyCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } else { return new SelectionOrderByCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } } } CombinePlanNode(List<PlanNode> planNodes, QueryContext queryContext, ExecutorService executorService,
      long endTimeMs, int numGroupsLimit, @Nullable StreamObserver<Server.ServerResponse> streamObserver); @SuppressWarnings({"rawtypes", "unchecked"}) @Override Operator<IntermediateResultsBlock> run(); }
CombinePlanNode implements PlanNode { @SuppressWarnings({"rawtypes", "unchecked"}) @Override public Operator<IntermediateResultsBlock> run() { int numPlanNodes = _planNodes.size(); List<Operator> operators = new ArrayList<>(numPlanNodes); if (numPlanNodes <= TARGET_NUM_PLANS_PER_THREAD) { for (PlanNode planNode : _planNodes) { operators.add(planNode.run()); } } else { int numThreads = Math.min((numPlanNodes + TARGET_NUM_PLANS_PER_THREAD - 1) / TARGET_NUM_PLANS_PER_THREAD, MAX_NUM_THREADS_PER_QUERY); Phaser phaser = new Phaser(1); Future[] futures = new Future[numThreads]; for (int i = 0; i < numThreads; i++) { int index = i; futures[i] = _executorService.submit(new TraceCallable<List<Operator>>() { @Override public List<Operator> callJob() { try { if (phaser.register() < 0) { return Collections.emptyList(); } List<Operator> operators = new ArrayList<>(); for (int i = index; i < numPlanNodes; i += numThreads) { operators.add(_planNodes.get(i).run()); } return operators; } finally { phaser.arriveAndDeregister(); } } }); } try { for (Future future : futures) { List<Operator> ops = (List<Operator>) future.get(_endTimeMs - System.currentTimeMillis(), TimeUnit.MILLISECONDS); operators.addAll(ops); } } catch (Exception e) { Throwable cause = e.getCause(); if (cause instanceof BadQueryRequestException) { throw (BadQueryRequestException) cause; } else { throw new RuntimeException("Caught exception while running CombinePlanNode.", e); } } finally { for (Future future : futures) { if (!future.isDone()) { future.cancel(true); } } phaser.awaitAdvance(phaser.arriveAndDeregister()); } } if (_streamObserver != null) { return new StreamingSelectionOnlyCombineOperator(operators, _queryContext, _executorService, _endTimeMs, _streamObserver); } if (QueryContextUtils.isAggregationQuery(_queryContext)) { if (_queryContext.getGroupByExpressions() == null) { return new AggregationOnlyCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } else { QueryOptions queryOptions = new QueryOptions(_queryContext.getQueryOptions()); if (queryOptions.isGroupByModeSQL()) { return new GroupByOrderByCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } return new GroupByCombineOperator(operators, _queryContext, _executorService, _endTimeMs, _numGroupsLimit); } } else { if (_queryContext.getLimit() == 0 || _queryContext.getOrderByExpressions() == null) { return new SelectionOnlyCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } else { return new SelectionOrderByCombineOperator(operators, _queryContext, _executorService, _endTimeMs); } } } CombinePlanNode(List<PlanNode> planNodes, QueryContext queryContext, ExecutorService executorService,
      long endTimeMs, int numGroupsLimit, @Nullable StreamObserver<Server.ServerResponse> streamObserver); @SuppressWarnings({"rawtypes", "unchecked"}) @Override Operator<IntermediateResultsBlock> run();  }