/*
 * 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.BufferedInputStream;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.io.StringReader;
import java.net.URL;
import java.nio.charset.Charset;
import java.text.ParseException;
import java.util.AbstractList;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.jar.Attributes;
import java.util.jar.JarFile;
import java.util.jar.Manifest;

import jline.console.ConsoleReader;
import jline.console.history.FileHistory;

import org.antlr.runtime.RecognitionException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;
import org.apache.pig.PigRunner.ReturnCode;
import org.apache.pig.backend.BackendException;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
import org.apache.pig.classification.InterfaceAudience;
import org.apache.pig.classification.InterfaceStability;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.PigImplConstants;
import org.apache.pig.impl.io.FileLocalizer;
import org.apache.pig.impl.util.JarManager;
import org.apache.pig.impl.util.LogUtils;
import org.apache.pig.impl.util.ObjectSerializer;
import org.apache.pig.impl.util.PropertiesUtil;
import org.apache.pig.impl.util.UDFContext;
import org.apache.pig.impl.util.Utils;
import org.apache.pig.parser.DryRunGruntParser;
import org.apache.pig.scripting.ScriptEngine;
import org.apache.pig.scripting.ScriptEngine.SupportedScriptLang;
import org.apache.pig.tools.cmdline.CmdLineParser;
import org.apache.pig.tools.grunt.ConsoleReaderInputStream;
import org.apache.pig.tools.grunt.Grunt;
import org.apache.pig.tools.pigstats.PigProgressNotificationListener;
import org.apache.pig.tools.pigstats.PigStats;
import org.apache.pig.tools.pigstats.PigStatsUtil;
import org.apache.pig.tools.pigstats.ScriptState;
import org.apache.pig.tools.timer.PerformanceTimerFactory;
import org.joda.time.DateTime;
import org.joda.time.Duration;
import org.joda.time.Period;
import org.joda.time.PeriodType;
import org.joda.time.format.PeriodFormat;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Strings;
import com.google.common.io.Closeables;

/**
 * Main class for Pig engine.
 */
@InterfaceAudience.LimitedPrivate({"Oozie"})
@InterfaceStability.Stable
public class Main {

    static {
        Utils.addShutdownHookWithPriority(new Runnable() {
            @Override
            public void run() {
                FileLocalizer.deleteTempResourceFiles();
            }
        }, PigImplConstants.SHUTDOWN_HOOK_TMP_FILES_CLEANUP_PRIORITY);
    }

    private final static Log log = LogFactory.getLog(Main.class);

    private static final String LOG4J_CONF = "log4jconf";
    private static final String BRIEF = "brief";
    private static final String DEBUG = "debug";
    private static final String VERBOSE = "verbose";

    private static final String version;
    private static final String majorVersion;
    private static final String minorVersion;
    private static final String patchVersion;
    private static final String svnRevision;
    private static final String buildTime;

    private enum ExecMode {STRING, FILE, SHELL, UNKNOWN}

    protected static final String PROGRESS_NOTIFICATION_LISTENER_KEY = "pig.notification.listener";

    protected static final String PROGRESS_NOTIFICATION_LISTENER_ARG_KEY = "pig.notification.listener.arg";

    static {
       Attributes attr=null;
       try {
            String findContainingJar = JarManager.findContainingJar(Main.class);
            if (findContainingJar != null) {
                JarFile jar = new JarFile(findContainingJar);
                final Manifest manifest = jar.getManifest();
                final Map<String,Attributes> attrs = manifest.getEntries();
                attr = attrs.get("org/apache/pig");
            } else {
                log.info("Unable to read pigs manifest file as we are not running from a jar, version information unavailable");
            }
        } catch (Exception e) {
            log.warn("Unable to read pigs manifest file, version information unavailable", e);
        }
        if (attr!=null) {
            version = attr.getValue("Implementation-Version");
            svnRevision = attr.getValue("Svn-Revision");
            buildTime = attr.getValue("Build-TimeStamp");
            String[] split = version.split("\\.");
            majorVersion=split[0];
            minorVersion=split[1];
            patchVersion=split[2];
        } else {
            version=null;
            majorVersion=null;
            minorVersion=null;
            patchVersion=null;
            svnRevision=null;
            buildTime=null;
        }
    }

    /**
     * The Main-Class for the Pig Jar that will provide a shell and setup a classpath appropriate
     * for executing Jar files.  Warning, this method calls System.exit().
     *
     * @param args
     *            -jar can be used to add additional jar files (colon separated). - will start a
     *            shell. -e will execute the rest of the command line as if it was input to the
     *            shell.
     * @throws IOException
     */
    public static void main(String args[]) {
        System.exit(run(args, null));
    }

