Java 类org.apache.hadoop.hbase.filter.RowFilter 实例源码

public HBaseScanSpec parseTree() {
  HBaseScanSpec parsedSpec = le.accept(this, null);
  if (parsedSpec != null) {
    parsedSpec = mergeScanSpecs("booleanAnd", this.groupScan.getHBaseScanSpec(), parsedSpec);
    /*
     * If RowFilter is THE filter attached to the scan specification,
     * remove it since its effect is also achieved through startRow and stopRow.
     */
    Filter parsedFilter = HBaseUtils.deserializeFilter(parsedSpec.filter);
    if (parsedFilter instanceof RowFilter &&
        ((RowFilter)parsedFilter).getComparator() instanceof BinaryComparator) {
      parsedSpec.filter = null;
    }
  }
  return parsedSpec;
}

@Override
public void configure(JobConf job) {
  try {
    HTable exampleTable = new HTable(HBaseConfiguration.create(job),
      Bytes.toBytes("exampleDeprecatedTable"));
    // mandatory
    setHTable(exampleTable);
    byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
      Bytes.toBytes("columnB") };
    // mandatory
    setInputColumns(inputColumns);
    Filter exampleFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("aa.*"));
    // optional
    setRowFilter(exampleFilter);
  } catch (IOException exception) {
    throw new RuntimeException("Failed to configure for job.", exception);
  }
}

@Test
public void testJira6912() throws Exception {
  TableName TABLE = TableName.valueOf("testJira6912");
  Table foo = TEST_UTIL.createTable(TABLE, new byte[][] {FAMILY}, 10);

  List<Put> puts = new ArrayList<Put>();
  for (int i=0;i !=100; i++){
    Put put = new Put(Bytes.toBytes(i));
    put.add(FAMILY, FAMILY, Bytes.toBytes(i));
    puts.add(put);
  }
  foo.put(puts);
  // If i comment this out it works
  TEST_UTIL.flush();

  Scan scan = new Scan();
  scan.setStartRow(Bytes.toBytes(1));
  scan.setStopRow(Bytes.toBytes(3));
  scan.addColumn(FAMILY, FAMILY);
  scan.setFilter(new RowFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator(Bytes.toBytes(1))));

  ResultScanner scanner = foo.getScanner(scan);
  Result[] bar = scanner.next(100);
  assertEquals(1, bar.length);
}

@Override
public void configure(JobConf job) {
  try {
    HTable exampleTable = new HTable(HBaseConfiguration.create(job),
      Bytes.toBytes("exampleDeprecatedTable"));
    // mandatory
    setHTable(exampleTable);
    byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
      Bytes.toBytes("columnB") };
    // optional
    Scan scan = new Scan();
    for (byte[] family : inputColumns) {
      scan.addFamily(family);
    }
    Filter exampleFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("aa.*"));
    scan.setFilter(exampleFilter);
    setScan(scan);
  } catch (IOException exception) {
    throw new RuntimeException("Failed to configure for job.", exception);
  }
}

@Override
public void configure(JobConf job) {
  try {
    Connection connection = ConnectionFactory.createConnection(HBaseConfiguration.create(job));
    TableName tableName = TableName.valueOf("exampleJobConfigurableTable");
    // mandatory
    initializeTable(connection, tableName);
    byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
      Bytes.toBytes("columnB") };
    //optional
    Scan scan = new Scan();
    for (byte[] family : inputColumns) {
      scan.addFamily(family);
    }
    Filter exampleFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("aa.*"));
    scan.setFilter(exampleFilter);
    setScan(scan);
  } catch (IOException exception) {
    throw new RuntimeException("Failed to initialize.", exception);
  }
}

@Override
protected void initialize(JobContext job) throws IOException {
  Connection connection = ConnectionFactory.createConnection(HBaseConfiguration.create(
      job.getConfiguration()));
  TableName tableName = TableName.valueOf("exampleTable");
  // mandatory
  initializeTable(connection, tableName);
  byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
    Bytes.toBytes("columnB") };
  //optional
  Scan scan = new Scan();
  for (byte[] family : inputColumns) {
    scan.addFamily(family);
  }
  Filter exampleFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("aa.*"));
  scan.setFilter(exampleFilter);
  setScan(scan);
}

