/**
 * Copyright 2010 Karthik Kumar
 * 
 * Based off the original code by Lucandra project, (C): Jake Luciani
 * 
 * 
 * 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.hbasene.index;

import java.io.IOException;
import java.io.StringReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.util.Bytes;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.TokenStream;
import org.apache.lucene.analysis.tokenattributes.PositionIncrementAttribute;
import org.apache.lucene.analysis.tokenattributes.TermAttribute;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.document.Fieldable;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.store.LockObtainFailedException;

/**
 * Index Writer in the noSQL world.
 * 
 */
public class HBaseIndexWriter { // TODO: extends IndexWriter {

  /**
   * Abstraction of the index store
   */
  private final AbstractIndexStore indexStore;

  /**
   * Field representing the primary key in the given search index.
   */
  private final String primaryKeyField;

  /**
   * The default capacity of the number of terms in the given document.
   */
  private static final int DEFAULT_TERM_CAPACITY = 50;

  /**
   * List of empty term vectors
   */
  private static final List<Integer> EMPTY_TERM_POSITIONS = Arrays
      .asList(new Integer[] { 0 });


  /**
   * 
   * @param indexTransactionLog
   * @param primaryKeyField
   *          The primary key field of the lucene Document schema to be used.
   * @throws CorruptIndexException
   * @throws LockObtainFailedException
   * @throws IOException
   */
  public HBaseIndexWriter(
      final AbstractIndexStore indexTransactionLog,
      final String primaryKeyField) throws 
      IOException {
    // super(d, a, create, deletionPolicy, mfl);
    // TODO: bring super ctor in when we inherit from IndexWriter.

    this.indexStore = indexTransactionLog;
    this.primaryKeyField = primaryKeyField;

    // Reset the transaction Log.
    //this.indexStore.init();
  }

  public void addDocument(Document doc, Analyzer analyzer)
      throws CorruptIndexException, IOException {
    String docId = doc.get(this.primaryKeyField);
    if (docId == null) {
      throw new IllegalArgumentException("Primary Key " + this.primaryKeyField
          + " not present in the document to be added ");
      // TODO: Special type of exception needed ?

    }

    long assignedDocId = indexStore.docId(Bytes.toBytes(docId));
    List<String> allIndexedTerms = new ArrayList<String>(DEFAULT_TERM_CAPACITY);

    int position = 0;

    for (Fieldable field : doc.getFields()) {

      // Indexed field
      if (field.isIndexed() && field.isTokenized()) {

        TokenStream tokens = field.tokenStreamValue();

        if (tokens == null) {
          tokens = analyzer.tokenStream(field.name(), new StringReader(field
              .stringValue()));
        }
        tokens.addAttribute(TermAttribute.class);
        tokens.addAttribute(PositionIncrementAttribute.class);

        // collect term frequencies per doc
        Map<String, List<Integer>> termPositions = new HashMap<String, List<Integer>>();
        if (position > 0) {
          position += analyzer.getPositionIncrementGap(field.name());
        }

        // Build the termPositions vector for all terms
        while (tokens.incrementToken()) {
          String term = createColumnName(field.name(), tokens.getAttribute(
              TermAttribute.class).term());
          allIndexedTerms.add(term);

          List<Integer> pvec = termPositions.get(term);

          if (pvec == null) {
            pvec = new ArrayList<Integer>();
            termPositions.put(term, pvec);
          }

          position += (tokens.getAttribute(PositionIncrementAttribute.class)
              .getPositionIncrement() - 1);
          pvec.add(++position);

        }

        for (Map.Entry<String, List<Integer>> term : termPositions.entrySet()) {
          String key = term.getKey();
          indexStore.addTermPositions(key, assignedDocId,
              term.getValue());
        }
      }

      // Untokenized fields go in without a termPosition
      if (field.isIndexed() && !field.isTokenized()) {
        String term = this.createColumnName(field.name(), field.stringValue());
        allIndexedTerms.add(term);

        String key = term;

        indexStore.addTermPositions(key, assignedDocId,
            EMPTY_TERM_POSITIONS);
      }

      // Stores each field as a column under this doc key
      if (field.isStored()) {

        byte[] _value = field.isBinary() ? field.getBinaryValue() : field
            .stringValue().getBytes();

        // last byte flags if binary or not
        byte[] value = new byte[_value.length + 1];
        System.arraycopy(_value, 0, value, 0, _value.length);

        value[value.length - 1] = (byte) (field.isBinary() ? 'B' : 'T');


        this.indexStore.storeField(assignedDocId, field.name(),
            value);
      }
    }

    //this.indexStore.commit();
  }

  public void commit() throws IOException { 
    this.indexStore.commit();
  }
  
  public void close() throws IOException {
    this.indexStore.close();
  }
  
  String createColumnName(final String fieldName, final String term) {
    return fieldName + "/" + term;
  }
}

@@ -1124,16 +1124,101 @@
 l.Map;%0A%0A
+import org.apache.commons.logging.Log;%0Aimport org.apache.commons.logging.LogFactory;%0A
 import o
@@ -1252,16 +1252,16 @@
 .Bytes;%0A
-
 import o
@@ -1841,16 +1841,93 @@
 iter %7B%0A%0A
+  private static final Log LOG = LogFactory.getLog(HBaseIndexWriter.class);%0A%0A
   /**%0A  
@@ -3686,25 +3686,31 @@
 Fields()) %7B%0A
+      
 %0A
-
       // Ind
@@ -3769,24 +3769,82 @@
 kenized()) %7B
+%0A        LOG.info(%22Processing - %5BI, T%5D %22 + field.name() );
 %0A%0A        To
@@ -4994,33 +4994,65 @@
 on);%0A%0A        %7D%0A
+        tokens.close();%0A        
 %0A
-
         for (Map
@@ -5320,16 +5320,16 @@
 osition%0A
-
       if
@@ -5367,32 +5367,96 @@
 sTokenized()) %7B%0A
+        LOG.info(%22Processing - %5BI,  not T%5D %22 + field.name() );%0A%0A
         String t

add logging statements