/*
 * 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.pig;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.PrintStream;
import java.io.PrintWriter;
import java.io.StringReader;
import java.io.StringWriter;
import java.lang.reflect.Constructor;
import java.lang.reflect.Method;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Deque;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.compress.BZip2Codec;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.pig.backend.datastorage.ContainerDescriptor;
import org.apache.pig.backend.datastorage.DataStorage;
import org.apache.pig.backend.datastorage.ElementDescriptor;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.executionengine.ExecJob;
import org.apache.pig.backend.executionengine.ExecJob.JOB_STATUS;
import org.apache.pig.backend.hadoop.PigATSClient;
import org.apache.pig.backend.hadoop.executionengine.HJob;
import org.apache.pig.builtin.PigStorage;
import org.apache.pig.classification.InterfaceAudience;
import org.apache.pig.classification.InterfaceStability;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.io.FileLocalizer;
import org.apache.pig.impl.io.FileLocalizer.FetchFileRet;
import org.apache.pig.impl.io.compress.BZip2CodecWithExtensionBZ;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.impl.streaming.StreamingCommand;
import org.apache.pig.impl.util.LogUtils;
import org.apache.pig.impl.util.PropertiesUtil;
import org.apache.pig.impl.util.UDFContext;
import org.apache.pig.impl.util.UriUtil;
import org.apache.pig.impl.util.Utils;
import org.apache.pig.newplan.DependencyOrderWalker;
import org.apache.pig.newplan.Operator;
import org.apache.pig.newplan.logical.Util;
import org.apache.pig.newplan.logical.expression.LogicalExpressionPlan;
import org.apache.pig.newplan.logical.expression.LogicalExpressionVisitor;
import org.apache.pig.newplan.logical.expression.ScalarExpression;
import org.apache.pig.newplan.logical.optimizer.AllExpressionVisitor;
import org.apache.pig.newplan.logical.relational.LOForEach;
import org.apache.pig.newplan.logical.relational.LOLoad;
import org.apache.pig.newplan.logical.relational.LOStore;
import org.apache.pig.newplan.logical.relational.LogicalPlan;
import org.apache.pig.newplan.logical.relational.LogicalPlanData;
import org.apache.pig.newplan.logical.relational.LogicalRelationalOperator;
import org.apache.pig.newplan.logical.relational.LogicalSchema;
import org.apache.pig.parser.QueryParserDriver;
import org.apache.pig.parser.QueryParserUtils;
import org.apache.pig.pen.ExampleGenerator;
import org.apache.pig.scripting.ScriptEngine;
import org.apache.pig.tools.grunt.GruntParser;
import org.apache.pig.tools.pigstats.EmptyPigStats;
import org.apache.pig.tools.pigstats.JobStats;
import org.apache.pig.tools.pigstats.OutputStats;
import org.apache.pig.tools.pigstats.PigStats;
import org.apache.pig.tools.pigstats.PigStats.JobGraph;
import org.apache.pig.tools.pigstats.ScriptState;
import org.apache.pig.validator.BlackAndWhitelistFilter;
import org.apache.pig.validator.PigCommandFilter;

import com.google.common.annotations.VisibleForTesting;

/**
 *
 * A class for Java programs to connect to Pig. Typically a program will create a PigServer
 * instance. The programmer then registers queries using registerQuery() and
 * retrieves results using openIterator() or store(). After doing so, the
 * shutdown() method should be called to free any resources used by the current
 * PigServer instance. Not doing so could result in a memory leak.
 *
 */
@InterfaceAudience.Public
@InterfaceStability.Stable
public class PigServer {

    protected final Log log = LogFactory.getLog(getClass());

    public static final String PRETTY_PRINT_SCHEMA_PROPERTY = "pig.pretty.print.schema";
    private static final String PIG_LOCATION_CHECK_STRICT = "pig.location.check.strict";

    /*
     * The data structure to support grunt shell operations.
     * The grunt shell can only work on one graph at a time.
     * If a script is contained inside another script, the grunt
     * shell first saves the current graph on the stack and works
     * on a new graph. After the nested script is done, the grunt
     * shell pops up the saved graph and continues working on it.
     */
    protected final Deque<Graph> graphs = new LinkedList<Graph>();

    /*
     * The current Graph the grunt shell is working on.
     */
    private Graph currDAG;

    protected final PigContext pigContext;

    private String jobName;

    private String jobPriority;

    private final static AtomicInteger scopeCounter = new AtomicInteger(0);

    protected final String scope = constructScope();

    private boolean validateEachStatement = false;
    private boolean skipParseInRegisterForBatch = false;

    private final BlackAndWhitelistFilter filter;

    private String constructScope() {
        // scope servers for now as a session id

        // String user = System.getProperty("user.name", "DEFAULT_USER_ID");
        // String date = (new Date()).toString();

        // scope is not really used in the system right now. It will
        // however make your explain statements look lengthy if set to
        // username-date. For now let's simplify the scope, if a real
        // scope is needed again, we might need to update all the
        // operators to not include scope in their name().
        return "" + scopeCounter.incrementAndGet();
    }

    @VisibleForTesting
    public static void resetScope() {
        scopeCounter.set(0);
    }

    /**
     * @param execTypeString can be 'mapreduce' or 'local'.  Local mode will
     * use Hadoop's local job runner to execute the job on the local machine.
     * Mapreduce mode will connect to a cluster to execute the job. If
     * execTypeString is not one of these two, Pig will deduce the ExecutionEngine
     * if it is on the classpath and use it for the backend execution.
     * @throws ExecException
     * @throws IOException
     */
    public PigServer(String execTypeString) throws ExecException, IOException {
        this(addExecTypeProperty(PropertiesUtil.loadDefaultProperties(), execTypeString));
    }

    public PigServer(String execTypeString, Properties properties) throws ExecException, IOException {
        this(addExecTypeProperty(properties, execTypeString));
    }

    public PigServer(Properties properties) throws ExecException, IOException {
        this(new PigContext(properties));
    }

    private static Properties addExecTypeProperty(Properties properties, String execType) {
        properties.setProperty("exectype", execType);
        return properties;
    }

    /**
     * @param execType execution type to start the engine.  Local mode will
     * use Hadoop's local job runner to execute the job on the local machine.
     * Mapreduce mode will connect to a cluster to execute the job.
     * @throws ExecException
     */
    public PigServer(ExecType execType) throws ExecException {
        this(execType, PropertiesUtil.loadDefaultProperties());
    }

    public PigServer(ExecType execType, Properties properties) throws ExecException {
        this(new PigContext(execType, properties));
    }

    public PigServer(ExecType execType, Configuration conf) throws ExecException {
        this(new PigContext(execType, conf));
    }

    public PigServer(PigContext context) throws ExecException {
        this(context, true);
    }

