/**
 * Just for debug
 */
@SuppressWarnings("unused")
private void dump(String table, Scan scan) throws IOException {
    System.out.println(String.format(">>>> scan table %s with %s",
                                     table, scan));
    RowIterator iterator = this.scan(table, scan);
    while (iterator.hasNext()) {
        Result row = iterator.next();
        System.out.println(StringEncoding.format(row.getRow()));
        CellScanner cellScanner = row.cellScanner();
        while (cellScanner.advance()) {
            Cell cell = cellScanner.current();
            byte[] key = CellUtil.cloneQualifier(cell);
            byte[] val = CellUtil.cloneValue(cell);
            System.out.println(String.format("  %s=%s",
                               StringEncoding.format(key),
                               StringEncoding.format(val)));
        }
    }
}
 

@Override
public List<String> getSearchClicks() {
	LOG.debug("Checking searchclicks table content!");
	Scan scan = new Scan();
	scan.addFamily(HbaseJsonEventSerializer.COLUMFAMILY_CLIENT_BYTES);
	scan.addFamily(HbaseJsonEventSerializer.COLUMFAMILY_SEARCH_BYTES);
	scan.addFamily(HbaseJsonEventSerializer.COLUMFAMILY_FILTERS_BYTES);
	List<String> rows = hbaseTemplate.find("searchclicks", scan,
			new RowMapper<String>() {
				@Override
				public String mapRow(Result result, int rowNum)
						throws Exception {
					return Arrays.toString(result.rawCells());
				}
			});
	for (String row : rows) {
		LOG.debug("searchclicks table content, Table returned row: {}", row);
	}
	LOG.debug("Checking searchclicks table content done!");
	return rows;
}
 

/**
 * Verifies that result contains all the key values within expKvList. Fails the test otherwise
 * @param result
 * @param expKvList
 * @param msg
 */
static void verifyResult(Result result, List<Cell> expKvList, String msg) {
  if (LOG.isInfoEnabled()) {
    LOG.info(msg);
    LOG.info("Expected count: " + expKvList.size());
    LOG.info("Actual count: " + result.size());
  }

  if (expKvList.isEmpty()) return;

  int i = 0;
  for (Cell kv : result.rawCells()) {
    if (i >= expKvList.size()) {
      break; // we will check the size later
    }

    Cell kvExp = expKvList.get(i++);
    assertTrue("Not equal. get kv: " + kv.toString() + " exp kv: " + kvExp.toString(),
        kvExp.equals(kv));
  }

  assertEquals(expKvList.size(), result.size());
}
 

/**
 * Delete prune upper bounds for the regions that are not in the given exclude set, and the
 * prune upper bound is less than the given value.
 * After the invalid list is pruned up to deletionPruneUpperBound, we do not need entries for regions that have
 * prune upper bound less than deletionPruneUpperBound. We however limit the deletion to only regions that are
 * no longer in existence (due to deletion, etc.), to avoid update/delete race conditions.
 *
 * @param deletionPruneUpperBound prune upper bound below which regions will be deleted
 * @param excludeRegions set of regions that should not be deleted
 * @throws IOException when not able to delete data in HBase
 */
public void deletePruneUpperBounds(long deletionPruneUpperBound, SortedSet<byte[]> excludeRegions)
  throws IOException {
  try (HTableInterface stateTable = stateTableSupplier.get()) {
    byte[] startRow = makeRegionKey(EMPTY_BYTE_ARRAY);
    Scan scan = new Scan(startRow, REGION_KEY_PREFIX_STOP);
    scan.addColumn(FAMILY, PRUNE_UPPER_BOUND_COL);

    try (ResultScanner scanner = stateTable.getScanner(scan)) {
      Result next;
      while ((next = scanner.next()) != null) {
        byte[] region = getRegionFromKey(next.getRow());
        if (!excludeRegions.contains(region)) {
          byte[] timeBytes = next.getValue(FAMILY, PRUNE_UPPER_BOUND_COL);
          if (timeBytes != null) {
            long pruneUpperBoundRegion = Bytes.toLong(timeBytes);
            if (pruneUpperBoundRegion < deletionPruneUpperBound) {
              stateTable.delete(new Delete(next.getRow()));
            }
          }
        }
      }
    }
  }
}
 