public HBaseScanSpec parseTree() {
  HBaseScanSpec parsedSpec = le.accept(this, null);
  if (parsedSpec != null) {
    parsedSpec = mergeScanSpecs("booleanAnd", this.groupScan.getHBaseScanSpec(), parsedSpec);
    /*
     * If RowFilter is THE filter attached to the scan specification,
     * remove it since its effect is also achieved through startRow and stopRow.
     */
    Filter filter = parsedSpec.getFilter();
    if (filter instanceof RowFilter &&
        ((RowFilter)filter).getOperator() != CompareOp.NOT_EQUAL &&
        ((RowFilter)filter).getComparator() instanceof BinaryComparator) {
      parsedSpec = new HBaseScanSpec(parsedSpec.getTableName(), parsedSpec.getStartRow(), parsedSpec.getStopRow(), null);
    }
  }
  return parsedSpec;
}

@Test
public void testJira6912() throws Exception {
  byte [] TABLE = Bytes.toBytes("testJira6912");
  HTable foo = TEST_UTIL.createTable(TABLE, new byte[][] {FAMILY}, 10);

  List<Put> puts = new ArrayList<Put>();
  for (int i=0;i !=100; i++){
    Put put = new Put(Bytes.toBytes(i));
    put.add(FAMILY, FAMILY, Bytes.toBytes(i));
    puts.add(put);
  }
  foo.put(puts);
  // If i comment this out it works
  TEST_UTIL.flush();

  Scan scan = new Scan();
  scan.setStartRow(Bytes.toBytes(1));
  scan.setStopRow(Bytes.toBytes(3));
  scan.addColumn(FAMILY, FAMILY);
  scan.setFilter(new RowFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator(Bytes.toBytes(1))));

  ResultScanner scanner = foo.getScanner(scan);
  Result[] bar = scanner.next(100);
  assertEquals(1, bar.length);
}

@Override
public void configure(JobConf job) {
  try {
    HTable exampleTable = new HTable(HBaseConfiguration.create(job),
      Bytes.toBytes("exampleDeprecatedTable"));
    // mandatory
    setHTable(exampleTable);
    byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
      Bytes.toBytes("columnB") };
    // mandatory
    setInputColumns(inputColumns);
    Filter exampleFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("aa.*"));
    // optional
    setRowFilter(exampleFilter);
  } catch (IOException exception) {
    throw new RuntimeException("Failed to configure for job.", exception);
  }
}

@Test
public void testJira6912() throws Exception {
  TableName TABLE = TableName.valueOf("testJira6912");
  Table foo = TEST_UTIL.createTable(TABLE, new byte[][] {FAMILY}, 10);

  List<Put> puts = new ArrayList<Put>();
  for (int i=0;i !=100; i++){
    Put put = new Put(Bytes.toBytes(i));
    put.add(FAMILY, FAMILY, Bytes.toBytes(i));
    puts.add(put);
  }
  foo.put(puts);
  // If i comment this out it works
  TEST_UTIL.flush();

  Scan scan = new Scan();
  scan.setStartRow(Bytes.toBytes(1));
  scan.setStopRow(Bytes.toBytes(3));
  scan.addColumn(FAMILY, FAMILY);
  scan.setFilter(new RowFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator(Bytes.toBytes(1))));

  ResultScanner scanner = foo.getScanner(scan);
  Result[] bar = scanner.next(100);
  assertEquals(1, bar.length);
}

@Override
public void configure(JobConf job) {
  try {
    HTable exampleTable = new HTable(HBaseConfiguration.create(job),
      Bytes.toBytes("exampleDeprecatedTable"));
    // mandatory
    setHTable(exampleTable);
    byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
      Bytes.toBytes("columnB") };
    // optional
    Scan scan = new Scan();
    for (byte[] family : inputColumns) {
      scan.addFamily(family);
    }
    Filter exampleFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("aa.*"));
    scan.setFilter(exampleFilter);
    setScan(scan);
  } catch (IOException exception) {
    throw new RuntimeException("Failed to configure for job.", exception);
  }
}