    public PigServer(PigContext context, boolean connect) throws ExecException {
        this.pigContext = context;
        currDAG = new Graph(false);

        jobName = pigContext.getProperties().getProperty(
                PigContext.JOB_NAME,
                PigContext.JOB_NAME_PREFIX + ":DefaultJobName");

        if (connect) {
            pigContext.connect();
        }

        this.filter = new BlackAndWhitelistFilter(this);

        addHadoopProperties();
        addJarsFromProperties();
        markPredeployedJarsFromProperties();

        if (ScriptState.get() == null) {
            // If Pig was started via command line, ScriptState should have been
            // already initialized in Main. If so, we should not overwrite it.
            ScriptState.start(pigContext.getExecutionEngine().instantiateScriptState());
        }
        PigStats.start(pigContext.getExecutionEngine().instantiatePigStats());

        // log ATS event includes the caller context
        String auditId = PigATSClient.getPigAuditId(pigContext);
        String callerId = (String)pigContext.getProperties().get(PigConfiguration.CALLER_ID);
        log.info("Pig Script ID for the session: " + auditId);
        if (callerId != null) {
            log.info("Caller ID for session: " + callerId);
        }
        if (Boolean.parseBoolean(pigContext.getProperties()
                .getProperty(PigConfiguration.ENABLE_ATS))) {
            if (Boolean.parseBoolean(pigContext.getProperties()
                    .getProperty("yarn.timeline-service.enabled", "false"))) {
                PigATSClient.ATSEvent event = new PigATSClient.ATSEvent(auditId, callerId);
                try {
                    PigATSClient.getInstance().logEvent(event);
                } catch (Exception e) {
                    log.warn("Error posting to ATS: ", e);
                }
            } else {
                log.warn("ATS is disabled since"
                        + " yarn.timeline-service.enabled set to false");
            }
            
        }

        // set hdfs caller context
        Class callerContextClass = null;
        try {
            callerContextClass = Class.forName("org.apache.hadoop.ipc.CallerContext");
        } catch (ClassNotFoundException e) {
            // If pre-Hadoop 2.8.0, skip setting CallerContext
        }
        if (callerContextClass != null) {
            try {
                // Reflection for the following code since it is only available since hadoop 2.8.0:
                // CallerContext hdfsContext = new CallerContext.Builder(auditId).build();
                // CallerContext.setCurrent(hdfsContext);
                Class callerContextBuilderClass = Class.forName("org.apache.hadoop.ipc.CallerContext$Builder");
                Constructor callerContextBuilderConstruct = callerContextBuilderClass.getConstructor(String.class);
                Object builder = callerContextBuilderConstruct.newInstance(auditId);
                Method builderBuildMethod = builder.getClass().getMethod("build");
                Object hdfsContext = builderBuildMethod.invoke(builder);
                Method callerContextSetCurrentMethod = callerContextClass.getMethod("setCurrent", hdfsContext.getClass());
                callerContextSetCurrentMethod.invoke(callerContextClass, hdfsContext);
            } catch (Exception e) {
                // Shall not happen unless API change in future Hadoop commons
                throw new ExecException(e);
            }
        }
    }

    private void addHadoopProperties() throws ExecException {
        // For BZip input on hadoop 0.23/2.X
        // with PIG_BZIP_USE_HADOOP_INPUTFORMAT turned on,
        // PigTextInputFormat depends on hadoop's TextInputFormat
        // for handling bzip2 input. One problem is it only recognize 'bz2'
        // as extension and not 'bz'.
        // Adding custom BZip2 codec that returns 'bz' as extension
        // for backward compatibility.
        String codecs =
            pigContext.getProperties().getProperty("io.compression.codecs");

        if( codecs != null
            && codecs.contains(BZip2Codec.class.getCanonicalName() ) ) {
            pigContext.getProperties().setProperty("io.compression.codecs",
                codecs + ","
                + BZip2CodecWithExtensionBZ.class.getCanonicalName() );
        }
    }

    private void addJarsFromProperties() throws ExecException {
        //add jars from properties to extraJars
        String jar_str = pigContext.getProperties().getProperty("pig.additional.jars");
        if (jar_str==null) {
            jar_str = "";
        }
        jar_str = jar_str.replaceAll(File.pathSeparator, ",");
        if (!jar_str.isEmpty()) {
            jar_str += ",";
        }

        String jar_str_comma = pigContext.getProperties().getProperty("pig.additional.jars.uris");
        if (jar_str_comma!=null && !jar_str_comma.isEmpty()) {
            jar_str = jar_str + jar_str_comma;
        }

        if(jar_str != null && !jar_str.isEmpty()){
            // Use File.pathSeparator (":" on Linux, ";" on Windows)
            // to correctly handle path aggregates as they are represented
            // on the Operating System.
            for(String jar : jar_str.split(",")){
                try {
                    registerJar(jar);
                } catch (IOException e) {
                    int errCode = 4010;
                    String msg =
                        "Failed to register jar :" + jar + ". Caught exception.";
                    throw new ExecException(
                            msg,
                            errCode,
                            PigException.USER_ENVIRONMENT,
                            e
                    );
                }
            }
        }
    }

    private void markPredeployedJarsFromProperties() throws ExecException {
        // mark jars as predeployed from properties
        String jar_str = pigContext.getProperties().getProperty("pig.predeployed.jars");

        if(jar_str != null){
            // Use File.pathSeparator (":" on Linux, ";" on Windows)
            // to correctly handle path aggregates as they are represented
            // on the Operating System.
            for(String jar : jar_str.split(File.pathSeparator)){
                if (jar.length() > 0) {
                    pigContext.markJarAsPredeployed(jar);
                }
            }
        }
    }

    public PigContext getPigContext(){
        return pigContext;
    }

    /**
     * Current DAG
     *
     * @return
     */
    public Graph getCurrentDAG() {
        return this.currDAG;
    }
    /**
     * Set the logging level to DEBUG.
     */
    public void debugOn() {
        Logger.getLogger("org.apache.pig").setLevel(Level.DEBUG);
        pigContext.getLog4jProperties().setProperty("log4j.logger.org.apache.pig", Level.DEBUG.toString());
    }

    /**
     * Set the logging level to the default.
     */
    public void debugOff() {
        Logger.getLogger("org.apache.pig").setLevel(pigContext.getDefaultLogLevel());
        pigContext.getLog4jProperties().setProperty("log4j.logger.org.apache.pig", pigContext.getDefaultLogLevel().toString());
    }

    /**
     * Set the default parallelism for this job
     * @param p default number of reducers to use for this job.
     */
    public void setDefaultParallel(int p) {
        pigContext.defaultParallel = p;
    }

    /**
     * Starts batch execution mode.
     */
    public void setBatchOn() {
        log.debug("Create a new graph.");

        if (currDAG != null) {
            graphs.push(currDAG);
        }
        currDAG = new Graph(true);
    }

    /**
     * Retrieve the current execution mode.
     *
     * @return true if the execution mode is batch; false otherwise.
     */
    public boolean isBatchOn() {
        // Batch is on when there are multiple graphs on the
        // stack. That gives the right response even if multiquery was
        // turned off.
        return graphs.size() > 0;
    }

    /**
     * Returns whether there is anything to process in the current batch.
     * @throws FrontendException
     * @return true if there are no stores to process in the current
     * batch, false otherwise.
     */
    public boolean isBatchEmpty() throws FrontendException {
        if (currDAG == null) {
            int errCode = 1083;
            String msg = "setBatchOn() must be called first.";
            throw new FrontendException(msg, errCode, PigException.INPUT);
        }

        return currDAG.isBatchEmpty();
    }

    /**
     * This method parses the scripts and builds the LogicalPlan. This method
     * should be followed by {@link PigServer#executeBatch(boolean)} with
     * argument as false. Do Not use {@link PigServer#executeBatch()} after
     * calling this method as that will re-parse and build the script.
     *
     * @throws IOException
     */
    public void parseAndBuild() throws IOException {
        if (currDAG == null || !isBatchOn()) {
            int errCode = 1083;
            String msg = "setBatchOn() must be called first.";
            throw new FrontendException(msg, errCode, PigException.INPUT);
        }
        currDAG.parseQuery();
        currDAG.buildPlan( null );
    }

    /**
     * Submits a batch of Pig commands for execution.
     *
     * @return list of jobs being executed
     * @throws IOException
     */
    public List<ExecJob> executeBatch() throws IOException {
        return executeBatch(true);
    }

    /**
     * Submits a batch of Pig commands for execution. Parse and build of script
     * should be skipped if user called {@link PigServer#parseAndBuild()}
     * before. Pass false as an argument in which case.
     *
     * @param parseAndBuild
     * @return
     * @throws IOException
     */
    public List<ExecJob> executeBatch(boolean parseAndBuild) throws IOException {
        if (parseAndBuild) {
            parseAndBuild();
        }

        PigStats stats = execute();

        return getJobs(stats);
    }

