public class FileSystemHelper {
    public static String makeCommaSeparated(Path[] strs) {
        if(strs == null) {
            return null;
        }
        
        StringBuilder sb = new StringBuilder();
        for(int i=0;i<strs.length;i++) {
            sb.append(strs[i].toString());
            if(i < strs.length - 1) {
                sb.append(",");
            }
        }
        return sb.toString();
    }
    
    public static String makeCommaSeparated(Collection<String> strs) {
        if(strs == null) {
            return null;
        }
        
        StringBuilder sb = new StringBuilder();
        int i=0;
        for(String str : strs) {
            sb.append(str);
            i++;
            
            if(i<strs.size()) {
                sb.append(",");
            }
        }
        return sb.toString();
    }
    
    public static String makeCommaSeparated(String[] strs) {
        if(strs == null) {
            return null;
        }
        
        StringBuilder sb = new StringBuilder();
        for(int i=0;i<strs.length;i++) {
            sb.append(strs[i]);
            if(i < strs.length - 1) {
                sb.append(",");
            }
        }
        return sb.toString();
    }
    
    public static String[] splitCommaSeparated(String comma_separated_input) {
        String[] inputs = comma_separated_input.split(",");
        return inputs;
    }
    
    public static Path[] makePathFromString(Configuration conf, Collection<String> pathStrings) throws IOException {
        return makePathFromString(conf, pathStrings.toArray(new String[0]));
    }
    
    public static Path[] makePathFromString(Configuration conf, String[] pathStrings) throws IOException {
        List<Path> paths = new ArrayList<Path>();
        for(String path : pathStrings) {
            if(hasWildcard(path)) {
                Path[] patharr = resolveWildcard(conf, path);
                for(Path pathentry : patharr) {
                    paths.add(pathentry);
                }
            } else {
                paths.add(new Path(path));
            }
        }
        return paths.toArray(new Path[0]);
    }
    
    public static String[] makeStringFromPath(Path[] paths) {
        String[] pathStrings = new String[paths.length];
        for(int i=0;i<paths.length;i++) {
            pathStrings[i] = paths[i].toString();
        }
        return pathStrings;
    }
    
    public static Path[] getAllSamplePaths(Configuration conf, String inputPathsCommaSeparated) throws IOException {
        return FileSystemHelper.getAllSamplePaths(conf, makePathFromString(conf, splitCommaSeparated(inputPathsCommaSeparated)));
    }
    
    public static Path[] getAllSamplePaths(Configuration conf, String[] inputPaths) throws IOException {
        return FileSystemHelper.getAllSamplePaths(conf, makePathFromString(conf, inputPaths));
    }
    
    public static Path[] getAllSamplePaths(Configuration conf, Collection<String> inputPaths) throws IOException {
        return FileSystemHelper.getAllSamplePaths(conf, makePathFromString(conf, inputPaths));
    }
    
    public static Path[] getAllSamplePaths(Configuration conf, Path[] inputPaths) throws IOException {
        List<Path> inputFiles = new ArrayList<Path>();
        SamplePathFilter filter = new SamplePathFilter();
        
        for(Path path : inputPaths) {
            FileSystem fs = path.getFileSystem(conf);
            FileStatus status = fs.getFileStatus(path);
            if(status.isDirectory()) {
                FileStatus[] entries = fs.listStatus(path);
                entries = sortFileStatusList(entries);
                for(FileStatus entry : entries) {
                    if(filter.accept(entry.getPath())) {
                        inputFiles.add(entry.getPath());
                    }
                }
            } else {
                FileStatus entry = fs.getFileStatus(path);
                if(filter.accept(entry.getPath())) {
                    inputFiles.add(entry.getPath());
                }
            }
        }
        
        Path[] files = inputFiles.toArray(new Path[0]);
        return files;
    }
    
    public static boolean hasWildcard(String path) {
        if(path.indexOf("*") >= 0) {
            return true;
        }
        
        return false;
    }

    private static Path[] resolveWildcard(Configuration conf, String path) throws IOException {
        List<Path> paths = new ArrayList<Path>();
        
        int idxWildcard = path.lastIndexOf("*");
        if(idxWildcard >= 0) {
            String left = null;
            String right = path.substring(idxWildcard+1);
            
            String parentPath = path.substring(0, idxWildcard);
            int idxParent = parentPath.lastIndexOf("/");
            if(idxParent >= 0) {
                parentPath = parentPath.substring(0, idxParent);
                left = path.substring(idxParent+1, idxWildcard);
            } else {
                parentPath = "";
                left = path.substring(0, idxWildcard);
            }
            
            Path parent = new Path(parentPath);
            FileSystem fs = parent.getFileSystem(conf);
            FileStatus status = fs.getFileStatus(parent);
            if(status.isDirectory()) {
                FileStatus[] entries = fs.listStatus(parent);
                entries = sortFileStatusList(entries);
                for(FileStatus entry : entries) {
                    if(!left.isEmpty()) {
                        if(!entry.getPath().getName().startsWith(left)) {
                            // skip
                            continue;
                        }
                    }
                    
                    if(!right.isEmpty()) {
                        if(!entry.getPath().getName().endsWith(right)) {
                            // skip
                            continue;
                        }
                    }
                    
                    paths.add(entry.getPath());
                }
            }
        }
        
        return paths.toArray(new Path[0]);
    }
    