    static int run(String args[], PigProgressNotificationListener listener) {
        DateTime startTime = new DateTime();
        int rc = 1;
        boolean verbose = false;
        boolean gruntCalled = false;
        boolean deleteTempFiles = true;
        String logFileName = null;
        boolean printScriptRunTime = true;
        PigContext pigContext = null;

        try {
            Configuration conf = new Configuration(false);
            GenericOptionsParser parser = new GenericOptionsParser(conf, args);
            conf = parser.getConfiguration();

            Properties properties = new Properties();
            PropertiesUtil.loadDefaultProperties(properties);
            properties.putAll(ConfigurationUtil.toProperties(conf));

            if (listener == null) {
                listener = makeListener(properties);
            }
            String[] pigArgs = parser.getRemainingArgs();

            boolean userSpecifiedLog = false;
            boolean checkScriptOnly = false;

            BufferedReader pin = null;
            boolean debug = false;
            boolean dryrun = false;
            boolean embedded = false;
            List<String> params = new ArrayList<String>();
            List<String> paramFiles = new ArrayList<String>();
            HashSet<String> disabledOptimizerRules = new HashSet<String>();

            CmdLineParser opts = new CmdLineParser(pigArgs);
            opts.registerOpt('4', "log4jconf", CmdLineParser.ValueExpected.REQUIRED);
            opts.registerOpt('b', "brief", CmdLineParser.ValueExpected.NOT_ACCEPTED);
            opts.registerOpt('c', "check", CmdLineParser.ValueExpected.NOT_ACCEPTED);
            opts.registerOpt('d', "debug", CmdLineParser.ValueExpected.REQUIRED);
            opts.registerOpt('e', "execute", CmdLineParser.ValueExpected.NOT_ACCEPTED);
            opts.registerOpt('f', "file", CmdLineParser.ValueExpected.REQUIRED);
            opts.registerOpt('g', "embedded", CmdLineParser.ValueExpected.REQUIRED);
            opts.registerOpt('h', "help", CmdLineParser.ValueExpected.OPTIONAL);
            opts.registerOpt('i', "version", CmdLineParser.ValueExpected.OPTIONAL);
            opts.registerOpt('l', "logfile", CmdLineParser.ValueExpected.REQUIRED);
            opts.registerOpt('m', "param_file", CmdLineParser.ValueExpected.OPTIONAL);
            opts.registerOpt('p', "param", CmdLineParser.ValueExpected.OPTIONAL);
            opts.registerOpt('r', "dryrun", CmdLineParser.ValueExpected.NOT_ACCEPTED);
            opts.registerOpt('t', "optimizer_off", CmdLineParser.ValueExpected.REQUIRED);
            opts.registerOpt('v', "verbose", CmdLineParser.ValueExpected.NOT_ACCEPTED);
            opts.registerOpt('w', "warning", CmdLineParser.ValueExpected.NOT_ACCEPTED);
            opts.registerOpt('x', "exectype", CmdLineParser.ValueExpected.REQUIRED);
            opts.registerOpt('F', "stop_on_failure", CmdLineParser.ValueExpected.NOT_ACCEPTED);
            opts.registerOpt('M', "no_multiquery", CmdLineParser.ValueExpected.NOT_ACCEPTED);
            opts.registerOpt('N', "no_fetch", CmdLineParser.ValueExpected.NOT_ACCEPTED);
            opts.registerOpt('P', "propertyFile", CmdLineParser.ValueExpected.REQUIRED);

            ExecMode mode = ExecMode.UNKNOWN;
            String file = null;
            String engine = null;

            // set up client side system properties in UDF context
            UDFContext.getUDFContext().setClientSystemProps(properties);

            char opt;
            while ((opt = opts.getNextOpt()) != CmdLineParser.EndOfOpts) {
                switch (opt) {
                case '4':
                    String log4jconf = opts.getValStr();
                    if(log4jconf != null){
                        properties.setProperty(LOG4J_CONF, log4jconf);
                    }
                    break;

                case 'b':
                    properties.setProperty(BRIEF, "true");
                    break;

                case 'c':
                    checkScriptOnly = true;
                    break;

                case 'd':
                    String logLevel = opts.getValStr();
                    if (logLevel != null) {
                        properties.setProperty(DEBUG, logLevel);
                    }
                    debug = true;
                    break;

                case 'e':
                    mode = ExecMode.STRING;
                    break;

                case 'f':
                    mode = ExecMode.FILE;
                    file = opts.getValStr();
                    break;

                case 'g':
                    embedded = true;
                    engine = opts.getValStr();
                    break;

                case 'F':
                    properties.setProperty("stop.on.failure", ""+true);
                    break;

                case 'h':
                    String topic = opts.getValStr();
                    if (topic != null)
                        if (topic.equalsIgnoreCase("properties"))
                            printProperties();
                        else{
                            System.out.println("Invalide help topic - " + topic);
                            usage();
                        }
                    else
                        usage();
                    return ReturnCode.SUCCESS;

                case 'i':
                    printScriptRunTime = false;
                    System.out.println(getVersionString());
                    return ReturnCode.SUCCESS;

                case 'l':
                    //call to method that validates the path to the log file
                    //and sets up the file to store the client side log file
                    String logFileParameter = opts.getValStr();
                    if (logFileParameter != null && logFileParameter.length() > 0) {
                        logFileName = validateLogFile(logFileParameter, null);
                    } else {
                        logFileName = validateLogFile(logFileName, null);
                    }
                    userSpecifiedLog = true;
                    properties.setProperty("pig.logfile", (logFileName == null? "": logFileName));
                    break;

                case 'm':
                    paramFiles.add(opts.getValStr());
                    break;

                case 'M':
                    // turns off multiquery optimization
                    properties.setProperty(PigConfiguration.PIG_OPT_MULTIQUERY,""+false);
                    break;

                case 'N':
                    properties.setProperty(PigConfiguration.PIG_OPT_FETCH,""+false);
                    break;

                case 'p':
                    params.add(opts.getValStr());
                    break;

                case 'r':
                    // currently only used for parameter substitution
                    // will be extended in the future
                    dryrun = true;
                    break;

                case 't':
                    disabledOptimizerRules.add(opts.getValStr());
                    break;

                case 'v':
                    properties.setProperty(VERBOSE, ""+true);
                    verbose = true;
                    break;

                case 'w':
                    properties.setProperty("aggregate.warning", ""+false);
                    break;

                case 'x':
                    properties.setProperty("exectype", opts.getValStr());
                    if (opts.getValStr().toLowerCase().contains("local")) {
                        UserGroupInformation.setConfiguration(new Configuration(false));
                    }
                    break;

                case 'P':
                {
                    InputStream inputStream = null;
                    try {
                        FileLocalizer.FetchFileRet localFileRet = FileLocalizer.fetchFile(properties, opts.getValStr());
                        inputStream = new BufferedInputStream(new FileInputStream(localFileRet.file));
                        properties.load(inputStream) ;
                    } catch (IOException e) {
                        throw new RuntimeException("Unable to parse properties file '" + opts.getValStr() + "'");
                    } finally {
                        if (inputStream != null) {
                            try {
                                inputStream.close();
                            } catch (IOException e) {
                            }
                        }
                    }
                }
                break;

                default: {
                    Character cc = Character.valueOf(opt);
                    throw new AssertionError("Unhandled option " + cc.toString());
                         }
                }
            }

            // create the context with the parameter
            pigContext = new PigContext(properties);

            // create the static script state object
            ScriptState scriptState = pigContext.getExecutionEngine().instantiateScriptState();
            String commandLine = LoadFunc.join((AbstractList<String>)Arrays.asList(args), " ");
            scriptState.setCommandLine(commandLine);
            if (listener != null) {
                scriptState.registerListener(listener);
            }
            ScriptState.start(scriptState);

            pigContext.getProperties().setProperty("pig.cmd.args", commandLine);

            if(logFileName == null && !userSpecifiedLog) {
                logFileName = validateLogFile(properties.getProperty("pig.logfile"), null);
            }

            pigContext.getProperties().setProperty("pig.logfile", (logFileName == null? "": logFileName));

            // configure logging
            configureLog4J(properties, pigContext);

            log.info(getVersionString().replace("\n", ""));

            if(logFileName != null) {
                log.info("Logging error messages to: " + logFileName);
            }

            deleteTempFiles = Boolean.valueOf(properties.getProperty(
                    PigConfiguration.PIG_DELETE_TEMP_FILE, "true"));

            pigContext.getProperties().setProperty(PigImplConstants.PIG_OPTIMIZER_RULES_KEY,
                    ObjectSerializer.serialize(disabledOptimizerRules));

            PigContext.setClassLoader(pigContext.createCl(null));

            // construct the parameter substitution preprocessor
            Grunt grunt = null;
            BufferedReader in;
            String substFile = null;

            paramFiles = fetchRemoteParamFiles(paramFiles, properties);
            pigContext.setParams(params);
            pigContext.setParamFiles(paramFiles);

            switch (mode) {

            case FILE: {
                String remainders[] = opts.getRemainingArgs();
                if (remainders != null) {
                    pigContext.getProperties().setProperty(PigContext.PIG_CMD_ARGS_REMAINDERS,
                            ObjectSerializer.serialize(remainders));
                }
                FileLocalizer.FetchFileRet localFileRet = FileLocalizer.fetchFile(properties, file);
                if (localFileRet.didFetch) {
                    properties.setProperty("pig.jars.relative.to.dfs", "true");
                }

                scriptState.setFileName(file);

                if (embedded) {
                    return runEmbeddedScript(pigContext, localFileRet.file.getPath(), engine);
                } else {
                    SupportedScriptLang type = determineScriptType(localFileRet.file.getPath());
                    if (type != null) {
                        return runEmbeddedScript(pigContext, localFileRet.file
                                    .getPath(), type.name().toLowerCase());
                    }
                }
                //Reader is created by first loading "pig.load.default.statements" or .pigbootup file if available
                in = new BufferedReader(new InputStreamReader(Utils.getCompositeStream(new FileInputStream(localFileRet.file), properties)));

                // run parameter substitution preprocessor first
                substFile = file + ".substituted";

                pin = runParamPreprocessor(pigContext, in, substFile, debug || dryrun || checkScriptOnly);
                if (dryrun) {
                    if (dryrun(substFile, pigContext)) {
                        log.info("Dry run completed. Substituted pig script is at "
                                + substFile
                                + ". Expanded pig script is at "
                                + file + ".expanded");
                    } else {
                        log.info("Dry run completed. Substituted pig script is at "
                                    + substFile);
                    }
                    return ReturnCode.SUCCESS;
                }


                logFileName = validateLogFile(logFileName, localFileRet.file);
                pigContext.getProperties().setProperty("pig.logfile", (logFileName == null? "": logFileName));

                // Set job name based on name of the script
                pigContext.getProperties().setProperty(PigContext.JOB_NAME,
                                                       "PigLatin:" +new File(file).getName()
                );
                if (!debug) {
                    new File(substFile).deleteOnExit();
                }

                scriptState.setScript(localFileRet.file);

                grunt = new Grunt(pin, pigContext);
                gruntCalled = true;

                if(checkScriptOnly) {
                    grunt.checkScript(substFile);
                    System.err.println(file + " syntax OK");
                    rc = ReturnCode.SUCCESS;
                } else {
                    int results[] = grunt.exec();
                    rc = getReturnCodeForStats(results);
                }

                return rc;
            }

            case STRING: {
                if(checkScriptOnly) {
                    System.err.println("ERROR:" +
                            "-c (-check) option is only valid " +
                            "when executing pig with a pig script file)");
                    return ReturnCode.ILLEGAL_ARGS;
                }
                // Gather up all the remaining arguments into a string and pass them into
                // grunt.
                StringBuffer sb = new StringBuffer();
                String remainders[] = opts.getRemainingArgs();
                for (int i = 0; i < remainders.length; i++) {
                    if (i != 0) sb.append(' ');
                    sb.append(remainders[i]);
                }

                sb.append('\n');

                scriptState.setScript(sb.toString());

                in = new BufferedReader(new StringReader(sb.toString()));

                grunt = new Grunt(in, pigContext);
                gruntCalled = true;
                int results[] = grunt.exec();
                return getReturnCodeForStats(results);
            }

            default:
                break;
            }

            // If we're here, we don't know yet what they want.  They may have just
            // given us a jar to execute, they might have given us a pig script to
            // execute, or they might have given us a dash (or nothing) which means to
            // run grunt interactive.
            String remainders[] = opts.getRemainingArgs();
            if (remainders == null) {
                if(checkScriptOnly) {
                    System.err.println("ERROR:" +
                            "-c (-check) option is only valid " +
                            "when executing pig with a pig script file)");
                    return ReturnCode.ILLEGAL_ARGS;
                }
                // Interactive
                mode = ExecMode.SHELL;
                //Reader is created by first loading "pig.load.default.statements" or .pigbootup file if available
                ConsoleReader reader = new ConsoleReader(Utils.getCompositeStream(System.in, properties), System.out);
                reader.setExpandEvents(false);
                reader.setPrompt("grunt> ");
                final String HISTORYFILE = ".pig_history";
                String historyFile = System.getProperty("user.home") + File.separator  + HISTORYFILE;
                reader.setHistory(new FileHistory(new File(historyFile)));
                ConsoleReaderInputStream inputStream = new ConsoleReaderInputStream(reader);
                grunt = new Grunt(new BufferedReader(new InputStreamReader(inputStream)), pigContext);
                grunt.setConsoleReader(reader);
                gruntCalled = true;
                grunt.run();
                return ReturnCode.SUCCESS;
            } else {
                pigContext.getProperties().setProperty(PigContext.PIG_CMD_ARGS_REMAINDERS, ObjectSerializer.serialize(remainders));

                // They have a pig script they want us to run.
                mode = ExecMode.FILE;

                FileLocalizer.FetchFileRet localFileRet = FileLocalizer.fetchFile(properties, remainders[0]);
                if (localFileRet.didFetch) {
                    properties.setProperty("pig.jars.relative.to.dfs", "true");
                }

                scriptState.setFileName(remainders[0]);

                if (embedded) {
                    return runEmbeddedScript(pigContext, localFileRet.file.getPath(), engine);
                } else {
                    SupportedScriptLang type = determineScriptType(localFileRet.file.getPath());
                    if (type != null) {
                        return runEmbeddedScript(pigContext, localFileRet.file
                                    .getPath(), type.name().toLowerCase());
                    }
                }
                //Reader is created by first loading "pig.load.default.statements" or .pigbootup file if available
                InputStream seqInputStream = Utils.getCompositeStream(new FileInputStream(localFileRet.file), properties);
                in = new BufferedReader(new InputStreamReader(seqInputStream));

                // run parameter substitution preprocessor first
                substFile = remainders[0] + ".substituted";
                pin = runParamPreprocessor(pigContext, in, substFile, debug || dryrun || checkScriptOnly);
                if (dryrun) {
                    if (dryrun(substFile, pigContext)) {
                        log.info("Dry run completed. Substituted pig script is at "
                                + substFile
                                + ". Expanded pig script is at "
                                + remainders[0] + ".expanded");
                    } else {
                        log.info("Dry run completed. Substituted pig script is at "
                                + substFile);
                    }
                    return ReturnCode.SUCCESS;
                }

                logFileName = validateLogFile(logFileName, localFileRet.file);
                pigContext.getProperties().setProperty("pig.logfile", (logFileName == null? "": logFileName));

                if (!debug) {
                    new File(substFile).deleteOnExit();
                }

                // Set job name based on name of the script
                pigContext.getProperties().setProperty(PigContext.JOB_NAME,
                                                       "PigLatin:" +new File(remainders[0]).getName()
                );

                scriptState.setScript(localFileRet.file);

                grunt = new Grunt(pin, pigContext);
                gruntCalled = true;

                if(checkScriptOnly) {
                    grunt.checkScript(substFile);
                    System.err.println(remainders[0] + " syntax OK");
                    rc = ReturnCode.SUCCESS;
                } else {
                    int results[] = grunt.exec();
                    rc = getReturnCodeForStats(results);
                }
                return rc;
            }

            // Per Utkarsh and Chris invocation of jar file via pig depricated.
        } catch (ParseException e) {
            usage();
            rc = ReturnCode.PARSE_EXCEPTION;
            PigStatsUtil.setErrorMessage(e.getMessage());
            PigStatsUtil.setErrorThrowable(e);
        } catch (org.apache.pig.tools.parameters.ParseException e) {
           // usage();
            rc = ReturnCode.PARSE_EXCEPTION;
            PigStatsUtil.setErrorMessage(e.getMessage());
            PigStatsUtil.setErrorThrowable(e);
        } catch (IOException e) {
            if (e instanceof PigException) {
                PigException pe = (PigException)e;
                rc = (pe.retriable()) ? ReturnCode.RETRIABLE_EXCEPTION
                        : ReturnCode.PIG_EXCEPTION;
                PigStatsUtil.setErrorMessage(pe.getMessage());
                PigStatsUtil.setErrorCode(pe.getErrorCode());
            } else {
                rc = ReturnCode.IO_EXCEPTION;
                PigStatsUtil.setErrorMessage(e.getMessage());
            }
            PigStatsUtil.setErrorThrowable(e);

            if(!gruntCalled) {
                LogUtils.writeLog(e, logFileName, log, verbose, "Error before Pig is launched");
            }
            killRunningJobsIfInterrupted(e, pigContext);
        } catch (Throwable e) {
            rc = ReturnCode.THROWABLE_EXCEPTION;
            PigStatsUtil.setErrorMessage(e.getMessage());
            PigStatsUtil.setErrorThrowable(e);

            if(!gruntCalled) {
                LogUtils.writeLog(e, logFileName, log, verbose, "Error before Pig is launched");
            }
            killRunningJobsIfInterrupted(e, pigContext);
        } finally {
            if (printScriptRunTime) {
                printScriptRunTime(startTime);
            }
            if (deleteTempFiles) {
                // clear temp files
                FileLocalizer.deleteTempFiles();
            }
            if (pigContext != null) {
                pigContext.getExecutionEngine().destroy();
            }
            PerformanceTimerFactory.getPerfTimerFactory().dumpTimers();
        }

        return rc;
    }