    /**
     * Retrieves a list of Job objects from the PigStats object
     * @param stats
     * @return A list of ExecJob objects
     */
    protected List<ExecJob> getJobs(PigStats stats) {
        LinkedList<ExecJob> jobs = new LinkedList<ExecJob>();
        if (stats instanceof EmptyPigStats) {
            HJob job = new HJob(HJob.JOB_STATUS.COMPLETED, pigContext, stats.result(null)
                    .getPOStore(), null);
            jobs.add(job);
            return jobs;
        }
        JobGraph jGraph = stats.getJobGraph();
        Iterator<JobStats> iter = jGraph.iterator();
        while (iter.hasNext()) {
            JobStats js = iter.next();
            for (OutputStats output : js.getOutputs()) {
                if (js.isSuccessful()) {
                    jobs.add(new HJob(HJob.JOB_STATUS.COMPLETED, pigContext, output
                            .getPOStore(), output.getAlias(), stats));
                } else {
                    HJob hjob = new HJob(HJob.JOB_STATUS.FAILED, pigContext, output
                            .getPOStore(), output.getAlias(), stats);
                    hjob.setException(js.getException());
                    jobs.add(hjob);
                }
            }
        }
        return jobs;
    }

    /**
     * Discards a batch of Pig commands.
     *
     * @throws FrontendException
     */
    public void discardBatch() throws FrontendException {
        if (currDAG == null || !isBatchOn()) {
            int errCode = 1083;
            String msg = "setBatchOn() must be called first.";
            throw new FrontendException(msg, errCode, PigException.INPUT);
        }

        currDAG = graphs.pop();
    }

    /**
     * Add a path to be skipped while automatically shipping binaries for
     * streaming.
     *
     * @param path path to be skipped
     */
    public void addPathToSkip(String path) {
        pigContext.addPathToSkip(path);
    }

    /**
     * Defines an alias for the given function spec. This
     * is useful for functions that require arguments to the
     * constructor.
     *
     * @param function - the new function alias to define.
     * @param funcSpec - the FuncSpec object representing the name of
     * the function class and any arguments to constructor.
     */
    public void registerFunction(String function, FuncSpec funcSpec) {
        pigContext.registerFunction(function, funcSpec);
    }

    /**
     * Defines an alias for the given streaming command.
     *
     * @param commandAlias - the new command alias to define
     * @param command - streaming command to be executed
     */
    public void registerStreamingCommand(String commandAlias, StreamingCommand command) {
        pigContext.registerStreamCmd(commandAlias, command);
    }

    private URL locateJarFromResources(String jarName) throws IOException {
        Enumeration<URL> urls = ClassLoader.getSystemResources(jarName);
        URL resourceLocation = null;

        if (urls.hasMoreElements()) {
            resourceLocation = urls.nextElement();
        }

        if (urls.hasMoreElements()) {
            StringBuffer sb = new StringBuffer("Found multiple resources that match ");
            sb.append(jarName);
            sb.append(": ");
            sb.append(resourceLocation);

            while (urls.hasMoreElements()) {
                sb.append(urls.nextElement());
                sb.append("; ");
            }

            log.debug(sb.toString());
        }

        return resourceLocation;
    }

    /**
     * Registers a jar file. Name of the jar file can be an absolute or
     * relative path.
     *
     * If multiple resources are found with the specified name, the
     * first one is registered as returned by getSystemResources.
     * A warning is issued to inform the user.
     *
     * @param name of the jar file to register
     * @throws IOException
     */
    public void registerJar(String name) throws IOException {
        // Check if this operation is permitted
        filter.validate(PigCommandFilter.Command.REGISTER);

        if (pigContext.hasJar(name)) {
            log.debug("Ignoring duplicate registration for jar " + name);
            return;
        }

        // first try to locate jar via system resources
        // if this fails, try by using "name" as File (this preserves
        // compatibility with case when user passes absolute path or path
        // relative to current working directory.)
        if (name != null) {
            if (name.isEmpty()) {
                log.warn("Empty string specified for jar path");
                return;
            }

            URL resource = locateJarFromResources(name);

            if (resource == null) {
                FetchFileRet[] files = FileLocalizer.fetchFiles(pigContext.getProperties(), name);
                for (FetchFileRet file : files) {
                    File f = file.file;
                    if (!f.canRead()) {
                        int errCode = 4002;
                        String msg = "Can't read jar file: " + name;
                        throw new FrontendException(msg, errCode, PigException.USER_ENVIRONMENT);
                    }

                    pigContext.addJar(f.toURI().toURL(), name);
                }
            } else {
                pigContext.addJar(resource, name);
            }
        }
    }

    /**
     * Universal Scripting Language Support, see PIG-928
     *
     * @param path path of the script file
     * @param scriptingLang language keyword or scriptingEngine used to interpret the script
     * @param namespace namespace defined for functions of this script
     * @throws IOException
     */
    public void registerCode(String path, String scriptingLang, String namespace)
                             throws IOException {
        if (pigContext.scriptingUDFs.containsKey(path) &&
            pigContext.scriptingUDFs.get(path).equals(namespace)) {
            log.debug("Ignoring duplicate registration for scripting udf file " + path + " in namespace " + namespace);
            return;
        } else {
            pigContext.scriptingUDFs.put(path, namespace);
        }

        FetchFileRet ret = FileLocalizer.fetchFile(pigContext.getProperties(), path);
        File f = ret.file;
        if (!f.canRead()) {
            int errCode = 4002;
            String msg = "Can't read file: " + path;
            throw new FrontendException(msg, errCode,
                    PigException.USER_ENVIRONMENT);
        }
        String cwd = new File(".").getCanonicalPath();
        String filePath = f.getCanonicalPath();
        String nameInJar = filePath;
        // Use the relative path in the jar, if the path specified is relative
        if (!ret.didFetch) {
            if (!new File(path).isAbsolute()) {
                // In case of Oozie, the localized files are in a different
                // directory symlinked to the current directory. Canonical path will not point to cwd.
                nameInJar = path;
            } else if (filePath.equals(cwd + File.separator + path)) {
                // If user specified absolute path and it refers to cwd
                nameInJar = filePath.substring(cwd.length() + 1);
            }
        }

        pigContext.addScriptFile(nameInJar, filePath);
        if(scriptingLang != null) {
            ScriptEngine se = ScriptEngine.getInstance(scriptingLang);
            se.registerFunctions(nameInJar, namespace, pigContext);
        }
    }

    /**
     * Register a query with the Pig runtime. The query is parsed and registered, but it is not
     * executed until it is needed.
     *
     * @param query
     *            a Pig Latin expression to be evaluated.
     * @param startLine
     *            line number of the query within the whole script
     * @throws IOException
     */
    public void registerQuery(String query, int startLine) throws IOException {
        currDAG.registerQuery(query, startLine, validateEachStatement, skipParseInRegisterForBatch);
    }

    /**
     * Register a query with the Pig runtime. The query is parsed and registered, but it is not
     * executed until it is needed.  Equivalent to calling {@link #registerQuery(String, int)}
     * with startLine set to 1.
     *
     * @param query
     *            a Pig Latin expression to be evaluated.
     * @throws IOException
     */
    public void registerQuery(String query) throws IOException {
        registerQuery(query, 1);
    }

    /**
     * Register a pig script from InputStream source which is more general and extensible
     * the pig script can be from local file, then you can use FileInputStream.
     * or pig script can be in memory which you build it dynamically, the you can use ByteArrayInputStream
     * even pig script can be in remote machine, which you get wrap it as SocketInputStream
     * @param in
     * @throws IOException
     */
    public void registerScript(InputStream in) throws IOException{
        registerScript(in, null, null);
    }

    /**
     * Register a pig script from InputStream source which is more general and extensible
     * the pig script can be from local file, then you can use FileInputStream.
     * or pig script can be in memory which you build it dynamically, the you can use ByteArrayInputStream
     * even pig script can be in remote machine, which you get wrap it as SocketInputStream.
     * The parameters in the pig script will be substituted with the values in params
     * @param in
     * @param params the key is the parameter name, and the value is the parameter value
     * @throws IOException
     */
    public void registerScript(InputStream in, Map<String,String> params) throws IOException{
        registerScript(in, params, null);
    }