@Override
protected void parseRowColumns(Result row, BackendEntry entry,
                               Query query) throws IOException {
    /*
     * Collapse owner-vertex id from edge id, NOTE: unneeded to
     * collapse if BinarySerializer.keyWithIdPrefix set to true
     */
    byte[] key = row.getRow();
    key = Arrays.copyOfRange(key, entry.id().length(), key.length);

    long total = query.total();
    CellScanner cellScanner = row.cellScanner();
    while (cellScanner.advance() && total-- > 0) {
        Cell cell = cellScanner.current();
        assert CellUtil.cloneQualifier(cell).length == 0;
        entry.columns(BackendColumn.of(key, CellUtil.cloneValue(cell)));
    }
}
 

@Override
public List<TRowResult> scannerGetList(int id,int nbRows)
    throws IllegalArgument, IOError {
  LOG.debug("scannerGetList: id={}", id);
  ResultScannerWrapper resultScannerWrapper = getScanner(id);
  if (null == resultScannerWrapper) {
    String message = "scanner ID is invalid";
    LOG.warn(message);
    throw new IllegalArgument("scanner ID is invalid");
  }

  Result [] results;
  try {
    results = resultScannerWrapper.getScanner().next(nbRows);
    if (null == results) {
      return new ArrayList<>();
    }
  } catch (IOException e) {
    LOG.warn(e.getMessage(), e);
    throw getIOError(e);
  }
  return ThriftUtilities.rowResultFromHBase(results, resultScannerWrapper.isColumnSorted());
}
 

public void verifyNumericRows(Table table, final byte[] f, int startRow, int endRow,
    int replicaId)
    throws IOException {
  for (int i = startRow; i < endRow; i++) {
    String failMsg = "Failed verification of row :" + i;
    byte[] data = Bytes.toBytes(String.valueOf(i));
    Get get = new Get(data);
    get.setReplicaId(replicaId);
    get.setConsistency(Consistency.TIMELINE);
    Result result = table.get(get);
    assertTrue(failMsg, result.containsColumn(f, null));
    assertEquals(failMsg, 1, result.getColumnCells(f, null).size());
    Cell cell = result.getColumnLatestCell(f, null);
    assertTrue(failMsg,
      Bytes.equals(data, 0, data.length, cell.getValueArray(), cell.getValueOffset(),
        cell.getValueLength()));
  }
}
 

public Map<String, byte[]> get_(String ns, String tn, byte[] key) 
throws IOException {
    ns = (ns.equals(Orca.SYSNS)) ? this.sysns : ns;
        TableName tableName = TableName.valueOf(ns, tn);
    Result r = Helper.get(this.hbaseConnection, tableName, key);
    if (r.isEmpty()) {
        return null;
    }
    Map<String, byte[]> row = new HashMap<>();
    for (Map.Entry<byte[],NavigableMap<byte[],byte[]>> i:r.getNoVersionMap().entrySet()) {
        String cf =  new String(i.getKey());
        for (Map.Entry<byte[],byte[]> j:i.getValue().entrySet()) {
            String q = new String(j.getKey());
            row.put(cf + ":" + q, j.getValue());
        }
    }
    return row;
}
 

private void verifyRows(Table table, Get get, List<byte[]> expectedValues) throws Exception {
  Result result = table.get(get);
  if (expectedValues == null) {
    assertTrue(result.isEmpty());
  } else {
    assertFalse(result.isEmpty());
    byte[] family = TestBytes.family;
    byte[] col = TestBytes.qualifier;
    if (get.hasFamilies()) {
      family = get.getFamilyMap().keySet().iterator().next();
      col = get.getFamilyMap().get(family).first();
    }
    Iterator<Cell> it = result.getColumnCells(family, col).iterator();
    for (byte[] expectedValue : expectedValues) {
      Assert.assertTrue(it.hasNext());
      assertArrayEquals(expectedValue, CellUtil.cloneValue(it.next()));
    }
  }
}
 

@Test
public void testVisibilityLabelsOnWALReplay() throws Exception {
  final TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
  try (Table table = createTableAndWriteDataWithLabels(tableName,
      "(" + SECRET + "|" + CONFIDENTIAL + ")", PRIVATE)) {
    List<RegionServerThread> regionServerThreads = TEST_UTIL.getHBaseCluster()
        .getRegionServerThreads();
    for (RegionServerThread rsThread : regionServerThreads) {
      rsThread.getRegionServer().abort("Aborting ");
    }
    // Start one new RS
    RegionServerThread rs = TEST_UTIL.getHBaseCluster().startRegionServer();
    waitForLabelsRegionAvailability(rs.getRegionServer());
    Scan s = new Scan();
    s.setAuthorizations(new Authorizations(SECRET));
    ResultScanner scanner = table.getScanner(s);
    Result[] next = scanner.next(3);
    assertTrue(next.length == 1);
  }
}
 