@Override
public void configure(JobConf job) {
  try {
    Connection connection = ConnectionFactory.createConnection(HBaseConfiguration.create(job));
    TableName tableName = TableName.valueOf("exampleJobConfigurableTable");
    // mandatory
    initializeTable(connection, tableName);
    byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
      Bytes.toBytes("columnB") };
    //optional
    Scan scan = new Scan();
    for (byte[] family : inputColumns) {
      scan.addFamily(family);
    }
    Filter exampleFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("aa.*"));
    scan.setFilter(exampleFilter);
    setScan(scan);
  } catch (IOException exception) {
    throw new RuntimeException("Failed to initialize.", exception);
  }
}

@Override
protected void initialize(JobContext job) throws IOException {
  Connection connection = ConnectionFactory.createConnection(HBaseConfiguration.create(
      job.getConfiguration()));
  TableName tableName = TableName.valueOf("exampleTable");
  // mandatory
  initializeTable(connection, tableName);
  byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
    Bytes.toBytes("columnB") };
  //optional
  Scan scan = new Scan();
  for (byte[] family : inputColumns) {
    scan.addFamily(family);
  }
  Filter exampleFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("aa.*"));
  scan.setFilter(exampleFilter);
  setScan(scan);
}

@Test
public void testJira6912() throws Exception {
  byte [] TABLE = Bytes.toBytes("testJira6912");
  HTable foo = TEST_UTIL.createTable(TABLE, new byte[][] {FAMILY}, 10);

  List<Put> puts = new ArrayList<Put>();
  for (int i=0;i !=100; i++){
    Put put = new Put(Bytes.toBytes(i));
    put.add(FAMILY, FAMILY, Bytes.toBytes(i));
    puts.add(put);
  }
  foo.put(puts);
  // If i comment this out it works
  TEST_UTIL.flush();

  Scan scan = new Scan();
  scan.setStartRow(Bytes.toBytes(1));
  scan.setStopRow(Bytes.toBytes(3));
  scan.addColumn(FAMILY, FAMILY);
  scan.setFilter(new RowFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator(Bytes.toBytes(1))));

  ResultScanner scanner = foo.getScanner(scan);
  Result[] bar = scanner.next(100);
  assertEquals(1, bar.length);
}

@Test
public void testJira6912() throws Exception {
  byte [] TABLE = Bytes.toBytes("testJira6912");
  HTable foo = TEST_UTIL.createTable(TABLE, new byte[][] {FAMILY}, 10);

  List<Put> puts = new ArrayList<Put>();
  for (int i=0;i !=100; i++){
    Put put = new Put(Bytes.toBytes(i));
    put.add(FAMILY, FAMILY, Bytes.toBytes(i));
    puts.add(put);
  }
  foo.put(puts);
  // If i comment this out it works
  TEST_UTIL.flush();

  Scan scan = new Scan();
  scan.setStartRow(Bytes.toBytes(1));
  scan.setStopRow(Bytes.toBytes(3));
  scan.addColumn(FAMILY, FAMILY);
  scan.setFilter(new RowFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator(Bytes.toBytes(1))));

  ResultScanner scanner = foo.getScanner(scan);
  Result[] bar = scanner.next(100);
  assertEquals(1, bar.length);
}

@Test
public void testJira6912() throws Exception {
  final TableName tableName = TableName.valueOf(name.getMethodName());
  Table foo = TEST_UTIL.createTable(tableName, new byte[][] {FAMILY}, 10);

  List<Put> puts = new ArrayList<Put>();
  for (int i=0;i !=100; i++){
    Put put = new Put(Bytes.toBytes(i));
    put.addColumn(FAMILY, FAMILY, Bytes.toBytes(i));
    puts.add(put);
  }
  foo.put(puts);
  // If i comment this out it works
  TEST_UTIL.flush();

  Scan scan = new Scan();
  scan.setStartRow(Bytes.toBytes(1));
  scan.setStopRow(Bytes.toBytes(3));
  scan.addColumn(FAMILY, FAMILY);
  scan.setFilter(new RowFilter(CompareOperator.NOT_EQUAL,
      new BinaryComparator(Bytes.toBytes(1))));

  ResultScanner scanner = foo.getScanner(scan);
  Result[] bar = scanner.next(100);
  assertEquals(1, bar.length);
}

