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

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintStream;
import java.io.PrintWriter;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Properties;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.GnuParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.OptionBuilder;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.commons.cli.Parser;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hive.cli.CliSessionState;
import org.apache.hadoop.hive.common.LogUtils;
import org.apache.hadoop.hive.common.LogUtils.LogInitializationException;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
import org.apache.hadoop.hive.ql.CommandNeedRetryException;
import org.apache.hadoop.hive.ql.processors.DfsProcessor;
import org.apache.hadoop.hive.ql.processors.SetProcessor;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hcatalog.cli.SemanticAnalysis.HCatSemanticAnalyzer;
import org.apache.hcatalog.common.HCatConstants;
import org.apache.hcatalog.common.HCatUtil;

public class HCatCli {

    @SuppressWarnings("static-access")
    public static void main(String[] args) {

        try {
            LogUtils.initHiveLog4j();
        } catch (LogInitializationException e) {

        }

        CliSessionState ss = new CliSessionState(new HiveConf(SessionState.class));
        ss.in = System.in;
        try {
            ss.out = new PrintStream(System.out, true, "UTF-8");
            ss.err = new PrintStream(System.err, true, "UTF-8");
        } catch (UnsupportedEncodingException e) {
            System.exit(1);
        }

        HiveConf conf = ss.getConf();

        HiveConf.setVar(conf, ConfVars.SEMANTIC_ANALYZER_HOOK, HCatSemanticAnalyzer.class.getName());

        SessionState.start(ss);

        Options options = new Options();

        // -e 'quoted-query-string'
        options.addOption(OptionBuilder
                .hasArg()
                .withArgName("exec")
                .withDescription("hcat command given from command line")
                .create('e'));

        // -f <query-file>
        options.addOption(OptionBuilder
                .hasArg()
                .withArgName("file")
                .withDescription("hcat commands in file")
                .create('f'));

        // -g
        options.addOption(OptionBuilder
                .hasArg().
                withArgName("group").
                withDescription("group for the db/table specified in CREATE statement").
                create('g'));

        // -p
        options.addOption(OptionBuilder
                .hasArg()
                .withArgName("perms")
                .withDescription("permissions for the db/table specified in CREATE statement")
                .create('p'));

        // -D
        options.addOption(OptionBuilder
                .hasArgs(2)
                .withArgName("property=value")
                .withValueSeparator()
                .withDescription("use hadoop value for given property")
                .create('D'));

        // [-h|--help]
        options.addOption(new Option("h", "help", false, "Print help information"));

        Parser parser = new GnuParser();
        CommandLine cmdLine = null;

        try {
            cmdLine = parser.parse(options, args);

        } catch (ParseException e) {
            printUsage(options, ss.err);
            System.exit(1);
        }
        // -e
        String execString = (String) cmdLine.getOptionValue('e');
        // -f
        String fileName = (String) cmdLine.getOptionValue('f');
        // -h
        if (cmdLine.hasOption('h')) {
            printUsage(options, ss.out);
            System.exit(0);
        }

        if (execString != null && fileName != null) {
            ss.err.println("The '-e' and '-f' options cannot be specified simultaneously");
            printUsage(options, ss.err);
            System.exit(1);
        }

        // -p
        String perms = (String) cmdLine.getOptionValue('p');
        if (perms != null) {
            validatePermissions(ss, conf, perms);
        }

        // -g
        String grp = (String) cmdLine.getOptionValue('g');
        if (grp != null) {
            conf.set(HCatConstants.HCAT_GROUP, grp);
        }

        // -D
        setConfProperties(conf, cmdLine.getOptionProperties("D"));

        if (execString != null) {
            System.exit(processLine(execString));
        }

        try {
            if (fileName != null) {
                System.exit(processFile(fileName));
            }
        } catch (FileNotFoundException e) {
            ss.err.println("Input file not found. (" + e.getMessage() + ")");
            System.exit(1);
        } catch (IOException e) {
            ss.err.println("Could not open input file for reading. (" + e.getMessage() + ")");
            System.exit(1);
        }

        // -h
        printUsage(options, ss.err);
        System.exit(1);
    }

    private static void setConfProperties(HiveConf conf, Properties props) {
        for (java.util.Map.Entry<Object, Object> e : props.entrySet())
            conf.set((String) e.getKey(), (String) e.getValue());
    }

    private static int processLine(String line) {
        int ret = 0;

        String command = "";
        for (String oneCmd : line.split(";")) {

            if (StringUtils.endsWith(oneCmd, "\\")) {
                command += StringUtils.chop(oneCmd) + ";";
                continue;
            } else {
                command += oneCmd;
            }
            if (StringUtils.isBlank(command)) {
                continue;
            }

            ret = processCmd(command);
            command = "";
        }
        return ret;
    }