    /**
     * Register a pig script from InputStream source which is more general and extensible
     * the pig script can be from local file, then you can use FileInputStream.
     * or pig script can be in memory which you build it dynamically, the you can use ByteArrayInputStream
     * even pig script can be in remote machine, which you get wrap it as SocketInputStream
     * The parameters in the pig script will be substituted with the values in the parameter files
     * @param in
     * @param paramsFiles  files which have the parameter setting
     * @throws IOException
     */
    public void registerScript(InputStream in, List<String> paramsFiles) throws IOException {
        registerScript(in, null, paramsFiles);
    }

    /**
     * Register a pig script from InputStream.<br>
     * The pig script can be from local file, then you can use FileInputStream.
     * Or pig script can be in memory which you build it dynamically, the you can use ByteArrayInputStream
     * Pig script can even be in remote machine, which you get wrap it as SocketInputStream.<br>
     * The parameters in the pig script will be substituted with the values in the map and the parameter files.
     * The values in params Map will override the value in parameter file if they have the same parameter
     * @param in
     * @param params the key is the parameter name, and the value is the parameter value
     * @param paramsFiles  files which have the parameter setting
     * @throws IOException
     */
    public void registerScript(InputStream in, Map<String,String> params,List<String> paramsFiles) throws IOException {
        try {
            String substituted = pigContext.doParamSubstitution(in, paramMapToList(params), paramsFiles);
            GruntParser grunt = new GruntParser(new StringReader(substituted), this);
            grunt.setInteractive(false);
            grunt.parseStopOnError(true);
        } catch (org.apache.pig.tools.pigscript.parser.ParseException e) {
            log.error(e.getLocalizedMessage());
            throw new IOException(e);
        }
    }

    protected List<String> paramMapToList(Map<String, String> params) {
        List<String> paramList = new ArrayList<String>();
        if (params != null) {
            for (Map.Entry<String, String> entry : params.entrySet()) {
                paramList.add(entry.getKey() + "=" + entry.getValue());
            }
        }
        return paramList;
    }

    /**
     * Creates a clone of the current DAG
     * @return A Graph object which is a clone of the current DAG
     * @throws IOException
     */
    protected Graph getClonedGraph() throws IOException {
        Graph graph = currDAG.duplicate();

        if (graph == null) {
            int errCode = 2127;
            String msg = "Cloning of plan failed.";
            throw new FrontendException(msg, errCode, PigException.BUG);
        }
        return graph;
    }


    /**
     * Register a query with the Pig runtime.  The query will be read from the indicated file.
     * @param fileName file to read query from.
     * @throws IOException
     */
    public void registerScript(String fileName) throws IOException {
        registerScript(fileName, null, null);
    }

    /**
     * Register a pig script file.  The parameters in the file will be substituted with the values in params
     * @param fileName  pig script file
     * @param params  the key is the parameter name, and the value is the parameter value
     * @throws IOException
     */
    public void registerScript(String fileName, Map<String,String> params) throws IOException {
        registerScript(fileName, params, null);
    }


    /**
     * Register a pig script file.  The parameters in the file will be substituted with the values in the parameter files
     * @param fileName pig script file
     * @param paramsFiles  files which have the parameter setting
     * @throws IOException
     */
    public void registerScript(String fileName, List<String> paramsFiles) throws IOException {
        registerScript(fileName, null, paramsFiles);
    }

    /**
     * Register a pig script file.  The parameters in the file will be substituted with the values in the map and the parameter files
     * The values in params Map will override the value in parameter file if they have the same parameter
     * @param fileName  pig script
     * @param params  the key is the parameter name, and the value is the parameter value
     * @param paramsFiles   files which have the parameter setting
     * @throws IOException
     */
    public void registerScript(String fileName, Map<String,String> params,List<String> paramsFiles) throws IOException {
        FileInputStream fis = null;
        try{
            fis = new FileInputStream(fileName);
            registerScript(fis, params, paramsFiles);
        }catch (FileNotFoundException e){
            log.error(e.getLocalizedMessage());
            throw new IOException(e);
        } finally {
            if (fis != null) {
                fis.close();
            }
        }
    }

    /**
     * Intended to be used by unit tests only.
     * Print a list of all aliases in in the current Pig Latin script.  Output is written to
     * System.out.
     * @throws FrontendException
     */
    public void printAliases () throws FrontendException {
        System.out.println("aliases: " + currDAG.getAliasOp().keySet());
    }

    /**
     * Write the schema for an alias to System.out.
     * @param alias Alias whose schema will be written out
     * @return Schema of alias dumped
     * @throws IOException
     */
    public Schema dumpSchema(String alias) throws IOException {
        try {
            pigContext.inDumpSchema = true;
            if ("@".equals(alias)) {
                alias = getLastRel();
            }
            LogicalRelationalOperator op = getOperatorForAlias( alias );
            LogicalSchema schema = op.getSchema();

            boolean pretty = "true".equals(pigContext.getProperties()
                                   .getProperty(PRETTY_PRINT_SCHEMA_PROPERTY));

            if (schema != null) {
                Schema s = org.apache.pig.newplan.logical.Util.translateSchema(schema);
                System.out.println(alias + ": " + (pretty ? s.prettyPrint() : s.toString()));
                return s;
            } else {
                System.out.println("Schema for " + alias + " unknown.");
                return null;
            }
        } catch (FrontendException fee) {
            int errCode = 1001;
            String msg = "Unable to describe schema for alias " + alias;
            throw new FrontendException (msg, errCode, PigException.INPUT, false, null, fee);
        } finally {
            pigContext.inDumpSchema = false;
        }
    }

    /**
     * Write the schema for a nestedAlias to System.out. Denoted by
     * alias::nestedAlias.
     *
     * @param alias Alias whose schema has nestedAlias
     * @param nestedAlias Alias whose schema will be written out
     * @return Schema of alias dumped
     * @throws IOException
     */
    public Schema dumpSchemaNested(String alias, String nestedAlias) throws IOException {
        try {
            pigContext.inDumpSchema = true;
            if ("@".equals(alias)) {
                alias = getLastRel();
            }
            Operator op = getOperatorForAlias( alias );
            if( op instanceof LOForEach ) {
                LogicalSchema nestedSc = ((LOForEach)op).dumpNestedSchema(alias, nestedAlias);
                if (nestedSc!=null) {
                    Schema s = org.apache.pig.newplan.logical.Util.translateSchema(nestedSc);
                    System.out.println(alias+ "::" + nestedAlias + ": " + s.toString());
                    return s;
                }
                else {
                    System.out.println("Schema for "+ alias+ "::" + nestedAlias + " unknown.");
                    return null;
                }
            }
            else {
                int errCode = 1001;
                String msg = "Unable to describe schema for " + alias + "::" + nestedAlias;
                throw new FrontendException (msg, errCode, PigException.INPUT, false, null);
            }
        } finally {
            pigContext.inDumpSchema = false;
        }
    }

    /**
     * Set the name of the job.  This name will get translated to mapred.job.name.
     * @param name of job
     */
    public void setJobName(String name) {
        jobName = PigContext.JOB_NAME_PREFIX + ":" + name;
    }

    /**
     * Set Hadoop job priority.  This value will get translated to mapred.job.priority.
     * @param priority valid values are found in {@link org.apache.hadoop.mapred.JobPriority}
     */
    public void setJobPriority(String priority) {
        jobPriority = priority;
    }