@Override
public void configure(JobConf job) {
  try {
    Connection connection = ConnectionFactory.createConnection(job);
    Table exampleTable = connection.getTable(TableName.valueOf("exampleDeprecatedTable"));
    // mandatory
    initializeTable(connection, exampleTable.getName());
    byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
      Bytes.toBytes("columnB") };
    // mandatory
    setInputColumns(inputColumns);
    Filter exampleFilter =
      new RowFilter(CompareOperator.EQUAL, new RegexStringComparator("aa.*"));
    // optional
    setRowFilter(exampleFilter);
  } catch (IOException exception) {
    throw new RuntimeException("Failed to configure for job.", exception);
  }
}

@Override
public void configure(JobConf job) {
  try {
    Connection connection = ConnectionFactory.createConnection(job);
    Table exampleTable = connection.getTable(TableName.valueOf(("exampleDeprecatedTable")));
    // mandatory
    initializeTable(connection, exampleTable.getName());
    byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
      Bytes.toBytes("columnB") };
    // optional
    Scan scan = new Scan();
    for (byte[] family : inputColumns) {
      scan.addFamily(family);
    }
    Filter exampleFilter =
      new RowFilter(CompareOperator.EQUAL, new RegexStringComparator("aa.*"));
    scan.setFilter(exampleFilter);
    setScan(scan);
  } catch (IOException exception) {
    throw new RuntimeException("Failed to configure for job.", exception);
  }
}

@Override
public void configure(JobConf job) {
  try {
    Connection connection = ConnectionFactory.createConnection(HBaseConfiguration.create(job));
    TableName tableName = TableName.valueOf("exampleJobConfigurableTable");
    // mandatory
    initializeTable(connection, tableName);
    byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
      Bytes.toBytes("columnB") };
    //optional
    Scan scan = new Scan();
    for (byte[] family : inputColumns) {
      scan.addFamily(family);
    }
    Filter exampleFilter =
      new RowFilter(CompareOperator.EQUAL, new RegexStringComparator("aa.*"));
    scan.setFilter(exampleFilter);
    setScan(scan);
  } catch (IOException exception) {
    throw new RuntimeException("Failed to initialize.", exception);
  }
}

@Override
protected void initialize(JobContext job) throws IOException {
  Connection connection = ConnectionFactory.createConnection(HBaseConfiguration.create(
      job.getConfiguration()));
  TableName tableName = TableName.valueOf("exampleTable");
  // mandatory
  initializeTable(connection, tableName);
  byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
    Bytes.toBytes("columnB") };
  //optional
  Scan scan = new Scan();
  for (byte[] family : inputColumns) {
    scan.addFamily(family);
  }
  Filter exampleFilter =
    new RowFilter(CompareOperator.EQUAL, new RegexStringComparator("aa.*"));
  scan.setFilter(exampleFilter);
  setScan(scan);
}

@Test
public void testJira6912() throws Exception {
  byte [] TABLE = Bytes.toBytes("testJira6912");
  HTable foo = TEST_UTIL.createTable(TABLE, new byte[][] {FAMILY}, 10);

  List<Put> puts = new ArrayList<Put>();
  for (int i=0;i !=100; i++){
    Put put = new Put(Bytes.toBytes(i));
    put.add(FAMILY, FAMILY, Bytes.toBytes(i));
    puts.add(put);
  }
  foo.put(puts);
  // If i comment this out it works
  TEST_UTIL.flush();

  Scan scan = new Scan();
  scan.setStartRow(Bytes.toBytes(1));
  scan.setStopRow(Bytes.toBytes(3));
  scan.addColumn(FAMILY, FAMILY);
  scan.setFilter(new RowFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator(Bytes.toBytes(1))));

  ResultScanner scanner = foo.getScanner(scan);
  Result[] bar = scanner.next(100);
  assertEquals(1, bar.length);
}

