/*
 * 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.rya.accumulo.mr.merge;

import java.io.BufferedOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.PrintStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Paths;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.ClientConfiguration;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.TableExistsException;
import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.client.admin.TableOperations;
import org.apache.accumulo.core.client.mapreduce.AbstractInputFormat;
import org.apache.accumulo.core.client.mapreduce.AccumuloFileOutputFormat;
import org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat;
import org.apache.accumulo.core.client.mapreduce.AccumuloMultiTableInputFormat;
import org.apache.accumulo.core.client.mapreduce.AccumuloOutputFormat;
import org.apache.accumulo.core.client.mapreduce.InputFormatBase;
import org.apache.accumulo.core.client.mapreduce.InputTableConfig;
import org.apache.accumulo.core.client.mapreduce.lib.partition.KeyRangePartitioner;
import org.apache.accumulo.core.client.mock.MockInstance;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.file.rfile.bcfile.Compression.Algorithm;
import org.apache.accumulo.core.iterators.user.AgeOffFilter;
import org.apache.accumulo.core.iterators.user.TimestampFilter;
import org.apache.accumulo.core.security.TablePermission;
import org.apache.accumulo.core.util.TextUtil;
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.io.FilenameUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FsShell;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.BasicConfigurator;
import org.apache.log4j.Logger;
import org.apache.log4j.xml.DOMConfigurator;
import org.apache.rya.accumulo.AccumuloRdfConfiguration;
import org.apache.rya.accumulo.mr.AccumuloHDFSFileInputFormat;
import org.apache.rya.accumulo.mr.MRUtils;
import org.apache.rya.accumulo.mr.merge.common.InstanceType;
import org.apache.rya.accumulo.mr.merge.mappers.AccumuloCopyToolMapper;
import org.apache.rya.accumulo.mr.merge.mappers.AccumuloRyaRuleMapper;
import org.apache.rya.accumulo.mr.merge.mappers.FileCopyToolMapper;
import org.apache.rya.accumulo.mr.merge.mappers.MergeToolMapper;
import org.apache.rya.accumulo.mr.merge.mappers.RowRuleMapper;
import org.apache.rya.accumulo.mr.merge.reducers.MultipleFileReducer;
import org.apache.rya.accumulo.mr.merge.util.AccumuloInstanceDriver;
import org.apache.rya.accumulo.mr.merge.util.AccumuloQueryRuleset;
import org.apache.rya.accumulo.mr.merge.util.AccumuloRyaUtils;
import org.apache.rya.accumulo.mr.merge.util.GroupedRow;
import org.apache.rya.accumulo.mr.merge.util.TimeUtils;
import org.apache.rya.accumulo.mr.merge.util.ToolConfigUtils;
import org.apache.rya.api.RdfCloudTripleStoreConstants;
import org.apache.rya.api.RdfCloudTripleStoreUtils;
import org.apache.rya.api.layout.TablePrefixLayoutStrategy;
import org.apache.rya.api.path.PathUtils;
import org.apache.rya.indexing.accumulo.ConfigUtils;

import com.google.common.base.Joiner;

/**
 * Handles copying data from a parent instance into a child instance.
 */
public class CopyTool extends AbstractDualInstanceAccumuloMRTool {
    private static final Logger log = Logger.getLogger(CopyTool.class);

    /**
     * Use this property to set the tables that are going to be copied.  The list should
     * be a comma-separated string containing the full table names.  If not set, then all
     * tables will be copied.
     */
    public static final String COPY_TABLE_LIST_PROP = "copy.table.list";

    /**
     * Indicates the type of child instance to create.  {@code null} or empty to not create an
     * instance indicating that it already was created and exists.
     */
    public static final String CREATE_CHILD_INSTANCE_TYPE_PROP = "create.child.instance.type";

    /**
     * The time difference between the parent machine and the time server.
     */
    public static final String PARENT_TIME_OFFSET_PROP = "time.offset";

    /**
     * The time difference between the child machine and the time server.
     */
    public static final String CHILD_TIME_OFFSET_PROP = "time.offset.child";

    /**
     * The host name of the time server to use.
     */
    public static final String NTP_SERVER_HOST_PROP = "ntp.server.host";

    /**
     * The URL of the Apache Tomcat server web page running on the parent machine.
     */
    public static final String PARENT_TOMCAT_URL_PROP = "tomcat.url";

    /**
     * The URL of the Apache Tomcat server web page running on the child machine.
     */
    public static final String CHILD_TOMCAT_URL_PROP = "tomcat.url.child";

    /**
     * The run time of the copy process.
     */
    public static final String COPY_RUN_TIME_PROP = "copy.run.time";

    /**
     * "true" to use the NTP server to handle time synchronization.
     * "false" (or any other value) to not use the NTP server.
     */
    public static final String USE_NTP_SERVER_PROP = "use.ntp.server";

    /**
     * "true" to use file output. "false" to use Accumulo output.
     */
    public static final String USE_COPY_FILE_OUTPUT = "use.copy.file.output";

    /**
     * The file path to output the child data to.
     */
    public static final String COPY_FILE_OUTPUT_PATH = "copy.file.output.path";