private Map<FamCol, Integer> countColsInRows(ResultScanner rs, FamCol... famCols) throws IOException {
    Map<FamCol, Integer> colCount = new HashMap<>();
    Result r = rs.next();
    while (r != null) {
        for (FamCol col : famCols) {
            if (r.containsColumn(col.fam, col.col)) {
                Integer c = colCount.get(col);

                if (c == null) {
                    colCount.put(col, 1);
                } else {
                    colCount.put(col, c + 1);
                }
            }
        }
        r = rs.next();
    }
    return colCount;
}
 

private void verifyReplicationStuck() throws Exception {
  try (Table normalTable = utility1.getConnection().getTable(NORMAL_TABLE)) {
    Put put = new Put(ROW);
    put.addColumn(FAMILY, QUALIFIER, VALUE);
    normalTable.put(put);
  }
  try (Table normalTable = utility2.getConnection().getTable(NORMAL_TABLE)) {
    for (int i = 0; i < NB_RETRIES; i++) {
      Result result = normalTable.get(new Get(ROW).addColumn(FAMILY, QUALIFIER));
      if (result != null && !result.isEmpty()) {
        fail("Edit should have been stuck behind dropped tables, but value is " + Bytes
            .toString(result.getValue(FAMILY, QUALIFIER)));
      } else {
        LOG.info("Row not replicated, let's wait a bit more...");
        Thread.sleep(SLEEP_TIME);
      }
    }
  }
}
 

private void verifyRows(Table table, Get get, List<byte[]> expectedValues) throws Exception {
  Result result = table.get(get);
  if (expectedValues == null) {
    assertTrue(result.isEmpty());
  } else {
    assertFalse(result.isEmpty());
    byte[] family = TestBytes.family;
    byte[] col = TestBytes.qualifier;
    if (get.hasFamilies()) {
      family = get.getFamilyMap().keySet().iterator().next();
      col = get.getFamilyMap().get(family).first();
    }
    Iterator<Cell> it = result.getColumnCells(family, col).iterator();
    for (byte[] expectedValue : expectedValues) {
      Assert.assertTrue(it.hasNext());
      assertArrayEquals(expectedValue, CellUtil.cloneValue(it.next()));
    }
  }
}
 

@Override
public Result deserialize(Result mutation) throws IOException {
  int totalBuffer = in.readInt();
  if (totalBuffer == 0) {
    return Result.EMPTY_RESULT;
  }
  byte[] buf = new byte[totalBuffer];
  readChunked(in, buf, 0, totalBuffer);
  List<Cell> kvs = new ArrayList<>();
  int offset = 0;
  while (offset < totalBuffer) {
    int keyLength = Bytes.toInt(buf, offset);
    offset += Bytes.SIZEOF_INT;
    kvs.add(new KeyValue(buf, offset, keyLength));
    offset += keyLength;
  }
  return Result.create(kvs);
}
 

private String getValue(HBaseColumn hBaseColumn, Result result) {
  String value = null;
  if (result.isEmpty()) {
    return value;
  }
  if (!hBaseColumn.getCf().isPresent() || !hBaseColumn.getQualifier().isPresent()) {
    Map<String, String> columnMap = new HashMap<>();
    // parse column family, column, timestamp, and value
    for (Map.Entry<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> entry : result.getMap().entrySet()) {
      String columnFamily = Bytes.toString(entry.getKey());
      for (Map.Entry<byte[], NavigableMap<Long, byte[]>> cells : entry.getValue().entrySet()) {
        String column = Bytes.toString(cells.getKey());
        NavigableMap<Long, byte[]> cell = cells.getValue();
        Map.Entry<Long, byte[]> v = cell.lastEntry();
        String columnValue = Bytes.toString(v.getValue());
        columnMap.put(columnFamily + ":" + column, columnValue);
      }
    }
    JSONObject json = new JSONObject(columnMap);
    value = json.toString();
  } else {
    value = Bytes.toString(result.getValue(hBaseColumn.getCf().get(), hBaseColumn.getQualifier().get()));
  }
  return value;
}
 

/**
 * Invokes Table#get and prints out the contents of the retrieved row.
 *
 * @param table Standard Table object
 * @throws IOException If IO problem encountered
 */