@Test
public void testJira6912() throws Exception {
  byte [] TABLE = Bytes.toBytes("testJira6912");
  HTable foo = TEST_UTIL.createTable(TABLE, new byte[][] {FAMILY}, 10);

  List<Put> puts = new ArrayList<Put>();
  for (int i=0;i !=100; i++){
    Put put = new Put(Bytes.toBytes(i));
    put.add(FAMILY, FAMILY, Bytes.toBytes(i));
    puts.add(put);
  }
  foo.put(puts);
  // If i comment this out it works
  TEST_UTIL.flush();

  Scan scan = new Scan();
  scan.setStartRow(Bytes.toBytes(1));
  scan.setStopRow(Bytes.toBytes(3));
  scan.addColumn(FAMILY, FAMILY);
  scan.setFilter(new RowFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator(Bytes.toBytes(1))));

  ResultScanner scanner = foo.getScanner(scan);
  Result[] bar = scanner.next(100);
  assertEquals(1, bar.length);
}

@Test
public void testJira6912() throws Exception {
  byte [] TABLE = Bytes.toBytes("testJira6912");
  HTable foo = TEST_UTIL.createTable(TABLE, new byte[][] {FAMILY}, 10);

  List<Put> puts = new ArrayList<Put>();
  for (int i=0;i !=100; i++){
    Put put = new Put(Bytes.toBytes(i));
    put.add(FAMILY, FAMILY, Bytes.toBytes(i));
    puts.add(put);
  }
  foo.put(puts);
  // If i comment this out it works
  TEST_UTIL.flush();

  Scan scan = new Scan();
  scan.setStartRow(Bytes.toBytes(1));
  scan.setStopRow(Bytes.toBytes(3));
  scan.addColumn(FAMILY, FAMILY);
  scan.setFilter(new RowFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator(Bytes.toBytes(1))));

  ResultScanner scanner = foo.getScanner(scan);
  Result[] bar = scanner.next(100);
  assertEquals(1, bar.length);
}

@Test
public void testJira6912() throws Exception {
  byte [] TABLE = Bytes.toBytes("testJira6912");
  HTable foo = TEST_UTIL.createTable(TABLE, new byte[][] {FAMILY}, 10);

  List<Put> puts = new ArrayList<Put>();
  for (int i=0;i !=100; i++){
    Put put = new Put(Bytes.toBytes(i));
    put.add(FAMILY, FAMILY, Bytes.toBytes(i));
    puts.add(put);
  }
  foo.put(puts);
  // If i comment this out it works
  TEST_UTIL.flush();

  Scan scan = new Scan();
  scan.setStartRow(Bytes.toBytes(1));
  scan.setStopRow(Bytes.toBytes(3));
  scan.addColumn(FAMILY, FAMILY);
  scan.setFilter(new RowFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator(Bytes.toBytes(1))));

  ResultScanner scanner = foo.getScanner(scan);
  Result[] bar = scanner.next(100);
  assertEquals(1, bar.length);
}

@Test
public void testJira6912() throws Exception {
  byte [] TABLE = Bytes.toBytes("testJira6912");
  HTable foo = TEST_UTIL.createTable(TABLE, new byte[][] {FAMILY}, 10);

  List<Put> puts = new ArrayList<Put>();
  for (int i=0;i !=100; i++){
    Put put = new Put(Bytes.toBytes(i));
    put.add(FAMILY, FAMILY, Bytes.toBytes(i));
    puts.add(put);
  }
  foo.put(puts);
  // If i comment this out it works
  TEST_UTIL.flush();

  Scan scan = new Scan();
  scan.setStartRow(Bytes.toBytes(1));
  scan.setStopRow(Bytes.toBytes(3));
  scan.addColumn(FAMILY, FAMILY);
  scan.setFilter(new RowFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator(Bytes.toBytes(1))));

  ResultScanner scanner = foo.getScanner(scan);
  Result[] bar = scanner.next(100);
  assertEquals(1, bar.length);
}