    /**
     * The compression type to use for file output.  One of "none", "gz", "lzo", or "snappy".
     */
    public static final String COPY_FILE_OUTPUT_COMPRESSION_TYPE = "copy.file.output.compression.type";

    /**
     * "true" to clear the file output directory before copying. "false" to leave the output directory alone.
     */
    public static final String USE_COPY_FILE_OUTPUT_DIRECTORY_CLEAR = "use.copy.file.output.directory.clear";

    /**
     * The input directory for importing files into accumulo tables.
     */
    public static final String COPY_FILE_IMPORT_DIRECTORY = "copy.file.import.directory";

    /**
     * "true" to read from the input directory. "false" otherwise.
     */
    public static final String USE_COPY_FILE_IMPORT = "use.copy.file.import";

    /**
     * "true" to extract a set of rules from a SPARQL query, and only copy statements relevant to those rules. "false" otherwise.
     * If set, either the query itself or a query file should also be provided.
     */
    public static final String USE_COPY_QUERY_SPARQL = "use.copy.query.sparql";

    /**
     * The text of the query that defines which statements to copy.
     */
    public static final String QUERY_STRING_PROP = "ac.copy.query";

    /**
     * The path to a file containing the query that defines which statements to copy.
     */
    public static final String QUERY_FILE_PROP = "ac.copy.queryfile";

    /**
     *  startTime is the time of the data to copy. Only parent data AFTER the selected time will be copied to the child.
     */
    private String startTime = null;
    private boolean useCopyFileOutput = false;
    private String baseOutputDir = null;
    private String localBaseOutputDir = null;
    private String compressionType = null;
    private boolean useCopyFileOutputDirectoryClear = false;
    private String tempDir = null;
    private boolean useCopyFileImport = false;
    private boolean useQuery = false;
    private String localCopyFileImportDir = null;
    private String baseImportDir = null;

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

    private AccumuloInstanceDriver childAccumuloInstanceDriver = null;

    /**
     * Sets up and initializes the copy tool's configuration.
     * @throws Exception
     */
    public void setup() throws Exception {
        super.init();

        tempDir = PathUtils.clean(conf.get("hadoop.tmp.dir", null));
        if (tempDir == null) {
            throw new Exception("Invalid hadoop temp directory. \"hadoop.tmp.dir\" could not be found in the configuration.");
        }

        useCopyFileOutput = conf.getBoolean(USE_COPY_FILE_OUTPUT, false);
        baseOutputDir = tempDir + "/copy_tool_file_output/";
        localBaseOutputDir = PathUtils.clean(conf.get(COPY_FILE_OUTPUT_PATH, null));
        compressionType = conf.get(COPY_FILE_OUTPUT_COMPRESSION_TYPE, null);
        useCopyFileOutputDirectoryClear = conf.getBoolean(USE_COPY_FILE_OUTPUT_DIRECTORY_CLEAR, false);
        localCopyFileImportDir = PathUtils.clean(conf.get(COPY_FILE_IMPORT_DIRECTORY, null));
        baseImportDir = tempDir + "/copy_tool_import/";

        startTime = conf.get(MergeTool.START_TIME_PROP, null);

        if (!useCopyFileImport) {
            if (startTime != null) {
                try {
                    final Date date = MergeTool.START_TIME_FORMATTER.parse(startTime);
                    log.info("Will copy all data after " + date);
                } catch (final ParseException e) {
                    throw new Exception("Unable to parse the provided start time: " + startTime, e);
                }
            }

            Date copyRunTime = new Date();
            final boolean useTimeSync = conf.getBoolean(USE_NTP_SERVER_PROP, false);
            if (useTimeSync) {
                final String tomcatUrl = conf.get(PARENT_TOMCAT_URL_PROP, null);
                final String ntpServerHost = conf.get(NTP_SERVER_HOST_PROP, null);
                Long timeOffset = null;
                Date ntpDate = null;
                try {
                    log.info("Comparing parent machine's time to NTP server time...");
                    ntpDate = TimeUtils.getNtpServerDate(ntpServerHost);
                    final Date parentMachineDate = TimeUtils.getMachineDate(tomcatUrl);
                    final boolean isMachineLocal = TimeUtils.isUrlLocalMachine(tomcatUrl);
                    timeOffset = TimeUtils.getTimeDifference(ntpDate, parentMachineDate, isMachineLocal);
                } catch (IOException | ParseException e) {
                    throw new Exception("Unable to get time difference between machine and NTP server.", e);
                }
                if (timeOffset != null) {
                    conf.set(PARENT_TIME_OFFSET_PROP, "" + timeOffset);
                }
                copyRunTime = ntpDate;
            }
            final String copyRunTimeString = MergeTool.START_TIME_FORMATTER.format(copyRunTime);
            if (copyRunTime != null) {
                conf.set(COPY_RUN_TIME_PROP, copyRunTimeString);
            }
        }

        MergeTool.setDuplicateKeys(conf);

        final String copyTableListProperty = conf.get(COPY_TABLE_LIST_PROP);
        if (StringUtils.isNotBlank(copyTableListProperty)) {
            // Copy the tables specified in the config
            final String[] split = copyTableListProperty.split(",");
            tables.addAll(Arrays.asList(split));
        } else if (useCopyFileImport) {
            final File importDir = new File(localCopyFileImportDir);
            final String[] files = importDir.list();
            tables.addAll(Arrays.asList(files));
        } else {
            // By default copy all tables
            tables.add(tablePrefix + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX);
            tables.add(tablePrefix + RdfCloudTripleStoreConstants.TBL_OSP_SUFFIX);
            tables.add(tablePrefix + RdfCloudTripleStoreConstants.TBL_PO_SUFFIX);
            tables.add(tablePrefix + RdfCloudTripleStoreConstants.TBL_NS_SUFFIX);
            tables.add(tablePrefix + RdfCloudTripleStoreConstants.TBL_EVAL_SUFFIX);
            tables.add(tablePrefix + RdfCloudTripleStoreConstants.TBL_STATS_SUFFIX);
            tables.add(tablePrefix + RdfCloudTripleStoreConstants.TBL_SEL_SUFFIX);
            /* TODO: SEE RYA-160
            tables.add(ConfigUtils.getFreeTextDocTablename(conf));
            tables.add(ConfigUtils.getFreeTextTermTablename(conf));
            tables.add(ConfigUtils.getGeoTablename(conf));
            tables.add(ConfigUtils.getTemporalTableName(conf));
            tables.add(ConfigUtils.getEntityTableName(conf));
            */
        }
        if (tables.isEmpty()) {
            log.warn("No list of tables to copy was provided.");
        } else {
            final String tablesToCopy = Joiner.on("\r\n\t").join(tables);
            log.info("Will attempt to copy the following tables/indices from the parent:\r\n\t" + tablesToCopy);
        }
    }

