/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.hadoop.hive.ql;

import java.io.DataInput;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.JavaUtils;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.MetaStoreUtils;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Schema;
import org.apache.hadoop.hive.ql.exec.ConditionalTask;
import org.apache.hadoop.hive.ql.exec.ExecDriver;
import org.apache.hadoop.hive.ql.exec.FetchTask;
import org.apache.hadoop.hive.ql.exec.MapRedTask;
import org.apache.hadoop.hive.ql.exec.MoveTask;
import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.StatsTask;
import org.apache.hadoop.hive.ql.exec.TableScanOperator;
import org.apache.hadoop.hive.ql.exec.Task;
import org.apache.hadoop.hive.ql.exec.TaskFactory;
import org.apache.hadoop.hive.ql.exec.TaskResult;
import org.apache.hadoop.hive.ql.exec.TaskRunner;
import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.history.HiveHistory.Keys;
import org.apache.hadoop.hive.ql.hooks.ExecuteWithHookContext;
import org.apache.hadoop.hive.ql.hooks.Hook;
import org.apache.hadoop.hive.ql.hooks.HookContext;
import org.apache.hadoop.hive.ql.hooks.PostExecute;
import org.apache.hadoop.hive.ql.hooks.PreExecute;
import org.apache.hadoop.hive.ql.hooks.ReadEntity;
import org.apache.hadoop.hive.ql.hooks.WriteEntity;
import org.apache.hadoop.hive.ql.lockmgr.HiveLock;
import org.apache.hadoop.hive.ql.lockmgr.HiveLockManager;
import org.apache.hadoop.hive.ql.lockmgr.HiveLockManagerCtx;
import org.apache.hadoop.hive.ql.lockmgr.HiveLockMode;
import org.apache.hadoop.hive.ql.lockmgr.HiveLockObj;
import org.apache.hadoop.hive.ql.lockmgr.HiveLockObject;
import org.apache.hadoop.hive.ql.lockmgr.HiveLockObject.HiveLockObjectData;
import org.apache.hadoop.hive.ql.lockmgr.LockException;
import org.apache.hadoop.hive.ql.log.PerfLogger;
import org.apache.hadoop.hive.ql.metadata.AuthorizationException;
import org.apache.hadoop.hive.ql.metadata.DummyPartition;
import org.apache.hadoop.hive.ql.metadata.Hive;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.metadata.Partition;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.optimizer.ppr.PartitionPruner;
import org.apache.hadoop.hive.ql.parse.ASTNode;
import org.apache.hadoop.hive.ql.parse.AbstractSemanticAnalyzerHook;
import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer;
import org.apache.hadoop.hive.ql.parse.HiveSemanticAnalyzerHookContext;
import org.apache.hadoop.hive.ql.parse.HiveSemanticAnalyzerHookContextImpl;
import org.apache.hadoop.hive.ql.parse.ImportSemanticAnalyzer;
import org.apache.hadoop.hive.ql.parse.ParseContext;
import org.apache.hadoop.hive.ql.parse.ParseDriver;
import org.apache.hadoop.hive.ql.parse.ParseUtils;
import org.apache.hadoop.hive.ql.parse.PrunedPartitionList;
import org.apache.hadoop.hive.ql.parse.SemanticAnalyzer;
import org.apache.hadoop.hive.ql.parse.SemanticAnalyzerFactory;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.parse.VariableSubstitution;
import org.apache.hadoop.hive.ql.plan.ConditionalResolver;
import org.apache.hadoop.hive.ql.plan.ConditionalResolverMergeFiles;
import org.apache.hadoop.hive.ql.plan.HiveOperation;
import org.apache.hadoop.hive.ql.plan.OperatorDesc;
import org.apache.hadoop.hive.ql.plan.TableDesc;
import org.apache.hadoop.hive.ql.processors.CommandProcessor;
import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.hive.ql.session.SessionState.LogHelper;
import org.apache.hadoop.hive.serde2.ByteStream;
import org.apache.hadoop.hive.shims.ShimLoader;
import org.apache.hadoop.mapred.ClusterStatus;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.util.ReflectionUtils;

import edu.uci.ics.hivesterix.runtime.exec.HyracksExecutionEngine;
import edu.uci.ics.hivesterix.runtime.exec.IExecutionEngine;

@SuppressWarnings({ "deprecation", "unchecked", "rawtypes" })
public class Driver implements CommandProcessor {

    // hivesterix
    private IExecutionEngine engine;
    private boolean hivesterix = false;
    private Set<Task> executedConditionalTsks = new HashSet<Task>();

    static final private Log LOG = LogFactory.getLog(Driver.class.getName());
    static final private LogHelper console = new LogHelper(LOG);

    private static final Object compileMonitor = new Object();

    private int maxRows = 100;
    ByteStream.Output bos = new ByteStream.Output();

    private HiveConf conf;
    private DataInput resStream;
    private Context ctx;
    private QueryPlan plan;
    private Schema schema;
    private HiveLockManager hiveLockMgr;

    private String errorMessage;
    private String SQLState;

    // A limit on the number of threads that can be launched
    private int maxthreads;
    private static final int SLEEP_TIME = 2000;
    protected int tryCount = Integer.MAX_VALUE;

    /**
     * for backwards compatibility with current tests
     */
    public Driver(HiveConf conf) {
        this.conf = conf;

    }

    public Driver() {
        if (SessionState.get() != null) {
            conf = SessionState.get().getConf();
        }

        // hivesterix
        engine = new HyracksExecutionEngine(conf);
    }

    // hivesterix: plan printer
    public Driver(HiveConf conf, PrintWriter planPrinter) {
        this.conf = conf;
        engine = new HyracksExecutionEngine(conf, planPrinter);
    }

    public void clear() {
        this.hivesterix = false;
        this.executedConditionalTsks.clear();
    }