    /**
     * Executes a Pig Latin script up to and including indicated alias.  That is, if a user does:
     * <pre>
     * PigServer server = new PigServer();
     * server.registerQuery("A = load 'foo';");
     * server.registerQuery("B = filter A by $0 &gt; 0;");
     * server.registerQuery("C = order B by $1;");
     * </pre>
     * Then
     * <pre>
     * server.openIterator("B");
     * </pre>
     * filtered but unsorted data will be returned.  If instead a user does
     * <pre>
     * server.openIterator("C");
     * </pre>
     * filtered and sorted data will be returned.
     * @param id Alias to open iterator for
     * @return iterator of tuples returned from the script
     * @throws IOException
     */
    public Iterator<Tuple> openIterator(String id) throws IOException {
        try {
            pigContext.getProperties().setProperty( PigContext.JOB_NAME, jobName );
            if( jobPriority != null ) {
                pigContext.getProperties().setProperty( PigContext.JOB_PRIORITY, jobPriority );
            }
            ExecJob job = store(id, FileLocalizer.getTemporaryPath(pigContext)
                    .toString(), Utils.getTmpFileCompressorName(pigContext)
                    + "()");

            // invocation of "execute" is synchronous!

            if (job.getStatus() == JOB_STATUS.COMPLETED) {
                return job.getResults();
            } else if (job.getStatus() == JOB_STATUS.FAILED
                    && job.getException() != null) {
                // throw the backend exception in the failed case
                Exception e = job.getException();
                int errCode = 1066;
                String msg = "Unable to open iterator for alias " + id
                        + ". Backend error : " + e.getMessage();
                throw new FrontendException(msg, errCode, PigException.INPUT, e);
            } else {
                throw new IOException("Job terminated with anomalous status "
                        + job.getStatus().toString());
            }
        } catch (FrontendException e) {
            throw e;
        } catch (Exception e) {
            int errCode = 1066;
            String msg = "Unable to open iterator for alias " + id;
            throw new FrontendException(msg, errCode, PigException.INPUT, e);
        }
    }

    /**
     * Executes a Pig Latin script up to and including indicated alias and stores the resulting
     * records into a file.  That is, if a user does:
     * <pre>
     * PigServer server = new PigServer();
     * server.registerQuery("A = load 'foo';");
     * server.registerQuery("B = filter A by $0 &gt; 0;");
     * server.registerQuery("C = order B by $1;");
     * </pre>
     * Then
     * <pre>
     * server.store("B", "bar");
     * </pre>
     * filtered but unsorted data will be stored to the file <tt>bar</tt>.  If instead a user does
     * <pre>
     * server.store("C", "bar");
     * </pre>
     * filtered and sorted data will be stored to the file <tt>bar</tt>.
     * Equivalent to calling {@link #store(String, String, String)} with
     * <tt>org.apache.pig.PigStorage</tt> as the store function.
     * @param id The alias to store
     * @param filename The file to which to store to
     * @return {@link ExecJob} containing information about this job
     * @throws IOException
     */
    public ExecJob store(String id, String filename) throws IOException {
        return store(id, filename, PigStorage.class.getName() + "()");   // SFPig is the default store function
    }

    /**
     * Executes a Pig Latin script up to and including indicated alias and stores the resulting
     * records into a file.  That is, if a user does:
     * <pre>
     * PigServer server = new PigServer();
     * server.registerQuery("A = load 'foo';");
     * server.registerQuery("B = filter A by $0 &gt; 0;");
     * server.registerQuery("C = order B by $1;");
     * </pre>
     * Then
     * <pre>
     * server.store("B", "bar", "mystorefunc");
     * </pre>
     * filtered but unsorted data will be stored to the file <tt>bar</tt> using
     * <tt>mystorefunc</tt>.  If instead a user does
     * <pre>
     * server.store("C", "bar", "mystorefunc");
     * </pre>
     * filtered and sorted data will be stored to the file <tt>bar</tt> using
     * <tt>mystorefunc</tt>.
     * <p>
     * @param id The alias to store
     * @param filename The file to which to store to
     * @param func store function to use
     * @return {@link ExecJob} containing information about this job
     * @throws IOException
     */
    public ExecJob store(String id, String filename, String func)
            throws IOException {
        PigStats stats = storeEx(id, filename, func);
        if (stats.getOutputStats().size() < 1) {
            throw new IOException("Couldn't retrieve job.");
        }
        OutputStats output = stats.getOutputStats().get(0);

        if(stats.isSuccessful()){
            return  new HJob(JOB_STATUS.COMPLETED, pigContext, output
                    .getPOStore(), output.getAlias(), stats);
        }else{
            HJob job = new HJob(JOB_STATUS.FAILED, pigContext,
                    output.getPOStore(), output.getAlias(), stats);

            //check for exception
            Exception ex = null;
            for(JobStats js : stats.getJobGraph()){
                if(js.getException() != null) {
                    ex = js.getException();
                }
            }
            job.setException(ex);
            return job;
        }
    }

    private PigStats storeEx(String alias, String filename, String func)
    throws IOException {
        if ("@".equals(alias)) {
            alias = getLastRel();
        }
        currDAG.parseQuery();
        currDAG.skipStores(); // skip the stores that have already been processed
        currDAG.buildPlan( alias );

        try {
            QueryParserUtils.attachStorePlan(scope, currDAG.lp, filename, func, currDAG.getOperator( alias ), alias, pigContext);
            currDAG.compile();
            return executeCompiledLogicalPlan();
        } catch (PigException e) {
            int errCode = 1002;
            String msg = "Unable to store alias " + alias;
            throw new PigException(msg, errCode, PigException.INPUT, e);
        }
    }

    /**
     * Provide information on how a pig query will be executed.  For now
     * this information is very developer focussed, and probably not very
     * useful to the average user.
     * @param alias Name of alias to explain.
     * @param stream PrintStream to write explanation to.
     * @throws IOException if the requested alias cannot be found.
     */
    public void explain(String alias,
                        PrintStream stream) throws IOException {
        explain(alias, "text", true, false, stream, stream, null, null);
    }

    /**
     * Provide information on how a pig query will be executed.
     * @param alias Name of alias to explain.
     * @param format Format in which the explain should be printed.  If text, then the plan will
     * be printed in plain text.  Otherwise, the execution plan will be printed in
     * <a href="http://en.wikipedia.org/wiki/DOT_language">DOT</a> format.
     * @param verbose Controls the amount of information printed
     * @param markAsExecute When set will treat the explain like a
     * call to execute in the respoect that all the pending stores are
     * marked as complete.
     * @param lps Stream to print the logical tree
     * @param eps Stream to print the ExecutionEngine trees. If null, then will print to files
     * @param dir Directory to print ExecutionEngine trees. If null, will use eps
     * @param suffix Suffix of file names
     * @throws IOException if the requested alias cannot be found.
     */
    public void explain(String alias,
                        String format,
                        boolean verbose,
                        boolean markAsExecute,
                        PrintStream lps,
                        PrintStream eps,
                        File dir,
                        String suffix) throws IOException {
        try {
            pigContext.inExplain = true;
            buildStorePlan( alias );
            currDAG.lp.optimize(pigContext);

            //Only add root xml node if all plans are being written to same stream.
            if (format == "xml" && lps == eps) {
                lps.println("<plan>");
            }

            currDAG.lp.explain(lps, format, verbose);

            if( currDAG.lp.size() == 0 ) {
                if (format == "xml" && lps == eps) {
                    lps.println("</plan>");
                }
                return;
            }

            pigContext.getExecutionEngine().explain(currDAG.lp, pigContext, eps, format, verbose, dir, suffix );

            if (format.equals("xml") && lps == eps) {
                lps.println("</plan>");
            }

            if (markAsExecute) {
                currDAG.markAsExecuted();
            }
        } catch (Exception e) {
            int errCode = 1067;
            String msg = "Unable to explain alias " + alias;
            throw new FrontendException(msg, errCode, PigException.INPUT, e);
        } finally {
            pigContext.inExplain = false;
        }
    }