    private static FileStatus[] sortFileStatusList(FileStatus[] list) {
        List<FileStatus> l = new ArrayList<FileStatus>();
        for(FileStatus status : list) {
            l.add(status);
        }
        
        Collections.sort(l, new Comparator<FileStatus>(){

            @Override
            public int compare(FileStatus arg0, FileStatus arg1) {
                return arg0.getPath().getName().compareTo(arg1.getPath().getName());
            }
        });
        
        return l.toArray(new FileStatus[0]);
    }
} public class Report {
    private List<Job> jobs;
    
    public Report() {
        this.jobs = new ArrayList<Job>();
    }

    public void addJob(org.apache.hadoop.mapreduce.Job job) {
        this.jobs.add(new Job(job));
    }
    
    public void addJob(Job job) {
        this.jobs.add(job);
    }
    
    public void addJob(Job[] jobs) {
        for(Job job : jobs) {
            this.jobs.add(job);
        }
    }
    
    public void addJob(List<Job> jobs) {
        this.jobs.addAll(jobs);
    }
    
    public void writeTo(String filename) throws IOException {
        writeTo(new File("./", filename));
    }
    
    public void writeTo(File f) throws IOException {
        if(f.getParentFile() != null) {
            if(!f.getParentFile().exists()) {
                f.getParentFile().mkdirs();
            }
        }
        
        Writer writer = new FileWriter(f, true);
        boolean first = true;
        for(Job job : this.jobs) {
            if(first) {
                first = false;
            }
            writer.write(makeText(job));
            writer.write("\n\n");
        }
        
        writer.close();
    }
    
    private String makeText(Job job) {
        return "Job : " + job.getName() + "\n" +
                "JobID : " + job.getID() + "\n" + 
                "Status : " + job.getStatus() + "\n" +
                "StartTime : " + TimeHelper.getTimeString(job.getStartTime()) + "\n" +
                "FinishTime : " + TimeHelper.getTimeString(job.getFinishTime()) + "\n" + 
                "TimeTaken : " + TimeHelper.getDiffTimeString(job.getStartTime(), job.getFinishTime()) + "\n\n" +
                job.getCountersStr();
    }
} public class MapReduceHelper {
    public static String getOutputNameFromMapReduceOutput(Path mapreduceOutputPath) {
        return getOutputNameFromMapReduceOutput(mapreduceOutputPath.getName());
    }
    
    public static String getOutputNameFromMapReduceOutput(String mapreduceOutputName) {
        int midx = mapreduceOutputName.indexOf("-m-");
        if(midx > 0) {
            return mapreduceOutputName.substring(0, midx);
        }
        
        int ridx = mapreduceOutputName.indexOf("-r-");
        if(ridx > 0) {
            return mapreduceOutputName.substring(0, ridx);
        }
        
        return mapreduceOutputName;
    }
    
    public static int getMapReduceID(Path mapreduceOutputName) {
        return getMapReduceID(mapreduceOutputName.getName());
    }
    
    public static int getMapReduceID(String mapreduceOutputName) {
        int midx = mapreduceOutputName.indexOf("-m-");
        if(midx > 0) {
            return Integer.parseInt(mapreduceOutputName.substring(midx + 3));
        }
        
        int ridx = mapreduceOutputName.indexOf("-r-");
        if(ridx > 0) {
            return Integer.parseInt(mapreduceOutputName.substring(ridx + 3));
        }
        
        return 0;
    }
    
    public static boolean isLogFiles(Path path) {
        if(path.getName().equals("_SUCCESS")) {
            return true;
        } else if(path.getName().equals("_logs")) {
            return true;
        }
        return false;
    }
    
    public static boolean isPartialOutputFiles(Path path) {
        if(path.getName().startsWith("part-r-")) {
            return true;
        } else if(path.getName().startsWith("part-m-")) {
            return true;
        }
        return false;
    }
} public class KmerMatchFileMapping {
    
    private static final Log LOG = LogFactory.getLog(KmerMatchFileMapping.class);
    
    private static final String HADOOP_CONFIG_KEY = "libra.common.kmermatch.kmermatcherfilemapping";
    
    private Hashtable<String, Integer> idTable;
    private List<String> objList;

    public static KmerMatchFileMapping createInstance(File file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerMatchFileMapping) serializer.fromJsonFile(file, KmerMatchFileMapping.class);
    }
    
    public static KmerMatchFileMapping createInstance(String json) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerMatchFileMapping) serializer.fromJson(json, KmerMatchFileMapping.class);
    }
    
    public static KmerMatchFileMapping createInstance(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerMatchFileMapping) serializer.fromJsonConfiguration(conf, HADOOP_CONFIG_KEY, KmerMatchFileMapping.class);
    }
    
    public static KmerMatchFileMapping createInstance(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerMatchFileMapping) serializer.fromJsonFile(fs, file, KmerMatchFileMapping.class);
    }
    
    public KmerMatchFileMapping() {
        this.idTable = new Hashtable<String, Integer>();
        this.objList = new ArrayList<String>();
    }
    
    @JsonProperty("sample_files")
    public Collection<String> getSampleFiles() {
        return this.objList;
    }
    
    @JsonProperty("sample_files")
    public void addSampleFile(Collection<String> sampleFiles) {
        for(String sampleFile : sampleFiles) {
            addSampleFile(sampleFile);
        }
    }
    
    @JsonIgnore
    public void addSampleFile(String sampleFile) {
        this.idTable.put(sampleFile, this.objList.size());
        this.objList.add(sampleFile);
    }
    
    @JsonIgnore
    public int getIDFromSampleFile(String sampleFile) throws IOException {
        if(this.idTable.get(sampleFile) == null) {
            throw new IOException("could not find id from " + sampleFile);
        } else {
            return this.idTable.get(sampleFile);
        }
    }
    
    @JsonIgnore
    public String getSampleFileFromID(int id) throws IOException {
        if(this.objList.size() <= id) {
            throw new IOException("could not find a sample file from " + id);
        } else {
            return this.objList.get(id);
        }
    }
    
    @JsonIgnore
    public int getSize() {
        return this.objList.size();
    }
    
    @JsonIgnore
    public void saveTo(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonConfiguration(conf, HADOOP_CONFIG_KEY, this);
    }
    
    @JsonIgnore
    public void saveTo(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonFile(fs, file, this);
    }
} public class KmerMatchInputFormat extends SequenceFileInputFormat<CompressedSequenceWritable, KmerMatchResult> {

    private static final Log LOG = LogFactory.getLog(KmerMatchInputFormat.class);

    private final static String NUM_INPUT_FILES = "mapreduce.input.num.files";
    
    @Override
    public RecordReader<CompressedSequenceWritable, KmerMatchResult> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException {
        return new KmerMatchRecordReader();
    }
    
    public static void setInputFormatConfig(JobContext job, KmerMatchInputFormatConfig inputFormatConfig) throws IOException {
        inputFormatConfig.saveTo(job.getConfiguration());
    }
    
    public List<InputSplit> getSplits(JobContext job) throws IOException {
        Configuration conf = job.getConfiguration();
        KmerMatchInputFormatConfig inputFormatConfig = KmerMatchInputFormatConfig.createInstance(conf);
        
        // generate splits
        List<InputSplit> splits = new ArrayList<InputSplit>();
        List<FileStatus> files = listStatus(job);
        List<Path> kmerIndexTableFiles = new ArrayList<Path>();
        for (FileStatus file : files) {
            Path path = file.getPath();
            kmerIndexTableFiles.add(path);
        }
        
        LOG.info("# of Split input file : " + kmerIndexTableFiles.size());
        for(int i=0;i<kmerIndexTableFiles.size();i++) {
            LOG.info("> " + kmerIndexTableFiles.get(i).toString());
        }
        
        int partitions = 0;
        for(Path kmerIndexTableFile : kmerIndexTableFiles) {
            FileSystem fs = kmerIndexTableFile.getFileSystem(conf);
            KmerIndexTable indexTable = KmerIndexTable.createInstance(fs, kmerIndexTableFile);
            if(partitions == 0) {
                partitions = indexTable.getSize();
            } else {
                if(partitions != indexTable.getSize()) {
                    throw new IOException(String.format("# of partitions are different between input files - %d expected, but %d got", partitions, indexTable.getSize()));
                }
            }
        }
        
        if(partitions == 0) {
            throw new IOException("There is no partition");
        }
        
        for(int i=0;i<partitions;i++) {
            splits.add(new KmerMatchInputSplit(inputFormatConfig.getKmerSize(), kmerIndexTableFiles.toArray(new Path[0]), i));
        }
        
        // Save the number of input files in the job-conf
        job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());

        LOG.debug("Total # of splits: " + splits.size());
        return splits;
    }
    
    @Override
    protected List<FileStatus> listStatus(JobContext job) throws IOException {
        Configuration conf = job.getConfiguration();
        KmerMatchInputFormatConfig inputFormatConfig = KmerMatchInputFormatConfig.createInstance(conf);
        
        List<FileStatus> result = new ArrayList<FileStatus>();
        Path[] fileTablePaths = getInputPaths(job);
        if (fileTablePaths.length == 0) {
            throw new IOException("No input paths specified in job");
        }
        
        // get tokens for all the required FileSystems..
        TokenCache.obtainTokensForNamenodes(job.getCredentials(), fileTablePaths, conf);

        List<FileTable> fileTables = new ArrayList<FileTable>();
        List<Path> inputKmerIndexTableFiles = new ArrayList<Path>();
        for(Path fileTablePath : fileTablePaths) {
            FileSystem fs = fileTablePath.getFileSystem(conf);
            FileTable ft = FileTable.createInstance(fs, fileTablePath);
            fileTables.add(ft);
            
            String kmerIndexTableFileName = KmerIndexHelper.makeKmerIndexTableFileName(ft.getName());
            Path kmerIndexTableFilePath = new Path(inputFormatConfig.getKmerIndexPath(), kmerIndexTableFileName);
            inputKmerIndexTableFiles.add(kmerIndexTableFilePath);
        }
        
        // get tokens for all the required FileSystems..
        TokenCache.obtainTokensForNamenodes(job.getCredentials(), inputKmerIndexTableFiles.toArray(new Path[0]), conf);
        
        // creates a MultiPathFilter with the hiddenFileFilter and the
        // user provided one (if any).
        List<PathFilter> filters = new ArrayList<PathFilter>();
        PathFilter jobFilter = getInputPathFilter(job);
        if (jobFilter != null) {
            filters.add(jobFilter);
        }
        filters.add(new KmerIndexTablePathFilter());
        PathFilter inputFilter = new MultiPathFilter(filters);
        
        for (int i = 0; i < inputKmerIndexTableFiles.size(); ++i) {
            Path inputKmerIndexTableFile = inputKmerIndexTableFiles.get(i);
            if(inputFilter.accept(inputKmerIndexTableFile)) {
                FileSystem fs = inputKmerIndexTableFile.getFileSystem(conf);
                FileStatus status = fs.getFileStatus(inputKmerIndexTableFile);
                result.add(status);
            }
        }

        LOG.info("Total input paths to process : " + result.size());
        return result;
    }
    
    private static class MultiPathFilter implements PathFilter {

        private List<PathFilter> filters;

        public MultiPathFilter(List<PathFilter> filters) {
            this.filters = filters;
        }

        public boolean accept(Path path) {
            for (PathFilter filter : filters) {
                if (!filter.accept(path)) {
                    return false;
                }
            }
            return true;
        }
    }
} public class KmerMatchInputFormatConfig {
    
    private static final Log LOG = LogFactory.getLog(KmerMatchInputFormatConfig.class);
    
    private static final String HADOOP_CONFIG_KEY = "libra.common.kmermatch.kmermatchinputformatconfig";
    
    private int kmerSize;
    private String fileTablePath;
    private String kmerIndexPath;
    
    public static KmerMatchInputFormatConfig createInstance(File file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerMatchInputFormatConfig) serializer.fromJsonFile(file, KmerMatchInputFormatConfig.class);
    }
    
    public static KmerMatchInputFormatConfig createInstance(String json) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerMatchInputFormatConfig) serializer.fromJson(json, KmerMatchInputFormatConfig.class);
    }
    
    public static KmerMatchInputFormatConfig createInstance(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerMatchInputFormatConfig) serializer.fromJsonConfiguration(conf, HADOOP_CONFIG_KEY, KmerMatchInputFormatConfig.class);
    }
    
    public static KmerMatchInputFormatConfig createInstance(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerMatchInputFormatConfig) serializer.fromJsonFile(fs, file, KmerMatchInputFormatConfig.class);
    }
    
    public KmerMatchInputFormatConfig() {
    }
    
    @JsonProperty("kmer_size")
    public void setKmerSize(int kmerSize) {
        this.kmerSize = kmerSize;
    }
    
    @JsonProperty("kmer_size")
    public int getKmerSize() {
        return this.kmerSize;
    }
    
    @JsonProperty("file_table_path")
    public String getFileTablePath() {
        return this.fileTablePath;
    }
    
    @JsonProperty("file_table_path")
    public void setFileTablePath(String fileTablePath) {
        this.fileTablePath = fileTablePath;
    }
    
    @JsonProperty("kmer_index_path")
    public String getKmerIndexPath() {
        return this.kmerIndexPath;
    }
    
    @JsonProperty("kmer_index_path")
    public void setKmerIndexPath(String kmerIndexPath) {
        this.kmerIndexPath = kmerIndexPath;
    }
    
    @JsonIgnore
    public void saveTo(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonConfiguration(conf, HADOOP_CONFIG_KEY, this);
    }
    
    @JsonIgnore
    public void saveTo(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonFile(fs, file, this);
    }
} public class DistanceMatrixConfig {
    
    public static final String DEFAULT_OUTPUT_PATH = "./libra_output";
    public static WeightAlgorithm DEFAULT_WEIGHT_ALGORITHM = WeightAlgorithm.LOGARITHM;
    public static ScoreAlgorithm DEFAULT_SCORE_ALGORITHM = ScoreAlgorithm.COSINESIMILARITY;
    public static RunMode DEFAULT_RUN_MODE = RunMode.MAP;
    
    private static final String HADOOP_CONFIG_KEY = "libra.distancematrix.common.distancematrixconfig";
    
    private String reportFilePath;
    
    private String fileTablePath;
    private String kmerIndexPath;
    private String kmerStatisticsPath;
    private WeightAlgorithm weightAlgorithm = WeightAlgorithm.LOGARITHM;
    private ScoreAlgorithm scoreAlgorithm = ScoreAlgorithm.COSINESIMILARITY;
    private RunMode runMode = RunMode.MAP;
    private String outputPath = DEFAULT_OUTPUT_PATH;
    
    private List<FileTable> fileTables = new ArrayList<FileTable>();
    
    public static DistanceMatrixConfig createInstance(File file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (DistanceMatrixConfig) serializer.fromJsonFile(file, DistanceMatrixConfig.class);
    }
    
    public static DistanceMatrixConfig createInstance(String json) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (DistanceMatrixConfig) serializer.fromJson(json, DistanceMatrixConfig.class);
    }
    
    public static DistanceMatrixConfig createInstance(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (DistanceMatrixConfig) serializer.fromJsonConfiguration(conf, HADOOP_CONFIG_KEY, DistanceMatrixConfig.class);
    }
    
    public static DistanceMatrixConfig createInstance(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (DistanceMatrixConfig) serializer.fromJsonFile(fs, file, DistanceMatrixConfig.class);
    }
    
    public DistanceMatrixConfig() {
        
    }

    @JsonIgnore
    public void setPreprocessRootPath(String preprocessRootPath) {
        this.fileTablePath = FileTableHelper.makeFileTableDirPath(preprocessRootPath);
        this.kmerIndexPath = KmerIndexHelper.makeKmerIndexDirPath(preprocessRootPath);
        this.kmerStatisticsPath = KmerStatisticsHelper.makeKmerStatisticsDirPath(preprocessRootPath);
    }
    
    @JsonProperty("file_table_path")
    public String getFileTablePath() {
        return this.fileTablePath;
    }
    
    @JsonProperty("file_table_path")
    public void setFileTablePath(String fileTablePath) {
        this.fileTablePath = fileTablePath;
    }
    
    @JsonProperty("kmer_index_path")
    public String getKmerIndexPath() {
        return this.kmerIndexPath;
    }
    
    @JsonProperty("kmer_index_path")
    public void setKmerIndexPath(String kmerIndexPath) {
        this.kmerIndexPath = kmerIndexPath;
    }
    
    @JsonProperty("statistics_path")
    public String getKmerStatisticsPath() {
        return this.kmerStatisticsPath;
    }
    
    @JsonProperty("statistics_path")
    public void setKmerStatisticsPath(String kmerStatisticsPath) {
        this.kmerStatisticsPath = kmerStatisticsPath;
    }
    
    @JsonProperty("weight_algorithm")
    public WeightAlgorithm getWeightAlgorithm() {
        return this.weightAlgorithm;
    }
    
    @JsonProperty("weight_algorithm")
    public void setWeightAlgorithm(WeightAlgorithm weightAlgorithm) {
        this.weightAlgorithm = weightAlgorithm;
    }
    
    @JsonProperty("score_algorithm")
    public ScoreAlgorithm getScoreAlgorithm() {
        return this.scoreAlgorithm;
    }
    
    @JsonProperty("score_algorithm")
    public void setScoreAlgorithm(ScoreAlgorithm scoreAlgorithm) {
        this.scoreAlgorithm = scoreAlgorithm;
    }
    
    @JsonProperty("run_mode")
    public RunMode getRunMode() {
        return this.runMode;
    }
    
    @JsonProperty("run_mode")
    public void setRunMode(RunMode runMode) {
        this.runMode = runMode;
    }
    
    @JsonProperty("output_path")
    public void setOutputPath(String outputPath) {
        this.outputPath = outputPath;
    }
    
    @JsonProperty("output_path")
    public String getOutputPath() {
        return this.outputPath;
    }
    
    @JsonProperty("report_path")
    public void setReportPath(String reportFilePath) {
        this.reportFilePath = reportFilePath;
    }
    
    @JsonProperty("report_path")
    public String getReportPath() {
        return this.reportFilePath;
    }
    
    @JsonProperty("file_table")
    public Collection<FileTable> getFileTables() {
        return this.fileTables;
    }
    
    @JsonProperty("file_table")
    public void addFileTables(Collection<FileTable> fileTable) {
        this.fileTables.addAll(fileTable);
    }
    
    @JsonIgnore
    public void addFileTable(FileTable fileTable) {
        this.fileTables.add(fileTable);
    }
    
    @JsonIgnore
    public void saveTo(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonConfiguration(conf, HADOOP_CONFIG_KEY, this);
    }
    
    @JsonIgnore
    public void saveTo(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonFile(fs, file, this);
    }
} public class DistanceMatrixConfigException extends Exception {
    static final long serialVersionUID = 7818375828146090155L;

    public DistanceMatrixConfigException() {
        super();
    }

    public DistanceMatrixConfigException(String string) {
        super(string);
    }

    public DistanceMatrixConfigException(String string, Throwable thrwbl) {
        super(string, thrwbl);
    }

    public DistanceMatrixConfigException(Throwable thrwbl) {
        super(thrwbl);
    }
} public class KmerSimilarityHelper {
    private final static String KMER_SIMILARITY_RESULT_PART_PATH_EXP = ".+\\." + DistanceMatrixConstants.KMER_SIMILARITY_RESULT_FILENAME_EXTENSION + "\\.\\d+$";
    private final static Pattern KMER_SIMILARITY_RESULT_PART_PATH_PATTERN = Pattern.compile(KMER_SIMILARITY_RESULT_PART_PATH_EXP);
    
    public static boolean isKmerSimilarityFileMappingTableFile(Path path) {
        return isKmerSimilarityFileMappingTableFile(path.getName());
    }
    
    public static boolean isKmerSimilarityFileMappingTableFile(String path) {
        if(path.compareToIgnoreCase(DistanceMatrixConstants.KMER_SIMILARITY_FILE_MAPPING_TABLE_FILENAME) == 0) {
            return true;
        }
        return false;
    }
    
    public static boolean isKmerSimilarityResultPartFile(Path path) {
        return KmerSimilarityHelper.isKmerSimilarityResultPartFile(path.getName());
    }
    
    public static boolean isKmerSimilarityResultPartFile(String path) {
        Matcher matcher = KMER_SIMILARITY_RESULT_PART_PATH_PATTERN.matcher(path.toLowerCase());
        if(matcher.matches()) {
            return true;
        }
        return false;
    }
    
    public static String makeKmerSimilarityFileMappingTableFileName() {
        return DistanceMatrixConstants.KMER_SIMILARITY_FILE_MAPPING_TABLE_FILENAME;
    }
    
    public static String makeKmerSimilarityResultPartFileName(int mapreduceID) {
        return DistanceMatrixConstants.KMER_SIMILARITY_RESULT_FILENAME_PREFIX + "." + DistanceMatrixConstants.KMER_SIMILARITY_RESULT_FILENAME_EXTENSION + "." + mapreduceID;
    }
    
    public static String makeKmerSimilarityResultFileName() {
        return DistanceMatrixConstants.KMER_SIMILARITY_RESULT_FILENAME_PREFIX + "." + DistanceMatrixConstants.KMER_SIMILARITY_RESULT_FILENAME_EXTENSION;
    }
    
    public static Path[] getKmerSimilarityResultPartFilePath(Configuration conf, Path inputPath) throws IOException {
        List<Path> inputFiles = new ArrayList<Path>();
        KmerSimilarityResultPartPathFilter filter = new KmerSimilarityResultPartPathFilter();
        
        FileSystem fs = inputPath.getFileSystem(conf);
        if(fs.exists(inputPath)) {
            FileStatus status = fs.getFileStatus(inputPath);
            if(status.isDirectory()) {
                // check child
                FileStatus[] entries = fs.listStatus(inputPath);
                for (FileStatus entry : entries) {
                    if(entry.isFile()) {
                        if (filter.accept(entry.getPath())) {
                            inputFiles.add(entry.getPath());
                        }
                    }
                }
            } else {
                if (filter.accept(status.getPath())) {
                    inputFiles.add(status.getPath());
                }
            }
        }
        
        Path[] files = inputFiles.toArray(new Path[0]);
        return files;
    }
} public abstract class AbstractScore {
    public abstract void setParam(int size, double[] param_array);
    public abstract void setParam(int size, WeightAlgorithm algorithm, KmerStatistics[] statistics) throws IOException;
    public abstract void contributeScore(int size, double[] score_matrix, double[] score_array);
    public abstract double accumulateScore(double score1, double score2);
    public abstract double finalizeScore(double score);
} public class KmerSimilarityResultPartRecord {
    
    private static final Log LOG = LogFactory.getLog(KmerSimilarityResultPartRecord.class);
    
    private static final String HADOOP_CONFIG_KEY = "libra.core.common.kmersimilarity.kmersimilarityresultpartrecord";
    
    private int file1ID;
    private int file2ID;
    private double score;
    
    public static KmerSimilarityResultPartRecord createInstance(File file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerSimilarityResultPartRecord) serializer.fromJsonFile(file, KmerSimilarityResultPartRecord.class);
    }
    
    public static KmerSimilarityResultPartRecord createInstance(String json) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerSimilarityResultPartRecord) serializer.fromJson(json, KmerSimilarityResultPartRecord.class);
    }
    
    public static KmerSimilarityResultPartRecord createInstance(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerSimilarityResultPartRecord) serializer.fromJsonConfiguration(conf, HADOOP_CONFIG_KEY, KmerSimilarityResultPartRecord.class);
    }
    
    public static KmerSimilarityResultPartRecord createInstance(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerSimilarityResultPartRecord) serializer.fromJsonFile(fs, file, KmerSimilarityResultPartRecord.class);
    }
    
    public KmerSimilarityResultPartRecord() {
    }
    
    @JsonProperty("file1_id")
    public int getFile1ID() {
        return this.file1ID;
    }
    
    @JsonProperty("file1_id")
    public void setFile1ID(int file1ID) {
        this.file1ID = file1ID;
    }
    
    @JsonProperty("file2_id")
    public int getFile2ID() {
        return this.file2ID;
    }
    
    @JsonProperty("file2_id")
    public void setFile2ID(int file2ID) {
        this.file2ID = file2ID;
    }
    
    @JsonProperty("score")
    public double getScore() {
        return this.score;
    }
    
    @JsonProperty("score")
    public void setScore(double score) {
        this.score = score;
    }
    
    @JsonIgnore
    public void addScore(double score) {
        this.score += score;
    }
    
    @JsonIgnore
    public String toJson() {
        try {
            JsonSerializer serializer = new JsonSerializer();
            return serializer.toJson(this);
        } catch(Exception ex) {
            return null;
        }
    }
    
    @JsonIgnore
    public String toString() {
        try {
            JsonSerializer serializer = new JsonSerializer();
            return serializer.toJson(this);
        } catch(Exception ex) {
            return null;
        }
    }
    
    @JsonIgnore
    public void saveTo(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonConfiguration(conf, HADOOP_CONFIG_KEY, this);
    }
    
    @JsonIgnore
    public void saveTo(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonFile(fs, file, this);
    }
} public class KmerSimilarityResultPartRecordGroup {
    
    private static final Log LOG = LogFactory.getLog(KmerSimilarityResultPartRecordGroup.class);
    
    private static final String HADOOP_CONFIG_KEY = "libra.core.common.kmersimilarity.kmersimilarityresultpartrecordgroup";
    
    private List<KmerSimilarityResultPartRecord> scores = new ArrayList<KmerSimilarityResultPartRecord>();
    
    public static KmerSimilarityResultPartRecordGroup createInstance(File file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerSimilarityResultPartRecordGroup) serializer.fromJsonFile(file, KmerSimilarityResultPartRecordGroup.class);
    }
    
    public static KmerSimilarityResultPartRecordGroup createInstance(String json) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerSimilarityResultPartRecordGroup) serializer.fromJson(json, KmerSimilarityResultPartRecordGroup.class);
    }
    
    public static KmerSimilarityResultPartRecordGroup createInstance(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerSimilarityResultPartRecordGroup) serializer.fromJsonConfiguration(conf, HADOOP_CONFIG_KEY, KmerSimilarityResultPartRecordGroup.class);
    }
    
    public static KmerSimilarityResultPartRecordGroup createInstance(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (KmerSimilarityResultPartRecordGroup) serializer.fromJsonFile(fs, file, KmerSimilarityResultPartRecordGroup.class);
    }
    
    public KmerSimilarityResultPartRecordGroup() {
    }
    
    @JsonProperty("score")
    public Collection<KmerSimilarityResultPartRecord> getScore() {
        return this.scores;
    }
    
    @JsonProperty("score")
    public void addScore(Collection<KmerSimilarityResultPartRecord> scores) {
        this.scores.addAll(scores);
    }
    
    @JsonIgnore
    public void addScore(KmerSimilarityResultPartRecord score) {
        this.scores.add(score);
    }
    
    @JsonIgnore
    public String toJson() {
        try {
            JsonSerializer serializer = new JsonSerializer();
            return serializer.toJson(this);
        } catch(Exception ex) {
            return null;
        }
    }
    
    @JsonIgnore
    public String toString() {
        try {
            JsonSerializer serializer = new JsonSerializer();
            return serializer.toJson(this);
        } catch(Exception ex) {
            return null;
        }
    }
    
    @JsonIgnore
    public void saveTo(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonConfiguration(conf, HADOOP_CONFIG_KEY, this);
    }
    
    @JsonIgnore
    public void saveTo(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonFile(fs, file, this);
    }
} public class ScoreFactory {
    private static final Log LOG = LogFactory.getLog(ScoreFactory.class);
    
    public static AbstractScore getScore(ScoreAlgorithm algorithm) throws IOException {
        switch(algorithm) {
            case COSINESIMILARITY:
                return new CosineSimilarity();
            case BRAYCURTIS:
                return new BrayCurtis();
            case JENSENSHANNON:
                return new JensenShannon();
            default:
                LOG.info("Unknown score algorithm specified : " + algorithm.toString());
                throw new IOException("Unknown score algorithm specified : " + algorithm.toString());
        }
    }
} public class FileTable {
    
    private static final Log LOG = LogFactory.getLog(FileTable.class);
    
    private static final String HADOOP_CONFIG_KEY = "libra.preprocess.common.filetable.filetable";
    
    private String name;
    private int kmerSize;
    private List<String> samples = new ArrayList<String>();
    private Hashtable<String, Integer> sampleIDCacheTable = new Hashtable<String, Integer>();
    
    public static FileTable createInstance(File file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (FileTable) serializer.fromJsonFile(file, FileTable.class);
    }
    
    public static FileTable createInstance(String json) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (FileTable) serializer.fromJson(json, FileTable.class);
    }
    
    public static FileTable createInstance(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (FileTable) serializer.fromJsonConfiguration(conf, HADOOP_CONFIG_KEY, FileTable.class);
    }
    
    public static FileTable createInstance(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        return (FileTable) serializer.fromJsonFile(fs, file, FileTable.class);
    }
    
    public FileTable() {
    }
    
    public FileTable(String name, int kmerSize) {
        this.name = name;
        this.kmerSize = kmerSize;
    }
    
    public FileTable(SampleGroup group, int kmerSize) {
        this.name = group.getName();
        this.kmerSize = kmerSize;
        
        Collection<SampleInfo> samples = group.getSamples();
        for(SampleInfo sample : samples) {
            this.samples.add(sample.getPath());
        }
    }
    
    @JsonProperty("name")
    public String getName() {
        return this.name;
    }
    
    @JsonProperty("name")
    public void setName(String name) {
        this.name = name;
    }
    
    @JsonProperty("kmer_size")
    public int getKmerSize() {
        return this.kmerSize;
    }
    
    @JsonProperty("kmer_size")
    public void setKmerSize(int kmerSize) {
        this.kmerSize = kmerSize;
    }
    
    @JsonIgnore
    public int samples() {
        return this.samples.size();
    }
    
    @JsonProperty("samples")
    public Collection<String> getSamples() {
        return this.samples;
    }
    
    @JsonIgnore
    public int getSampleID(String sample) {
        Integer sampleID = this.sampleIDCacheTable.get(sample);
        if(sampleID == null) {
            int sid = -1; // not found
            for(int i=0;i<this.samples.size();i++) {
                if(this.samples.get(i).compareTo(sample) == 0) {
                    sid = i;
                    break;
                }
            }
            
            if(sid < 0) {
                String sample_target_name = sample;
                int idx = sample_target_name.lastIndexOf("/");
                if(idx >= 0) {
                    sample_target_name = sample_target_name.substring(idx + 1);
                }
                
                for(int j=0;j<this.samples.size();j++) {
                    String sample_name = this.samples.get(j);
                    idx = sample_name.lastIndexOf("/");
                    if(idx >= 0) {
                        sample_name = sample_name.substring(idx + 1);
                    }
                    
                    if(sample_name.compareTo(sample_target_name) == 0) {
                        sid = j;
                        break;
                    }
                }
            }
            
            sampleID = sid;
            // cache
            this.sampleIDCacheTable.put(sample, sampleID);
        }
        return sampleID;
    }
    
    @JsonProperty("samples")
    public void addSample(Collection<String> samples) {
        this.samples.addAll(samples);
    }
    
    @JsonIgnore
    public void addSample(String sample) {
        this.samples.add(sample);
    }
    
    @JsonIgnore
    public void clearSample() {
        this.samples.clear();
    }
    
    @JsonIgnore
    public void saveTo(Configuration conf) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonConfiguration(conf, HADOOP_CONFIG_KEY, this);
    }
    
    @JsonIgnore
    public void saveTo(FileSystem fs, Path file) throws IOException {
        JsonSerializer serializer = new JsonSerializer();
        serializer.toJsonFile(fs, file, this);
    }
} public class FileTableHelper {
    private final static String FILE_TABLE_PATH_EXP = ".+\\." + PreprocessorConstants.FILE_TABLE_FILENAME_EXTENSION + "$";
    private final static Pattern FILE_TABLE_PATH_PATTERN = Pattern.compile(FILE_TABLE_PATH_EXP);
    
    public static String makeFileTableFileName(String sampleFileName) {
        return sampleFileName + "." + PreprocessorConstants.FILE_TABLE_FILENAME_EXTENSION;
    }
    
    public static String makeFileTableDirPath(String rootPath) {
        return PathHelper.concatPath(rootPath, PreprocessorConstants.FILE_TABLE_DIRNAME);
    }
    
    public static boolean isFileTableFile(Path path) {
        return isFileTableFile(path.getName());
    }
    
    public static boolean isFileTableFile(String path) {
        Matcher matcher = FILE_TABLE_PATH_PATTERN.matcher(path.toLowerCase());
        if(matcher.matches()) {
            return true;
        }
        return false;
    }
    
    public static Path[] getFileTableFilePaths(Configuration conf, Path inputPath) throws IOException {
        List<Path> inputFiles = new ArrayList<Path>();
        FileTablePathFilter filter = new FileTablePathFilter();
        
        FileSystem fs = inputPath.getFileSystem(conf);
        if(fs.exists(inputPath)) {
            FileStatus status = fs.getFileStatus(inputPath);
            if(status.isDirectory()) {
                // check child
                FileStatus[] entries = fs.listStatus(inputPath);
                for (FileStatus entry : entries) {
                    if(entry.isFile()) {
                        if (filter.accept(entry.getPath())) {
                            inputFiles.add(entry.getPath());
                        }
                    }
                }
            } else {
                if (filter.accept(inputPath)) {
                    inputFiles.add(inputPath);
                }
            }
        }
        
        Path[] files = inputFiles.toArray(new Path[0]);
        return files;
    }
}
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import libra.common.helpers.FileSystemHelper;
import libra.common.report.Report;
import libra.common.helpers.MapReduceHelper;
import libra.common.kmermatch.KmerMatchFileMapping;
import libra.common.kmermatch.KmerMatchInputFormat;
import libra.common.kmermatch.KmerMatchInputFormatConfig;
import libra.distancematrix.common.DistanceMatrixConfig;
import libra.distancematrix.common.DistanceMatrixConfigException;
import libra.distancematrix.common.helpers.KmerSimilarityHelper;
import libra.distancematrix.common.kmersimilarity.AbstractScore;
import libra.distancematrix.common.kmersimilarity.KmerSimilarityResultPartRecord;
import libra.distancematrix.common.kmersimilarity.KmerSimilarityResultPartRecordGroup;
import libra.distancematrix.common.kmersimilarity.ScoreFactory;
import libra.preprocess.common.filetable.FileTable;
import libra.preprocess.common.helpers.FileTableHelper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.LineRecordReader;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
/*
 * Copyright 2016 iychoi.
 *
 * Licensed 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 libra.distancematrix.kmersimilarity_m;


/**
 *
 * @author iychoi
 */