    @Override
    public int run(final String[] strings) throws Exception {
        useCopyFileImport = conf.getBoolean(USE_COPY_FILE_IMPORT, false);
        useQuery = conf.getBoolean(USE_COPY_QUERY_SPARQL, false);

        if (useCopyFileImport) {
            return runImport();
        } else if (useQuery) {
            return runQueryCopy();
        } else {
            return runCopy();
        }
    }

    private int runCopy() throws Exception {
        log.info("Setting up Copy Tool...");

        setup();

        if (!useCopyFileOutput) {
            createChildInstance(conf);
        }

        final AccumuloRdfConfiguration parentAccumuloRdfConfiguration = new AccumuloRdfConfiguration(conf);
        parentAccumuloRdfConfiguration.setTablePrefix(tablePrefix);
        final Connector parentConnector = AccumuloRyaUtils.setupConnector(parentAccumuloRdfConfiguration);
        final TableOperations parentTableOperations = parentConnector.tableOperations();

        for (final String table : tables) {
            // Check if the parent table exists before creating a job on it
            if (parentTableOperations.exists(table)) {
                final String childTable = table.replaceFirst(tablePrefix, childTablePrefix);
                final String jobName = "Copy Tool, copying Parent Table: " + table + ", into Child Table: " + childTable + ", " + System.currentTimeMillis();
                log.info("Initializing job: " + jobName);
                conf.set(MRUtils.JOB_NAME_PROP, jobName);
                conf.set(MergeTool.TABLE_NAME_PROP, table);

                final Job job = Job.getInstance(conf);
                job.setJarByClass(CopyTool.class);

                setupAccumuloInput(job);

                InputFormatBase.setInputTableName(job, table);

                // Set input output of the particular job
                if (useCopyFileOutput) {
                    job.setMapOutputKeyClass(Key.class);
                    job.setMapOutputValueClass(Value.class);
                    job.setOutputKeyClass(Key.class);
                    job.setOutputValueClass(Value.class);
                } else {
                    job.setMapOutputKeyClass(Text.class);
                    job.setMapOutputValueClass(Mutation.class);
                    job.setOutputKeyClass(Text.class);
                    job.setOutputValueClass(Mutation.class);
                }

                setupAccumuloOutput(job, childTable);

                // Set mapper and reducer classes
                if (useCopyFileOutput) {
                    setupSplitsFile(job, parentTableOperations, table, childTable);
                    job.setMapperClass(FileCopyToolMapper.class);
                } else {
                    job.setMapperClass(AccumuloCopyToolMapper.class);
                }
                job.setReducerClass(Reducer.class);

                // Submit the job
                final Date beginTime = new Date();
                log.info("Job for table \"" + table + "\" started: " + beginTime);
                final int exitCode = job.waitForCompletion(true) ? 0 : 1;

                if (exitCode == 0) {
                    if (useCopyFileOutput) {
                        log.info("Moving data from HDFS to the local file system for the table: " + childTable);
                        final Path hdfsPath = getPath(baseOutputDir, childTable);
                        final Path localPath = getPath(localBaseOutputDir, childTable);
                        log.info("HDFS directory: " + hdfsPath.toString());
                        log.info("Local directory: " + localPath.toString());
                        copyHdfsToLocal(hdfsPath, localPath);
                    }

                    final Date endTime = new Date();
                    log.info("Job for table \"" + table + "\" finished: " + endTime);
                    log.info("The job took " + (endTime.getTime() - beginTime.getTime()) / 1000 + " seconds.");
                } else {
                    log.error("Job for table \"" + table + "\" Failed!!!");
                    return exitCode;
                }
            } else {
                log.warn("The table \"" + table + "\" was NOT found in the parent instance and cannot be copied.");
            }
        }

        return 0;
    }

