package org.hbasene.index;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

import jsr166y.RecursiveAction;

import org.apache.lucene.util.OpenBitSet;

public class TermVectorAppendTask extends RecursiveAction {

  static final int THRESHOLD = 500;

  private final int termVectorArrayThreshold;

  private final long docBase;

  private final String[] inputTerms;

  private final long docId;

  private final ConcurrentHashMap<String, Object> termVectorMap;

  private final int low;

  private final int high;

  public TermVectorAppendTask(final String[] inputTerms, int low, int high,
      final long docId, final ConcurrentHashMap<String, Object> termVectorMap,
      final int termVectorArrayThreshold, final long docBase) {
    this.docId = docId;
    this.inputTerms = inputTerms;
    this.termVectorMap = termVectorMap;
    this.low = low;
    this.high = high;
    this.termVectorArrayThreshold = termVectorArrayThreshold;
    this.docBase = docBase;
  }

  @Override
  protected void compute() {
    if ((high - low) < THRESHOLD) {
      processAssign();
    } else {
      int mid = (low + high) >>> 1;

      invokeAll(new TermVectorAppendTask(this.inputTerms, low, mid, docId,
          this.termVectorMap, this.termVectorArrayThreshold, this.docBase),
          new TermVectorAppendTask(this.inputTerms, mid, high, docId,
              this.termVectorMap, this.termVectorArrayThreshold, this.docBase));

    }
  }

  void processAssign() {
    for (int i = low; i < high; ++i) {
      final String fieldTerm = this.inputTerms[i];
      Object docs = this.termVectorMap.get(fieldTerm);
      if (docs == null) {
        docs = new ArrayList<Integer>();
        this.termVectorMap.put(fieldTerm, docs);
      }
      long relativeId = docId - docBase;
      if (docs instanceof List) {
        List<Integer> listImpl = (List<Integer>) docs;
        listImpl.add((int) relativeId);
        if (listImpl.size() > this.termVectorArrayThreshold) {
          OpenBitSet bitset = new OpenBitSet();
          for (Integer value : listImpl) {
            bitset.set(value);
          }
          listImpl.clear();
          this.termVectorMap.put(fieldTerm, bitset);
        }
      } else if (docs instanceof OpenBitSet) {
        ((OpenBitSet) docs).set(relativeId);
      }
    }
  }
}

@@ -345,39 +345,8 @@
 d;%0A%0A
-  private final long docBase;%0A%0A
   pr
@@ -400,19 +400,24 @@
 al long 
-doc
+relative
 Id;%0A%0A  p
@@ -617,26 +617,8 @@
     
- final long docId,
  fin
@@ -726,42 +726,21 @@
 ong 
-docBase) %7B%0A    this.docId = docId;
+relativeId) %7B
 %0A   
@@ -927,25 +927,31 @@
 his.
-docBase = docBase
+relativeId = relativeId
 ;%0A  
@@ -1040,24 +1040,93 @@
     
-processAssign();
+for (int i = low; i %3C high; ++i) %7B%0A        processAssign(this.inputTerms%5Bi%5D);%0A      %7D
 %0A   
@@ -1240,22 +1240,16 @@
 w, mid, 
-docId,
 %0A       
@@ -1307,23 +1307,26 @@
 d, this.
-docBase
+relativeId
 ),%0A     
@@ -1387,14 +1387,8 @@
 gh, 
-docId,
 %0A   
@@ -1454,23 +1454,26 @@
 d, this.
-docBase
+relativeId
 ));%0A%0A   
@@ -1505,104 +1505,34 @@
 ign(
-) %7B%0A    for (int i = low; i %3C high; ++i) %7B%0A      final String fieldTerm = this.inputTerms%5Bi%5D;%0A  
+final String fieldTerm) %7B%0A
     
@@ -1580,26 +1580,24 @@
 dTerm);%0A    
- 
- 
 if (docs == 
@@ -1600,26 +1600,24 @@
  == null) %7B%0A
-  
       docs =
@@ -1645,26 +1645,24 @@
 r%3E();%0A      
- 
- 
 this.termVec
@@ -1698,55 +1698,10 @@
     
-  %7D%0A      long relativeId = docId - docBase;%0A  
+%7D%0A
     
@@ -1724,26 +1724,24 @@
 eof List) %7B%0A
-  
       List%3CI
@@ -1781,18 +1781,16 @@
 ) docs;%0A
-  
       li
@@ -1825,18 +1825,16 @@
 ;%0A      
-  
 if (list
@@ -1876,26 +1876,24 @@
 hreshold) %7B%0A
-  
         Open
@@ -1930,26 +1930,24 @@
 ();%0A        
-  
 for (Integer
@@ -1977,18 +1977,16 @@
         
-  
 bitset.s
@@ -2000,30 +2000,26 @@
 e);%0A        
-  %7D%0A  
+%7D%0A
         list
@@ -2032,34 +2032,32 @@
 lear();%0A        
-  
 this.termVectorM
@@ -2083,18 +2083,16 @@
 itset);%0A
-  
       %7D%0A
@@ -2095,18 +2095,16 @@
   %7D%0A    
-  
 %7D else i
@@ -2132,26 +2132,24 @@
 enBitSet) %7B%0A
-  
       ((Open
@@ -2179,24 +2179,16 @@
 iveId);%0A
-      %7D%0A
     %7D%0A  
@@ -2189,10 +2189,11 @@
   %7D%0A  %7D%0A
+%0A
 %7D%0A

modify task for reuse outside the task