static void getAndPrintRowContents(final Table table) throws IOException {

  Result row = table.get(new Get(MY_ROW_ID));

  System.out.println("Row [" + Bytes.toString(row.getRow())
          + "] was retrieved from Table ["
          + table.getName().getNameAsString()
          + "] in HBase, with the following content:");

  for (Entry<byte[], NavigableMap<byte[], byte[]>> colFamilyEntry
          : row.getNoVersionMap().entrySet()) {
    String columnFamilyName = Bytes.toString(colFamilyEntry.getKey());

    System.out.println("  Columns in Column Family [" + columnFamilyName
            + "]:");

    for (Entry<byte[], byte[]> columnNameAndValueMap
            : colFamilyEntry.getValue().entrySet()) {

      System.out.println("    Value of Column [" + columnFamilyName + ":"
              + Bytes.toString(columnNameAndValueMap.getKey()) + "] == "
              + Bytes.toString(columnNameAndValueMap.getValue()));
    }
  }
}
 

@Override
public Notification getNotification(String notificationId) {
    try {
        String tableName = notificationMapper.getTableName();
        LOG.debug("getting notification with notificationId {} from table {}", notificationId, tableName);
        Get get = new Get(notificationId.getBytes(StandardCharsets.UTF_8));
        Result result = tables.get(tableName).get().get(get);
        return result.isEmpty() ? null : notificationMapper.entity(result);
    } catch (IOException ex) {
        throw new NotificationStoreException("Error getting notification id: " + notificationId, ex);
    }
}
 

/**
 * Read the timestamp for each region server log after the last successful backup. Each table has
 * its own set of the timestamps. The info is stored for each table as a concatenated string of
 * rs->timestapmp
 * @param backupRoot root directory path to backup
 * @return the timestamp for each region server. key: tableName value:
 *         RegionServer,PreviousTimeStamp
 * @throws IOException exception
 */
public HashMap<TableName, HashMap<String, Long>> readLogTimestampMap(String backupRoot)
    throws IOException {
  if (LOG.isTraceEnabled()) {
    LOG.trace("read RS log ts from backup system table for root=" + backupRoot);
  }

  HashMap<TableName, HashMap<String, Long>> tableTimestampMap = new HashMap<>();

  Scan scan = createScanForReadLogTimestampMap(backupRoot);
  try (Table table = connection.getTable(tableName);
      ResultScanner scanner = table.getScanner(scan)) {
    Result res;
    while ((res = scanner.next()) != null) {
      res.advance();
      Cell cell = res.current();
      byte[] row = CellUtil.cloneRow(cell);
      String tabName = getTableNameForReadLogTimestampMap(row);
      TableName tn = TableName.valueOf(tabName);
      byte[] data = CellUtil.cloneValue(cell);
      if (data == null) {
        throw new IOException("Data of last backup data from backup system table "
            + "is empty. Create a backup first.");
      }
      if (data != null && data.length > 0) {
        HashMap<String, Long> lastBackup =
            fromTableServerTimestampProto(BackupProtos.TableServerTimestamp.parseFrom(data));
        tableTimestampMap.put(tn, lastBackup);
      }
    }
    return tableTimestampMap;
  }
}
 

private void verifyTableCount(final Table table, final int count) throws IOException {
    Scan scan = new Scan();
    scan.setMaxVersions(1);
    ResultScanner scanner = table.getScanner(scan);
    int i = 0;
    for (Result r : scanner) {
        i++;
    }
    assertEquals(count, i);
    scanner.close();
}
 

@Override
public void run() {
  for (int i = 0; i < numIncrements; i++) {
    try {
      Increment inc = new Increment(row);
      inc.addColumn(fam1, qual1, amount);
      inc.addColumn(fam1, qual2, amount*2);
      inc.addColumn(fam2, qual3, amount*3);
      inc.setDurability(Durability.ASYNC_WAL);
      Result result = region.increment(inc);
      if (result != null) {
        assertEquals(Bytes.toLong(result.getValue(fam1, qual1))*2,
          Bytes.toLong(result.getValue(fam1, qual2)));
        assertTrue(result.getValue(fam2, qual3) != null);
        assertEquals(Bytes.toLong(result.getValue(fam1, qual1))*3,
          Bytes.toLong(result.getValue(fam2, qual3)));
        assertEquals(Bytes.toLong(result.getValue(fam1, qual1))*2,
           Bytes.toLong(result.getValue(fam1, qual2)));
        long fam1Increment = Bytes.toLong(result.getValue(fam1, qual1))*3;
        long fam2Increment = Bytes.toLong(result.getValue(fam2, qual3));
        assertEquals("fam1=" + fam1Increment + ", fam2=" + fam2Increment,
          fam1Increment, fam2Increment);
      }
    } catch (IOException e) {
      e.printStackTrace();
    }
  }
}
 

/**
 * Pass the key, and reversed value to reduce
 */
public void map(ImmutableBytesWritable key, Result value,
  OutputCollector<ImmutableBytesWritable, Put> output,
  Reporter reporter)