    private int runImport() throws Exception {
        log.info("Setting up Copy Tool for importing...");

        setup();

        createChildInstance(conf);

        for (final String childTable : tables) {
            final String jobName = "Copy Tool, importing Exported Parent Table files from: " + getPath(localCopyFileImportDir, childTable).toString() + ", into Child Table: " + childTable + ", " + System.currentTimeMillis();
            log.info("Initializing job: " + jobName);
            conf.set(MRUtils.JOB_NAME_PROP, jobName);

            // Submit the job
            final Date beginTime = new Date();
            log.info("Job for table \"" + childTable + "\" started: " + beginTime);

            createTableIfNeeded(childTable);
            importFilesToChildTable(childTable);

            final Date endTime = new Date();
            log.info("Job for table \"" + childTable + "\" finished: " + endTime);
            log.info("The job took " + (endTime.getTime() - beginTime.getTime()) / 1000 + " seconds.");
        }

        return 0;
    }

    private int runQueryCopy() throws Exception {
        log.info("Setting up Copy Tool with a query-based ruleset...");
        setup();
        if (!useCopyFileOutput) {
            createChildInstance(conf);
        }

        // Set up the configuration
        final AccumuloRdfConfiguration aconf = new AccumuloRdfConfiguration(conf);
        aconf.setBoolean(ConfigUtils.USE_MOCK_INSTANCE, mock);
        aconf.setTablePrefix(tablePrefix);
        aconf.setFlush(false);
        ConfigUtils.setIndexers(aconf);

        // Since we're copying at the statement-level, ignore any given list of tables and determine
        // which tables we might need to create based on which indexers are desired.
        final TablePrefixLayoutStrategy prefixStrategy = new TablePrefixLayoutStrategy(tablePrefix);
        tables.clear();
        // Always include core tables
        tables.add(prefixStrategy.getSpo());
        tables.add(prefixStrategy.getOsp());
        tables.add(prefixStrategy.getPo());
        // Copy namespaces if they exist
        tables.add(prefixStrategy.getNs());
        // Add tables associated with any configured indexers
        /* TODO: SEE RYA-160
        if (aconf.getBoolean(ConfigUtils.USE_FREETEXT, false)) {
            tables.add(ConfigUtils.getFreeTextDocTablename(conf));
            tables.add(ConfigUtils.getFreeTextTermTablename(conf));
        }
        if (aconf.getBoolean(ConfigUtils.USE_GEO, false)) {
            tables.add(ConfigUtils.getGeoTablename(conf));
        }
        if (aconf.getBoolean(ConfigUtils.USE_TEMPORAL, false)) {
            tables.add(ConfigUtils.getTemporalTableName(conf));
        }
        if (aconf.getBoolean(ConfigUtils.USE_ENTITY, false)) {
            tables.add(ConfigUtils.getEntityTableName(conf));
        }
        */
        // Ignore anything else, e.g. statistics -- must be recalculated for the child if desired

        // Extract the ruleset, and copy the namespace table directly
        final AccumuloQueryRuleset ruleset = new AccumuloQueryRuleset(aconf);
        ruleset.addTable(prefixStrategy.getNs());
        for (final String line : ruleset.toString().split("\n")) {
            log.info(line);
        }

        // Create a Job and configure its input and output
        final Job job = Job.getInstance(aconf);
        job.setJarByClass(this.getClass());
        setupMultiTableInputFormat(job, ruleset);
        setupAccumuloOutput(job, "");

        if (useCopyFileOutput) {
            // Configure job for file output
            job.setJobName("Ruleset-based export to file: " + tablePrefix + " -> " + localBaseOutputDir);
            // Map (row) to (table+key, key+value)
            job.setMapperClass(RowRuleMapper.class);
            job.setMapOutputKeyClass(GroupedRow.class);
            job.setMapOutputValueClass(GroupedRow.class);
            // Group according to table and and sort according to key
            job.setGroupingComparatorClass(GroupedRow.GroupComparator.class);
            job.setSortComparatorClass(GroupedRow.SortComparator.class);
            // Reduce ([table+row], rows): output each row to the file for that table, in sorted order
            job.setReducerClass(MultipleFileReducer.class);
            job.setOutputKeyClass(Key.class);
            job.setOutputValueClass(Value.class);
        }
        else {
            // Configure job for table output
            job.setJobName("Ruleset-based copy: " + tablePrefix + " -> " + childTablePrefix);
            // Map (row): convert to statement, insert to child (for namespace table, output row directly)
            job.setMapperClass(AccumuloRyaRuleMapper.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Mutation.class);
            job.setNumReduceTasks(0);
            // Create the child tables, so mappers don't try to do this in parallel
            for (final String parentTable : tables) {
                final String childTable = parentTable.replaceFirst(tablePrefix, childTablePrefix);
                createTableIfNeeded(childTable);
            }
        }

        // Run the job and copy files to local filesystem if needed
        final Date beginTime = new Date();
        log.info("Job started: " + beginTime);
        final boolean success = job.waitForCompletion(true);
        if (success) {
            if (useCopyFileOutput) {
                log.info("Moving data from HDFS to the local file system");
                final Path baseOutputPath = new Path(baseOutputDir);
                for (final FileStatus status : FileSystem.get(conf).listStatus(baseOutputPath)) {
                    if (status.isDirectory()) {
                        final String tableName = status.getPath().getName();
                        final Path hdfsPath = getPath(baseOutputDir, tableName);
                        final Path localPath = getPath(localBaseOutputDir, tableName);
                        log.info("HDFS directory: " + hdfsPath.toString());
                        log.info("Local directory: " + localPath.toString());
                        copyHdfsToLocal(hdfsPath, localPath);
                    }
                }
            }
            final Date endTime = new Date();
            log.info("Job finished: " + endTime);
            log.info("The job took " + (endTime.getTime() - beginTime.getTime()) / 1000 + " seconds.");
            return 0;
        } else {
            log.error("Job failed!!!");
            return 1;
        }
    }

