/*
 * 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.apache.accumulo.examples.wikisearch.iterator;

import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;

import org.apache.accumulo.core.data.ArrayByteSequence;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.PartialKey;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.IteratorEnvironment;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.util.TextUtil;
import org.apache.commons.codec.binary.Base64;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Logger;

public class AndIterator implements SortedKeyValueIterator<Key,Value> {
  
  protected static final Logger log = Logger.getLogger(AndIterator.class);
  private TermSource[] sources;
  private int sourcesCount = 0;
  protected Text nullText = new Text();
  protected final byte[] emptyByteArray = new byte[0];
  private Key topKey = null;
  protected Value value = new Value(emptyByteArray);
  private Range overallRange;
  private Text currentRow = null;
  private Text currentTerm = new Text(emptyByteArray);
  private Text currentDocID = new Text(emptyByteArray);
  private static boolean SEEK_INCLUSIVE = true;
  private Text parentEndRow;
  
  /**
   * Used in representing a Term that is intersected on.
   */
  protected static class TermSource {
    
    public SortedKeyValueIterator<Key,Value> iter;
    public Text dataLocation;
    public Text term;
    public boolean notFlag;
    private Collection<ByteSequence> seekColumnFamilies;
    
    private TermSource(TermSource other) {
      this(other.iter, other.dataLocation, other.term, other.notFlag);
    }
    
    public TermSource(SortedKeyValueIterator<Key,Value> iter, Text dataLocation, Text term) {
      this(iter, dataLocation, term, false);
    }
    
    public TermSource(SortedKeyValueIterator<Key,Value> iter, Text dataLocation, Text term, boolean notFlag) {
      this.iter = iter;
      this.dataLocation = dataLocation;
      ByteSequence bs = new ArrayByteSequence(dataLocation.getBytes(), 0, dataLocation.getLength());
      this.seekColumnFamilies = Collections.singletonList(bs);
      this.term = term;
      this.notFlag = notFlag;
    }
    
    public String getTermString() {
      return (this.term == null) ? new String("Iterator") : this.term.toString();
    }
  }
  
  /*
   * | Row     | Column Family  | Column Qualifier          | Value 
   * | {RowID} | {dataLocation} | {term}\0{dataType}\0{UID} | Empty
   */
  protected Text getPartition(Key key) {
    return key.getRow();
  }
  
  /**
   * Returns the given key's dataLocation
   * 
   * @param key
   * @return The given key's dataLocation
   */
  protected Text getDataLocation(Key key) {
    return key.getColumnFamily();
  }
  
  /**
   * Returns the given key's term
   * 
   * @param key
   * @return The given key's term
   */
  protected Text getTerm(Key key) {
    int idx = 0;
    String sKey = key.getColumnQualifier().toString();
    
    idx = sKey.indexOf("\0");
    return new Text(sKey.substring(0, idx));
  }
  
  /**
   * Returns the given key's DocID
   * 
   * @param key
   * @return The given key's DocID
   */
  protected Text getDocID(Key key) {
    int idx = 0;
    String sKey = key.getColumnQualifier().toString();
    
    idx = sKey.indexOf("\0");
    return new Text(sKey.substring(idx + 1));
  }
  
  /**
   * Returns the given key's UID
   * 
   * @param key
   * @return The given key's UID
   */
  protected String getUID(Key key) {
    int idx = 0;
    String sKey = key.getColumnQualifier().toString();
    
    idx = sKey.indexOf("\0");
    return sKey.substring(idx + 1);
  }
  
  /**
   * Build a key from the given row and dataLocation
   * 
   * @param row
   *          The desired row
   * @param dataLocation
   *          The desired dataLocation
   * @return A Key object built from the given row and dataLocation.
   */
  protected Key buildKey(Text row, Text dataLocation) {
    return new Key(row, (dataLocation == null) ? nullText : dataLocation);
  }
  
  /**
   * Build a key from the given row, dataLocation, and term
   * 
   * @param row
   *          The desired row
   * @param dataLocation
   *          The desired dataLocation
   * @param term
   *          The desired term
   * @return A Key object built from the given row, dataLocation, and term.
   */
  protected Key buildKey(Text row, Text dataLocation, Text term) {
    return new Key(row, (dataLocation == null) ? nullText : dataLocation, (term == null) ? nullText : term);
  }
  
  /**
   * Return the key that directly follows the given key
   * 
   * @param key
   *          The key who will be directly before the returned key
   * @return The key directly following the given key.
   */
  protected Key buildFollowingPartitionKey(Key key) {
    return key.followingKey(PartialKey.ROW);
  }
  
  /**
   * Empty default constructor
   */
  public AndIterator() {}
  
  public SortedKeyValueIterator<Key,Value> deepCopy(IteratorEnvironment env) {
    return new AndIterator(this, env);
  }
  
  public AndIterator(AndIterator other, IteratorEnvironment env) {
    if (other.sources != null) {
      sourcesCount = other.sourcesCount;
      sources = new TermSource[sourcesCount];
      for (int i = 0; i < sourcesCount; i++) {
        sources[i] = new TermSource(other.sources[i].iter.deepCopy(env), other.sources[i].dataLocation, other.sources[i].term);
      }
    }
  }
  
  public Key getTopKey() {
    return topKey;
  }
  
  public Value getTopValue() {
    return value;
  }
  
  public boolean hasTop() {
    return currentRow != null;
  }
  
  /**
   * Find the next key in the current TermSource that is at or beyond the cursor (currentRow, currentTerm, currentDocID).
   * 
   * @param sourceID
   *          The index of the current source in <code>sources</code>
   * @return True if the source advanced beyond the cursor
   * @throws IOException
   */
  private boolean seekOneSource(TermSource ts) throws IOException {
    /*
     * Within this loop progress must be made in one of the following forms: - currentRow, currentTerm, or curretDocID must be increased - the given source must
     * advance its iterator This loop will end when any of the following criteria are met - the iterator for the given source is pointing to the key
     * (currentRow, columnFamilies[sourceID], currentTerm, currentDocID) - the given source is out of data and currentRow is set to null - the given source has
     * advanced beyond the endRow and currentRow is set to null
     */
    
    // precondition: currentRow is not null
    boolean advancedCursor = false;
    
    while (true) {
      if (ts.iter.hasTop() == false) {
        if (log.isDebugEnabled()) {
          log.debug("The current iterator no longer has a top");
        }
        
        // If we got to the end of an iterator, found a Match if it's a NOT
        if (ts.notFlag) {
          break;
        }
        
        currentRow = null;
        // setting currentRow to null counts as advancing the cursor
        return true;
      }
      
      // check if we're past the end key
      int endCompare = -1;
      
      if (log.isDebugEnabled()) {
        log.debug("Current topKey = " + ts.iter.getTopKey());
      }
      
      // we should compare the row to the end of the range
      if (overallRange.getEndKey() != null) {
        if (log.isDebugEnabled()) {
          log.debug("II.seekOneSource overallRange.getEndKey() != null");
        }
        
        endCompare = overallRange.getEndKey().getRow().compareTo(ts.iter.getTopKey().getRow());
        
        if ((!overallRange.isEndKeyInclusive() && endCompare <= 0) || endCompare < 0) {
          if (log.isDebugEnabled()) {
            log.debug("II.seekOneSource at the end of the tablet server");
          }
          
          currentRow = null;
          
          // setting currentRow to null counts as advancing the cursor
          return true;
        }
      } else {
        if (log.isDebugEnabled()) {
          log.debug("II.seekOneSource overallRange.getEndKey() == null");
        }
      }
      
      // Compare the Row IDs
      int partitionCompare = currentRow.compareTo(getPartition(ts.iter.getTopKey()));
      if (log.isDebugEnabled()) {
        log.debug("Current partition: " + currentRow);
      }
      
      // check if this source is already at or beyond currentRow
      // if not, then seek to at least the current row
      if (partitionCompare > 0) {
        if (log.isDebugEnabled()) {
          log.debug("Need to seek to the current row");
          
          // seek to at least the currentRow
          log.debug("ts.dataLocation = " + ts.dataLocation.getBytes());
          log.debug("Term = " + new Text(ts.term + "\0" + currentDocID).getBytes());
        }
        
        Key seekKey = buildKey(currentRow, ts.dataLocation, nullText);// new Text(ts.term + "\0" + currentDocID));
        
        if (log.isDebugEnabled()) {
          log.debug("Seeking to: " + seekKey);
        }
        ts.iter.seek(new Range(seekKey, true, null, false), ts.seekColumnFamilies, SEEK_INCLUSIVE);
        continue;
      }
      
      // check if this source has gone beyond currentRow
      // if so, advance currentRow
      if (partitionCompare < 0) {
        if (log.isDebugEnabled()) {
          log.debug("Went too far beyond the currentRow");
        }
        
        if (ts.notFlag) {
          break;
        }
        
        currentRow.set(getPartition(ts.iter.getTopKey()));
        currentDocID.set(emptyByteArray);
        
        advancedCursor = true;
        continue;
      }
      
      // we have verified that the current source is positioned in currentRow
      // now we must make sure we're in the right columnFamily in the current row
      if (ts.dataLocation != null) {
        int dataLocationCompare = ts.dataLocation.compareTo(getDataLocation(ts.iter.getTopKey()));
        
        if (log.isDebugEnabled()) {
          log.debug("Comparing dataLocations");
          log.debug("dataLocation = " + ts.dataLocation);
          log.debug("newDataLocation = " + getDataLocation(ts.iter.getTopKey()));
        }
        
        // check if this source is already on the right columnFamily
        // if not, then seek forwards to the right columnFamily
        if (dataLocationCompare > 0) {
          if (log.isDebugEnabled()) {
            log.debug("Need to seek to the right dataLocation");
          }
          
          Key seekKey = buildKey(currentRow, ts.dataLocation, nullText);// , new Text(ts.term + "\0" + currentDocID));
          
          if (log.isDebugEnabled()) {
            log.debug("Seeking to: " + seekKey);
          }
          
          ts.iter.seek(new Range(seekKey, true, null, false), ts.seekColumnFamilies, SEEK_INCLUSIVE);
          if (!ts.iter.hasTop()) {
            currentRow = null;
            return true;
          }
          
          continue;
        }
        // check if this source is beyond the right columnFamily
        // if so, then seek to the next row
        if (dataLocationCompare < 0) {
          if (log.isDebugEnabled()) {
            log.debug("Went too far beyond the dataLocation");
          }
          
          if (endCompare == 0) {
            // we're done
            currentRow = null;
            
            // setting currentRow to null counts as advancing the cursor
            return true;
          }
          
          // Seeking beyond the current dataLocation gives a valid negated result
          if (ts.notFlag) {
            break;
          }
          
          Key seekKey = buildFollowingPartitionKey(ts.iter.getTopKey());
          
          if (log.isDebugEnabled()) {
            log.debug("Seeking to: " + seekKey);
          }
          ts.iter.seek(new Range(seekKey, true, null, false), ts.seekColumnFamilies, SEEK_INCLUSIVE);
          if (!ts.iter.hasTop()) {
            currentRow = null;
            return true;
          }
          continue;
        }
      }
      
      // Compare the Terms
      int termCompare = ts.term.compareTo(getTerm(ts.iter.getTopKey()));
      if (log.isDebugEnabled()) {
        log.debug("term = " + ts.term);
        log.debug("newTerm = " + getTerm(ts.iter.getTopKey()));
      }
      
      // We need to seek down farther into the data
      if (termCompare > 0) {
        if (log.isDebugEnabled()) {
          log.debug("Need to seek to the right term");
        }
        Key seekKey = buildKey(currentRow, ts.dataLocation, new Text(ts.term + "\0"));// new Text(ts.term + "\0" + currentDocID));
        
        if (log.isDebugEnabled()) {
          log.debug("Seeking to: " + seekKey);
        }
        
        ts.iter.seek(new Range(seekKey, true, null, false), ts.seekColumnFamilies, SEEK_INCLUSIVE);
        if (!ts.iter.hasTop()) {
          currentRow = null;
          return true;
        }
        
        // currentTerm = getTerm(ts.iter.getTopKey());
        
        if (log.isDebugEnabled()) {
          log.debug("topKey after seeking to correct term: " + ts.iter.getTopKey());
        }
        
        continue;
      }
      
      // We've jumped out of the current term, set the new term as currentTerm and start looking again
      if (termCompare < 0) {
        if (log.isDebugEnabled()) {
          log.debug("TERM: Need to jump to the next row");
        }
        
        if (endCompare == 0) {
          currentRow = null;
          
          return true;
        }
        
        if (ts.notFlag) {
          break;
        }
        
        Key seekKey = buildFollowingPartitionKey(ts.iter.getTopKey());
        if (log.isDebugEnabled()) {
          log.debug("Using this key to find the next key: " + ts.iter.getTopKey());
          log.debug("Seeking to: " + seekKey);
        }
        
        ts.iter.seek(new Range(seekKey, true, null, false), ts.seekColumnFamilies, SEEK_INCLUSIVE);
        
        if (!ts.iter.hasTop()) {
          currentRow = null;
          return true;
        }
        
        currentTerm = getTerm(ts.iter.getTopKey());
        
        continue;
      }
      
      // Compare the DocIDs
      Text docid = getDocID(ts.iter.getTopKey());
      int docidCompare = currentDocID.compareTo(docid);
      
      if (log.isDebugEnabled()) {
        log.debug("Comparing DocIDs");
        log.debug("currentDocID = " + currentDocID);
        log.debug("docid = " + docid);
      }
      
      // The source isn't at the right DOC
      if (docidCompare > 0) {
        if (log.isDebugEnabled()) {
          log.debug("Need to seek to the correct docid");
        }
        
        // seek forwards
        Key seekKey = buildKey(currentRow, ts.dataLocation, new Text(ts.term + "\0" + currentDocID));
        
        if (log.isDebugEnabled()) {
          log.debug("Seeking to: " + seekKey);
        }
        
        ts.iter.seek(new Range(seekKey, true, null, false), ts.seekColumnFamilies, SEEK_INCLUSIVE);
        
        continue;
      }
      
      // if this source has advanced beyond the current column qualifier then advance currentCQ and return true
      if (docidCompare < 0) {
        if (ts.notFlag) {
          break;
        }
        
        if (log.isDebugEnabled()) {
          log.debug("We went too far, update the currentDocID to be the location of where were seek'ed to");
        }
        
        currentDocID.set(docid);
        advancedCursor = true;
        break;
      }
      
      // Set the term as currentTerm (in case we found this record on the first try)
      currentTerm = getTerm(ts.iter.getTopKey());
      
      if (log.isDebugEnabled()) {
        log.debug("currentTerm = " + currentTerm);
      }
      
      // If we're negated, next() the first TermSource since we guaranteed it was not a NOT term
      if (ts.notFlag) {
        sources[0].iter.next();
        advancedCursor = true;
      }
      
      // If we got here, we have a match
      break;
    }
    
    return advancedCursor;
  }
  
  public void next() throws IOException {
    if (log.isDebugEnabled()) {
      log.debug("In ModifiedIntersectingIterator.next()");
    }
    
    if (currentRow == null) {
      return;
    }
    
    // precondition: the current row is set up and the sources all have the same column qualifier
    // while we don't have a match, seek in the source with the smallest column qualifier
    sources[0].iter.next();
    
    advanceToIntersection();
    
    if (hasTop()) {
      if (overallRange != null && !overallRange.contains(topKey)) {
        topKey = null;
      }
    }
  }
  
  protected void advanceToIntersection() throws IOException {
    if (log.isDebugEnabled()) {
      log.debug("In AndIterator.advanceToIntersection()");
    }
    
    boolean cursorChanged = true;
    while (cursorChanged) {
      // seek all of the sources to at least the highest seen column qualifier in the current row
      cursorChanged = false;
      for (TermSource ts : sources) {
        if (currentRow == null) {
          topKey = null;
          return;
        }
        if (seekOneSource(ts)) {
          cursorChanged = true;
          break;
        }
      }
    }
    
    topKey = buildKey(currentRow, currentTerm, currentDocID);
    
    if (log.isDebugEnabled()) {
      log.debug("ModifiedIntersectingIterator: Got a match: " + topKey);
    }
  }
  
  public static String stringTopKey(SortedKeyValueIterator<Key,Value> iter) {
    if (iter.hasTop()) {
      return iter.getTopKey().toString();
    }
    return "";
  }
  
  public static final String columnFamiliesOptionName = "columnFamilies";
  public static final String termValuesOptionName = "termValues";
  public static final String notFlagsOptionName = "notFlags";
  
  /**
   * Encode a <code>Text</code> array of all the columns to intersect on
   * 
   * @param columns
   *          The columns to be encoded
   * @return A Base64 encoded string (using a \n delimiter) of all columns to intersect on.
   */
  public static String encodeColumns(Text[] columns) {
    StringBuilder sb = new StringBuilder();
    for (int i = 0; i < columns.length; i++) {
      sb.append(new String(Base64.encodeBase64(TextUtil.getBytes(columns[i]))));
      sb.append('\n');
    }
    return sb.toString();
  }
  
  /**
   * Encode a <code>Text</code> array of all of the terms to intersect on. The terms should match the columns in a one-to-one manner
   * 
   * @param terms
   *          The terms to be encoded
   * @return A Base64 encoded string (using a \n delimiter) of all terms to intersect on.
   */
  public static String encodeTermValues(Text[] terms) {
    StringBuilder sb = new StringBuilder();
    for (int i = 0; i < terms.length; i++) {
      sb.append(new String(Base64.encodeBase64(TextUtil.getBytes(terms[i]))));
      sb.append('\n');
    }
    
    return sb.toString();
  }
  
  /**
   * Encode an array of <code>booleans</code> denoted which columns are NOT'ed
   * 
   * @param flags
   *          The array of NOTs
   * @return A base64 encoded string of which columns are NOT'ed
   */
  public static String encodeBooleans(boolean[] flags) {
    byte[] bytes = new byte[flags.length];
    for (int i = 0; i < flags.length; i++) {
      if (flags[i]) {
        bytes[i] = 1;
      } else {
        bytes[i] = 0;
      }
    }
    return new String(Base64.encodeBase64(bytes));
  }
  
  /**
   * Decode the encoded columns into a <code>Text</code> array
   * 
   * @param columns
   *          The Base64 encoded String of the columns
   * @return A Text array of the decoded columns
   */
  public static Text[] decodeColumns(String columns) {
    String[] columnStrings = columns.split("\n");
    Text[] columnTexts = new Text[columnStrings.length];
    for (int i = 0; i < columnStrings.length; i++) {
      columnTexts[i] = new Text(Base64.decodeBase64(columnStrings[i].getBytes()));
    }
    
    return columnTexts;
  }
  
  /**
   * Decode the encoded terms into a <code>Text</code> array
   * 
   * @param terms
   *          The Base64 encoded String of the terms
   * @return A Text array of decoded terms.
   */
  public static Text[] decodeTermValues(String terms) {
    String[] termStrings = terms.split("\n");
    Text[] termTexts = new Text[termStrings.length];
    for (int i = 0; i < termStrings.length; i++) {
      termTexts[i] = new Text(Base64.decodeBase64(termStrings[i].getBytes()));
    }
    
    return termTexts;
  }
  
  /**
   * Decode the encoded NOT flags into a <code>boolean</code> array
   * 
   * @param flags
   * @return A boolean array of decoded NOT flags
   */
  public static boolean[] decodeBooleans(String flags) {
    // return null of there were no flags
    if (flags == null) {
      return null;
    }
    
    byte[] bytes = Base64.decodeBase64(flags.getBytes());
    boolean[] bFlags = new boolean[bytes.length];
    for (int i = 0; i < bytes.length; i++) {
      if (bytes[i] == 1) {
        bFlags[i] = true;
      } else {
        bFlags[i] = false;
      }
    }
    
    return bFlags;
  }
  
  public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options, IteratorEnvironment env) throws IOException {
    if (log.isDebugEnabled()) {
      log.debug("In AndIterator.init()");
    }
    
    Text[] dataLocations = decodeColumns(options.get(columnFamiliesOptionName));
    Text[] terms = decodeTermValues(options.get(termValuesOptionName));
    boolean[] notFlags = decodeBooleans(options.get(notFlagsOptionName));
    
    if (terms.length < 2) {
      throw new IllegalArgumentException("AndIterator requires two or more columns families");
    }
    
    // Scan the not flags.
    // There must be at least one term that isn't negated
    // And we are going to re-order such that the first term is not a ! term
    if (notFlags == null) {
      notFlags = new boolean[terms.length];
      for (int i = 0; i < terms.length; i++) {
        notFlags[i] = false;
      }
    }
    
    // Make sure that the first dataLocation/Term is not a NOT by swapping it with a later dataLocation/Term
    if (notFlags[0]) {
      for (int i = 1; i < notFlags.length; i++) {
        if (notFlags[i] == false) {
          // Swap the terms
          Text swap = new Text(terms[0]);
          terms[0].set(terms[i]);
          terms[i].set(swap);
          
          // Swap the dataLocations
          swap.set(dataLocations[0]);
          dataLocations[0].set(dataLocations[i]);
          dataLocations[i].set(swap);
          
          // Flip the notFlags
          notFlags[0] = false;
          notFlags[i] = true;
          break;
        }
      }
      
      if (notFlags[0]) {
        throw new IllegalArgumentException("AndIterator requires at least one column family without not");
      }
    }
    
    // Build up the array of sources that are to be intersected
    sources = new TermSource[dataLocations.length];
    for (int i = 0; i < dataLocations.length; i++) {
      sources[i] = new TermSource(source.deepCopy(env), dataLocations[i], terms[i], notFlags[i]);
    }
    
    sourcesCount = dataLocations.length;
  }
  
  public void seek(Range range, Collection<ByteSequence> seekColumnFamilies, boolean inclusive) throws IOException {
    if (log.isDebugEnabled()) {
      log.debug("In AndIterator.seek()");
      log.debug("AndIterator.seek Given range => " + range);
    }
    currentRow = new Text();
    currentDocID.set(emptyByteArray);
    doSeek(range);
  }
  
  private void doSeek(Range range) throws IOException {

    overallRange = new Range(range);

    if (range.getEndKey() != null && range.getEndKey().getRow() != null) {
      this.parentEndRow = range.getEndKey().getRow();
    }
    
    // seek each of the sources to the right column family within the row given by key
    for (int i = 0; i < sourcesCount; i++) {
      Key sourceKey;
      Text dataLocation = (sources[i].dataLocation == null) ? nullText : sources[i].dataLocation;
      if (range.getStartKey() != null) {
        // Build a key with the DocID if one is given
        if (range.getStartKey().getColumnFamily() != null) {
          sourceKey = buildKey(getPartition(range.getStartKey()), dataLocation,
              (sources[i].term == null) ? nullText : new Text(sources[i].term + "\0" + range.getStartKey().getColumnFamily()));
        } // Build a key with just the term.
        else {
          sourceKey = buildKey(getPartition(range.getStartKey()), dataLocation,
              (sources[i].term == null) ? nullText : sources[i].term);
        }
        if (!range.isStartKeyInclusive())
          sourceKey = sourceKey.followingKey(PartialKey.ROW_COLFAM_COLQUAL);
        sources[i].iter.seek(new Range(sourceKey, true, null, false), sources[i].seekColumnFamilies, SEEK_INCLUSIVE);
      } else {
        sources[i].iter.seek(range, sources[i].seekColumnFamilies, SEEK_INCLUSIVE);
      }
    }
    
    advanceToIntersection();
    
    if (hasTop()) {
      if (overallRange != null && !overallRange.contains(topKey)) {
        topKey = null;
        if (log.isDebugEnabled()) {
          log.debug("doSeek, topKey is outside of overall range: " + overallRange);
        }
      }
    }
  }
  
  public void addSource(SortedKeyValueIterator<Key,Value> source, IteratorEnvironment env, Text term, boolean notFlag) {
    addSource(source, env, null, term, notFlag);
  }
  
  public void addSource(SortedKeyValueIterator<Key,Value> source, IteratorEnvironment env, Text dataLocation, Text term, boolean notFlag) {
    // Check if we have space for the added Source
    if (sources == null) {
      sources = new TermSource[1];
    } else {
      // allocate space for node, and copy current tree.
      // TODO: Should we change this to an ArrayList so that we can just add() ?
      TermSource[] localSources = new TermSource[sources.length + 1];
      int currSource = 0;
      for (TermSource myTerm : sources) {
        // TODO: Do I need to call new here? or can I just re-use the term?
        localSources[currSource] = new TermSource(myTerm);
        currSource++;
      }
      sources = localSources;
    }
    
    sources[sourcesCount] = new TermSource(source.deepCopy(env), dataLocation, term, notFlag);
    sourcesCount++;
  }
  
  public boolean jump(Key jumpKey) throws IOException {
    if (log.isDebugEnabled()) {
      log.debug("jump: " + jumpKey);
    }
    
    // is the jumpKey outside my overall range?
    if (parentEndRow != null && parentEndRow.compareTo(jumpKey.getRow()) < 0) {
      // can't go there.
      if (log.isDebugEnabled()) {
        log.debug("jumpRow: " + jumpKey.getRow() + " is greater than my parentEndRow: " + parentEndRow);
      }
      return false;
    }
    
    if (!hasTop()) {
      // TODO: will need to add current/last row if you want to measure if
      // we don't have topkey because we hit end of tablet.
      
      if (log.isDebugEnabled()) {
        log.debug("jump called, but topKey is null, must need to move to next row");
      }
      return false;
    } else {
      
      int comp = this.topKey.getRow().compareTo(jumpKey.getRow());
      // compare rows
      if (comp > 0) {
        if (log.isDebugEnabled()) {
          log.debug("jump, our row is ahead of jumpKey.");
          log.debug("jumpRow: " + jumpKey.getRow() + " myRow: " + topKey.getRow() + " parentEndRow" + parentEndRow);
        }
        return hasTop(); // do nothing, we're ahead of jumpKey row
      } else if (comp < 0) { // a row behind jump key, need to move forward
      
        if (log.isDebugEnabled()) {
          log.debug("II jump, row jump");
        }
        Key endKey = null;
        if (parentEndRow != null) {
          endKey = new Key(parentEndRow);
        }
        Key sKey = new Key(jumpKey.getRow());
        Range fake = new Range(sKey, true, endKey, false);
        this.seek(fake, null, false);
        return hasTop();
      } else {
        // need to check uid
        String myUid = this.topKey.getColumnQualifier().toString();
        String jumpUid = getUID(jumpKey);
        if (log.isDebugEnabled()) {
          if (myUid == null) {
            log.debug("myUid is null");
          } else {
            log.debug("myUid: " + myUid);
          }
          
          if (jumpUid == null) {
            log.debug("jumpUid is null");
          } else {
            log.debug("jumpUid: " + jumpUid);
          }
        }
        
        int ucomp = myUid.compareTo(jumpUid);
        if (ucomp < 0) { // need to move all sources forward
          if (log.isDebugEnabled()) {
            log.debug("jump, uid jump");
          }
          Text row = jumpKey.getRow();
          Range range = new Range(row);
          this.currentRow = row;
          this.currentDocID = new Text(this.getUID(jumpKey));
          
          doSeek(range);

          // make sure it is in the range if we have one.
          if (hasTop() && parentEndRow != null && topKey.getRow().compareTo(parentEndRow) > 0) {
            topKey = null;
          }
          if (log.isDebugEnabled() && hasTop()) {
            log.debug("jump, topKey is now: " + topKey);
          }
          return hasTop();
          
        }// else do nothing
        if (hasTop() && parentEndRow != null && topKey.getRow().compareTo(parentEndRow) > 0) {
          topKey = null;
        }
        return hasTop();
      }
    }
  }
}