    private static void printScriptRunTime(DateTime startTime) {
        DateTime endTime = new DateTime();
        Duration duration = new Duration(startTime, endTime);
        Period period = duration.toPeriod().normalizedStandard(PeriodType.time());
        log.info("Pig script completed in "
                + PeriodFormat.getDefault().print(period)
                + " (" + duration.getMillis() + " ms)");
    }

    private static void killRunningJobsIfInterrupted(Throwable e, PigContext pigContext) {
        Throwable cause = e.getCause();
        // Kill running job when we get InterruptedException
        // Pig thread is interrupted by mapreduce when Oozie launcher job is killed
        // Shutdown hook kills running jobs, but sometimes NodeManager can issue
        // a SIGKILL after AM unregisters and before shutdown hook gets to execute
        // causing orphaned jobs that continue to run.
        if (e instanceof InterruptedException || (cause != null && cause instanceof InterruptedException)) {
            try {
                pigContext.getExecutionEngine().kill();
            } catch (BackendException be) {
                log.error("Error while killing running jobs", be);
            }
        }
    }

    protected static PigProgressNotificationListener makeListener(Properties properties) {

        try {
            return PigContext.instantiateObjectFromParams(
                        ConfigurationUtil.toConfiguration(properties),
                        PROGRESS_NOTIFICATION_LISTENER_KEY,
                        PROGRESS_NOTIFICATION_LISTENER_ARG_KEY,
                        PigProgressNotificationListener.class);
        } catch (ExecException e) {
            throw new RuntimeException(e);
        }
    }

