@Override
public void scan(final String tableName, final Collection<Column> columns, final String filterExpression, final long minTime, final ResultHandler handler)
        throws IOException {

    Filter filter = null;
    if (!StringUtils.isBlank(filterExpression)) {
        ParseFilter parseFilter = new ParseFilter();
        filter = parseFilter.parseFilterString(filterExpression);
    }

    try (final Table table = connection.getTable(TableName.valueOf(tableName));
         final ResultScanner scanner = getResults(table, columns, filter, minTime)) {

        for (final Result result : scanner) {
            final byte[] rowKey = result.getRow();
            final Cell[] cells = result.rawCells();

            if (cells == null) {
                continue;
            }

            // convert HBase cells to NiFi cells
            final ResultCell[] resultCells = new ResultCell[cells.length];
            for (int i=0; i < cells.length; i++) {
                final Cell cell = cells[i];
                final ResultCell resultCell = getResultCell(cell);
                resultCells[i] = resultCell;
            }

            // delegate to the handler
            handler.handle(rowKey, resultCells);
        }
    }
}
 

public static void registerFilters(Configuration conf) {
  String[] filters = conf.getStrings(THRIFT_FILTERS);
  Splitter splitter = Splitter.on(':');
  if(filters != null) {
    for(String filterClass: filters) {
      List<String> filterPart = splitter.splitToList(filterClass);
      if(filterPart.size() != 2) {
        LOG.warn("Invalid filter specification " + filterClass + " - skipping");
      } else {
        ParseFilter.registerFilter(filterPart.get(0), filterPart.get(1));
      }
    }
  }
}
 

@Test
public void testFilterRegistration() throws Exception {
  Configuration conf = UTIL.getConfiguration();
  conf.set("hbase.thrift.filters", "MyFilter:filterclass");
  ThriftServer.registerFilters(conf);
  Map<String, String> registeredFilters = ParseFilter.getAllFilters();
  assertEquals("filterclass", registeredFilters.get("MyFilter"));
}
 

private void registerCustomFilter(Configuration conf) {
  String[] filterList = conf.getStrings(Constants.CUSTOM_FILTERS);
  if (filterList != null) {
    for (String filterClass : filterList) {
      String[] filterPart = filterClass.split(":");
      if (filterPart.length != 2) {
        LOG.warn(
          "Invalid filter specification " + filterClass + " - skipping");
      } else {
        ParseFilter.registerFilter(filterPart[0], filterPart[1]);
      }
    }
  }
}
 

@Override
public void scan(String tableName, Collection<Column> columns, String filterExpression, long minTime, List<String> visibilityLabels, ResultHandler handler) throws IOException {
    Filter filter = null;
    if (!StringUtils.isBlank(filterExpression)) {
        ParseFilter parseFilter = new ParseFilter();
        filter = parseFilter.parseFilterString(filterExpression);
    }

    try (final Table table = connection.getTable(TableName.valueOf(tableName));
         final ResultScanner scanner = getResults(table, columns, filter, minTime, visibilityLabels)) {

        for (final Result result : scanner) {
            final byte[] rowKey = result.getRow();
            final Cell[] cells = result.rawCells();

            if (cells == null) {
                continue;
            }

            // convert HBase cells to NiFi cells
            final ResultCell[] resultCells = new ResultCell[cells.length];
            for (int i=0; i < cells.length; i++) {
                final Cell cell = cells[i];
                final ResultCell resultCell = getResultCell(cell);
                resultCells[i] = resultCell;
            }

            // delegate to the handler
            handler.handle(rowKey, resultCells);
        }
    }
}
 