@Test
public void testJira6912() throws Exception {
  byte [] TABLE = Bytes.toBytes("testJira6912");
  HTable foo = TEST_UTIL.createTable(TABLE, new byte[][] {FAMILY}, 10);

  List<Put> puts = new ArrayList<Put>();
  for (int i=0;i !=100; i++){
    Put put = new Put(Bytes.toBytes(i));
    put.add(FAMILY, FAMILY, Bytes.toBytes(i));
    puts.add(put);
  }
  foo.put(puts);
  // If i comment this out it works
  TEST_UTIL.flush();

  Scan scan = new Scan();
  scan.setStartRow(Bytes.toBytes(1));
  scan.setStopRow(Bytes.toBytes(3));
  scan.addColumn(FAMILY, FAMILY);
  scan.setFilter(new RowFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator(Bytes.toBytes(1))));

  ResultScanner scanner = foo.getScanner(scan);
  Result[] bar = scanner.next(100);
  assertEquals(1, bar.length);
}

private void scanWithFilter(byte[] tableName, byte[] columnFamilyName, String val, int expected) throws IOException {
    HTable table = new HTable(hbaseCluster.getConfiguration(), tableName);
    LOGGER.info("Scanning with filter from HBase table.");
    Filter filter = new RowFilter(CompareFilter.CompareOp.LESS_OR_EQUAL, new BinaryComparator(Bytes.toBytes(val)));
    Scan scan = new Scan();
    LOGGER.info("Getting from HBase table.");
    scan.setFilter(filter);
    int resultCount = 0;
    ResultScanner resultScanner = table.getScanner(scan);
    for (Result result : resultScanner) {
        logResult(result, columnFamilyName);
        resultCount++;
    }
    resultScanner.close();
    table.close();
}

public List<GroupedData> getRecommendedUserItem(String cp, String collection, long user, Long from, Long startDate,
        Long endDate, int size)
{
    FilterList filters = new FilterList();

    if (from == null)
    {
        filters.addFilter(new RowFilter(CompareOp.EQUAL, new BinaryPrefixComparator(RowKeys.getStatRecommendedItemKey(
                collection, user))));
    }
    else
    {
        filters.addFilter(new RowFilter(CompareOp.GREATER_OR_EQUAL, new BinaryPrefixComparator(RowKeys
                .getStatRecommendedItemKey(collection, user, from))));
    }

    setDateLimit(STATS_RECOMMENDED_USERITEM, startDate, endDate, filters);

    Scan scan = new Scan().addFamily(STATS_RECOMMENDED_USERITEM).setFilter(filters);

    return getResults(cp, scan, STATS_RECOMMENDED_USERITEM, size);
}

public List<GroupedData> getUserRated(String cp, String collection, Long from, Long startDate, Long endDate, int size)
{
    // Put put = new Put(RowKeys.getStatRatingsPerItemKey(collection, item, date));
    Scan scan = new Scan();
    scan.addFamily(STATS_USER_RATINGS);
    FilterList filters = new FilterList();
    if (from == null)
    {
        filters.addFilter(new RowFilter(CompareOp.EQUAL, new BinaryPrefixComparator(RowKeys
                .getStatRatingsPerUserKey(collection))));
    }
    else
    {
        filters.addFilter(new RowFilter(CompareOp.GREATER_OR_EQUAL, new BinaryPrefixComparator(RowKeys
                .getStatRatingsPerUserKey(collection, from.longValue()))));
    }

    setDateLimit(STATS_USER_RATINGS, startDate, endDate, filters);

    scan.setFilter(filters);

    return getResults(cp, scan, STATS_USER_RATINGS, size);
}

public List<GroupedData> getItemRated(String cp, String collection, Long from, Long startDate, Long endDate, int size)
{
    // Put put = new Put(RowKeys.getStatRatingsPerItemKey(collection, item, date));
    Scan scan = new Scan();
    scan.addFamily(STATS_ITEM_RATINGS);
    FilterList filters = new FilterList();
    if (from == null)
    {
        filters.addFilter(new RowFilter(CompareOp.EQUAL, new BinaryPrefixComparator(RowKeys
                .getStatRatingsPerItemKey(collection))));
    }
    else
    {
        filters.addFilter(new RowFilter(CompareOp.GREATER_OR_EQUAL, new BinaryPrefixComparator(RowKeys
                .getStatRatingsPerItemKey(collection, from.longValue()))));
    }

    setDateLimit(STATS_ITEM_RATINGS, startDate, endDate, filters);

    scan.setFilter(filters);

    return getResults(cp, scan, STATS_ITEM_RATINGS, size);
}