    private static int getReturnCodeForStats(int[] stats) {
        return (stats[1] == 0) ? ReturnCode.SUCCESS         // no failed jobs
                    : (stats[0] == 0) ? ReturnCode.FAILURE  // no succeeded jobs
                            : ReturnCode.PARTIAL_FAILURE;   // some jobs have failed
    }

    public static boolean dryrun(String scriptFile, PigContext pigContext)
            throws RecognitionException, IOException {
        BufferedReader rd = new BufferedReader(new FileReader(scriptFile));

        DryRunGruntParser dryrun = new DryRunGruntParser(rd, scriptFile,
                pigContext);

        boolean hasMacro = dryrun.parseStopOnError();

        if (hasMacro) {
            String expandedFile = scriptFile.replace(".substituted",
                    ".expanded");
            BufferedWriter fw = new BufferedWriter(new FileWriter(expandedFile));
            fw.append(dryrun.getResult());
            fw.close();
        }
        return hasMacro;
    }

    //TODO jz: log4j.properties should be used instead
    private static void configureLog4J(Properties properties, PigContext pigContext) {
        // TODO Add a file appender for the logs
        // TODO Need to create a property in the properties file for it.
        // sgroschupf, 25Feb2008: this method will be obsolete with PIG-115.

        String log4jconf = properties.getProperty(LOG4J_CONF);
        String trueString = "true";
        boolean brief = trueString.equalsIgnoreCase(properties.getProperty(BRIEF));
        Level logLevel = Level.INFO;

        String logLevelString = properties.getProperty(DEBUG);
        if (logLevelString != null){
            logLevel = Level.toLevel(logLevelString, Level.INFO);
        }

        final Properties props = log4jConfAsProperties(log4jconf);
        if (props.size() == 0) {
            props.setProperty("log4j.logger.org.apache.pig", logLevel.toString());
            if((logLevelString = System.getProperty("pig.logfile.level")) == null){
                props.setProperty("log4j.rootLogger", "INFO, PIGCONSOLE");
            }
            else{
                logLevel = Level.toLevel(logLevelString, Level.INFO);
                props.setProperty("log4j.logger.org.apache.pig", logLevel.toString());
                props.setProperty("log4j.rootLogger", "INFO, PIGCONSOLE, F");
                props.setProperty("log4j.appender.F","org.apache.log4j.RollingFileAppender");
                props.setProperty("log4j.appender.F.File",properties.getProperty("pig.logfile"));
                props.setProperty("log4j.appender.F.layout","org.apache.log4j.PatternLayout");
                props.setProperty("log4j.appender.F.layout.ConversionPattern", brief ? "%m%n" : "%d [%t] %-5p %c - %m%n");
            }

            props.setProperty("log4j.appender.PIGCONSOLE","org.apache.log4j.ConsoleAppender");
            props.setProperty("log4j.appender.PIGCONSOLE.target", "System.err");
            props.setProperty("log4j.appender.PIGCONSOLE.layout","org.apache.log4j.PatternLayout");
            props.setProperty("log4j.appender.PIGCONSOLE.layout.ConversionPattern", brief ? "%m%n" : "%d [%t] %-5p %c - %m%n");
        }

        PropertyConfigurator.configure(props);
        logLevel = Logger.getLogger("org.apache.pig").getLevel();
        if (logLevel==null) {
            logLevel = Logger.getLogger("org.apache.pig").getEffectiveLevel();
        }
        Properties backendProps = pigContext.getLog4jProperties();
        backendProps.setProperty("log4j.logger.org.apache.pig", logLevel.toString());
        pigContext.setLog4jProperties(backendProps);
        pigContext.setDefaultLogLevel(logLevel);
    }