@Override
public void scan(String tableName, Collection<Column> columns, String filterExpression, long minTime, List<String> visibilityLabels, ResultHandler handler) throws IOException {
    Filter filter = null;
    if (!StringUtils.isBlank(filterExpression)) {
        ParseFilter parseFilter = new ParseFilter();
        filter = parseFilter.parseFilterString(filterExpression);
    }

    try (final Table table = connection.getTable(TableName.valueOf(tableName));
         final ResultScanner scanner = getResults(table, columns, filter, minTime, visibilityLabels)) {

        for (final Result result : scanner) {
            final byte[] rowKey = result.getRow();
            final Cell[] cells = result.rawCells();

            if (cells == null) {
                continue;
            }

            // convert HBase cells to NiFi cells
            final ResultCell[] resultCells = new ResultCell[cells.length];
            for (int i=0; i < cells.length; i++) {
                final Cell cell = cells[i];
                final ResultCell resultCell = getResultCell(cell);
                resultCells[i] = resultCell;
            }

            // delegate to the handler
            handler.handle(rowKey, resultCells);
        }
    }
}
 

/**
 * Creates a {@link Get} (HBase) from a {@link TGet} (Thrift).
 *
 * This ignores any timestamps set on {@link TColumn} objects.
 *
 * @param in the <code>TGet</code> to convert
 *
 * @return <code>Get</code> object
 *
 * @throws IOException if an invalid time range or max version parameter is given
 */
public static Get getFromThrift(TGet in) throws IOException {
  Get out = new Get(in.getRow());

  // Timestamp overwrites time range if both are set
  if (in.isSetTimestamp()) {
    out.setTimestamp(in.getTimestamp());
  } else if (in.isSetTimeRange()) {
    out.setTimeRange(in.getTimeRange().getMinStamp(), in.getTimeRange().getMaxStamp());
  }

  if (in.isSetMaxVersions()) {
    out.readVersions(in.getMaxVersions());
  }

  if (in.isSetFilterString()) {
    ParseFilter parseFilter = new ParseFilter();
    out.setFilter(parseFilter.parseFilterString(in.getFilterString()));
  }

  if (in.isSetAttributes()) {
    addAttributes(out,in.getAttributes());
  }

  if (in.isSetAuthorizations()) {
    out.setAuthorizations(new Authorizations(in.getAuthorizations().getLabels()));
  }

  if (in.isSetConsistency()) {
    out.setConsistency(consistencyFromThrift(in.getConsistency()));
  }

  if (in.isSetTargetReplicaId()) {
    out.setReplicaId(in.getTargetReplicaId());
  }

  if (in.isSetCacheBlocks()) {
    out.setCacheBlocks(in.isCacheBlocks());
  }
  if (in.isSetStoreLimit()) {
    out.setMaxResultsPerColumnFamily(in.getStoreLimit());
  }
  if (in.isSetStoreOffset()) {
    out.setRowOffsetPerColumnFamily(in.getStoreOffset());
  }
  if (in.isSetExistence_only()) {
    out.setCheckExistenceOnly(in.isExistence_only());
  }

  if (in.isSetColumns()) {
    for (TColumn column : in.getColumns()) {
      if (column.isSetQualifier()) {
        out.addColumn(column.getFamily(), column.getQualifier());
      } else {
        out.addFamily(column.getFamily());
      }
    }
  }

  if (in.isSetFilterBytes()) {
    out.setFilter(filterFromThrift(in.getFilterBytes()));
  }
  return out;
}
 