    /**
     * Returns the unused byte capacity of an HDFS filesystem. This value does
     * not take into account a replication factor, as that can vary from file
     * to file. Thus if you are using this to determine if you data set will fit
     * in the HDFS, you need to divide the result of this call by your specific replication
     * setting.
     * @return unused byte capacity of the file system.
     * @throws IOException
     */
    public long capacity() throws IOException {
        if (pigContext.getExecType().isLocal()) {
            throw new IOException("capacity only supported for non-local execution");
        }
        else {
            DataStorage dds = pigContext.getDfs();

            Map<String, Object> stats = dds.getStatistics();

            String rawCapacityStr = (String) stats.get(DataStorage.RAW_CAPACITY_KEY);
            String rawUsedStr = (String) stats.get(DataStorage.RAW_USED_KEY);

            if ((rawCapacityStr == null) || (rawUsedStr == null)) {
                throw new IOException("Failed to retrieve capacity stats");
            }

            long rawCapacityBytes = new Long(rawCapacityStr).longValue();
            long rawUsedBytes = new Long(rawUsedStr).longValue();

            return rawCapacityBytes - rawUsedBytes;
        }
    }

    /**
     * Returns the length of a file in bytes which exists in the HDFS (accounts for replication).
     * @param filename
     * @return length of the file in bytes
     * @throws IOException
     */
    public long fileSize(String filename) throws IOException {
        DataStorage dfs = pigContext.getDfs();
        ElementDescriptor elem = dfs.asElement(filename);
        Map<String, Object> stats = elem.getStatistics();
        long length = (Long) stats.get(ElementDescriptor.LENGTH_KEY);
        int replication = (Short) stats
                .get(ElementDescriptor.BLOCK_REPLICATION_KEY);

        return length * replication;
    }

    /**
     * Test whether a file exists.
     * @param filename to test
     * @return true if file exists, false otherwise
     * @throws IOException
     */
    public boolean existsFile(String filename) throws IOException {
        ElementDescriptor elem = pigContext.getDfs().asElement(filename);
        return elem.exists();
    }

    /**
     * Delete a file.
     * @param filename to delete
     * @return true
     * @throws IOException
     */
    public boolean deleteFile(String filename) throws IOException {
        // Check if this operation is permitted
        filter.validate(PigCommandFilter.Command.RM);
        filter.validate(PigCommandFilter.Command.RMF);

        ElementDescriptor elem = pigContext.getDfs().asElement(filename);
        elem.delete();
        return true;
   }

    /**
     * Rename a file.
     * @param source file to rename
     * @param target new file name
     * @return true
     * @throws IOException
     */
    public boolean renameFile(String source, String target) throws IOException {
        // Check if this operation is permitted
        filter.validate(PigCommandFilter.Command.MV);

        pigContext.rename(source, target);
        return true;
    }

    /**
     * Make a directory.
     * @param dirs directory to make
     * @return true
     * @throws IOException
     */
    public boolean mkdirs(String dirs) throws IOException {
        // Check if this operation is permitted
        filter.validate(PigCommandFilter.Command.MKDIR);

        ContainerDescriptor container = pigContext.getDfs().asContainer(dirs);
        container.create();
        return true;
    }

    /**
     * List the contents of a directory.
     * @param dir name of directory to list
     * @return array of strings, one for each file name
     * @throws IOException
     */
    public String[] listPaths(String dir) throws IOException {
        // Check if this operation is permitted
        filter.validate(PigCommandFilter.Command.LS);

        Collection<String> allPaths = new ArrayList<String>();
        ContainerDescriptor container = pigContext.getDfs().asContainer(dir);
        Iterator<ElementDescriptor> iter = container.iterator();

        while (iter.hasNext()) {
            ElementDescriptor elem = iter.next();
            allPaths.add(elem.toString());
        }

        String[] type = new String[1];
        return allPaths.toArray(type);
    }

    /**
     * Return a map containing the logical plan associated with each alias.
     *
     * @return map
     */
    public Map<String, LogicalPlan> getAliases() {
        Map<String, LogicalPlan> aliasPlans = new HashMap<String, LogicalPlan>();
        for (LogicalRelationalOperator op : currDAG.getAliases().keySet()) {
            String alias = op.getAlias();
            if(null != alias) {
                aliasPlans.put(alias, currDAG.getAliases().get(op));
            }
        }
        return aliasPlans;
    }

    /**
     * Reclaims resources used by this instance of PigServer. This method
     * deletes all temporary files generated by the current thread while
     * executing Pig commands.
     */
    public void shutdown() {
        // clean-up activities
        // TODO: reclaim scope to free up resources. Currently
        // this is not implemented and throws an exception
        // hence, for now, we won't call it.
        //
        // pigContext.getExecutionEngine().reclaimScope(this.scope);

        FileLocalizer.deleteTempFiles();
    }

    /**
     * Get the set of all current aliases.
     * @return set
     */
    public Set<String> getAliasKeySet() {
        return currDAG.getAliasOp().keySet();
    }

    public Map<Operator, DataBag> getExamples(String alias) throws IOException {
        try {
            if (currDAG.isBatchOn() && alias != null) {
                currDAG.parseQuery();
                currDAG.buildPlan( null );
                execute();
            }
            currDAG.parseQuery();
            currDAG.skipStores();
            currDAG.buildPlan( alias );
            currDAG.compile();
        } catch (IOException e) {
            //Since the original script is parsed anyway, there should not be an
            //error in this parsing. The only reason there can be an error is when
            //the files being loaded in load don't exist anymore.
            e.printStackTrace();
        }

        ExampleGenerator exgen = new ExampleGenerator( currDAG.lp, pigContext );
        try {
            return exgen.getExamples();
        } catch (ExecException e) {
            e.printStackTrace(System.out);
            throw new IOException("ExecException" , e);
        } catch (Exception e) {
            e.printStackTrace(System.out);
            throw new IOException("Exception ", e);
        }

    }

    public void printHistory(boolean withNumbers) {

        List<String> sc = currDAG.getScriptCache();

        if(!sc.isEmpty()) {
            for(int i = 0 ; i < sc.size(); i++) {
                if(withNumbers) System.out.print((i+1)+"   ");
                System.out.println(sc.get(i));
            }
        }

    }

    private void buildStorePlan(String alias) throws IOException {
        currDAG.parseQuery();
        currDAG.buildPlan( alias );

        if( !isBatchOn() || alias != null ) {
            // MRCompiler needs a store to be the leaf - hence
            // add a store to the plan to explain
            QueryParserUtils.attachStorePlan(scope, currDAG.lp, "fakefile", null, currDAG.getOperator( alias ),
                    "fake", pigContext );
        }
        currDAG.compile();
    }

    /**
     * Compile and execute the current plan.
     * @return
     * @throws IOException
     */
    private PigStats execute() throws IOException {
        pigContext.getProperties().setProperty( PigContext.JOB_NAME, jobName );
        if( jobPriority != null ) {
            pigContext.getProperties().setProperty( PigContext.JOB_PRIORITY, jobPriority );
        }

        // In this plan, all stores in the plan will be executed. They should be ignored if the plan is reused.
        currDAG.countExecutedStores();

        currDAG.compile();

        if( currDAG.lp.size() == 0 ) {
           return PigStats.get();
        }

        pigContext.getProperties().setProperty("pig.logical.plan.signature", currDAG.lp.getSignature());

        PigStats stats = executeCompiledLogicalPlan();

        return stats;
    }

    private PigStats executeCompiledLogicalPlan() throws ExecException,
            FrontendException {
        // discover pig features used in this script
        ScriptState.get().setScriptFeatures(currDAG.lp);
        currDAG.lp.optimize(pigContext);

        return launchPlan(currDAG.lp, "job_pigexec_");
    }