    /**
     * Creates the child table if it doesn't already exist.
     * @param childTableName the name of the child table.
     * @throws IOException
     */
    public void createTableIfNeeded(final String childTableName) throws IOException {
        try {
            final Configuration childConfig = MergeToolMapper.getChildConfig(conf);
            final AccumuloRdfConfiguration childAccumuloRdfConfiguration = new AccumuloRdfConfiguration(childConfig);
            childAccumuloRdfConfiguration.setTablePrefix(childTablePrefix);
            final Connector childConnector = AccumuloRyaUtils.setupConnector(childAccumuloRdfConfiguration);
            if (!childConnector.tableOperations().exists(childTableName)) {
                log.info("Creating table: " + childTableName);
                childConnector.tableOperations().create(childTableName);
                log.info("Created table: " + childTableName);
                log.info("Granting authorizations to table: " + childTableName);
                childConnector.securityOperations().grantTablePermission(childUserName, childTableName, TablePermission.WRITE);
                log.info("Granted authorizations to table: " + childTableName);
            }
        } catch (TableExistsException | AccumuloException | AccumuloSecurityException e) {
            throw new IOException(e);
        }
    }

    private void setupSplitsFile(final Job job, final TableOperations parentTableOperations, final String parentTableName, final String childTableName) throws Exception {
        final FileSystem fs = FileSystem.get(conf);
        fs.setPermission(getPath(baseOutputDir, childTableName), new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
        final Path splitsPath = getPath(baseOutputDir, childTableName, "splits.txt");
        final Collection<Text> splits = parentTableOperations.listSplits(parentTableName, 100);
        log.info("Creating splits file at: " + splitsPath);
        try (PrintStream out = new PrintStream(new BufferedOutputStream(fs.create(splitsPath)), false, StandardCharsets.UTF_8.name())) {
            for (final Text split : splits) {
                final String encoded = new String(Base64.encodeBase64(TextUtil.getBytes(split)), StandardCharsets.UTF_8);
                out.println(encoded);
            }
        }
        fs.setPermission(splitsPath, new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));

        final String userDir = PathUtils.clean(System.getProperty("user.dir"));
        // The splits file has a symlink created in the user directory for some reason.
        // It might be better to copy the entire file for Windows but it doesn't seem to matter if
        // the user directory symlink is broken.
        java.nio.file.Files.deleteIfExists(new File(userDir, "splits.txt").toPath());
        //Files.copy(new File(splitsPath.toString()), new File(userDir, "splits.txt"));
        job.setPartitionerClass(KeyRangePartitioner.class);
        KeyRangePartitioner.setSplitFile(job, splitsPath.toString());
        job.setNumReduceTasks(splits.size() + 1);
    }

    /**
     * Converts a path string, or a sequence of strings that when joined form a path string,
     * to a {@link org.apache.hadoop.fs.Path}.
     * @param first The path string or initial part of the path string.
     * @param more Additional strings to be joined to form the path string.
     * @return the resulting {@link org.apache.hadoop.fs.Path}.
     */
    public static Path getPath(final String first, final String... more) {
        final java.nio.file.Path path = Paths.get(first, more);
        final String stringPath = FilenameUtils.separatorsToUnix(path.toAbsolutePath().toString());
        final Path hadoopPath = new Path(stringPath);
        return hadoopPath;
    }