public static Scan scanFromThrift(TScan in) throws IOException {
  Scan out = new Scan();

  if (in.isSetStartRow()) {
    out.withStartRow(in.getStartRow());
  }
  if (in.isSetStopRow()) {
    out.withStopRow(in.getStopRow());
  }
  if (in.isSetCaching()) {
    out.setCaching(in.getCaching());
  }
  if (in.isSetMaxVersions()) {
    out.readVersions(in.getMaxVersions());
  }

  if (in.isSetColumns()) {
    for (TColumn column : in.getColumns()) {
      if (column.isSetQualifier()) {
        out.addColumn(column.getFamily(), column.getQualifier());
      } else {
        out.addFamily(column.getFamily());
      }
    }
  }

  TTimeRange timeRange = in.getTimeRange();
  if (timeRange != null &&
      timeRange.isSetMinStamp() && timeRange.isSetMaxStamp()) {
    out.setTimeRange(timeRange.getMinStamp(), timeRange.getMaxStamp());
  }

  if (in.isSetBatchSize()) {
    out.setBatch(in.getBatchSize());
  }

  if (in.isSetFilterString()) {
    ParseFilter parseFilter = new ParseFilter();
    out.setFilter(parseFilter.parseFilterString(in.getFilterString()));
  }

  if (in.isSetAttributes()) {
    addAttributes(out,in.getAttributes());
  }

  if (in.isSetAuthorizations()) {
    out.setAuthorizations(new Authorizations(in.getAuthorizations().getLabels()));
  }

  if (in.isSetReversed()) {
    out.setReversed(in.isReversed());
  }

  if (in.isSetCacheBlocks()) {
    out.setCacheBlocks(in.isCacheBlocks());
  }

  if (in.isSetColFamTimeRangeMap()) {
    Map<ByteBuffer, TTimeRange> colFamTimeRangeMap = in.getColFamTimeRangeMap();
    if (MapUtils.isNotEmpty(colFamTimeRangeMap)) {
      for (Map.Entry<ByteBuffer, TTimeRange> entry : colFamTimeRangeMap.entrySet()) {
        out.setColumnFamilyTimeRange(Bytes.toBytes(entry.getKey()),
            entry.getValue().getMinStamp(), entry.getValue().getMaxStamp());
      }
    }
  }

  if (in.isSetReadType()) {
    out.setReadType(readTypeFromThrift(in.getReadType()));
  }

  if (in.isSetLimit()) {
    out.setLimit(in.getLimit());
  }

  if (in.isSetConsistency()) {
    out.setConsistency(consistencyFromThrift(in.getConsistency()));
  }

  if (in.isSetTargetReplicaId()) {
    out.setReplicaId(in.getTargetReplicaId());
  }

  if (in.isSetFilterBytes()) {
    out.setFilter(filterFromThrift(in.getFilterBytes()));
  }

  return out;
}
 

@Override
public int scannerOpenWithScan(ByteBuffer tableName, TScan tScan,
    Map<ByteBuffer, ByteBuffer> attributes)
    throws IOError {

  Table table = null;
  try {
    table = getTable(tableName);
    Scan scan = new Scan();
    addAttributes(scan, attributes);
    if (tScan.isSetStartRow()) {
      scan.withStartRow(tScan.getStartRow());
    }
    if (tScan.isSetStopRow()) {
      scan.withStopRow(tScan.getStopRow());
    }
    if (tScan.isSetTimestamp()) {
      scan.setTimeRange(0, tScan.getTimestamp());
    }
    if (tScan.isSetCaching()) {
      scan.setCaching(tScan.getCaching());
    }
    if (tScan.isSetBatchSize()) {
      scan.setBatch(tScan.getBatchSize());
    }
    if (tScan.isSetColumns() && !tScan.getColumns().isEmpty()) {
      for(ByteBuffer column : tScan.getColumns()) {
        byte [][] famQf = CellUtil.parseColumn(getBytes(column));
        if(famQf.length == 1) {
          scan.addFamily(famQf[0]);
        } else {
          scan.addColumn(famQf[0], famQf[1]);
        }
      }
    }
    if (tScan.isSetFilterString()) {
      ParseFilter parseFilter = new ParseFilter();
      scan.setFilter(
          parseFilter.parseFilterString(tScan.getFilterString()));
    }
    if (tScan.isSetReversed()) {
      scan.setReversed(tScan.isReversed());
    }
    if (tScan.isSetCacheBlocks()) {
      scan.setCacheBlocks(tScan.isCacheBlocks());
    }
    return addScanner(table.getScanner(scan), tScan.sortColumns);
  } catch (IOException e) {
    LOG.warn(e.getMessage(), e);
    throw getIOError(e);
  } finally{
    closeTable(table);
  }
}
 