    private boolean checkLockManager() {
        boolean supportConcurrency = conf.getBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY);
        if (!supportConcurrency) {
            return false;
        }
        if ((hiveLockMgr == null)) {
            try {
                setLockManager();
            } catch (SemanticException e) {
                errorMessage = "FAILED: Error in semantic analysis: " + e.getMessage();
                SQLState = ErrorMsg.findSQLState(e.getMessage());
                console.printError(errorMessage, "\n" + org.apache.hadoop.util.StringUtils.stringifyException(e));
                return false;
            }
        }
        // the reason that we set the lock manager for the cxt here is because each
        // query has its own ctx object. The hiveLockMgr is shared accross the
        // same instance of Driver, which can run multiple queries.
        ctx.setHiveLockMgr(hiveLockMgr);
        return hiveLockMgr != null;
    }

    private void setLockManager() throws SemanticException {
        boolean supportConcurrency = conf.getBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY);
        if (supportConcurrency) {
            String lockMgr = conf.getVar(HiveConf.ConfVars.HIVE_LOCK_MANAGER);
            if ((lockMgr == null) || (lockMgr.isEmpty())) {
                throw new SemanticException(ErrorMsg.LOCKMGR_NOT_SPECIFIED.getMsg());
            }

            try {
                hiveLockMgr = (HiveLockManager) ReflectionUtils.newInstance(conf.getClassByName(lockMgr), conf);
                hiveLockMgr.setContext(new HiveLockManagerCtx(conf));
            } catch (Exception e) {
                // set hiveLockMgr to null just in case this invalid manager got set to
                // next query's ctx.
                if (hiveLockMgr != null) {
                    try {
                        hiveLockMgr.close();
                    } catch (LockException e1) {
                        //nothing can do here
                    }
                    hiveLockMgr = null;
                }
                throw new SemanticException(ErrorMsg.LOCKMGR_NOT_INITIALIZED.getMsg() + e.getMessage());
            }
        }
    }

    public void init() {
        Operator.resetId();
    }

    /**
     * Return the status information about the Map-Reduce cluster
     */
    public ClusterStatus getClusterStatus() throws Exception {
        ClusterStatus cs;
        try {
            JobConf job = new JobConf(conf, ExecDriver.class);
            JobClient jc = new JobClient(job);
            cs = jc.getClusterStatus();
        } catch (Exception e) {
            e.printStackTrace();
            throw e;
        }
        LOG.info("Returning cluster status: " + cs.toString());
        return cs;
    }

    public Schema getSchema() {
        return schema;
    }

    /**
     * Get a Schema with fields represented with native Hive types
     */
    public static Schema getSchema(BaseSemanticAnalyzer sem, HiveConf conf) {
        Schema schema = null;

        // If we have a plan, prefer its logical result schema if it's
        // available; otherwise, try digging out a fetch task; failing that,
        // give up.
        if (sem == null) {
            // can't get any info without a plan
        } else if (sem.getResultSchema() != null) {
            List<FieldSchema> lst = sem.getResultSchema();
            schema = new Schema(lst, null);
        } else if (sem.getFetchTask() != null) {
            FetchTask ft = sem.getFetchTask();
            TableDesc td = ft.getTblDesc();
            // partitioned tables don't have tableDesc set on the FetchTask. Instead
            // they have a list of PartitionDesc objects, each with a table desc.
            // Let's
            // try to fetch the desc for the first partition and use it's
            // deserializer.
            if (td == null && ft.getWork() != null && ft.getWork().getPartDesc() != null) {
                if (ft.getWork().getPartDesc().size() > 0) {
                    td = ft.getWork().getPartDesc().get(0).getTableDesc();
                }
            }

            if (td == null) {
                LOG.info("No returning schema.");
            } else {
                String tableName = "result";
                List<FieldSchema> lst = null;
                try {
                    lst = MetaStoreUtils.getFieldsFromDeserializer(tableName, td.getDeserializer());
                } catch (Exception e) {
                    LOG.warn("Error getting schema: " + org.apache.hadoop.util.StringUtils.stringifyException(e));
                }
                if (lst != null) {
                    schema = new Schema(lst, null);
                }
            }
        }
        if (schema == null) {
            schema = new Schema();
        }
        LOG.info("Returning Hive schema: " + schema);
        return schema;
    }

    /**
     * Get a Schema with fields represented with Thrift DDL types
     */
    public Schema getThriftSchema() throws Exception {
        Schema schema;
        try {
            schema = getSchema();
            if (schema != null) {
                List<FieldSchema> lst = schema.getFieldSchemas();
                // Go over the schema and convert type to thrift type
                if (lst != null) {
                    for (FieldSchema f : lst) {
                        f.setType(MetaStoreUtils.typeToThriftType(f.getType()));
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            throw e;
        }
        LOG.info("Returning Thrift schema: " + schema);
        return schema;
    }

    /**
     * Return the maximum number of rows returned by getResults
     */
    public int getMaxRows() {
        return maxRows;
    }

    /**
     * Set the maximum number of rows returned by getResults
     */
    public void setMaxRows(int maxRows) {
        this.maxRows = maxRows;
    }

    public boolean hasReduceTasks(List<Task<? extends Serializable>> tasks) {
        if (tasks == null) {
            return false;
        }

        boolean hasReduce = false;
        for (Task<? extends Serializable> task : tasks) {
            if (task.hasReduce()) {
                return true;
            }

            hasReduce = (hasReduce || hasReduceTasks(task.getChildTasks()));
        }
        return hasReduce;
    }

    /**
     * Compile a new query. Any currently-planned query associated with this Driver is discarded.
     * 
     * @param command
     *            The SQL query to compile.
     */
    public int compile(String command) {
        return compile(command, true);
    }

    /**
     * Hold state variables specific to each query being executed, that may not
     * be consistent in the overall SessionState
     */
    private static class QueryState {
        private HiveOperation op;
        private String cmd;
        private boolean init = false;

        /**
         * Initialize the queryState with the query state variables
         */
        public void init(HiveOperation op, String cmd) {
            this.op = op;
            this.cmd = cmd;
            this.init = true;
        }

        public boolean isInitialized() {
            return this.init;
        }

        public HiveOperation getOp() {
            return this.op;
        }

        public String getCmd() {
            return this.cmd;
        }
    }

    public void saveSession(QueryState qs) {
        SessionState oldss = SessionState.get();
        if (oldss != null && oldss.getHiveOperation() != null) {
            qs.init(oldss.getHiveOperation(), oldss.getCmd());
        }
    }

    public void restoreSession(QueryState qs) {
        SessionState ss = SessionState.get();
        if (ss != null && qs != null && qs.isInitialized()) {
            ss.setCmd(qs.getCmd());
            ss.setCommandType(qs.getOp());
        }
    }

    /**
     * Compile a new query, but potentially reset taskID counter. Not resetting task counter
     * is useful for generating re-entrant QL queries.
     * 
     * @param command
     *            The HiveQL query to compile
     * @param resetTaskIds
     *            Resets taskID counter if true.
     * @return 0 for ok
     */
    public int compile(String command, boolean resetTaskIds) {
        PerfLogger perfLogger = PerfLogger.getPerfLogger();
        perfLogger.PerfLogBegin(LOG, PerfLogger.COMPILE);

        //holder for parent command type/string when executing reentrant queries
        QueryState queryState = new QueryState();

        if (plan != null) {
            close();
            plan = null;
        }

        if (resetTaskIds) {
            TaskFactory.resetId();
        }
        saveSession(queryState);

        try {
            command = new VariableSubstitution().substitute(conf, command);
            ctx = new Context(conf);
            ctx.setTryCount(getTryCount());
            ctx.setCmd(command);
            ctx.setHDFSCleanup(true);

            ParseDriver pd = new ParseDriver();
            ASTNode tree = pd.parse(command, ctx);
            tree = ParseUtils.findRootNonNullToken(tree);

            BaseSemanticAnalyzer sem = SemanticAnalyzerFactory.get(conf, tree);
            List<AbstractSemanticAnalyzerHook> saHooks = getHooks(HiveConf.ConfVars.SEMANTIC_ANALYZER_HOOK,
                    AbstractSemanticAnalyzerHook.class);

            // Do semantic analysis and plan generation
            if (saHooks != null) {
                HiveSemanticAnalyzerHookContext hookCtx = new HiveSemanticAnalyzerHookContextImpl();
                hookCtx.setConf(conf);
                for (AbstractSemanticAnalyzerHook hook : saHooks) {
                    tree = hook.preAnalyze(hookCtx, tree);
                }
                sem.analyze(tree, ctx);
                hookCtx.update(sem);
                for (AbstractSemanticAnalyzerHook hook : saHooks) {
                    hook.postAnalyze(hookCtx, sem.getRootTasks());
                }
            } else {
                sem.analyze(tree, ctx);
            }

            LOG.info("Semantic Analysis Completed");

            // validate the plan
            sem.validate();

            plan = new QueryPlan(command, sem, perfLogger.getStartTime(PerfLogger.DRIVER_RUN));

            // test Only - serialize the query plan and deserialize it
            if ("true".equalsIgnoreCase(System.getProperty("test.serialize.qplan"))) {

                String queryPlanFileName = ctx.getLocalScratchDir(true) + Path.SEPARATOR_CHAR + "queryplan.xml";
                LOG.info("query plan = " + queryPlanFileName);
                queryPlanFileName = new Path(queryPlanFileName).toUri().getPath();

                // serialize the queryPlan
                FileOutputStream fos = new FileOutputStream(queryPlanFileName);
                Utilities.serializeQueryPlan(plan, fos);
                fos.close();

                // deserialize the queryPlan
                FileInputStream fis = new FileInputStream(queryPlanFileName);
                QueryPlan newPlan = Utilities.deserializeQueryPlan(fis, conf);
                fis.close();

                // Use the deserialized plan
                plan = newPlan;
            }

            // initialize FetchTask right here
            if (plan.getFetchTask() != null) {
                plan.getFetchTask().initialize(conf, plan, null);
            }

            // get the output schema
            schema = getSchema(sem, conf);

            //do the authorization check
            if (HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_AUTHORIZATION_ENABLED)) {
                try {
                    perfLogger.PerfLogBegin(LOG, PerfLogger.DO_AUTHORIZATION);
                    doAuthorization(sem);
                } catch (AuthorizationException authExp) {
                    console.printError("Authorization failed:" + authExp.getMessage()
                            + ". Use show grant to get more details.");
                    return 403;
                } finally {
                    perfLogger.PerfLogEnd(LOG, PerfLogger.DO_AUTHORIZATION);
                }
            }

            //restore state after we're done executing a specific query
            // hyracks run
            if (sem instanceof SemanticAnalyzer && command.toLowerCase().indexOf("create") < 0) {
                int engineRet = engine.compileJob(sem.getRootTasks());
                if (engineRet == 0) {
                    hivesterix = true;
                }
            }
            return 0;
        } catch (Exception e) {
            ErrorMsg error = ErrorMsg.getErrorMsg(e.getMessage());
            errorMessage = "FAILED: " + e.getClass().getSimpleName();
            if (error != ErrorMsg.GENERIC_ERROR) {
                errorMessage += " [Error " + error.getErrorCode() + "]:";
            }
            errorMessage += " " + e.getMessage();
            SQLState = error.getSQLState();
            console.printError(errorMessage, "\n" + org.apache.hadoop.util.StringUtils.stringifyException(e));
            return error.getErrorCode();
        } finally {
            perfLogger.PerfLogEnd(LOG, PerfLogger.COMPILE);
            restoreSession(queryState);
        }
    }

    private void doAuthorization(BaseSemanticAnalyzer sem) throws HiveException, AuthorizationException {
        HashSet<ReadEntity> inputs = sem.getInputs();
        HashSet<WriteEntity> outputs = sem.getOutputs();
        SessionState ss = SessionState.get();
        HiveOperation op = ss.getHiveOperation();
        Hive db = sem.getDb();
        if (op != null) {
            if (op.equals(HiveOperation.CREATETABLE_AS_SELECT) || op.equals(HiveOperation.CREATETABLE)) {
                ss.getAuthorizer().authorize(db.getDatabase(db.getCurrentDatabase()), null,
                        HiveOperation.CREATETABLE_AS_SELECT.getOutputRequiredPrivileges());
            } else {
                if (op.equals(HiveOperation.IMPORT)) {
                    ImportSemanticAnalyzer isa = (ImportSemanticAnalyzer) sem;
                    if (!isa.existsTable()) {
                        ss.getAuthorizer().authorize(db.getDatabase(db.getCurrentDatabase()), null,
                                HiveOperation.CREATETABLE_AS_SELECT.getOutputRequiredPrivileges());
                    }
                }
            }
            if (outputs != null && outputs.size() > 0) {
                for (WriteEntity write : outputs) {

                    if (write.getType() == WriteEntity.Type.PARTITION) {
                        Partition part = db.getPartition(write.getTable(), write.getPartition().getSpec(), false);
                        if (part != null) {
                            ss.getAuthorizer().authorize(write.getPartition(), null, op.getOutputRequiredPrivileges());
                            continue;
                        }
                    }

                    if (write.getTable() != null) {
                        ss.getAuthorizer().authorize(write.getTable(), null, op.getOutputRequiredPrivileges());
                    }
                }

            }
        }

        if (inputs != null && inputs.size() > 0) {

            Map<Table, List<String>> tab2Cols = new HashMap<Table, List<String>>();
            Map<Partition, List<String>> part2Cols = new HashMap<Partition, List<String>>();

            Map<String, Boolean> tableUsePartLevelAuth = new HashMap<String, Boolean>();
            for (ReadEntity read : inputs) {
                Table tbl = read.getTable();
                if ((read.getPartition() != null) || (tbl.isPartitioned())) {
                    String tblName = tbl.getTableName();
                    if (tableUsePartLevelAuth.get(tblName) == null) {
                        boolean usePartLevelPriv = (tbl.getParameters().get("PARTITION_LEVEL_PRIVILEGE") != null && ("TRUE"
                                .equalsIgnoreCase(tbl.getParameters().get("PARTITION_LEVEL_PRIVILEGE"))));
                        if (usePartLevelPriv) {
                            tableUsePartLevelAuth.put(tblName, Boolean.TRUE);
                        } else {
                            tableUsePartLevelAuth.put(tblName, Boolean.FALSE);
                        }
                    }
                }
            }

            if (op.equals(HiveOperation.CREATETABLE_AS_SELECT) || op.equals(HiveOperation.QUERY)) {
                SemanticAnalyzer querySem = (SemanticAnalyzer) sem;
                ParseContext parseCtx = querySem.getParseContext();
                Map<TableScanOperator, Table> tsoTopMap = parseCtx.getTopToTable();

                for (Map.Entry<String, Operator<? extends OperatorDesc>> topOpMap : querySem.getParseContext()
                        .getTopOps().entrySet()) {
                    Operator<? extends OperatorDesc> topOp = topOpMap.getValue();
                    if (topOp instanceof TableScanOperator && tsoTopMap.containsKey(topOp)) {
                        TableScanOperator tableScanOp = (TableScanOperator) topOp;
                        Table tbl = tsoTopMap.get(tableScanOp);
                        List<Integer> neededColumnIds = tableScanOp.getNeededColumnIDs();
                        List<FieldSchema> columns = tbl.getCols();
                        List<String> cols = new ArrayList<String>();
                        if (neededColumnIds != null && neededColumnIds.size() > 0) {
                            for (int i = 0; i < neededColumnIds.size(); i++) {
                                cols.add(columns.get(neededColumnIds.get(i)).getName());
                            }
                        } else {
                            for (int i = 0; i < columns.size(); i++) {
                                cols.add(columns.get(i).getName());
                            }
                        }
                        //map may not contain all sources, since input list may have been optimized out
                        //or non-existent tho such sources may still be referenced by the TableScanOperator
                        //if it's null then the partition probably doesn't exist so let's use table permission
                        if (tbl.isPartitioned() && tableUsePartLevelAuth.get(tbl.getTableName()) == Boolean.TRUE) {
                            String alias_id = topOpMap.getKey();
                            PrunedPartitionList partsList = PartitionPruner.prune(parseCtx.getTopToTable().get(topOp),
                                    parseCtx.getOpToPartPruner().get(topOp), parseCtx.getConf(), alias_id,
                                    parseCtx.getPrunedPartitions());
                            Set<Partition> parts = new HashSet<Partition>();
                            parts.addAll(partsList.getConfirmedPartns());
                            parts.addAll(partsList.getUnknownPartns());
                            for (Partition part : parts) {
                                List<String> existingCols = part2Cols.get(part);
                                if (existingCols == null) {
                                    existingCols = new ArrayList<String>();
                                }
                                existingCols.addAll(cols);
                                part2Cols.put(part, existingCols);
                            }
                        } else {
                            List<String> existingCols = tab2Cols.get(tbl);
                            if (existingCols == null) {
                                existingCols = new ArrayList<String>();
                            }
                            existingCols.addAll(cols);
                            tab2Cols.put(tbl, existingCols);
                        }
                    }
                }
            }

            // cache the results for table authorization
            Set<String> tableAuthChecked = new HashSet<String>();
            for (ReadEntity read : inputs) {
                Table tbl = read.getTable();
                if (read.getPartition() != null) {
                    Partition partition = read.getPartition();
                    tbl = partition.getTable();
                    // use partition level authorization
                    if (tableUsePartLevelAuth.get(tbl.getTableName()) == Boolean.TRUE) {
                        List<String> cols = part2Cols.get(partition);
                        if (cols != null && cols.size() > 0) {
                            ss.getAuthorizer().authorize(partition.getTable(), partition, cols,
                                    op.getInputRequiredPrivileges(), null);
                        } else {
                            ss.getAuthorizer().authorize(partition, op.getInputRequiredPrivileges(), null);
                        }
                        continue;
                    }
                }

                // if we reach here, it means it needs to do a table authorization
                // check, and the table authorization may already happened because of other
                // partitions
                if (tbl != null && !tableAuthChecked.contains(tbl.getTableName())
                        && !(tableUsePartLevelAuth.get(tbl.getTableName()) == Boolean.TRUE)) {
                    List<String> cols = tab2Cols.get(tbl);
                    if (cols != null && cols.size() > 0) {
                        ss.getAuthorizer().authorize(tbl, null, cols, op.getInputRequiredPrivileges(), null);
                    } else {
                        ss.getAuthorizer().authorize(tbl, op.getInputRequiredPrivileges(), null);
                    }
                    tableAuthChecked.add(tbl.getTableName());
                }
            }

        }
    }

    /**
     * @return The current query plan associated with this Driver, if any.
     */
    public QueryPlan getPlan() {
        return plan;
    }

    /**
     * @param t
     *            The table to be locked
     * @param p
     *            The partition to be locked
     * @param mode
     *            The mode of the lock (SHARED/EXCLUSIVE) Get the list of objects to be locked. If a
     *            partition needs to be locked (in any mode), all its parents should also be locked in
     *            SHARED mode.
     **/
    private List<HiveLockObj> getLockObjects(Table t, Partition p, HiveLockMode mode) throws SemanticException {
        List<HiveLockObj> locks = new LinkedList<HiveLockObj>();

        HiveLockObjectData lockData = new HiveLockObjectData(plan.getQueryId(), String.valueOf(System
                .currentTimeMillis()), "IMPLICIT", plan.getQueryStr());

        if (t != null) {
            locks.add(new HiveLockObj(new HiveLockObject(t, lockData), mode));
            mode = HiveLockMode.SHARED;
            locks.add(new HiveLockObj(new HiveLockObject(t.getDbName(), lockData), mode));
            return locks;
        }

        if (p != null) {
            if (!(p instanceof DummyPartition)) {
                locks.add(new HiveLockObj(new HiveLockObject(p, lockData), mode));
            }

            // All the parents are locked in shared mode
            mode = HiveLockMode.SHARED;

            // For dummy partitions, only partition name is needed
            String name = p.getName();

            if (p instanceof DummyPartition) {
                name = p.getName().split("@")[2];
            }

            String partialName = "";
            String[] partns = name.split("/");
            int len = p instanceof DummyPartition ? partns.length : partns.length - 1;
            Map<String, String> partialSpec = new LinkedHashMap<String, String>();
            for (int idx = 0; idx < len; idx++) {
                String partn = partns[idx];
                partialName += partn;
                String[] nameValue = partn.split("=");
                assert (nameValue.length == 2);
                partialSpec.put(nameValue[0], nameValue[1]);
                try {
                    locks.add(new HiveLockObj(new HiveLockObject(new DummyPartition(p.getTable(), p.getTable()
                            .getDbName() + "/" + p.getTable().getTableName() + "/" + partialName, partialSpec),
                            lockData), mode));
                    partialName += "/";
                } catch (HiveException e) {
                    throw new SemanticException(e.getMessage());
                }
            }

            locks.add(new HiveLockObj(new HiveLockObject(p.getTable(), lockData), mode));
            locks.add(new HiveLockObj(new HiveLockObject(p.getTable().getDbName(), lockData), mode));
        }
        return locks;
    }

    /**
     * Acquire read and write locks needed by the statement. The list of objects to be locked are
     * obtained from he inputs and outputs populated by the compiler. The lock acuisition scheme is
     * pretty simple. If all the locks cannot be obtained, error out. Deadlock is avoided by making
     * sure that the locks are lexicographically sorted.
     **/
    public int acquireReadWriteLocks() {
        PerfLogger perfLogger = PerfLogger.getPerfLogger();
        perfLogger.PerfLogBegin(LOG, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);

        try {
            boolean supportConcurrency = conf.getBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY);
            if (!supportConcurrency) {
                return 0;
            }

            List<HiveLockObj> lockObjects = new ArrayList<HiveLockObj>();

            // Sort all the inputs, outputs.
            // If a lock needs to be acquired on any partition, a read lock needs to be acquired on all
            // its parents also
            for (ReadEntity input : plan.getInputs()) {
                if (input.getType() == ReadEntity.Type.TABLE) {
                    lockObjects.addAll(getLockObjects(input.getTable(), null, HiveLockMode.SHARED));
                } else {
                    lockObjects.addAll(getLockObjects(null, input.getPartition(), HiveLockMode.SHARED));
                }
            }

            for (WriteEntity output : plan.getOutputs()) {
                List<HiveLockObj> lockObj = null;
                if (output.getTyp() == WriteEntity.Type.TABLE) {
                    lockObj = getLockObjects(output.getTable(), null, output.isComplete() ? HiveLockMode.EXCLUSIVE
                            : HiveLockMode.SHARED);
                } else if (output.getTyp() == WriteEntity.Type.PARTITION) {
                    lockObj = getLockObjects(null, output.getPartition(), HiveLockMode.EXCLUSIVE);
                }
                // In case of dynamic queries, it is possible to have incomplete dummy partitions
                else if (output.getTyp() == WriteEntity.Type.DUMMYPARTITION) {
                    lockObj = getLockObjects(null, output.getPartition(), HiveLockMode.SHARED);
                }

                if (lockObj != null) {
                    lockObjects.addAll(lockObj);
                    ctx.getOutputLockObjects().put(output, lockObj);
                }
            }

            if (lockObjects.isEmpty() && !ctx.isNeedLockMgr()) {
                return 0;
            }

            HiveLockObjectData lockData = new HiveLockObjectData(plan.getQueryId(), String.valueOf(System
                    .currentTimeMillis()), "IMPLICIT", plan.getQueryStr());

            // Lock the database also
            try {
                Hive db = Hive.get(conf);
                lockObjects.add(new HiveLockObj(new HiveLockObject(db.getCurrentDatabase(), lockData),
                        HiveLockMode.SHARED));
            } catch (HiveException e) {
                throw new SemanticException(e.getMessage());
            }

            List<HiveLock> hiveLocks = ctx.getHiveLockMgr().lock(lockObjects, false);

            if (hiveLocks == null) {
                throw new SemanticException(ErrorMsg.LOCK_CANNOT_BE_ACQUIRED.getMsg());
            } else {
                ctx.setHiveLocks(hiveLocks);
            }

            return (0);
        } catch (SemanticException e) {
            errorMessage = "FAILED: Error in acquiring locks: " + e.getMessage();
            SQLState = ErrorMsg.findSQLState(e.getMessage());
            console.printError(errorMessage, "\n" + org.apache.hadoop.util.StringUtils.stringifyException(e));
            return (10);
        } catch (LockException e) {
            errorMessage = "FAILED: Error in acquiring locks: " + e.getMessage();
            SQLState = ErrorMsg.findSQLState(e.getMessage());
            console.printError(errorMessage, "\n" + org.apache.hadoop.util.StringUtils.stringifyException(e));
            return (10);
        } finally {
            perfLogger.PerfLogEnd(LOG, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
        }
    }

    /**
     * @param hiveLocks
     *            list of hive locks to be released Release all the locks specified. If some of the
     *            locks have already been released, ignore them
     **/
    private void releaseLocks(List<HiveLock> hiveLocks) {
        PerfLogger perfLogger = PerfLogger.getPerfLogger();
        perfLogger.PerfLogBegin(LOG, PerfLogger.RELEASE_LOCKS);

        if (hiveLocks != null) {
            ctx.getHiveLockMgr().releaseLocks(hiveLocks);
        }
        ctx.setHiveLocks(null);

        perfLogger.PerfLogEnd(LOG, PerfLogger.RELEASE_LOCKS);
    }

    public CommandProcessorResponse run(String command) throws CommandNeedRetryException {
        errorMessage = null;
        SQLState = null;

        if (!validateConfVariables()) {
            return new CommandProcessorResponse(12, errorMessage, SQLState);
        }

        HiveDriverRunHookContext hookContext = new HiveDriverRunHookContextImpl(conf, command);
        // Get all the driver run hooks and pre-execute them.
        List<HiveDriverRunHook> driverRunHooks;
        try {
            driverRunHooks = getHooks(HiveConf.ConfVars.HIVE_DRIVER_RUN_HOOKS, HiveDriverRunHook.class);
            for (HiveDriverRunHook driverRunHook : driverRunHooks) {
                driverRunHook.preDriverRun(hookContext);
            }
        } catch (Exception e) {
            errorMessage = "FAILED: Hive Internal Error: " + Utilities.getNameMessage(e);
            SQLState = ErrorMsg.findSQLState(e.getMessage());
            console.printError(errorMessage + "\n" + org.apache.hadoop.util.StringUtils.stringifyException(e));
            return new CommandProcessorResponse(12, errorMessage, SQLState);
        }

        // Reset the perf logger
        PerfLogger perfLogger = PerfLogger.getPerfLogger(true);
        perfLogger.PerfLogBegin(LOG, PerfLogger.DRIVER_RUN);
        perfLogger.PerfLogBegin(LOG, PerfLogger.TIME_TO_SUBMIT);

        int ret;
        synchronized (compileMonitor) {
            ret = compile(command);
        }
        if (ret != 0) {
            releaseLocks(ctx.getHiveLocks());
            return new CommandProcessorResponse(ret, errorMessage, SQLState);
        }

        boolean requireLock = false;
        boolean ckLock = checkLockManager();

        if (ckLock) {
            boolean lockOnlyMapred = HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_LOCK_MAPRED_ONLY);
            if (lockOnlyMapred) {
                Queue<Task<? extends Serializable>> taskQueue = new LinkedList<Task<? extends Serializable>>();
                taskQueue.addAll(plan.getRootTasks());
                while (taskQueue.peek() != null) {
                    Task<? extends Serializable> tsk = taskQueue.remove();
                    requireLock = requireLock || tsk.requireLock();
                    if (requireLock) {
                        break;
                    }
                    if (tsk instanceof ConditionalTask) {
                        taskQueue.addAll(((ConditionalTask) tsk).getListTasks());
                    }
                    if (tsk.getChildTasks() != null) {
                        taskQueue.addAll(tsk.getChildTasks());
                    }
                    // does not add back up task here, because back up task should be the same
                    // type of the original task.
                }
            } else {
                requireLock = true;
            }
        }

        if (requireLock) {
            ret = acquireReadWriteLocks();
            if (ret != 0) {
                releaseLocks(ctx.getHiveLocks());
                return new CommandProcessorResponse(ret, errorMessage, SQLState);
            }
        }

        ret = execute();
        if (ret != 0) {
            //if needRequireLock is false, the release here will do nothing because there is no lock
            releaseLocks(ctx.getHiveLocks());
            return new CommandProcessorResponse(ret, errorMessage, SQLState);
        }

        //if needRequireLock is false, the release here will do nothing because there is no lock
        releaseLocks(ctx.getHiveLocks());

        perfLogger.PerfLogEnd(LOG, PerfLogger.DRIVER_RUN);
        perfLogger.close(LOG, plan);

        // Take all the driver run hooks and post-execute them.
        try {
            for (HiveDriverRunHook driverRunHook : driverRunHooks) {
                driverRunHook.postDriverRun(hookContext);
            }
        } catch (Exception e) {
            errorMessage = "FAILED: Hive Internal Error: " + Utilities.getNameMessage(e);
            SQLState = ErrorMsg.findSQLState(e.getMessage());
            console.printError(errorMessage + "\n" + org.apache.hadoop.util.StringUtils.stringifyException(e));
            return new CommandProcessorResponse(12, errorMessage, SQLState);
        }

        return new CommandProcessorResponse(ret);
    }

    /**
     * Validate configuration variables.
     * 
     * @return
     */
    private boolean validateConfVariables() {
        boolean valid = true;
        if ((!conf.getBoolVar(HiveConf.ConfVars.HIVE_HADOOP_SUPPORTS_SUBDIRECTORIES))
                && ((conf.getBoolVar(HiveConf.ConfVars.HADOOPMAPREDINPUTDIRRECURSIVE))
                        || (conf.getBoolVar(HiveConf.ConfVars.HIVEOPTLISTBUCKETING)) || ((conf
                            .getBoolVar(HiveConf.ConfVars.HIVE_OPTIMIZE_UNION_REMOVE))))) {
            errorMessage = "FAILED: Hive Internal Error: " + ErrorMsg.SUPPORT_DIR_MUST_TRUE_FOR_LIST_BUCKETING.getMsg();
            SQLState = ErrorMsg.findSQLState(errorMessage);
            console.printError(errorMessage + "\n");
            valid = false;
        }
        return valid;
    }

    /**
     * Returns a set of hooks specified in a configuration variable.
     * See getHooks(HiveConf.ConfVars hookConfVar, Class<T> clazz)
     * 
     * @param hookConfVar
     * @return
     * @throws Exception
     */
    private List<Hook> getHooks(HiveConf.ConfVars hookConfVar) throws Exception {
        return getHooks(hookConfVar, Hook.class);
    }

    /**
     * Returns the hooks specified in a configuration variable. The hooks are returned in a list in
     * the order they were specified in the configuration variable.
     * 
     * @param hookConfVar
     *            The configuration variable specifying a comma separated list of the hook
     *            class names.
     * @param clazz
     *            The super type of the hooks.
     * @return A list of the hooks cast as the type specified in clazz, in the order
     *         they are listed in the value of hookConfVar
     * @throws Exception
     */
    private <T extends Hook> List<T> getHooks(HiveConf.ConfVars hookConfVar, Class<T> clazz) throws Exception {

        List<T> hooks = new ArrayList<T>();
        String csHooks = conf.getVar(hookConfVar);
        if (csHooks == null) {
            return hooks;
        }

        csHooks = csHooks.trim();
        if (csHooks.equals("")) {
            return hooks;
        }

        String[] hookClasses = csHooks.split(",");

        for (String hookClass : hookClasses) {
            try {
                T hook = (T) Class.forName(hookClass.trim(), true, JavaUtils.getClassLoader()).newInstance();
                hooks.add(hook);
            } catch (ClassNotFoundException e) {
                console.printError(hookConfVar.varname + " Class not found:" + e.getMessage());
                throw e;
            }
        }

        return hooks;
    }

    public int execute() throws CommandNeedRetryException {
        // execute hivesterix plan
        if (hivesterix) {
            hivesterix = false;
            int ret = engine.executeJob();
            if (ret != 0)
                return ret;
        }

        PerfLogger perfLogger = PerfLogger.getPerfLogger();
        perfLogger.PerfLogBegin(LOG, PerfLogger.DRIVER_EXECUTE);

        boolean noName = StringUtils.isEmpty(conf.getVar(HiveConf.ConfVars.HADOOPJOBNAME));
        int maxlen = conf.getIntVar(HiveConf.ConfVars.HIVEJOBNAMELENGTH);

        String queryId = plan.getQueryId();
        String queryStr = plan.getQueryStr();

        conf.setVar(HiveConf.ConfVars.HIVEQUERYID, queryId);
        conf.setVar(HiveConf.ConfVars.HIVEQUERYSTRING, queryStr);

        conf.set("mapreduce.workflow.id", "hive_" + queryId);
        conf.set("mapreduce.workflow.name", queryStr);

        maxthreads = HiveConf.getIntVar(conf, HiveConf.ConfVars.EXECPARALLETHREADNUMBER);

        try {
            LOG.info("Starting command: " + queryStr);

            plan.setStarted();

            if (SessionState.get() != null) {
                SessionState.get().getHiveHistory().startQuery(queryStr, conf.getVar(HiveConf.ConfVars.HIVEQUERYID));
                SessionState.get().getHiveHistory().logPlanProgress(plan);
            }
            resStream = null;

            HookContext hookContext = new HookContext(plan, conf, ctx.getPathToCS());
            hookContext.setHookType(HookContext.HookType.PRE_EXEC_HOOK);

            for (Hook peh : getHooks(HiveConf.ConfVars.PREEXECHOOKS)) {
                if (peh instanceof ExecuteWithHookContext) {
                    perfLogger.PerfLogBegin(LOG, PerfLogger.PRE_HOOK + peh.getClass().getName());

                    ((ExecuteWithHookContext) peh).run(hookContext);

                    perfLogger.PerfLogEnd(LOG, PerfLogger.PRE_HOOK + peh.getClass().getName());
                } else if (peh instanceof PreExecute) {
                    perfLogger.PerfLogBegin(LOG, PerfLogger.PRE_HOOK + peh.getClass().getName());

                    ((PreExecute) peh).run(SessionState.get(), plan.getInputs(), plan.getOutputs(), ShimLoader
                            .getHadoopShims().getUGIForConf(conf));

                    perfLogger.PerfLogEnd(LOG, PerfLogger.PRE_HOOK + peh.getClass().getName());
                }
            }

            int jobs = Utilities.getMRTasks(plan.getRootTasks()).size();
            if (jobs > 0) {
                console.printInfo("Total MapReduce jobs = " + jobs);
            }
            if (SessionState.get() != null) {
                SessionState.get().getHiveHistory()
                        .setQueryProperty(queryId, Keys.QUERY_NUM_TASKS, String.valueOf(jobs));
                SessionState.get().getHiveHistory().setIdToTableMap(plan.getIdToTableNameMap());
            }
            String jobname = Utilities.abbreviate(queryStr, maxlen - 6);

            // A runtime that launches runnable tasks as separate Threads through
            // TaskRunners
            // As soon as a task isRunnable, it is put in a queue
            // At any time, at most maxthreads tasks can be running
            // The main thread polls the TaskRunners to check if they have finished.

            Queue<Task<? extends Serializable>> runnable = new ConcurrentLinkedQueue<Task<? extends Serializable>>();
            Map<TaskResult, TaskRunner> running = new HashMap<TaskResult, TaskRunner>();

            DriverContext driverCxt = new DriverContext(runnable, ctx);
            ctx.setHDFSCleanup(true);

            SessionState.get().setLastMapRedStatsList(new ArrayList<MapRedStats>());
            SessionState.get().setStackTraces(new HashMap<String, List<List<String>>>());
            SessionState.get().setLocalMapRedErrors(new HashMap<String, List<String>>());

            // Add root Tasks to runnable
            for (Task<? extends Serializable> tsk : plan.getRootTasks()) {
                // This should never happen, if it does, it's a bug with the potential to produce
                // incorrect results.
                assert tsk.getParentTasks() == null || tsk.getParentTasks().isEmpty();
                driverCxt.addToRunnable(tsk);
            }

            perfLogger.PerfLogEnd(LOG, PerfLogger.TIME_TO_SUBMIT);
            // Loop while you either have tasks running, or tasks queued up
            while (running.size() != 0 || runnable.peek() != null) {
                // Launch upto maxthreads tasks
                while (runnable.peek() != null && running.size() < maxthreads) {
                    Task<? extends Serializable> tsk = runnable.remove();
                    launchTask(tsk, queryId, noName, running, jobname, jobs, driverCxt);
                }

                // poll the Tasks to see which one completed
                TaskResult tskRes = pollTasks(running.keySet());
                TaskRunner tskRun = running.remove(tskRes);
                Task<? extends Serializable> tsk = tskRun.getTask();
                hookContext.addCompleteTask(tskRun);

                int exitVal = tskRes.getExitVal();
                if (exitVal != 0) {
                    if (tsk.ifRetryCmdWhenFail()) {
                        if (!running.isEmpty()) {
                            taskCleanup(running);
                        }
                        // in case we decided to run everything in local mode, restore the
                        // the jobtracker setting to its initial value
                        ctx.restoreOriginalTracker();
                        throw new CommandNeedRetryException();
                    }
                    Task<? extends Serializable> backupTask = tsk.getAndInitBackupTask();
                    if (backupTask != null) {
                        errorMessage = "FAILED: Execution Error, return code " + exitVal + " from "
                                + tsk.getClass().getName();
                        ErrorMsg em = ErrorMsg.getErrorMsg(exitVal);
                        if (em != null) {
                            errorMessage += ". " + em.getMsg();
                        }
                        console.printError(errorMessage);
                        errorMessage = "ATTEMPT: Execute BackupTask: " + backupTask.getClass().getName();
                        console.printError(errorMessage);

                        // add backup task to runnable
                        if (DriverContext.isLaunchable(backupTask)) {
                            driverCxt.addToRunnable(backupTask);
                        }
                        continue;

                    } else {
                        hookContext.setHookType(HookContext.HookType.ON_FAILURE_HOOK);
                        // Get all the failure execution hooks and execute them.
                        for (Hook ofh : getHooks(HiveConf.ConfVars.ONFAILUREHOOKS)) {
                            perfLogger.PerfLogBegin(LOG, PerfLogger.FAILURE_HOOK + ofh.getClass().getName());

                            ((ExecuteWithHookContext) ofh).run(hookContext);

                            perfLogger.PerfLogEnd(LOG, PerfLogger.FAILURE_HOOK + ofh.getClass().getName());
                        }

                        errorMessage = "FAILED: Execution Error, return code " + exitVal + " from "
                                + tsk.getClass().getName();
                        ErrorMsg em = ErrorMsg.getErrorMsg(exitVal);
                        if (em != null) {
                            errorMessage += ". " + em.getMsg();
                        }
                        SQLState = "08S01";
                        console.printError(errorMessage);
                        if (!running.isEmpty()) {
                            taskCleanup(running);
                        }
                        // in case we decided to run everything in local mode, restore the
                        // the jobtracker setting to its initial value
                        ctx.restoreOriginalTracker();
                        return exitVal;
                    }
                }

                if (SessionState.get() != null) {
                    SessionState.get().getHiveHistory()
                            .setTaskProperty(queryId, tsk.getId(), Keys.TASK_RET_CODE, String.valueOf(exitVal));
                    SessionState.get().getHiveHistory().endTask(queryId, tsk);
                }

                if (tsk.getChildTasks() != null) {
                    for (Task<? extends Serializable> child : tsk.getChildTasks()) {
                        // hivesterix: don't check launchable condition
                        //if(DriverContext.isLaunchable(tsk)){
                        driverCxt.addToRunnable(child);
                        //}
                    }
                }
            }

            // in case we decided to run everything in local mode, restore the
            // the jobtracker setting to its initial value
            ctx.restoreOriginalTracker();

            // remove incomplete outputs.
            // Some incomplete outputs may be added at the beginning, for eg: for dynamic partitions.
            // remove them
            HashSet<WriteEntity> remOutputs = new HashSet<WriteEntity>();
            for (WriteEntity output : plan.getOutputs()) {
                if (!output.isComplete()) {
                    remOutputs.add(output);
                }
            }

            for (WriteEntity output : remOutputs) {
                plan.getOutputs().remove(output);
            }

            hookContext.setHookType(HookContext.HookType.POST_EXEC_HOOK);
            // Get all the post execution hooks and execute them.
            for (Hook peh : getHooks(HiveConf.ConfVars.POSTEXECHOOKS)) {
                if (peh instanceof ExecuteWithHookContext) {
                    perfLogger.PerfLogBegin(LOG, PerfLogger.POST_HOOK + peh.getClass().getName());

                    ((ExecuteWithHookContext) peh).run(hookContext);

                    perfLogger.PerfLogEnd(LOG, PerfLogger.POST_HOOK + peh.getClass().getName());
                } else if (peh instanceof PostExecute) {
                    perfLogger.PerfLogBegin(LOG, PerfLogger.POST_HOOK + peh.getClass().getName());

                    ((PostExecute) peh)
                            .run(SessionState.get(), plan.getInputs(), plan.getOutputs(),
                                    (SessionState.get() != null ? SessionState.get().getLineageState().getLineageInfo()
                                            : null), ShimLoader.getHadoopShims().getUGIForConf(conf));

                    perfLogger.PerfLogEnd(LOG, PerfLogger.POST_HOOK + peh.getClass().getName());
                }
            }

            if (SessionState.get() != null) {
                SessionState.get().getHiveHistory().setQueryProperty(queryId, Keys.QUERY_RET_CODE, String.valueOf(0));
                SessionState.get().getHiveHistory().printRowCount(queryId);
            }
        } catch (CommandNeedRetryException e) {
            throw e;
        } catch (Exception e) {
            ctx.restoreOriginalTracker();
            if (SessionState.get() != null) {
                SessionState.get().getHiveHistory().setQueryProperty(queryId, Keys.QUERY_RET_CODE, String.valueOf(12));
            }
            // TODO: do better with handling types of Exception here
            errorMessage = "FAILED: Hive Internal Error: " + Utilities.getNameMessage(e);
            SQLState = "08S01";
            console.printError(errorMessage + "\n" + org.apache.hadoop.util.StringUtils.stringifyException(e));
            return (12);
        } finally {
            if (SessionState.get() != null) {
                SessionState.get().getHiveHistory().endQuery(queryId);
            }
            if (noName) {
                conf.setVar(HiveConf.ConfVars.HADOOPJOBNAME, "");
            }
            perfLogger.PerfLogEnd(LOG, PerfLogger.DRIVER_EXECUTE);

            if (SessionState.get().getLastMapRedStatsList() != null
                    && SessionState.get().getLastMapRedStatsList().size() > 0) {
                long totalCpu = 0;
                console.printInfo("MapReduce Jobs Launched: ");
                for (int i = 0; i < SessionState.get().getLastMapRedStatsList().size(); i++) {
                    console.printInfo("Job " + i + ": " + SessionState.get().getLastMapRedStatsList().get(i));
                    totalCpu += SessionState.get().getLastMapRedStatsList().get(i).getCpuMSec();
                }
                console.printInfo("Total MapReduce CPU Time Spent: " + Utilities.formatMsecToStr(totalCpu));
            }
        }
        plan.setDone();

        if (SessionState.get() != null) {
            try {
                SessionState.get().getHiveHistory().logPlanProgress(plan);
            } catch (Exception e) {
            }
        }
        console.printInfo("OK");

        return (0);
    }

    /**
     * Launches a new task
     * 
     * @param tsk
     *            task being launched
     * @param queryId
     *            Id of the query containing the task
     * @param noName
     *            whether the task has a name set
     * @param running
     *            map from taskresults to taskrunners
     * @param jobname
     *            name of the task, if it is a map-reduce job
     * @param jobs
     *            number of map-reduce jobs
     * @param cxt
     *            the driver context
     */

    public void launchTask(Task<? extends Serializable> tsk, String queryId, boolean noName,
            Map<TaskResult, TaskRunner> running, String jobname, int jobs, DriverContext cxt) {
        if (SessionState.get() != null) {
            SessionState.get().getHiveHistory().startTask(queryId, tsk, tsk.getClass().getName());
        }
        if (tsk.isMapRedTask() && !(tsk instanceof ConditionalTask)) {
            if (noName) {
                conf.setVar(HiveConf.ConfVars.HADOOPJOBNAME, jobname + "(" + tsk.getId() + ")");
            }
            conf.set("mapreduce.workflow.node.name", tsk.getId());
            Utilities.setWorkflowAdjacencies(conf, plan);
            cxt.incCurJobNo(1);
            console.printInfo("Launching Job " + cxt.getCurJobNo() + " out of " + jobs);
        }
        tsk.initialize(conf, plan, cxt);
        TaskResult tskRes = new TaskResult();
        TaskRunner tskRun = new TaskRunner(tsk, tskRes);

        // Launch Task
        //if (HiveConf.getBoolVar(conf, HiveConf.ConfVars.EXECPARALLEL) && tsk.isMapRedTask()) {
        // Launch it in the parallel mode, as a separate thread only for MR tasks
        //  tskRun.start();
        //} else {
        //  tskRun.runSequential();
        //}
        // Launch Task: hivesterix tweak
        if (tsk instanceof MapRedTask || tsk instanceof StatsTask) {
            // Launch it in the parallel mode, as a separate thread only for MR
            // tasks
            tskRes.setRunning(false);
            tskRes.setExitVal(0);
        } else if (tsk instanceof ConditionalTask) {
            ConditionalTask condTask = (ConditionalTask) tsk;
            ConditionalResolver crs = condTask.getResolver();
            if (crs instanceof ConditionalResolverMergeFiles) {
                tskRes.setRunning(false);
                tskRes.setExitVal(0);
                if (!executedConditionalTsks.contains(tsk)) {
                    List<Task<? extends Serializable>> children = condTask.getListTasks();
                    Task<? extends Serializable> selectedBranch = null;
                    for (Task<? extends Serializable> branch : children) {
                        if (branch instanceof MoveTask) {
                            selectedBranch = branch;
                            break;
                        }
                    }
                    if (selectedBranch == null) {
                        for (int i = children.size() - 1; i >= 0; i--) {
                            Task<? extends Serializable> child = children.get(i);
                            if (child instanceof MapRedTask) {
                                selectedBranch = child;
                                break;
                            }
                        }
                    }
                    executedConditionalTsks.add(tsk);
                    cxt.addToRunnable(selectedBranch);
                }
            }
        } else {
            tskRun.runSequential();
        }
        running.put(tskRes, tskRun);
        return;
    }

    /**
     * Cleans up remaining tasks in case of failure
     */
    public void taskCleanup(Map<TaskResult, TaskRunner> running) {
        for (Map.Entry<TaskResult, TaskRunner> entry : running.entrySet()) {
            if (entry.getKey().isRunning()) {
                Task<?> task = entry.getValue().getTask();
                try {
                    task.shutdown();
                } catch (Exception e) {
                    console.printError("Exception on shutting down task " + task.getId() + ": " + e);
                }
            }
        }
        running.clear();
    }

    /**
     * Polls running tasks to see if a task has ended.
     * 
     * @param results
     *            Set of result objects for running tasks
     * @return The result object for any completed/failed task
     */

    public TaskResult pollTasks(Set<TaskResult> results) {
        Iterator<TaskResult> resultIterator = results.iterator();
        while (true) {
            while (resultIterator.hasNext()) {
                TaskResult tskRes = resultIterator.next();
                if (tskRes.isRunning() == false) {
                    return tskRes;
                }
            }

            // In this loop, nothing was found
            // Sleep 10 seconds and restart
            try {
                Thread.sleep(SLEEP_TIME);
            } catch (InterruptedException ie) {
                // Do Nothing
                ;
            }
            resultIterator = results.iterator();
        }
    }

    public boolean getResults(ArrayList<String> res) throws IOException, CommandNeedRetryException {
        if (plan != null && plan.getFetchTask() != null) {
            FetchTask ft = plan.getFetchTask();
            ft.setMaxRows(maxRows);
            return ft.fetch(res);
        }

        if (resStream == null) {
            resStream = ctx.getStream();
        }
        if (resStream == null) {
            return false;
        }

        int numRows = 0;
        String row = null;

        while (numRows < maxRows) {
            if (resStream == null) {
                if (numRows > 0) {
                    return true;
                } else {
                    return false;
                }
            }

            bos.reset();
            Utilities.StreamStatus ss;
            try {
                ss = Utilities.readColumn(resStream, bos);
                if (bos.getCount() > 0) {
                    row = new String(bos.getData(), 0, bos.getCount(), "UTF-8");
                } else if (ss == Utilities.StreamStatus.TERMINATED) {
                    row = new String();
                }

                if (row != null) {
                    numRows++;
                    res.add(row);
                }
            } catch (IOException e) {
                console.printError("FAILED: Unexpected IO exception : " + e.getMessage());
                res = null;
                return false;
            }

            if (ss == Utilities.StreamStatus.EOF) {
                resStream = ctx.getStream();
            }
        }
        return true;
    }

    public int getTryCount() {
        return tryCount;
    }

    public void setTryCount(int tryCount) {
        this.tryCount = tryCount;
    }

    public int close() {
        try {
            if (plan != null) {
                FetchTask fetchTask = plan.getFetchTask();
                if (null != fetchTask) {
                    try {
                        fetchTask.clearFetch();
                    } catch (Exception e) {
                        LOG.debug(" Exception while clearing the Fetch task ", e);
                    }
                }
            }
            if (ctx != null) {
                ctx.clear();
            }
            if (null != resStream) {
                try {
                    ((FSDataInputStream) resStream).close();
                } catch (Exception e) {
                    LOG.debug(" Exception while closing the resStream ", e);
                }
            }
        } catch (Exception e) {
            console.printError("FAILED: Hive Internal Error: " + Utilities.getNameMessage(e) + "\n"
                    + org.apache.hadoop.util.StringUtils.stringifyException(e));
            return 13;
        }

        return 0;
    }

    public void destroy() {
        if (ctx != null) {
            releaseLocks(ctx.getHiveLocks());
        }

        if (hiveLockMgr != null) {
            try {
                hiveLockMgr.close();
            } catch (LockException e) {
                LOG.warn("Exception in closing hive lock manager. "
                        + org.apache.hadoop.util.StringUtils.stringifyException(e));
            }
        }
    }

    public org.apache.hadoop.hive.ql.plan.api.Query getQueryPlan() throws IOException {
        return plan.getQueryPlan();
    }
}