/**
 * Copyright 2010 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 com.hbasene.index.create.mapred;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.mapred.IdentityTableMap;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RunningJob;

import com.hbasene.index.create.IndexConfiguration;

/**
 * Example table column indexing class. Runs a mapreduce job to index specified
 * table columns.
 * <ul>
 * <li>Each row is modeled as a Lucene document: row key is indexed in its
 * untokenized form, column name-value pairs are Lucene field name-value pairs.</li>
 * <li>A file passed on command line is used to populate an
 * {@link IndexConfiguration} which is used to set various Lucene parameters,
 * specify whether to optimize an index and which columns to index and/or store,
 * in tokenized or untokenized form, etc. For an example, see the
 * <code>createIndexConfContent</code> method in TestTableIndex</li>
 * <li>The number of reduce tasks decides the number of indexes (partitions).
 * The index(es) is stored in the output path of job configuration.</li>
 * <li>The index build process is done in the reduce phase. Users can use the
 * map phase to join rows from different tables or to pre-parse/analyze column
 * content, etc.</li>
 * </ul>
 */
@Deprecated
public class BuildTableIndex {
  private static final String USAGE = "Usage: BuildTableIndex "
      + "-m <numMapTasks> -r <numReduceTasks>\n  -indexConf <iconfFile> "
      + "-indexDir <indexDir>\n  -table <tableName> -columns <columnName1> "
      + "[<columnName2> ...]";

  private static void printUsage(String message) {
    System.err.println(message);
    System.err.println(USAGE);
    System.exit(-1);
  }

  /** default constructor */
  public BuildTableIndex() {
    super();
  }

  /**
   * Block and do not return until the job is complete.
   * 
   * @param args
   * @throws IOException
   */
  public void run(String[] args) throws IOException {
    if (args.length < 6) {
      printUsage("Too few arguments");
    }

    int numMapTasks = 1;
    int numReduceTasks = 1;
    String iconfFile = null;
    String indexDir = null;
    String tableName = null;
    StringBuilder columnNames = null;

    // parse args
    for (int i = 0; i < args.length - 1; i++) {
      if ("-m".equals(args[i])) {
        numMapTasks = Integer.parseInt(args[++i]);
      } else if ("-r".equals(args[i])) {
        numReduceTasks = Integer.parseInt(args[++i]);
      } else if ("-indexConf".equals(args[i])) {
        iconfFile = args[++i];
      } else if ("-indexDir".equals(args[i])) {
        indexDir = args[++i];
      } else if ("-table".equals(args[i])) {
        tableName = args[++i];
      } else if ("-columns".equals(args[i])) {
        columnNames = new StringBuilder(args[++i]);
        while (i + 1 < args.length && !args[i + 1].startsWith("-")) {
          columnNames.append(" ");
          columnNames.append(args[++i]);
        }
      } else {
        printUsage("Unsupported option " + args[i]);
      }
    }

    if (indexDir == null || tableName == null || columnNames == null) {
      printUsage("Index directory, table name and at least one column must "
          + "be specified");
    }

    Configuration conf = HBaseConfiguration.create();
    if (iconfFile != null) {
      // set index configuration content from a file
      String content = readContent(iconfFile);
      IndexConfiguration iconf = new IndexConfiguration();
      // purely to validate, exception will be thrown if not valid
      iconf.addFromXML(content);
      conf.set("hbase.index.conf", content);
    }

    if (columnNames != null) {
      JobConf jobConf = createJob(conf, numMapTasks, numReduceTasks, indexDir,
          tableName, columnNames.toString());
      RunningJob runningJob = JobClient.runJob(jobConf);
      runningJob.waitForCompletion();
    }
  }

  /**
   * @param conf
   * @param numMapTasks
   * @param numReduceTasks
   * @param indexDir
   * @param tableName
   * @param columnNames
   * @return JobConf
   */
  public JobConf createJob(Configuration conf, int numMapTasks,
      int numReduceTasks, String indexDir, String tableName, String columnNames) {
    JobConf jobConf = new JobConf(conf, BuildTableIndex.class);
    jobConf.setJobName("build index for table " + tableName);
    jobConf.setNumMapTasks(numMapTasks);
    // number of indexes to partition into
    jobConf.setNumReduceTasks(numReduceTasks);

    // use identity map (a waste, but just as an example)
    IdentityTableMap.initJob(tableName, columnNames, IdentityTableMap.class,
        jobConf);

    // use IndexTableReduce to build a Lucene index
    jobConf.setReducerClass(IndexTableReduce.class);
    FileOutputFormat.setOutputPath(jobConf, new Path(indexDir));
    jobConf.setOutputFormat(IndexOutputFormat.class);

    return jobConf;
  }

  /*
   * Read xml file of indexing configurations. The xml format is similar to
   * hbase-default.xml and hadoop-default.xml. For an example configuration, see
   * the <code>createIndexConfContent</code> method in TestTableIndex
   * 
   * @param fileName File to read.
   * 
   * @return XML configuration read from file
   * 
   * @throws IOException
   */
  private String readContent(String fileName) throws IOException {
    File file = new File(fileName);
    int length = (int) file.length();
    if (length == 0) {
      printUsage("Index configuration file " + fileName + " does not exist");
    }

    int bytesRead = 0;
    byte[] bytes = new byte[length];
    FileInputStream fis = new FileInputStream(file);

    try {
      // read entire file into content
      while (bytesRead < length) {
        int read = fis.read(bytes, bytesRead, length - bytesRead);
        if (read > 0) {
          bytesRead += read;
        } else {
          break;
        }
      }
    } finally {
      fis.close();
    }

    return new String(bytes, 0, bytesRead, HConstants.UTF8_ENCODING);
  }

  /**
   * @param args
   * @throws IOException
   */
  public static void main(String[] args) throws IOException {
    BuildTableIndex build = new BuildTableIndex();
    build.run(args);
  }
}
@@ -5938,24 +5938,73 @@
 mat.class);%0A
+    jobConf.setJarByClass(BuildTableIndex.class);
 %0A    return 

set JarByClass