    /**
     * Imports the files that hold the table data into the child instance.
     * @param childTableName the name of the child table to import.
     * @throws Exception
     */
    public void importFilesToChildTable(final String childTableName) throws Exception {
        final String normalizedChildTableName = FilenameUtils.normalize(childTableName);
        if (normalizedChildTableName == null) {
            throw new Exception("Invalid child table name: " + childTableName);
        }

        final Configuration childConfig = MergeToolMapper.getChildConfig(conf);
        final AccumuloRdfConfiguration childAccumuloRdfConfiguration = new AccumuloRdfConfiguration(childConfig);
        childAccumuloRdfConfiguration.setTablePrefix(childTablePrefix);
        final Connector childConnector = AccumuloRyaUtils.setupConnector(childAccumuloRdfConfiguration);
        final TableOperations childTableOperations = childConnector.tableOperations();

        final Path localWorkDir = getPath(localCopyFileImportDir, normalizedChildTableName);
        final Path hdfsBaseWorkDir = getPath(baseImportDir, normalizedChildTableName);

        final FileSystem fs = FileSystem.get(conf);
        if (fs.exists(hdfsBaseWorkDir)) {
            fs.delete(hdfsBaseWorkDir, true);
        }

        log.info("Importing from the local directory: " + localWorkDir);
        log.info("Importing to the HDFS directory: " + hdfsBaseWorkDir);
        copyLocalToHdfs(localWorkDir, hdfsBaseWorkDir);

        final Path files = getPath(hdfsBaseWorkDir.toString(), "files");
        final Path failures = getPath(hdfsBaseWorkDir.toString(), "failures");

        // With HDFS permissions on, we need to make sure the Accumulo user can read/move the files
        final FsShell hdfs = new FsShell(conf);
        if (!fs.isDirectory(hdfsBaseWorkDir)) {
            throw new IllegalArgumentException("Configured working directory is not a valid directory" + hdfsBaseWorkDir.toString());
        }
        hdfs.run(new String[] {"-chmod", "777", hdfsBaseWorkDir.toString()});
        if (fs.exists(failures)) {
            fs.delete(failures, true);
        }
        fs.mkdirs(failures);

        childTableOperations.importDirectory(normalizedChildTableName, files.toString(), failures.toString(), false);
    }

    /**
     * Copies the file from the local file system into the HDFS.
     * @param localInputPath the local system input {@link Path}.
     * @param hdfsOutputPath the HDFS output {@link Path}.
     * @throws IOException
     */
    public void copyLocalToHdfs(final Path localInputPath, final Path hdfsOutputPath) throws IOException {
        copyLocalToHdfs(localInputPath, hdfsOutputPath, conf);
    }

    /**
     * Copies the file from the local file system into the HDFS.
     * @param localInputPath the local system input {@link Path}.
     * @param hdfsOutputPath the HDFS output {@link Path}.
     * @param configuration the {@link Configuration} to use.
     * @throws IOException
     */
    public static void copyLocalToHdfs(final Path localInputPath, final Path hdfsOutputPath, final Configuration configuration) throws IOException {
        final FileSystem fs = FileSystem.get(configuration);
        fs.copyFromLocalFile(localInputPath, hdfsOutputPath);
    }

    /**
     * Copies the file from HDFS into the local file system.
     * @param hdfsInputPath the HDFS input {@link Path}.
     * @param localOutputPath the local system output {@link Path}.
     * @throws IOException
     */
    public void copyHdfsToLocal(final Path hdfsInputPath, final Path localOutputPath) throws IOException {
        copyHdfsToLocal(hdfsInputPath, localOutputPath, conf);
    }

    /**
     * Copies the file from HDFS into the local file system.
     * @param hdfsInputPath the HDFS input {@link Path}.
     * @param localOutputPath the local system output {@link Path}.
     * @param configuration the {@link Configuration} to use.
     * @throws IOException
     */
    public static void copyHdfsToLocal(final Path hdfsInputPath, final Path localOutputPath, final Configuration configuration) throws IOException {
        final FileSystem fs = FileSystem.get(configuration);
        fs.copyToLocalFile(hdfsInputPath, localOutputPath);
    }

    @Override
    protected void setupAccumuloInput(final Job job) throws AccumuloSecurityException {
        if (useCopyFileImport) {
            try {
                FileInputFormat.setInputPaths(job, localCopyFileImportDir);
            } catch (final IOException e) {
                log.error("Failed to set copy file import directory", e);
            }
        } else {
            // set up accumulo input
            if (!hdfsInput) {
                job.setInputFormatClass(AccumuloInputFormat.class);
            } else {
                job.setInputFormatClass(AccumuloHDFSFileInputFormat.class);
            }
            AbstractInputFormat.setConnectorInfo(job, userName, new PasswordToken(pwd));
            InputFormatBase.setInputTableName(job, RdfCloudTripleStoreUtils.layoutPrefixToTable(rdfTableLayout, tablePrefix));
            AbstractInputFormat.setScanAuthorizations(job, authorizations);
            if (!mock) {
                AbstractInputFormat.setZooKeeperInstance(job, new ClientConfiguration().withInstance(instance).withZkHosts(zk));
            } else {
                AbstractInputFormat.setMockInstance(job, instance);
            }
            if (ttl != null) {
                final IteratorSetting setting = new IteratorSetting(1, "fi", AgeOffFilter.class);
                AgeOffFilter.setTTL(setting, Long.valueOf(ttl));
                InputFormatBase.addIterator(job, setting);
            }
            if (startTime != null) {
                final IteratorSetting setting = getStartTimeSetting(startTime);
                InputFormatBase.addIterator(job, setting);
            }
            for (final IteratorSetting iteratorSetting : AccumuloRyaUtils.COMMON_REG_EX_FILTER_SETTINGS) {
                InputFormatBase.addIterator(job, iteratorSetting);
            }
        }
    }