/**
 * return the list of content providers
 * @param cp
 * @param recommender
 * @return
 * @throws IOException
 * @see nl.gridline.zieook.statistics.SourcesByRecommenderMap
 * @see nl.gridline.zieook.statistics.SourcesByRecommenderReduce
 */
public List<GroupedData> getSources(String cp, String recommender, String from, Long startDate, Long endDate,
        int size)
{
    Scan scan = new Scan();
    scan.addFamily(STATS_VIEWED_SOURCE);
    FilterList filters = new FilterList();

    if (from == null)
    {
        filters.addFilter(new RowFilter(CompareOp.EQUAL, new BinaryPrefixComparator(RowKeys
                .getStatSourcesKey(recommender))));
    }
    else
    {
        filters.addFilter(new RowFilter(CompareOp.GREATER_OR_EQUAL, new BinaryPrefixComparator(RowKeys
                .getStatSourcesKey(recommender, from))));
    }

    setDateLimit(STATS_VIEWED_SOURCE, startDate, endDate, filters);

    scan.setFilter(filters);

    return getResults(cp, scan, STATS_VIEWED_SOURCE, size);
}

public List<GroupedData> getViewed(String cp, String recommender, Long from, Long startDate, Long endDate, int size)
{
    Scan scan = new Scan();
    scan.addFamily(STATS_VIEWED_ITEM);

    FilterList filters = new FilterList();

    if (from == null)
    {
        filters.addFilter(new RowFilter(CompareOp.EQUAL, new BinaryPrefixComparator(RowKeys
                .getStatViewedKey(recommender))));
    }
    else
    {
        filters.addFilter(new RowFilter(CompareOp.EQUAL, new BinaryPrefixComparator(RowKeys.getStatViewedKey(
                recommender, from.longValue()))));
    }

    setDateLimit(STATS_VIEWED_ITEM, startDate, endDate, filters);

    scan.setFilter(filters);

    return getResults(cp, scan, STATS_VIEWED_ITEM, size);
}

public long deleteViews(String cp, String recommender)
{
    FilterList filters = new FilterList();

    // filter column-family & recommender name
    filters.addFilter(new FamilyFilter(CompareOp.EQUAL, new BinaryComparator(EVENTLOG_COLUMN_USERVIEW)));
    filters
            .addFilter(new RowFilter(CompareOp.EQUAL, new BinaryPrefixComparator(RowKeys.getUserViewKey(recommender))));

    Scan scan = new Scan().addFamily(EVENTLOG_COLUMN_USERVIEW).setFilter(filters);

    // TODO put this in a map-reduce delete.. that executes in background..
    // we only need to pass the table & a scan object. should be quite easy

    long count = deleteAll(scan, cp, EVENTLOG_COLUMN_RECOMMENDED);

    return count;
}

@Test
@Ignore
public void testItemFilter()
{
    FilterList itemFilter = new FilterList(Operator.MUST_PASS_ONE);

    // row filter is probably faster, (don't know for sure) - otherwise single column-value filter is also
    // possible.
    itemFilter.addFilter(new RowFilter(CompareOp.EQUAL, new BinaryComparator(RowKeys
            .getCollectionKey(COLLECTION, 402))));
    itemFilter.addFilter(new RowFilter(CompareOp.EQUAL, new BinaryComparator(RowKeys
            .getCollectionKey(COLLECTION, 403))));
    itemFilter.addFilter(new RowFilter(CompareOp.EQUAL, new BinaryComparator(RowKeys.getCollectionKey(COLLECTION,
            1312))));

    Scan scan = new Scan(RowKeys.getCollectionKey(COLLECTION, 402), RowKeys.getCollectionKey(COLLECTION, 1312 + 1))
            .addFamily(Bytes.toBytes(HBaseTableConstants.COLLECTION_TABLE_COLUMN_INTR)).setFilter(itemFilter);

    List<CollectionItem> result = new CollectionTable(collections).getItems(CP, scan, 100);
    for (CollectionItem i : result)
    {
        LOG.info("{}", i);
    }

}

/**
 * return a recommended scanner with an optional start date and end date
 * @param startDate start date
 * @param endDate end date
 * @return
 */