    /**
     * A common method for launching the jobs according to the logical plan
     * @param lp The logical plan
     * @param jobName A String containing the job name to be used
     * @return The PigStats object
     * @throws ExecException
     * @throws FrontendException
     */
    protected PigStats launchPlan(LogicalPlan lp, String jobName) throws ExecException, FrontendException {

        PigStats stats = null;
        try {
            stats = pigContext.getExecutionEngine().launchPig(lp, jobName, pigContext);
        } catch (ExecException e) {
            throw e;
        } catch (FrontendException e) {
            throw e;
        } catch (Exception e) {
            // There are a lot of exceptions thrown by the launcher.  If this
            // is an ExecException, just let it through.  Else wrap it.
            int errCode = 2043;
            String msg = "Unexpected error during execution.";
            throw new ExecException(msg, errCode, PigException.BUG, e);
        }

        return stats;
    }

    /**
     * NOTE: For testing only. Don't use.
     * @throws IOException
     */
    @SuppressWarnings("unused")
    private LogicalPlan buildLp() throws IOException {
        currDAG.buildPlan( null);
        currDAG.compile();
        return currDAG.lp;
    }

    private LogicalRelationalOperator getOperatorForAlias(String alias) throws IOException {
        buildStorePlan (alias);
        LogicalRelationalOperator op = (LogicalRelationalOperator)currDAG.getOperator( alias );
        if( op == null ) {
            int errCode = 1005;
            String msg = "No plan for " + alias + " to describe";
            throw new FrontendException(msg, errCode, PigException.INPUT, false, null);
        }
        return op;
    }

    /**
     * Returns data associated with LogicalPlan. It makes
     * sense to call this method only after a query/script
     * has been registered with one of the {@link #registerQuery(String)}
     * or {@link #registerScript(InputStream)} methods.
     *
     * @return LogicalPlanData
     */
    public LogicalPlanData getLogicalPlanData() {
        return new LogicalPlanData(currDAG.lp);
    }

    /*
     * This class holds the internal states of a grunt shell session.
     */
    protected class Graph {

        private final Map<LogicalRelationalOperator, LogicalPlan> aliases = new HashMap<LogicalRelationalOperator, LogicalPlan>();

        private Map<String, Operator> operators = new HashMap<String, Operator>();
        private String lastRel;

        private final List<String> scriptCache = new ArrayList<String>();

        // the fileNameMap contains filename to canonical filename
        // mappings. This is done so we can reparse the cached script
        // and remember the translation (current directory might only
        // be correct during the first parse
        private Map<String, String> fileNameMap = new HashMap<String, String>();

        private final boolean batchMode;

        private int processedStores = 0;

        private LogicalPlan lp;

        private int currentLineNum = 0;

        public Graph(boolean batchMode) {
            this.batchMode = batchMode;
            this.lp = new LogicalPlan();
        };

        /**
         * Call back method for counting executed stores.
         */
        private void countExecutedStores() throws FrontendException {
            List<LOStore> sinks = Util.getLogicalRelationalOperators(lp, LOStore.class);
            processedStores += sinks.size();
        }

        Map<LogicalRelationalOperator, LogicalPlan> getAliases() {
            return aliases;
        }

        Map<String, Operator> getAliasOp() {
            return operators;
        }

        boolean isBatchOn() {
            return batchMode;
        };

        boolean isBatchEmpty() {
            for( Operator op : lp.getSinks() ) {
                if( op instanceof LOStore )
                    return false;
            }
            return true;
        }

        void markAsExecuted() {
        }

        public LogicalPlan getLogicalPlan() {
            return this.lp;
        }

        /**
         * Get the operator with the given alias in the raw plan. Null if not
         * found.
         */
        Operator getOperator(String alias) throws FrontendException {
            return operators.get( alias );
        }

        public LogicalPlan getPlan(String alias) throws IOException {
            LogicalPlan plan = lp;

            if (alias != null) {
                LogicalRelationalOperator op = (LogicalRelationalOperator) operators.get(alias);
                if(op == null) {
                    int errCode = 1003;
                    String msg = "Unable to find an operator for alias " + alias;
                    throw new FrontendException(msg, errCode, PigException.INPUT);
                }
                plan = aliases.get(op);
            }
            return plan;
        }


        /**
         * Build a plan for the given alias. Extra branches and child branch under alias
         * will be ignored. Dependent branch (i.e. scalar) will be kept.
         * @throws IOException
         */
        void buildPlan(String alias) throws IOException {
            if( alias == null )
                skipStores();

            final Queue<Operator> queue = new LinkedList<Operator>();
            if( alias != null ) {
                Operator op = getOperator( alias );
                if (op == null) {
                    String msg = "Unable to find an operator for alias " + alias;
                    throw new FrontendException( msg, 1003, PigException.INPUT );
                }
                queue.add( op );
            } else {
                List<LOStore> stores = Util.getLogicalRelationalOperators(lp, LOStore.class);
                for (LOStore op : stores) {
                    boolean addSink = true;
                    // Only add if all the successors are loads
                    List<Operator> succs = lp.getSuccessors(op);
                    if (succs != null && succs.size() > 0) {
                        for (Operator succ : succs) {
                            if (!(succ instanceof LOLoad)) {
                                addSink = false;
                                break;
                            }
                        }
                    }
                    if (addSink) {
                        queue.add(op);
                    }
                }
            }

            LogicalPlan plan = new LogicalPlan();

            while( !queue.isEmpty() ) {
                Operator currOp = queue.poll();
                plan.add( currOp );

                List<Operator> preds = lp.getPredecessors( currOp );
                if( preds != null ) {
                    List<Operator> ops = new ArrayList<Operator>( preds );
                    for( Operator pred : ops ) {
                        if( !queue.contains( pred ) )
                            queue.add( pred );
                        plan.connect( pred, currOp );
                    }
                }

                // visit expression associated with currOp. If it refers to any other operator
                // that operator is also going to be enqueued.
                currOp.accept( new AllExpressionVisitor( plan, new DependencyOrderWalker( plan ) ) {
                        @Override
                        protected LogicalExpressionVisitor getVisitor(LogicalExpressionPlan exprPlan)
                        throws FrontendException {
                            return new LogicalExpressionVisitor( exprPlan, new DependencyOrderWalker( exprPlan ) ) {
                                @Override
                                public void visit(ScalarExpression expr) throws FrontendException {
                                    Operator refOp = expr.getImplicitReferencedOperator();
                                    if( !queue.contains( refOp ) )
                                        queue.add( refOp );
                                }
                            };
                        }
                    }
                );

                currOp.setPlan( plan );
            }
            lp = plan;
        }

        /**
         *  Remove stores that have been executed previously from the overall plan.
         */
        private void skipStores() throws IOException {
            // Get stores specifically
            List<LOStore> sinks = Util.getLogicalRelationalOperators(lp, LOStore.class);
            List<Operator> sinksToRemove = new ArrayList<Operator>();
            int skipCount = processedStores;
            if( skipCount > 0 ) {
                for( LOStore sink : sinks ) {
                    sinksToRemove.add( sink );
                    skipCount--;
                    if( skipCount == 0 )
                        break;
                }
            }

            for( Operator op : sinksToRemove ) {
                // It's fully possible in the multiquery case that
                // a store that is not a leaf (sink) and therefor has
                // successors that need to be removed.
                removeToLoad(op);
                Operator pred = lp.getPredecessors( op ).get(0);
                lp.disconnect( pred, op );
                lp.remove( op );
            }
        }

        private void removeToLoad(Operator toRemove) throws IOException {
            List<Operator> successors = lp.getSuccessors(toRemove);
            List<Operator> succToRemove = new ArrayList<Operator>();
            if (successors != null && successors.size() > 0) {
                succToRemove.addAll(successors);
                for (Operator succ : succToRemove) {
                    lp.disconnect( toRemove, succ );
                    if (!(succ instanceof LOLoad)) {
                        removeToLoad(succ);
                        lp.remove(succ);
                    }
                }
            }
        }