   @VisibleForTesting
   static Properties log4jConfAsProperties(String log4jconf) {
       final Properties properties = new Properties();
       if (!Strings.isNullOrEmpty(log4jconf)) {
           Reader propertyReader = null;
           try {
               final File file = new File(log4jconf);
               if (file.exists()) {
                   propertyReader = new FileReader(file);
                   properties.load(propertyReader);
                   log.info("Loaded log4j properties from file: " + file);
               } else {
                   final URL resource = Main.class.getClassLoader().getResource(log4jconf);
                   if (resource != null) {
                       propertyReader = new InputStreamReader(resource.openStream(), Charset.forName("UTF-8"));
                       properties.load(propertyReader);
                       log.info("Loaded log4j properties from resource: " +  resource);
                   } else {
                       log.warn("No file or resource found by the name: " + log4jconf);
                   }
               }
           } catch (IOException e)  {
               log.warn("Cannot open log4j properties file " + log4jconf + ", using default");
           } finally {
               Closeables.closeQuietly(propertyReader);
           }
       }
       return properties;
  }

  private static List<String> fetchRemoteParamFiles(List<String> paramFiles, Properties properties)
            throws IOException {
        List<String> paramFiles2 = new ArrayList<String>();
        for (String param: paramFiles) {
            FileLocalizer.FetchFileRet localFileRet = FileLocalizer.fetchFile(properties, param);
            paramFiles2.add(localFileRet.file.getAbsolutePath());
        }
        return paramFiles2;
    }
    // returns the stream of final pig script to be passed to Grunt
    private static BufferedReader runParamPreprocessor(PigContext context, BufferedReader origPigScript,
                                                String scriptFile, boolean createFile)
                                    throws org.apache.pig.tools.parameters.ParseException, IOException{
        if (createFile) {
            return context.doParamSubstitutionOutputToFile(origPigScript, scriptFile);
        } else {
            String substituted = context.doParamSubstitution(origPigScript);
            return new BufferedReader(new StringReader(substituted));
        }
    }

