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; } } }  void addColumnTimestamp(Column column, long timestamp); ReturnCode filterKeyValue(KeyValue v); @Override void reset(); @Override KeyValue getNextKeyHint(KeyValue kv); void readFields(DataInput arg0); void write(DataOutput arg0);  }
@Test public void testFilterKeyValue() { ColumnTimestampFilter filter = new ColumnTimestampFilter(); KeyValue kv = new KeyValue(ROW, FAMILY, QUALIFIER, PREWRITE_TS, Type.Put, VALUE); Assert.assertEquals(ReturnCode.NEXT_ROW, filter.filterKeyValue(kv)); filter = new ColumnTimestampFilter(); filter.addColumnTimestamp(COLUMN, PREWRITE_TS); kv = new KeyValue(ROW, ANOTHER_FAMILY, QUALIFIER, PREWRITE_TS, Type.Put, VALUE); Assert.assertEquals(ReturnCode.SEEK_NEXT_USING_HINT, filter.filterKeyValue(kv)); Assert.assertArrayEquals(filter.getNextKeyHint(kv).getFamily(), FAMILY); Assert.assertArrayEquals(filter.getNextKeyHint(kv).getQualifier(), QUALIFIER); filter = new ColumnTimestampFilter(); filter.addColumnTimestamp(COLUMN, PREWRITE_TS); kv = new KeyValue(ROW, FAMILY, ANOTHER_QUALIFIER, PREWRITE_TS, Type.Put, VALUE); Assert.assertEquals(ReturnCode.SEEK_NEXT_USING_HINT, filter.filterKeyValue(kv)); Assert.assertArrayEquals(filter.getNextKeyHint(kv).getFamily(), FAMILY); Assert.assertArrayEquals(filter.getNextKeyHint(kv).getQualifier(), QUALIFIER); filter = new ColumnTimestampFilter(); filter.addColumnTimestamp(COLUMN, PREWRITE_TS); byte[] laterFamily = Bytes.toBytes(Bytes.toString(FAMILY) + "#"); kv = new KeyValue(ROW, laterFamily, QUALIFIER, PREWRITE_TS, Type.Put, VALUE); Assert.assertEquals(ReturnCode.NEXT_ROW, filter.filterKeyValue(kv)); filter = new ColumnTimestampFilter(); filter.addColumnTimestamp(COLUMN, PREWRITE_TS); kv = new KeyValue(ROW, FAMILY, QUALIFIER, PREWRITE_TS - 1, Type.Put, VALUE); Assert.assertEquals(ReturnCode.NEXT_COL, filter.filterKeyValue(kv)); filter = new ColumnTimestampFilter(); filter.addColumnTimestamp(COLUMN, PREWRITE_TS); kv = new KeyValue(ROW, FAMILY, QUALIFIER, PREWRITE_TS, Type.Put, VALUE); Assert.assertEquals(ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(kv)); filter = new ColumnTimestampFilter(); filter.addColumnTimestamp(COLUMN, PREWRITE_TS); kv = new KeyValue(ROW, FAMILY, QUALIFIER, PREWRITE_TS + 1, Type.Put, VALUE); Assert.assertEquals(ReturnCode.SKIP, filter.filterKeyValue(kv)); filter = new ColumnTimestampFilter(); filter.addColumnTimestamp(COLUMN, PREWRITE_TS); kv = new KeyValue(ROW, FAMILY, ANOTHER_QUALIFIER, PREWRITE_TS, Type.Put, VALUE); Assert.assertEquals(ReturnCode.SEEK_NEXT_USING_HINT, filter.filterKeyValue(kv)); Assert.assertArrayEquals(filter.getNextKeyHint(kv).getFamily(), FAMILY); Assert.assertArrayEquals(filter.getNextKeyHint(kv).getQualifier(), QUALIFIER); kv = new KeyValue(ROW, FAMILY, QUALIFIER, PREWRITE_TS + 1, Type.Put, VALUE); Assert.assertEquals(ReturnCode.SKIP, filter.filterKeyValue(kv)); kv = new KeyValue(ROW, FAMILY, QUALIFIER, PREWRITE_TS, Type.Put, VALUE); Assert.assertEquals(ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(kv)); kv = new KeyValue(ROW, FAMILY, QUALIFIER, PREWRITE_TS - 1, Type.Put, VALUE); Assert.assertEquals(ReturnCode.NEXT_ROW, filter.filterKeyValue(kv)); byte[] laterQualifier = Bytes.toBytes(Bytes.toString(QUALIFIER) + "#"); kv = new KeyValue(ROW, FAMILY, laterQualifier, PREWRITE_TS + 1, Type.Put, VALUE); Assert.assertEquals(ReturnCode.NEXT_ROW, filter.filterKeyValue(kv)); }