        /**
         * Accumulate the given statement to previous query statements and generate
         * an overall (raw) plan.
         */
        void registerQuery(String query, int startLine, boolean validateEachStatement,
                boolean skipParseForBatch) throws IOException {
            if( batchMode ) {
                if( startLine == currentLineNum ) {
                    String line = scriptCache.remove( scriptCache.size() - 1 );
                    scriptCache.add( line + query );
                } else {
                    while( startLine > currentLineNum + 1 ) {
                        scriptCache.add( "" );
                        currentLineNum++;
                    }
                    BufferedReader br = new BufferedReader(new StringReader(query));
                    String line = br.readLine();
                    while (line != null) {
                        scriptCache.add(line);
                        currentLineNum++;
                        line = br.readLine();
                    }
                }
                if (skipParseForBatch) {
                    return;
                }
            } else {
                scriptCache.add( query );
            }

            if(validateEachStatement){
                validateQuery();
            }
            parseQuery();

            if( !batchMode ) {
                buildPlan( null );
                for( Operator sink : lp.getSinks() ) {
                    if( sink instanceof LOStore ) {
                        try {
                            execute();
                        } catch (Exception e) {
                            int errCode = 1002;
                            String msg = "Unable to store alias "
                                + ((LOStore) sink).getAlias();
                            throw new FrontendException(msg, errCode,
                                    PigException.INPUT, e);
                        }
                        break; // We should have at most one store, so break here.
                    }
                }
            }
        }

        private void validateQuery() throws FrontendException {
            String query = buildQuery();
            QueryParserDriver parserDriver = new QueryParserDriver( pigContext, scope, fileNameMap );
            try {
                LogicalPlan plan = parserDriver.parse( query );
                plan.validate(pigContext, scope, true);
            } catch(FrontendException ex) {
                scriptCache.remove( scriptCache.size() -1 );
                throw ex;
            }
        }

        public List<String> getScriptCache() {
            return scriptCache;
        }

        /**
         * Parse the accumulated pig statements and generate an overall plan.
         */
        private void parseQuery() throws FrontendException {
            UDFContext.getUDFContext().reset();
            UDFContext.getUDFContext().setClientSystemProps(pigContext.getProperties());

            String query = buildQuery();

            if( query.isEmpty() ) {
                lp = new LogicalPlan();
                return;
            }

            try {
                QueryParserDriver parserDriver = new QueryParserDriver( pigContext, scope, fileNameMap );
                lp = parserDriver.parse( query );
                operators = parserDriver.getOperators();
                lastRel = parserDriver.getLastRel();
            } catch(Exception ex) {
                scriptCache.remove( scriptCache.size() -1 ); // remove the bad script from the cache.
                PigException pe = LogUtils.getPigException(ex);
                int errCode = 1000;
                String msg = "Error during parsing. "
                        + (pe == null ? ex.getMessage() : pe.getMessage());
                log.error("exception during parsing: " + msg, ex);
                if (null == pe) {
                    throw new FrontendException (msg, errCode, PigException.INPUT , ex);
                } else {
                    throw new FrontendException (msg, errCode, PigException.INPUT , ex, pe.getSourceLocation() );
                }
            }
        }

        public String getLastRel() {
            return lastRel;
        }

        private String buildQuery() {
            StringBuilder accuQuery = new StringBuilder();
            for( String line : scriptCache ) {
                accuQuery.append( line + "\n" );
            }

            return accuQuery.toString();
        }

        private void compile() throws IOException {
            lp.validate(pigContext, scope, false);
            currDAG.postProcess();
        }

        private void postProcess() throws IOException {
            // The following code deals with store/load combination of
            // intermediate files. In this case we will replace the load
            // operator
            // with a (implicit) split operator, iff the load/store
            // func is reversible (because that's when we can safely
            // skip the load and keep going with the split output). If
            // the load/store func is not reversible (or they are
            // different functions), we connect the store and the load
            // to remember the dependency.

            Set<LOLoad> loadOps = new HashSet<LOLoad>();
            List<Operator> sources = lp.getSources();
            for (Operator source : sources) {
                if (source instanceof LOLoad) {
                    loadOps.add((LOLoad)source);
                }
            }

            Set<LOStore> storeOps = new HashSet<LOStore>();
            List<Operator> sinks = lp.getSinks();
            for (Operator sink : sinks) {
                if (sink instanceof LOStore) {
                    storeOps.add((LOStore)sink);
                }
            }

            if ("true".equals(pigContext.getProperties().getProperty(PIG_LOCATION_CHECK_STRICT))) {
                log.info("Output location strick check enabled");
                checkDuplicateStoreLoc(storeOps);
            }

            for (LOLoad load : loadOps) {
                for (LOStore store : storeOps) {
                    String ifile = load.getFileSpec().getFileName();
                    String ofile = store.getFileSpec().getFileName();
                    if (ofile.equals(ifile)) {
                        // if there is no path from the load to the store,
                        // then connect the store to the load to create the
                        // dependency of the store on the load. If there is
                        // a path from the load to the store, then we should
                        // not connect the store to the load and create a cycle
                        if (!store.getPlan().pathExists(load, store)) {
                            store.getPlan().connect(store, load);
                        }
                    }
                }
            }
        }

        /**
         * This method checks whether the multiple sinks (STORE) use the same
         * "file-based" location. If yes, throws a RuntimeException
         *
         * @param storeOps
         */
        private void checkDuplicateStoreLoc(Set<LOStore> storeOps) {
            Set<String> uniqueStoreLoc = new HashSet<String>();
            for(LOStore store : storeOps) {
                String fileName = store.getFileSpec().getFileName();
                if(!uniqueStoreLoc.add(fileName) && UriUtil.isHDFSFileOrLocalOrS3N(fileName, new Configuration(true))) {
                    throw new RuntimeException("Script contains 2 or more STORE statements writing to same location : "+ fileName);
                }
            }
        }

        protected Graph duplicate() {
            // There are two choices on how we duplicate the logical plan
            // 1 - we really clone each operator and connect up the cloned operators
            // 2 - we cache away the script till the point we need to clone
            // and then simply re-parse the script.
            // The latter approach is used here
            // FIXME: There is one open issue with this now:
            // Consider the following script:
            // A = load 'file:/somefile';
            // B = filter A by $0 > 10;
            // store B into 'bla';
            // rm 'file:/somefile';
            // A = load 'file:/someotherfile'
            // when we try to clone - we try to reparse
            // from the beginning and currently the parser
            // checks for file existence of files in the load
            // in the case where the file is a local one -i.e. with file: prefix
            // This will be a known issue now and we will need to revisit later

            // parse each line of the cached script
            int lineNumber = 1;

            // create data structures needed for parsing
            Graph graph = new Graph(isBatchOn());
            graph.processedStores = processedStores;
            graph.fileNameMap = new HashMap<String, String>(fileNameMap);

            try {
                for (Iterator<String> it = scriptCache.iterator(); it.hasNext(); lineNumber++) {
                    // always doing registerQuery irrespective of the batch mode
                    // TODO: Need to figure out if anything different needs to happen if batch
                    // mode is not on
                    // Don't have to do the validation again, so set validateEachStatement param to false
                    graph.registerQuery(it.next(), lineNumber, false, false);
                }
                graph.postProcess();
            } catch (IOException ioe) {
                ioe.printStackTrace();
                graph = null;
            }
            return graph;
        }
    }

    /**
     * This can be called to indicate if the query is being parsed/compiled
     * in a mode that expects each statement to be validated as it is
     * entered, instead of just doing it once for whole script.
     * @param validateEachStatement
     */
    public void setValidateEachStatement(boolean validateEachStatement) {
        this.validateEachStatement = validateEachStatement;
    }

    /**
     * Set whether to skip parsing while registering the query in batch mode
     * @param skipParseInRegisterForBatch
     */
    public void setSkipParseInRegisterForBatch(boolean skipParseInRegisterForBatch) {
        this.skipParseInRegisterForBatch = skipParseInRegisterForBatch;
    }

    public String getLastRel() {
        return currDAG.getLastRel();
    }
}