public static Filter createFilterFromArguments(ArrayList<byte[]> filterArguments) {
  byte[] prefix = ParseFilter.removeQuotesFromByteArray(filterArguments.get(0));
  return new CustomFilter(prefix);
}
 

protected ResultScanner getResults(final Table table, final String startRow, final String endRow, final String filterExpression, final Long timerangeMin, final Long timerangeMax,
        final Integer limitRows, final Boolean isReversed, final Boolean blockCache, final Collection<Column> columns, List<String> authorizations)  throws IOException {
    final Scan scan = new Scan();
    if (!StringUtils.isBlank(startRow)){
        scan.setStartRow(startRow.getBytes(StandardCharsets.UTF_8));
    }
    if (!StringUtils.isBlank(endRow)){
        scan.setStopRow(   endRow.getBytes(StandardCharsets.UTF_8));
    }

    if (authorizations != null && authorizations.size() > 0) {
        scan.setAuthorizations(new Authorizations(authorizations));
    }

    Filter filter = null;
    if (columns != null) {
        for (Column col : columns) {
            if (col.getQualifier() == null) {
                scan.addFamily(col.getFamily());
            } else {
                scan.addColumn(col.getFamily(), col.getQualifier());
            }
        }
    }
    if (!StringUtils.isBlank(filterExpression)) {
        ParseFilter parseFilter = new ParseFilter();
        filter = parseFilter.parseFilterString(filterExpression);
    }
    if (filter != null){
        scan.setFilter(filter);
    }

    if (timerangeMin != null && timerangeMax != null){
        scan.setTimeRange(timerangeMin, timerangeMax);
    }

    // ->>> reserved for HBase v 2 or later
    //if (limitRows != null && limitRows > 0){
    //    scan.setLimit(limitRows)
    //}

    if (isReversed != null){
        scan.setReversed(isReversed);
    }

    scan.setCacheBlocks(blockCache);

    return table.getScanner(scan);
}
 

protected ResultScanner getResults(final Table table, final String startRow, final String endRow, final String filterExpression, final Long timerangeMin, final Long timerangeMax,
        final Integer limitRows, final Boolean isReversed, final Boolean blockCache, final Collection<Column> columns, List<String> authorizations)  throws IOException {
    final Scan scan = new Scan();
    if (!StringUtils.isBlank(startRow)){
        scan.setStartRow(startRow.getBytes(StandardCharsets.UTF_8));
    }
    if (!StringUtils.isBlank(endRow)){
        scan.setStopRow(   endRow.getBytes(StandardCharsets.UTF_8));
    }

    if (authorizations != null && authorizations.size() > 0) {
        scan.setAuthorizations(new Authorizations(authorizations));
    }

    Filter filter = null;
    if (columns != null) {
        for (Column col : columns) {
            if (col.getQualifier() == null) {
                scan.addFamily(col.getFamily());
            } else {
                scan.addColumn(col.getFamily(), col.getQualifier());
            }
        }
    }
    if (!StringUtils.isBlank(filterExpression)) {
        ParseFilter parseFilter = new ParseFilter();
        filter = parseFilter.parseFilterString(filterExpression);
    }
    if (filter != null){
        scan.setFilter(filter);
    }

    if (timerangeMin != null && timerangeMax != null){
        scan.setTimeRange(timerangeMin, timerangeMax);
    }

    // ->>> reserved for HBase v 2 or later
    //if (limitRows != null && limitRows > 0){
    //    scan.setLimit(limitRows)
    //}

    if (isReversed != null){
        scan.setReversed(isReversed);
    }

    scan.setCacheBlocks(blockCache);

    return table.getScanner(scan);
}
 

public void doTestFilterRegistration() throws Exception {
  Configuration conf = UTIL.getConfiguration();

  conf.set("hbase.thrift.filters", "MyFilter:filterclass");

  ThriftServer.registerFilters(conf);

  Map<String, String> registeredFilters = ParseFilter.getAllFilters();

  assertEquals("filterclass", registeredFilters.get("MyFilter"));
}