ColumnPaginationFilterAdapter extends TypedFilterAdapterBase<ColumnPaginationFilter> { @Override public Filter adapt(FilterAdapterContext context, ColumnPaginationFilter filter) throws IOException { if (filter.getColumnOffset() != null) { byte[] family = context.getScan().getFamilies()[0]; ByteString startQualifier = ByteString.copyFrom(filter.getColumnOffset()); return createChain( filter, FILTERS .qualifier() .rangeWithinFamily(Bytes.toString(family)) .startClosed(startQualifier)); } else if (filter.getOffset() > 0) { return createChain(filter, FILTERS.offset().cellsPerRow(filter.getOffset())); } else { return createChain(filter, null); } }  @Override Filter adapt(FilterAdapterContext context, ColumnPaginationFilter filter); @Override FilterSupportStatus isFilterSupported(
      FilterAdapterContext context, ColumnPaginationFilter filter);  }
@Test public void integerLimitsAreApplied() throws IOException { ColumnPaginationFilter filter = new ColumnPaginationFilter(10, 20); Filters.Filter adaptedFilter = adapter.adapt(new FilterAdapterContext(new Scan(), null), filter); Filters.Filter expected = FILTERS .chain() .filter(FILTERS.limit().cellsPerColumn(1)) .filter(FILTERS.offset().cellsPerRow(20)) .filter(FILTERS.limit().cellsPerRow(10)); Assert.assertEquals(expected.toProto(), adaptedFilter.toProto()); }
@Test public void zeroOffsetLimitIsSupported() throws IOException { ColumnPaginationFilter filter = new ColumnPaginationFilter(10, 0); Filters.Filter adaptedFilter = adapter.adapt(new FilterAdapterContext(new Scan(), null), filter); Filters.Filter expected = FILTERS .chain() .filter(FILTERS.limit().cellsPerColumn(1)) .filter(FILTERS.limit().cellsPerRow(10)); Assert.assertEquals(expected.toProto(), adaptedFilter.toProto()); }
@Test public void qualifierOffsetIsPartiallySupported() throws IOException { Scan scan = new Scan().addFamily(Bytes.toBytes("f1")); Filters.Filter adaptedFilter = adapter.adapt( new FilterAdapterContext(scan, null), new ColumnPaginationFilter(10, Bytes.toBytes("q1"))); Filters.Filter expected = FILTERS .chain() .filter(FILTERS.limit().cellsPerColumn(1)) .filter( FILTERS .qualifier() .rangeWithinFamily("f1") .startClosed(ByteString.copyFromUtf8("q1"))) .filter(FILTERS.limit().cellsPerRow(10)); Assert.assertEquals(expected.toProto(), adaptedFilter.toProto()); }