    private static int processFile(String fileName) throws IOException {
        FileReader fileReader = null;
        BufferedReader reader = null;
        try {
            fileReader = new FileReader(fileName);
            reader = new BufferedReader(fileReader);
            String line;
            StringBuilder qsb = new StringBuilder();

            while ((line = reader.readLine()) != null) {
                qsb.append(line + "\n");
            }

            return (processLine(qsb.toString()));
        } finally {
            if (fileReader != null) {
                fileReader.close();
            }
            if (reader != null) {
                reader.close();
            }
        }
    }

    private static int processCmd(String cmd) {

        SessionState ss = SessionState.get();
        long start = System.currentTimeMillis();

        cmd = cmd.trim();
        String firstToken = cmd.split("\\s+")[0].trim();

        if (firstToken.equalsIgnoreCase("set")) {
            return new SetProcessor().run(cmd.substring(firstToken.length()).trim()).getResponseCode();
        } else if (firstToken.equalsIgnoreCase("dfs")) {
            return new DfsProcessor(ss.getConf()).run(cmd.substring(firstToken.length()).trim()).getResponseCode();
        }

        HCatDriver driver = new HCatDriver();

        int ret = driver.run(cmd).getResponseCode();

        if (ret != 0) {
            driver.close();
            System.exit(ret);
        }

        ArrayList<String> res = new ArrayList<String>();
        try {
            while (driver.getResults(res)) {
                for (String r : res) {
                    ss.out.println(r);
                }
                res.clear();
            }
        } catch (IOException e) {
            ss.err.println("Failed with exception " + e.getClass().getName() + ":"
                + e.getMessage() + "\n" + org.apache.hadoop.util.StringUtils.stringifyException(e));
            ret = 1;
        } catch (CommandNeedRetryException e) {
            ss.err.println("Failed with exception " + e.getClass().getName() + ":"
                + e.getMessage() + "\n" + org.apache.hadoop.util.StringUtils.stringifyException(e));
            ret = 1;
        }

        int cret = driver.close();
        if (ret == 0) {
            ret = cret;
        }

        long end = System.currentTimeMillis();
        if (end > start) {
            double timeTaken = (end - start) / 1000.0;
            ss.err.println("Time taken: " + timeTaken + " seconds");
        }
        return ret;
    }

    private static void printUsage(Options options, OutputStream os) {
        PrintWriter pw = new PrintWriter(os);
        new HelpFormatter().printHelp(pw, 2 * HelpFormatter.DEFAULT_WIDTH,
            "hcat { -e \"<query>\" | -f \"<filepath>\" } [ -g \"<group>\" ] [ -p \"<perms>\" ] [ -D\"<name>=<value>\" ]",
            null, options, HelpFormatter.DEFAULT_LEFT_PAD, HelpFormatter.DEFAULT_DESC_PAD,
            null, false);
        pw.flush();
    }

    private static void validatePermissions(CliSessionState ss, HiveConf conf, String perms) {
        perms = perms.trim();
        FsPermission fp = null;

        if (perms.matches("^\\s*([r,w,x,-]{9})\\s*$")) {
            fp = FsPermission.valueOf("d" + perms);
        } else if (perms.matches("^\\s*([0-7]{3})\\s*$")) {
            fp = new FsPermission(Short.decode("0" + perms));
        } else {
            ss.err.println("Invalid permission specification: " + perms);
            System.exit(1);
        }

        if (!HCatUtil.validateMorePermissive(fp.getUserAction(), fp.getGroupAction())) {
            ss.err.println("Invalid permission specification: " + perms + " : user permissions must be more permissive than group permission ");
            System.exit(1);
        }
        if (!HCatUtil.validateMorePermissive(fp.getGroupAction(), fp.getOtherAction())) {
            ss.err.println("Invalid permission specification: " + perms + " : group permissions must be more permissive than other permission ");
            System.exit(1);
        }
        if ((!HCatUtil.validateExecuteBitPresentIfReadOrWrite(fp.getUserAction())) ||
            (!HCatUtil.validateExecuteBitPresentIfReadOrWrite(fp.getGroupAction())) ||
            (!HCatUtil.validateExecuteBitPresentIfReadOrWrite(fp.getOtherAction()))) {
            ss.err.println("Invalid permission specification: " + perms + " : permissions must have execute permissions if read or write permissions are specified ");
            System.exit(1);
        }

        conf.set(HCatConstants.HCAT_PERMS, "d" + fp.toString());

    }


}