    /**
     * Returns the major version of Pig being run.
     */
    public static String getMajorVersion() {
        return majorVersion;
    }

    /**
     * Returns the major version of the Pig build being run.
     */
    public static String getMinorVersion() {
        return minorVersion;
    }

    /**
     * Returns the patch version of the Pig build being run.
     */
    public static String getPatchVersion() {
        return patchVersion;
    }

    /**
     * Returns the svn revision number of the Pig build being run.
     */
    public static String getSvnRevision() {
        return svnRevision;
    }

    /**
     * Returns the built time of the Pig build being run.
     */
    public static String getBuildTime() {
        return buildTime;
    }

    /**
     * Returns a version string formatted similarly to that of svn.
     * <pre>
     * Apache Pig version 0.11.0-SNAPSHOT (r1202387)
     * compiled Nov 15 2011, 15:22:09
     * </pre>
     */
    public static String getVersionString() {
        return "Apache Pig version " + version + " (r" + svnRevision + ") \ncompiled "+buildTime;
    }

    /**
     * Print usage string.
     */
    public static void usage()
    {
            System.out.println("\n"+getVersionString()+"\n");
            System.out.println("USAGE: Pig [options] [-] : Run interactively in grunt shell.");
            System.out.println("       Pig [options] -e[xecute] cmd [cmd ...] : Run cmd(s).");
            System.out.println("       Pig [options] [-f[ile]] file : Run cmds found in file.");
            System.out.println("  options include:");
            System.out.println("    -4, -log4jconf - Log4j configuration file, overrides log conf");
            System.out.println("    -b, -brief - Brief logging (no timestamps)");
            System.out.println("    -c, -check - Syntax check");
            System.out.println("    -d, -debug - Debug level, INFO is default");
            System.out.println("    -e, -execute - Commands to execute (within quotes)");
            System.out.println("    -f, -file - Path to the script to execute");
            System.out.println("    -g, -embedded - ScriptEngine classname or keyword for the ScriptEngine");
            System.out.println("    -h, -help - Display this message. You can specify topic to get help for that topic.");
            System.out.println("        properties is the only topic currently supported: -h properties.");
            System.out.println("    -i, -version - Display version information");
            System.out.println("    -l, -logfile - Path to client side log file; default is current working directory.");
            System.out.println("    -m, -param_file - Path to the parameter file");
            System.out.println("    -p, -param - Key value pair of the form param=val");
            System.out.println("    -r, -dryrun - Produces script with substituted parameters. Script is not executed.");
            System.out.println("    -t, -optimizer_off - Turn optimizations off. The following values are supported:");
            System.out.println("            ConstantCalculator - Calculate constants at compile time");
            System.out.println("            SplitFilter - Split filter conditions");
            System.out.println("            PushUpFilter - Filter as early as possible");
            System.out.println("            MergeFilter - Merge filter conditions");
            System.out.println("            PushDownForeachFlatten - Join or explode as late as possible");
            System.out.println("            LimitOptimizer - Limit as early as possible");
            System.out.println("            ColumnMapKeyPrune - Remove unused data");
            System.out.println("            AddForEach - Add ForEach to remove unneeded columns");
            System.out.println("            MergeForEach - Merge adjacent ForEach");
            System.out.println("            GroupByConstParallelSetter - Force parallel 1 for \"group all\" statement");
            System.out.println("            PartitionFilterOptimizer - Pushdown partition filter conditions to loader implementing LoadMetaData");
            System.out.println("            PredicatePushdownOptimizer - Pushdown filter predicates to loader implementing LoadPredicatePushDown");
            System.out.println("            All - Disable all optimizations");
            System.out.println("        All optimizations listed here are enabled by default. Optimization values are case insensitive.");
            System.out.println("    -v, -verbose - Print all error messages to screen");
            System.out.println("    -w, -warning - Turn warning logging on; also turns warning aggregation off");
            System.out.println("    -x, -exectype - Set execution mode: local|mapreduce|tez, default is mapreduce.");
            System.out.println("    -F, -stop_on_failure - Aborts execution on the first failed job; default is off");
            System.out.println("    -M, -no_multiquery - Turn multiquery optimization off; default is on");
            System.out.println("    -N, -no_fetch - Turn fetch optimization off; default is on");
            System.out.println("    -P, -propertyFile - Path to property file");
            System.out.println("    -printCmdDebug - Overrides anything else and prints the actual command used to run Pig, including");
            System.out.println("                     any environment variables that are set by the pig command.");
    }

