gt:     selectedColumns = plan.getInfo().getSelectedColumnNames();
context: /**
   * diqube: Distributed Query Base.
   *
   * Copyright (C) 2015 Bastian Gloeckle
   *
   * This file is part of diqube.
   *
   * diqube is free software: you can redistribute it and/or modify
   * it under the terms of the GNU Affero General Public License as
   * published by the Free Software Foundation, either version 3 of the
   * License, or (at your option) any later version.
   *
   * This program is distributed in the hope that it will be useful,
   * but WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   * GNU Affero General Public License for more details.
   *
   * You should have received a copy of the GNU Affero General Public License
   * along with this program.  If not, see <http://www.gnu.org/licenses/>.
   */
  package org.diqube.server.querymaster.query;
  
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Map.Entry;
  import java.util.Optional;
  import java.util.Set;
  import java.util.UUID;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.Executor;
  import java.util.concurrent.Future;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.TimeoutException;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicInteger;
  
  import org.diqube.diql.ParseException;
  import org.diqube.execution.ExecutablePlan;
  import org.diqube.execution.ExecutablePlanStep;
  import org.diqube.execution.ExecutionPercentage;
  import org.diqube.execution.RemotesTriggeredListener;
  import org.diqube.execution.consumers.AbstractThreadedColumnValueConsumer;
  import org.diqube.execution.consumers.AbstractThreadedOrderedRowIdConsumer;
  import org.diqube.execution.consumers.AbstractThreadedOverwritingRowIdConsumer;
  import org.diqube.execution.steps.ExecuteRemotePlanOnShardsStep;
  import org.diqube.executionenv.ExecutionEnvironment;
  import org.diqube.plan.ExecutionPlanBuilder;
  import org.diqube.plan.ExecutionPlanBuilderFactory;
  import org.diqube.plan.exception.ValidationException;
  import org.diqube.queries.QueryRegistry;
  import org.diqube.queries.QueryRegistry.QueryPercentHandler;
  import org.diqube.queries.QueryUuid.QueryUuidThreadState;
  import org.diqube.remote.query.thrift.RResultTable;
  import org.diqube.server.querymaster.query.validate.MasterExecutionRequestValidator;
  import org.diqube.threads.ExecutorManager;
  import org.diqube.thrift.base.thrift.RValue;
  import org.diqube.thrift.base.util.RValueUtil;
  import org.diqube.util.Triple;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.google.common.collect.Iterables;
  
  /**
   * Fully executes a diql query and provides a callback that has a {@link RResultTable} on a query master.
   * 
   * <p>
   * One instance of this class can be used only for executing one single query.
   *
   * @author Bastian Gloeckle
   */
  class MasterQueryExecutor {
    /** Milliseconds that need to be between sending out intermediary updates */
    private static final int MIN_INTERVAL_BETWEEN_INTERMEDIARY_UPDATES_MS = 300;
  
    private static final Logger logger = LoggerFactory.getLogger(MasterQueryExecutor.class);
  
    private MasterQueryExecutor.QueryExecutorCallback callback;
    private ExecutionPlanBuilderFactory executionPlanBuildeFactory;
  
    private volatile Map<Long, Map<String, Object>> valuesByRow = new ConcurrentHashMap<>();
    private List<Long> orderedRowIds;
    private Object orderedSync = new Object();
    /** Row IDs reported by a HAVING clause. These rowIDs restrict the rowIds reported by other consumers! */
    private Long[] havingRowIds;
    private Object havingRowIdsSync = new Object();
  
    private Object waiter = new Object();
  
    private AtomicBoolean valuesDone = new AtomicBoolean(false);
    private AtomicBoolean orderedDone = new AtomicBoolean(false);
    private AtomicBoolean havingDone = new AtomicBoolean(false);
    private AtomicInteger updatesWaiting = new AtomicInteger(0);
    private boolean isOrdered;
    private List<String> selectedColumns;
    private boolean createIntermediaryUpdates;
    private RResultTable lastIntermediaryResultSent = null;
    private long lastIntermediaryResultSentNanoTime = Long.MIN_VALUE;
  
    private ExecutorManager executorManager;
  
    private boolean isHaving;
  
    private QueryRegistry queryRegistry;
  
    private AtomicInteger numberOfRemotesTriggered = new AtomicInteger(100);
    private AtomicInteger percentDoneRemotesSum = new AtomicInteger(0);
  
    private ExecutionPercentage masterExecutionPercentage;
  
    private QueryPercentHandler remotePercentHandler = new QueryPercentHandler() {
      @Override
      public void newRemoteCompletionPercentDelta(short percentDeltaOfSingleRemote) {
        percentDoneRemotesSum.addAndGet(percentDeltaOfSingleRemote);
      }
    };
  
    private List<String> selectionRequests;
  
    private MasterExecutionRequestValidator masterExecutionRequestValidator;
  
    public MasterQueryExecutor(ExecutorManager executorManager, ExecutionPlanBuilderFactory executionPlanBuildeFactory,
        QueryRegistry queryRegistry, MasterExecutionRequestValidator masterExecutionRequestValidator,
        MasterQueryExecutor.QueryExecutorCallback callback, boolean createIntermediaryUpdates) {
      this.executorManager = executorManager;
      this.executionPlanBuildeFactory = executionPlanBuildeFactory;
      this.queryRegistry = queryRegistry;
      this.masterExecutionRequestValidator = masterExecutionRequestValidator;
      this.callback = callback;
      this.createIntermediaryUpdates = createIntermediaryUpdates;
    }
  
    /**
     * Prepares executing the given query.
     * 
     * <p>
     * The returned {@link Runnable} will, when called, execute the query, block the current thread until the execution is
     * done and will call the callback accordingly.
     * 
     * @return A triple consisting of the Runnable mentioned above, the {@link ExecutablePlan} that will be executed on
     *         query master and the {@link ExecuteRemotePlanOnShardsStep} if there is one available in the created plan
     *         (otherwise <code>null</code>). Note that the {@link Runnable} needs to be called in a thread that has
     *         correct {@link QueryUuidThreadState} set (e.g. using an Executor from
     *         {@link ExecutorManager#newQueryFixedThreadPoolWithTimeout(int, String, UUID, UUID)}).
     * @throws ParseException
     *           in case the query cannot be parsed.
     * @throws ValidationException
     *           in case the query cannot be validated.
     */
    public Triple<Runnable, ExecutablePlan, ExecuteRemotePlanOnShardsStep> prepareExecution(UUID queryUuid,
        UUID executionUuid, String diql) throws ParseException, ValidationException {
      ExecutionPlanBuilder planBuilder = executionPlanBuildeFactory.createExecutionPlanBuilder();
      planBuilder.fromDiql(diql);
      planBuilder.withFinalColumnValueConsumer(new AbstractThreadedColumnValueConsumer(null) {
        @Override
        protected void allSourcesAreDone() {
          valuesDone.set(true);
          scheduleUpdate();
        }
  
        @Override
        protected void doConsume(String colName, Map<Long, Object> values) {
          for (Entry<Long, Object> valueEntry : values.entrySet()) {
            if (!valuesByRow.containsKey(valueEntry.getKey())) {
              synchronized (valuesByRow) {
                if (!valuesByRow.containsKey(valueEntry.getKey())) {
                  valuesByRow.put(valueEntry.getKey(), new ConcurrentHashMap<>());
                }
              }
            }
  
            valuesByRow.get(valueEntry.getKey()).put(colName, valueEntry.getValue());
          }
          scheduleUpdate();
        }
      });
  
      planBuilder.withFinalOrderedRowIdConsumer(new AbstractThreadedOrderedRowIdConsumer(null) {
  
        @Override
        protected void allSourcesAreDone() {
          orderedDone.set(true);
          scheduleUpdate();
        }
  
        @Override
        protected void doConsumeOrderedRowIds(List<Long> rowIds) {
          synchronized (orderedSync) {
            orderedRowIds = rowIds;
          }
          scheduleUpdate();
        }
      });
  
      planBuilder.withHavingResultConsumer(new AbstractThreadedOverwritingRowIdConsumer(null) {
        @Override
        protected void allSourcesAreDone() {
          havingDone.set(true);
          scheduleUpdate();
        }
  
        @Override
        protected void doConsume(ExecutionEnvironment env, Long[] rowIds) {
          synchronized (havingRowIdsSync) {
            havingRowIds = rowIds;
          }
          scheduleUpdate();
        }
      });
  
      planBuilder.withRemotesTriggeredListener(new RemotesTriggeredListener() {
        @Override
        public void numberOfRemotesTriggered(AtomicInteger numberOfRemotes) {
          numberOfRemotesTriggered = numberOfRemotes;
        }
      });
  
      // additionally validate query according to our validator in diqube-server.
      planBuilder.withAdditionalRequestValidator(masterExecutionRequestValidator);
  
      ExecutablePlan plan = planBuilder.build();
  