class ColumnTimestampFilter extends FilterBase {

  public ReturnCode filterKeyValue(KeyValue v) {
    if (curColumnIdx == -1) {
      sortColumnsTs();
      curColumnIdx = 0;
    }
    
    if (curColumnIdx >= columnsTs.size()) {
      return ReturnCode.NEXT_ROW;
    }
    
    Column column = new Column(v.getFamily(), v.getQualifier());
    Column curColumn = null;
    Long curTs = null;
    int cmpRet = 0;
    do {
      curColumn = columnsTs.get(curColumnIdx).getFirst();
      curTs = columnsTs.get(curColumnIdx).getSecond();
    } while ((cmpRet = curColumn.compareTo(column)) < 0 && ++curColumnIdx < columnsTs.size());
    
    if (cmpRet < 0) {
      return ReturnCode.NEXT_ROW;
    } else if (cmpRet > 0){
      return ReturnCode.SEEK_NEXT_USING_HINT;
    } else {
      if (curTs.equals(v.getTimestamp())) {
        ++curColumnIdx;
        return ReturnCode.INCLUDE_AND_NEXT_COL;
      } else if (curTs > v.getTimestamp()) {
        return ReturnCode.NEXT_COL;      
      } else {
        return ReturnCode.SKIP;
      }
    }
  }

}