    /**
     * Set up job to use AccumuloMultiTableInput format, using the tables/ranges given by a ruleset.
     * @param job The Job to configure
     * @param rules The ruleset mapping a query to the appropriate tables and ranges
     */
    protected void setupMultiTableInputFormat(final Job job, final AccumuloQueryRuleset rules) throws AccumuloSecurityException {
        AbstractInputFormat.setConnectorInfo(job, userName, new PasswordToken(pwd));
        AbstractInputFormat.setScanAuthorizations(job, authorizations);
        if (!mock) {
            AbstractInputFormat.setZooKeeperInstance(job, new ClientConfiguration().withInstance(instance).withZkHosts(zk));
        } else {
            AbstractInputFormat.setMockInstance(job, instance);
        }
        final Map<String, InputTableConfig> configs = rules.getInputConfigs();
        // Add any relevant iterator settings
        final List<IteratorSetting> additionalSettings = new LinkedList<>(AccumuloRyaUtils.COMMON_REG_EX_FILTER_SETTINGS);
        if (ttl != null) {
            final IteratorSetting ttlSetting = new IteratorSetting(1, "fi", AgeOffFilter.class);
            AgeOffFilter.setTTL(ttlSetting, Long.valueOf(ttl));
            additionalSettings.add(ttlSetting);
        }
        if (startTime != null) {
            final IteratorSetting startTimeSetting = getStartTimeSetting(startTime);
            additionalSettings.add(startTimeSetting);
        }
        for (final Map.Entry<String, InputTableConfig> entry : configs.entrySet()) {
            final List<IteratorSetting> iterators = entry.getValue().getIterators();
            iterators.addAll(additionalSettings);
            entry.getValue().setIterators(iterators);
        }
        // Set the input format
        AccumuloMultiTableInputFormat.setInputTableConfigs(job, configs);
        job.setInputFormatClass(AccumuloMultiTableInputFormat.class);
    }