throws IOException {
  output.collect(key, TestTableMapReduceBase.map(key, value));
}
 

@Override
public Result[] get(List<Get> gets) throws IOException {
  List<TGet> tGets = ThriftUtilities.getsFromHBase(gets);
  try {
    List<TResult> results = client.getMultiple(tableNameInBytes, tGets);
    return ThriftUtilities.resultsFromThrift(results);
  }  catch (TException e) {
    throw new IOException(e);
  }
}
 

@Test
public void testScan() throws Exception {
    ValueFactory vf = SimpleValueFactory.getInstance();

    try (ResultScanner rs = table.getScanner(HalyardTableUtils.scan(s == null ? null : vf.createIRI(s), p == null ? null : vf.createIRI(p), o == null ? null : vf.createLiteral(o), c == null ? null : vf.createIRI(c)))) {
        Set<Statement> res = new HashSet<>();
        Result r;
        while ((r = rs.next()) != null) {
            res.addAll(HalyardTableUtils.parseStatements(r, vf));
        }
        assertTrue(allStatements.containsAll(res));
        assertEquals(expRes, res.size());
    }
}
 

private long getCurrentBalance(int id) throws IOException {
  Result r = txTable.get(new Get(Bytes.toBytes(id)));
  byte[] balanceBytes = r.getValue(FAMILY, COL);
  if (balanceBytes == null) {
    return 0;
  }
  return Bytes.toLong(balanceBytes);
}
 

/**
 * {@inheritDoc}
 */
@Override
public Result[] get(List<Get> gets) throws IOException {
    List<Result> results = new ArrayList<Result>();
    for (Get g : gets) {
        results.add(get(g));
    }
    return results.toArray(new Result[results.size()]);
}
 

public static long[] getReplicationBarrier(Connection conn, byte[] regionName)
  throws IOException {
  try (Table table = getMetaHTable(conn)) {
    Result result = table.get(new Get(regionName)
      .addColumn(HConstants.REPLICATION_BARRIER_FAMILY, HConstants.SEQNUM_QUALIFIER)
      .readAllVersions());
    return getReplicationBarriers(result);
  }
}
 

private void gotFailure(byte[] expected, Result res) {
  StringBuilder msg = new StringBuilder();
  msg.append("Failed after ").append(numRowsScanned).append("!");
  msg.append("Expected=").append(Bytes.toStringBinary(expected));
  msg.append("Got:\n");
  for (Cell kv : res.listCells()) {
    msg.append(kv.toString());
    msg.append(" val= ");
    msg.append(Bytes.toStringBinary(CellUtil.cloneValue(kv)));
    msg.append("\n");
  }
  throw new RuntimeException(msg.toString());
}
 

@Test
public void testPutRowToTable() throws IOException {
  Admin admin = TEST_UTIL.getAdmin();
  admin.createNamespace(NamespaceDescriptor.create(HelloHBase.MY_NAMESPACE_NAME).build());
  Table table
          = TEST_UTIL.createTable(HelloHBase.MY_TABLE_NAME, HelloHBase.MY_COLUMN_FAMILY_NAME);

  HelloHBase.putRowToTable(table);
  Result row = table.get(new Get(HelloHBase.MY_ROW_ID));
  assertEquals("#putRowToTable failed to store row.", false, row.isEmpty());

  TEST_UTIL.deleteTable(HelloHBase.MY_TABLE_NAME);
  admin.deleteNamespace(HelloHBase.MY_NAMESPACE_NAME);
}
 

@Override
public GetResponse get(RpcController controller, GetRequest request)
throws ServiceException {
  byte[] regionName = request.getRegion().getValue().toByteArray();
  Map<byte [], Result> m = this.gets.get(regionName);
  GetResponse.Builder builder = GetResponse.newBuilder();
  if (m != null) {
    byte[] row = request.getGet().getRow().toByteArray();
    builder.setResult(ProtobufUtil.toResult(m.get(row)));
  }
  return builder.build();
}
 

/**
 * Check if WAL file is eligible for deletion Future: to support all backup destinations
 * @param file name of a file to check
 * @return true, if deletable, false otherwise.
 * @throws IOException exception
 */
// TODO: multiple backup destination support
public boolean isWALFileDeletable(String file) throws IOException {
  if (LOG.isTraceEnabled()) {
    LOG.trace("Check if WAL file has been already backed up in backup system table " + file);
  }
  try (Table table = connection.getTable(tableName)) {
    Get get = createGetForCheckWALFile(file);
    Result res = table.get(get);
    return (!res.isEmpty());
  }
}