    public static void printProperties(){
            System.out.println("The following properties are supported:");
            System.out.println("    Logging:");
            System.out.println("        verbose=true|false; default is false. This property is the same as -v switch");
            System.out.println("        brief=true|false; default is false. This property is the same as -b switch");
            System.out.println("        debug=OFF|ERROR|WARN|INFO|DEBUG; default is INFO. This property is the same as -d switch");
            System.out.println("        aggregate.warning=true|false; default is true. If true, prints count of warnings");
            System.out.println("            of each type rather than logging each warning.");
            System.out.println("    Performance tuning:");
            System.out.println("        pig.cachedbag.memusage=<mem fraction>; default is 0.2 (20% of all memory).");
            System.out.println("            Note that this memory is shared across all large bags used by the application.");
            System.out.println("        pig.skewedjoin.reduce.memusagea=<mem fraction>; default is 0.3 (30% of all memory).");
            System.out.println("            Specifies the fraction of heap available for the reducer to perform the join.");
            System.out.println("        pig.exec.nocombiner=true|false; default is false. ");
            System.out.println("            Only disable combiner as a temporary workaround for problems.");
            System.out.println("        opt.multiquery=true|false; multiquery is on by default.");
            System.out.println("            Only disable multiquery as a temporary workaround for problems.");
            System.out.println("        opt.fetch=true|false; fetch is on by default.");
            System.out.println("            Scripts containing Filter, Foreach, Limit, Stream, and Union can be dumped without MR jobs.");
            System.out.println("        pig.tmpfilecompression=true|false; compression is off by default.");
            System.out.println("            Determines whether output of intermediate jobs is compressed.");
            System.out.println("        pig.tmpfilecompression.codec=lzo|gzip; default is gzip.");
            System.out.println("            Used in conjunction with pig.tmpfilecompression. Defines compression type.");
            System.out.println("        pig.noSplitCombination=true|false. Split combination is on by default.");
            System.out.println("            Determines if multiple small files are combined into a single map.");
            System.out.println("        pig.exec.mapPartAgg=true|false. Default is false.");
            System.out.println("            Determines if partial aggregation is done within map phase, ");
            System.out.println("            before records are sent to combiner.");
            System.out.println("        pig.exec.mapPartAgg.minReduction=<min aggregation factor>. Default is 10.");
            System.out.println("            If the in-map partial aggregation does not reduce the output num records");
            System.out.println("            by this factor, it gets disabled.");
            System.out.println("    Miscellaneous:");
            System.out.println("        exectype=mapreduce|tez|local; default is mapreduce. This property is the same as -x switch");
            System.out.println("        pig.additional.jars.uris=<comma seperated list of jars>. Used in place of register command.");
            System.out.println("        udf.import.list=<Colon seperated list of imports>. Used to avoid package names in UDF.");
            System.out.println("        stop.on.failure=true|false; default is false. Set to true to terminate on the first error.");
            System.out.println("        pig.datetime.default.tz=<UTC time offset>. e.g. +08:00. Default is the default timezone of the host.");
            System.out.println("            Determines the timezone used to handle datetime datatype and UDFs. ");
            System.out.println("        pig.artifacts.download.location=<path to download artifacts>; default is ~/.groovy/grapes");
            System.out.println("            Determines the location to download the artifacts when registering jars using ivy coordinates.");
            System.out.println("Additionally, any Hadoop property can be specified.");
    }