    @Override
    protected void setupAccumuloOutput(final Job job, final String outputTable) throws AccumuloSecurityException {
        AccumuloOutputFormat.setConnectorInfo(job, childUserName, new PasswordToken(childPwd));
        AccumuloOutputFormat.setCreateTables(job, true);
        AccumuloOutputFormat.setDefaultTableName(job, outputTable);
        if (!childMock) {
            AccumuloOutputFormat.setZooKeeperInstance(job, new ClientConfiguration().withInstance(childInstance).withZkHosts(childZk));
        } else {
            AccumuloOutputFormat.setMockInstance(job, childInstance);
        }
        if (useCopyFileOutput) {
            log.info("Using file output format mode.");
            if (StringUtils.isNotBlank(baseOutputDir)) {
                Path baseOutputPath;
                Path filesOutputPath;
                if (StringUtils.isNotBlank(outputTable)) {
                    filesOutputPath = getPath(baseOutputDir, outputTable, "files");
                    baseOutputPath = filesOutputPath.getParent();
                    job.setOutputFormatClass(AccumuloFileOutputFormat.class);
                }
                else {
                    // If table name is not given, configure output for one level higher:
                    // it's up to the job to handle subdirectories. Make sure the parent
                    // exists.
                    filesOutputPath = getPath(baseOutputDir);
                    baseOutputPath = filesOutputPath;
                    LazyOutputFormat.setOutputFormatClass(job, AccumuloFileOutputFormat.class);
                    MultipleOutputs.setCountersEnabled(job, true);
                }
                log.info("File output destination: " + filesOutputPath);
                if (useCopyFileOutputDirectoryClear) {
                    try {
                        clearOutputDir(baseOutputPath);
                    } catch (final IOException e) {
                        log.error("Error clearing out output path.", e);
                    }
                }
                try {
                    final FileSystem fs = FileSystem.get(conf);
                    fs.mkdirs(filesOutputPath.getParent());
                    fs.setPermission(filesOutputPath.getParent(), new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
                } catch (final IOException e) {
                    log.error("Failed to set permission for output path.", e);
                }
                FileOutputFormat.setOutputPath(job, filesOutputPath);

                if (StringUtils.isNotBlank(compressionType)) {
                    if (isValidCompressionType(compressionType)) {
                        log.info("File compression type: " + compressionType);
                        AccumuloFileOutputFormat.setCompressionType(job, compressionType);
                    } else {
                        log.warn("Invalid compression type: " + compressionType);
                    }
                }
            }
        } else {
            log.info("Using accumulo output format mode.");
            job.setOutputFormatClass(AccumuloOutputFormat.class);
        }
    }

    /**
     * Sets up and runs the copy tool with the provided args.
     * @param args the arguments list.
     * @return the execution result.
     */
    public int setupAndRun(final String[] args) {
        int returnCode = -1;
        try {
            final Configuration conf = new Configuration();
            final Set<String> toolArgs = ToolConfigUtils.getUserArguments(conf, args);
            if (!toolArgs.isEmpty()) {
                final String parameters = Joiner.on("\r\n\t").join(toolArgs);
                log.info("Running Copy Tool with the following parameters...\r\n\t" + parameters);
            }

            returnCode = ToolRunner.run(conf, this, args);
        } catch (final Exception e) {
            log.error("Error running copy tool", e);
        }
        return returnCode;
    }

    public static void main(final String[] args) {
        final String log4jConfiguration = System.getProperties().getProperty("log4j.configuration");
        if (StringUtils.isNotBlank(log4jConfiguration)) {
            final String parsedConfiguration = PathUtils.clean(StringUtils.removeStart(log4jConfiguration, "file:"));
            final File configFile = new File(parsedConfiguration);
            if (configFile.exists()) {
                DOMConfigurator.configure(parsedConfiguration);
            } else {
                BasicConfigurator.configure();
            }
        }
        log.info("Starting Copy Tool");

        Thread.setDefaultUncaughtExceptionHandler((thread, throwable) -> log.error("Uncaught exception in " + thread.getName(), throwable));

        final CopyTool copyTool = new CopyTool();
        final int returnCode = copyTool.setupAndRun(args);

        log.info("Finished running Copy Tool");

        System.exit(returnCode);
    }

    /**
     * Creates an {@link IteratorSetting} with a time stamp filter that starts with the specified data.
     * @param startTimeString the start time of the filter.
     * @return the {@link IteratorSetting}.
     */
    public static IteratorSetting getStartTimeSetting(final String startTimeString) {
        Date date = null;
        try {
            date = MergeTool.START_TIME_FORMATTER.parse(startTimeString);
        } catch (final ParseException e) {
            throw new IllegalArgumentException("Couldn't parse " + startTimeString, e);
        }
        return getStartTimeSetting(date);
    }

    /**
     * Creates an {@link IteratorSetting} with a time stamp filter that starts with the specified data.
     * @param date the start {@link Date} of the filter.
     * @return the {@link IteratorSetting}.
     */
    public static IteratorSetting getStartTimeSetting(final Date date) {
        return getStartTimeSetting(date.getTime());
    }

    /**
     * Creates an {@link IteratorSetting} with a time stamp filter that starts with the specified data.
     * @param time the start time of the filter.
     * @return the {@link IteratorSetting}.
     */
    public static IteratorSetting getStartTimeSetting(final long time) {
        final IteratorSetting setting = new IteratorSetting(1, "startTimeIterator", TimestampFilter.class);
        TimestampFilter.setStart(setting, time, true);
        TimestampFilter.setEnd(setting, Long.MAX_VALUE, true);
        return setting;
    }

    /**
     * Checks to see if the specified compression type is valid. The compression must be defined in
     * {@link Algorithm} to be valid.
     * @param compressionType the compression type to check.
     * @return {@code true} if the compression type is one of "none", "gz", "lzo", or "snappy".
     * {@code false} otherwise.
     */
    private static boolean isValidCompressionType(final String compressionType) {
        for (final Algorithm algorithm : Algorithm.values()) {
            if (algorithm.getName().equals(compressionType)) {
                return true;
            }
        }
        return false;
    }

    private void clearOutputDir(final Path path) throws IOException {
        final FileSystem fs = FileSystem.get(conf);
        fs.delete(path, true);
    }

    private Instance createChildInstance(final Configuration config) throws Exception {
        Instance instance = null;
        String instanceTypeProp = config.get(CREATE_CHILD_INSTANCE_TYPE_PROP);
        final String childAuth = config.get(MRUtils.AC_AUTH_PROP + MergeTool.CHILD_SUFFIX);

        // Default to distribution cluster if not specified
        if (StringUtils.isBlank(instanceTypeProp)) {
            instanceTypeProp = InstanceType.DISTRIBUTION.toString();
        }

        final InstanceType instanceType = InstanceType.fromName(instanceTypeProp);
        switch (instanceType) {
            case DISTRIBUTION:
                if (childInstance == null) {
                    throw new IllegalArgumentException("Must specify instance name for distributed mode");
                } else if (childZk == null) {
                    throw new IllegalArgumentException("Must specify ZooKeeper hosts for distributed mode");
                }
                instance = new ZooKeeperInstance(childInstance, childZk);
                break;

            case MINI:
                childAccumuloInstanceDriver = new AccumuloInstanceDriver("Child", false, true, false, false, childUserName, childPwd, childInstance, childTablePrefix, childAuth);
                childAccumuloInstanceDriver.setUpInstance();
                childAccumuloInstanceDriver.setUpTables();
                childZk = childAccumuloInstanceDriver.getZooKeepers();
                MergeTool.setDuplicateKeysForProperty(config, MRUtils.AC_ZK_PROP+ MergeTool.CHILD_SUFFIX, childZk);
                instance = new ZooKeeperInstance(childInstance, childZk);
                break;

            case MOCK:
                instance = new MockInstance(childInstance);
                break;

            default:
                throw new AccumuloException("Unexpected instance type: " + instanceType);
        }

        return instance;
    }

    /**
     * @return the child {@link AccumuloInstanceDriver} or {@code null}.
     */
    public AccumuloInstanceDriver getChildAccumuloInstanceDriver() {
        return childAccumuloInstanceDriver;
    }

    /**
     * Shuts down the child {@link AccumuloInstanceDriver} in the {@link CopyTool} if it exists.
     * @throws Exception
     */
    public void shutdown() throws Exception {
        if (childAccumuloInstanceDriver != null) {
            childAccumuloInstanceDriver.tearDown();
        }
    }
}