public Scan getEventLogRecommendedScanner(Long startDate, Long endDate)
{
    Scan scan = new Scan().addFamily(EVENTLOG_COLUMN_RECOMMENDED);
    FilterList filters = new FilterList();
    filters.addFilter(new RowFilter(CompareOp.EQUAL, new BinaryPrefixComparator(RowKeys.getRecommendedItemKey())));
    // timestamp filter:
    if (startDate != null)
    {
        SingleColumnValueFilter startFilter = new SingleColumnValueFilter(EVENTLOG_COLUMN_RECOMMENDED,
                ModelConstants.TIMESTAMP, CompareOp.GREATER_OR_EQUAL, Bytes.toBytes(startDate.longValue()));
        startFilter.setFilterIfMissing(true);
        filters.addFilter(startFilter);
    }

    if (endDate != null)
    {
        SingleColumnValueFilter endFilter = new SingleColumnValueFilter(EVENTLOG_COLUMN_RECOMMENDED,
                ModelConstants.TIMESTAMP, CompareOp.LESS, Bytes.toBytes(endDate.longValue()));
        endFilter.setFilterIfMissing(true);
        filters.addFilter(endFilter);
    }

    return scan.setFilter(filters);
}

@Test
public void testJira6912() throws Exception {
  byte [] TABLE = Bytes.toBytes("testJira6912");
  HTable foo = TEST_UTIL.createTable(TABLE, new byte[][] {FAMILY}, 10);

  List<Put> puts = new ArrayList<Put>();
  for (int i=0;i !=100; i++){
    Put put = new Put(Bytes.toBytes(i));
    put.add(FAMILY, FAMILY, Bytes.toBytes(i));
    puts.add(put);
  }
  foo.put(puts);
  // If i comment this out it works
  TEST_UTIL.flush();

  Scan scan = new Scan();
  scan.setStartRow(Bytes.toBytes(1));
  scan.setStopRow(Bytes.toBytes(3));
  scan.addColumn(FAMILY, FAMILY);
  scan.setFilter(new RowFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator(Bytes.toBytes(1))));

  ResultScanner scanner = foo.getScanner(scan);
  Result[] bar = scanner.next(100);
  assertEquals(1, bar.length);
}

public HBaseScanSpec parseTree() {
  HBaseScanSpec parsedSpec = le.accept(this, null);
  if (parsedSpec != null) {
    parsedSpec = mergeScanSpecs("booleanAnd", this.groupScan.getHBaseScanSpec(), parsedSpec);
    /*
     * If RowFilter is THE filter attached to the scan specification,
     * remove it since its effect is also achieved through startRow and stopRow.
     */
    if (parsedSpec.filter instanceof RowFilter &&
        ((RowFilter)parsedSpec.filter).getComparator() instanceof BinaryComparator) {
      parsedSpec.filter = null;
    }
  }
  return parsedSpec;
}

protected void initialize(JobConf job, String table) throws IOException {
  Connection connection = ConnectionFactory.createConnection(HBaseConfiguration.create(job));
  TableName tableName = TableName.valueOf(table);
  // mandatory
  initializeTable(connection, tableName);
  byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
    Bytes.toBytes("columnB") };
  // mandatory
  setInputColumns(inputColumns);
  Filter exampleFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("aa.*"));
  // optional
  setRowFilter(exampleFilter);
}

@Test public void testCompareFilter() throws Exception {
  Filter f = new RowFilter(CompareOp.EQUAL,
    new BinaryComparator(Bytes.toBytes("testRowOne-2")));
  byte [] bytes = f.toByteArray();
  Filter ff = RowFilter.parseFrom(bytes);
  assertNotNull(ff);
}

/** Tests reading all rows using a filter. */
@Test
public void testReadingWithFilter() throws Exception {
  final String table = "TEST-FILTER-TABLE";
  final int numRows = 1001;

  createTable(table);
  writeData(table, numRows);

  String regex = ".*17.*";
  Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(regex));
  HBaseIO.Read read =
      HBaseIO.read().withConfiguration(conf).withTableId(table).withFilter(filter);
  runReadTestLength(read, 20);
}