public class KmerSimilarityMap {
    private static final Log LOG = LogFactory.getLog(KmerSimilarityMap.class);
    
    public KmerSimilarityMap() {
        
    }
    
    private void validateDistanceMatrixConfig(DistanceMatrixConfig dmConfig) throws DistanceMatrixConfigException {
        if(dmConfig.getKmerIndexPath() == null) {
            throw new DistanceMatrixConfigException("cannot find input kmer index path");
        }
        
        if(dmConfig.getFileTables() == null || dmConfig.getFileTables().size() <= 0) {
            throw new DistanceMatrixConfigException("cannot find input path");
        }
        
        if(dmConfig.getKmerStatisticsPath() == null) {
            throw new DistanceMatrixConfigException("cannot find kmer statistics path");
        }
        
        if(dmConfig.getOutputPath() == null) {
            throw new DistanceMatrixConfigException("cannot find output path");
        }
    }
    
    public int runJob(Configuration conf, DistanceMatrixConfig dmConfig) throws Exception {
        // check config
        validateDistanceMatrixConfig(dmConfig);
        
        Job job = Job.getInstance(conf, "Libra - Computing distance matrix");
        conf = job.getConfiguration();
        
        // set user configuration
        dmConfig.saveTo(conf);
        
        Report report = new Report();
        
        job.setJarByClass(KmerSimilarityMap.class);
        
        // Mapper
        job.setMapperClass(KmerSimilarityMapper.class);
        job.setInputFormatClass(KmerMatchInputFormat.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        
        // Specify key / value
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        // Inputs
        List<Path> inputFileTableFiles = new ArrayList<Path>();
        for(FileTable fileTable : dmConfig.getFileTables()) {
            String fileTableFileName = FileTableHelper.makeFileTableFileName(fileTable.getName());
            Path fileTableFilePath = new Path(dmConfig.getFileTablePath(), fileTableFileName);
            inputFileTableFiles.add(fileTableFilePath);
        }
        
        KmerMatchInputFormat.addInputPaths(job, FileSystemHelper.makeCommaSeparated(inputFileTableFiles.toArray(new Path[0])));

        LOG.info("Input file table files : " + inputFileTableFiles.size());
        for(Path inputFile : inputFileTableFiles) {
            LOG.info("> " + inputFile.toString());
        }
        
        KmerMatchFileMapping fileMapping = new KmerMatchFileMapping();
        for(FileTable fileTable : dmConfig.getFileTables()) {
            for(String sample : fileTable.getSamples()) {
                fileMapping.addSampleFile(sample);
            }
        }
        fileMapping.saveTo(conf);
        
        int kmerSize = 0;
        Iterator<FileTable> iterator = dmConfig.getFileTables().iterator();
        if(iterator.hasNext()) {
            FileTable tbl = iterator.next();
            kmerSize = tbl.getKmerSize();
        }
        
        KmerMatchInputFormatConfig matchInputFormatConfig = new KmerMatchInputFormatConfig();
        matchInputFormatConfig.setKmerSize(kmerSize);
        matchInputFormatConfig.setFileTablePath(dmConfig.getFileTablePath());
        matchInputFormatConfig.setKmerIndexPath(dmConfig.getKmerIndexPath());
        
        KmerMatchInputFormat.setInputFormatConfig(job, matchInputFormatConfig);
        
        FileOutputFormat.setOutputPath(job, new Path(dmConfig.getOutputPath()));
        job.setOutputFormatClass(TextOutputFormat.class);

        // Reducer
        job.setNumReduceTasks(0);
        
        // Execute job and return status
        boolean result = job.waitForCompletion(true);

        // commit results
        if(result) {
            commit(new Path(dmConfig.getOutputPath()), conf);
            
            // create a file mapping table file
            Path fileMappingTablePath = new Path(dmConfig.getOutputPath(), KmerSimilarityHelper.makeKmerSimilarityFileMappingTableFileName());
            FileSystem fs = fileMappingTablePath.getFileSystem(conf);
            fileMapping.saveTo(fs, fileMappingTablePath);
            
            // combine results
            combineResults(fileMapping, new Path(dmConfig.getOutputPath()), ScoreFactory.getScore(dmConfig.getScoreAlgorithm()), conf);
        }
        
        report.addJob(job);
        
        // report
        if(dmConfig.getReportPath() != null && !dmConfig.getReportPath().isEmpty()) {
            report.writeTo(dmConfig.getReportPath());
        }
        
        return result ? 0 : 1;
    }
    
    private void commit(Path outputPath, Configuration conf) throws IOException {
        FileSystem fs = outputPath.getFileSystem(conf);
        
        FileStatus status = fs.getFileStatus(outputPath);
        if (status.isDirectory()) {
            FileStatus[] entries = fs.listStatus(outputPath);
            for (FileStatus entry : entries) {
                Path entryPath = entry.getPath();
                
                // remove unnecessary outputs
                if(MapReduceHelper.isLogFiles(entryPath)) {
                    fs.delete(entryPath, true);
                } else if(MapReduceHelper.isPartialOutputFiles(entryPath)) {
                    // rename outputs
                    int mapreduceID = MapReduceHelper.getMapReduceID(entryPath);
                    String newName = KmerSimilarityHelper.makeKmerSimilarityResultPartFileName(mapreduceID);
                    Path toPath = new Path(entryPath.getParent(), newName);

                    LOG.info(String.format("rename %s ==> %s", entryPath.toString(), toPath.toString()));
                    fs.rename(entryPath, toPath);
                } else {
                    // let it be
                }
            }
        } else {
            throw new IOException("path not found : " + outputPath.toString());
        }
    }
    
    private void combineResults(KmerMatchFileMapping fileMapping, Path outputPath, AbstractScore scoreFunction, Configuration conf) throws IOException {