/**
 * Copyright 2008 The Apache Software Foundation
 *
 * 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 net.sf.katta.index.indexer.merge;

import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.mapred.JobConf;
import net.sf.katta.util.Logger;

public class IndexMergeJob implements Configurable {

  private Configuration _configuration;

  public void merge(Path kattaIndices) throws Exception {

    Logger.info("collect all shards in this folder: " + kattaIndices);

    Path dedupPath = new Path("/tmp/katta.index.dedup", "" + System.currentTimeMillis());

    _configuration.setBoolean("mapred.map.tasks.speculative.execution", false);
    IndexToSequenceFileJob indexToSequenceFileJob = new IndexToSequenceFileJob();
    indexToSequenceFileJob.setConf(_configuration);
    indexToSequenceFileJob.indexToSequenceFile(kattaIndices, dedupPath);

    SequenceFileToIndexJob sequenceFileToIndexJob = new SequenceFileToIndexJob();
    sequenceFileToIndexJob.setConf(_configuration);
    sequenceFileToIndexJob.sequenceFileToIndex(dedupPath, kattaIndices);

    Logger.info("delete sequence file and extracted indices: " + dedupPath);
    FileSystem fileSystem = FileSystem.get(_configuration);
    fileSystem.delete(dedupPath);

    Logger.info("all shards in this folder are merged: " + kattaIndices);
    Logger.info("move these shards into a archive folder");
  }

  public void setConf(Configuration configuration) {
    _configuration = configuration;
  }

  public Configuration getConf() {
    return _configuration;
  }

  public static void main(String[] args) throws Exception {
    //TODO delete all merged katta indices
    Path kattaIndices = new Path(args[0]);
    IndexMergeJob job = new IndexMergeJob();
    JobConf jobConf = new JobConf();
    jobConf.setJarByClass(IndexMergeJob.class);
    job.setConf(jobConf);
    job.merge(kattaIndices);
  }
}

@@ -895,16 +895,49 @@
 merge;%0A%0A
+import net.sf.katta.util.Logger;%0A
 import o
@@ -1021,42 +1021,8 @@
 on;%0A
-import org.apache.hadoop.fs.Path;%0A
 impo
@@ -1057,16 +1057,50 @@
 System;%0A
+import org.apache.hadoop.fs.Path;%0A
 import o
@@ -1135,41 +1135,8 @@
 onf;
-%0Aimport net.sf.katta.util.Logger;
 %0A%0Apu
@@ -1265,16 +1265,50 @@
 aIndices
+, Path ouputPath, Path archivePath
 ) throws
@@ -1964,28 +1964,25 @@
 upPath, 
-kattaIndices
+ouputPath
 );%0A%0A    
@@ -2170,49 +2170,152 @@
 fo(%22
-all shards in this folder are merged: %22 +
+move katta index '%22 + kattaIndices + %22' into a archive folder '%22 + archivePath + %22'%22);%0A    fileSystem.rename(kattaIndices, new Path(archivePath,
  kat
@@ -2323,19 +2323,31 @@
 aIndices
+.getName())
 );%0A
+%0A
     Logg
@@ -2360,50 +2360,25 @@
 o(%22m
-ove these shards into a archive folder
+erging done.
 %22);%0A
+%0A
   %7D%0A
@@ -2607,68 +2607,97 @@
     
-//TODO delete all merged katta indices%0A    Path kattaIndices
+Path kattaIndices = new Path(args%5B0%5D);%0A    Path out = new Path(args%5B1%5D);%0A    Path archive
  = n
@@ -2701,33 +2701,33 @@
 = new Path(args%5B
-0
+2
 %5D);%0A    IndexMer
@@ -2896,16 +2896,30 @@
 aIndices
+, out, archive
 );%0A  %7D%0A%7D

extend merge api to set a outputfolder and a archivepath to move merged indices in this archive path