    private static String validateLogFile(String logFileName, File scriptFile) {
        String strippedDownScriptName = null;

        if (scriptFile != null) {
            if(!scriptFile.isDirectory()) {
                String scriptFileAbsPath;
                try {
                    scriptFileAbsPath = scriptFile.getCanonicalPath();
                    strippedDownScriptName = getFileFromCanonicalPath(scriptFileAbsPath);
                } catch (IOException ioe) {
                    log.warn("Could not compute canonical path to the script file " + ioe.getMessage());
                    strippedDownScriptName = null;
                }
            }
        }

        String defaultLogFileName = (strippedDownScriptName == null ? "pig_" : strippedDownScriptName) + new Date().getTime() + ".log";
        File logFile;

        if(logFileName != null) {
            logFile = new File(logFileName);

            //Check if the file name is a directory
            //append the default file name to the file
            if(logFile.isDirectory()) {
                if(logFile.canWrite()) {
                    try {
                        logFileName = logFile.getCanonicalPath() + File.separator + defaultLogFileName;
                    } catch (IOException ioe) {
                        log.warn("Could not compute canonical path to the log file " + ioe.getMessage());
                        return null;
                    }
                    return logFileName;
                } else {
                    log.warn("Need write permission in the directory: " + logFileName + " to create log file.");
                    return null;
                }
            } else {
                //we have a relative path or an absolute path to the log file
                //check if we can write to the directory where this file is/will be stored

                if (logFile.exists()) {
                    if(logFile.canWrite()) {
                        try {
                            logFileName = new File(logFileName).getCanonicalPath();
                        } catch (IOException ioe) {
                            log.warn("Could not compute canonical path to the log file " + ioe.getMessage());
                            return null;
                        }
                        return logFileName;
                    } else {
                        //do not have write permissions for the log file
                        //bail out with an error message
                        log.warn("Cannot write to file: " + logFileName + ". Need write permission.");
                        return logFileName;
                    }
                } else {
                    logFile = logFile.getParentFile();

                    if(logFile != null) {
                        //if the directory is writable we are good to go
                        if(logFile.canWrite()) {
                            try {
                                logFileName = new File(logFileName).getCanonicalPath();
                            } catch (IOException ioe) {
                                log.warn("Could not compute canonical path to the log file " + ioe.getMessage());
                                return null;
                            }
                            return logFileName;
                        } else {
                            log.warn("Need write permission in the directory: " + logFile + " to create log file.");
                            return logFileName;
                        }
                    }//end if logFile != null else is the default in fall through
                }//end else part of logFile.exists()
            }//end else part of logFile.isDirectory()
        }//end if logFileName != null

        //file name is null or its in the current working directory
        //revert to the current working directory
        String currDir = System.getProperty("user.dir");
        logFile = new File(currDir);
        logFileName = currDir + File.separator + (logFileName == null? defaultLogFileName : logFileName);
        if(logFile.canWrite()) {
            return logFileName;
        }
        log.warn("Cannot write to log file: " + logFileName);
        return null;
    }

    private static String getFileFromCanonicalPath(String canonicalPath) {
        return canonicalPath.substring(canonicalPath.lastIndexOf(File.separator));
    }

    private static SupportedScriptLang determineScriptType(String file)
            throws IOException {
        return ScriptEngine.getSupportedScriptLang(file);
    }

    private static int runEmbeddedScript(PigContext pigContext, String file, String engine)
    throws IOException {
        log.info("Run embedded script: " + engine);
        pigContext.connect();
        ScriptEngine scriptEngine = ScriptEngine.getInstance(engine);
        Map<String, List<PigStats>> statsMap = scriptEngine.run(pigContext, file);
        PigStatsUtil.setStatsMap(statsMap);

        int failCount = 0;
        int totalCount = 0;
        for (List<PigStats> lst : statsMap.values()) {
            if (lst != null && !lst.isEmpty()) {
                for (PigStats stats : lst) {
                    if (!stats.isSuccessful()) failCount++;
                    totalCount++;
                }
            }
        }
        return (totalCount > 0 && failCount == totalCount) ? ReturnCode.FAILURE
                : (failCount > 0) ? ReturnCode.PARTIAL_FAILURE
                        : ReturnCode.SUCCESS;
    }

}