org.apache.hadoop.hbase.client.Result#rawCells ( )源码实例Demo

public long incrementValue(Result result, ValueOp op) throws SQLException {
    // In this case, we don't definitely know the timestamp of the deleted sequence,
    // but we know anything older is likely deleted. Worse case, we remove a sequence
    // from the cache that we shouldn't have which will cause a gap in sequence values.
    // In that case, we might get an error that a curr value was done on a sequence
    // before a next val was. Not sure how to prevent that.
    if (result.rawCells().length == 1) {
        Cell errorKV = result.rawCells()[0];
        int errorCode = PInteger.INSTANCE.getCodec().decodeInt(errorKV.getValueArray(), errorKV.getValueOffset(), SortOrder.getDefault());
        SQLExceptionCode code = SQLExceptionCode.fromErrorCode(errorCode);
        // TODO: We could have the server return the timestamps of the
        // delete markers and we could insert them here, but this seems
        // like overkill.
        // if (code == SQLExceptionCode.SEQUENCE_UNDEFINED) {
        // }
        throw new SQLExceptionInfo.Builder(code)
            .setSchemaName(key.getSchemaName())
            .setTableName(key.getSequenceName())
            .build().buildException();
    }
    // If we found the sequence, we update our cache with the new value
    SequenceValue value = new SequenceValue(result, op);
    insertSequenceValue(value);
    return increment(value, op);
}
 

public boolean returnValue(Result result) throws SQLException {
    Cell statusKV = result.rawCells()[0];
    if (statusKV.getValueLength() == 0) { // No error, but unable to return sequence values
        return false;
    }
    long timestamp = statusKV.getTimestamp();
    int statusCode = PInteger.INSTANCE.getCodec().decodeInt(statusKV.getValueArray(), statusKV.getValueOffset(), SortOrder.getDefault());
    if (statusCode == SUCCESS) {  // Success - update nextValue down to currentValue
        SequenceValue value = findSequenceValue(timestamp);
        if (value == null) {
            throw new EmptySequenceCacheException(key.getSchemaName(),key.getSequenceName());
        }
        return true;
    }
    SQLExceptionCode code = SQLExceptionCode.fromErrorCode(statusCode);
    // TODO: We could have the server return the timestamps of the
    // delete markers and we could insert them here, but this seems
    // like overkill.
    // if (code == SQLExceptionCode.SEQUENCE_UNDEFINED) {
    // }
    throw new SQLExceptionInfo.Builder(code)
        .setSchemaName(key.getSchemaName())
        .setTableName(key.getSequenceName())
        .build().buildException();
}
 

private void assertICV(byte [] row,
                       byte [] familiy,
                       byte[] qualifier,
                       long amount,
                       boolean fast) throws IOException {
  // run a get and see?
  Get get = new Get(row);
  if (fast) get.setIsolationLevel(IsolationLevel.READ_UNCOMMITTED);
  get.addColumn(familiy, qualifier);
  Result result = region.get(get);
  assertEquals(1, result.size());

  Cell kv = result.rawCells()[0];
  long r = Bytes.toLong(CellUtil.cloneValue(kv));
  assertEquals(amount, r);
}
 

/**
 * print info for Result
 *
 * @param r
 */
public static void printResultInfo(Result r) {
	System.out.print(">>>> cell rowkey= [" + new String(r.getRow()) + "]");
	for (Cell cell : r.rawCells()) {
		System.out.print(">>>> cell rowkey= " + new String(CellUtil.cloneRow(cell)));
		System.out.print(",family= " + new String(CellUtil.cloneFamily(cell)) + ":" + new String(CellUtil.cloneQualifier(cell)));
		System.out.println(", value= [" + new String(CellUtil.cloneValue(cell)) + "]");
	}
}
 

@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);
        }
    }
}
 

private void verifyRows(Table t, byte[] family, byte[] column) throws IOException {
  for (int i = 0; i < 10; i++) {
    byte[] row = Bytes.toBytes("row" + i);
    Get g = new Get(row).addFamily(family);
    Result r = t.get(g);
    Assert.assertNotNull(r);
    Assert.assertEquals(1, r.size());
    Cell cell = r.rawCells()[0];
    Assert.assertTrue(CellUtil.matchingQualifier(cell, column));
    Assert.assertEquals(Bytes.compareTo(cell.getValueArray(), cell.getValueOffset(),
      cell.getValueLength(), row, 0, row.length), 0);
  }
}
 

/**
 * Parser method returning all Statements from a single HBase Scan Result
 * @param res HBase Scan Result
 * @param vf ValueFactory to construct Statement and its Values
 * @return List of Statements
 */
public static List<Statement> parseStatements(Result res, ValueFactory vf) {
	// multiple triples may have the same hash (i.e. row key)
    ArrayList<Statement> st = new ArrayList<>();
    if (res.rawCells() != null) for (Cell c : res.rawCells()) {
        st.add(parseStatement(c, vf));
    }
    return st;
}
 

@Test
public void testPartialResultWhenRegionMove() throws IOException {
  Table table = createTestTable(TableName.valueOf(name.getMethodName()),
      ROWS, FAMILIES, QUALIFIERS, VALUE);

  moveRegion(table, 1);

  Scan scan = new Scan();
  scan.setMaxResultSize(1);
  scan.setAllowPartialResults(true);
  ResultScanner scanner = table.getScanner(scan);
  for (int i = 0; i < NUM_FAMILIES * NUM_QUALIFIERS - 1; i++) {
    scanner.next();
  }
  Result result1 = scanner.next();
  assertEquals(1, result1.rawCells().length);
  Cell c1 = result1.rawCells()[0];
  assertCell(c1, ROWS[0], FAMILIES[NUM_FAMILIES - 1], QUALIFIERS[NUM_QUALIFIERS - 1]);
  assertFalse(result1.mayHaveMoreCellsInRow());

  moveRegion(table, 2);

  Result result2 = scanner.next();
  assertEquals(1, result2.rawCells().length);
  Cell c2 = result2.rawCells()[0];
  assertCell(c2, ROWS[1], FAMILIES[0], QUALIFIERS[0]);
  assertTrue(result2.mayHaveMoreCellsInRow());

  moveRegion(table, 3);

  Result result3 = scanner.next();
  assertEquals(1, result3.rawCells().length);
  Cell c3 = result3.rawCells()[0];
  assertCell(c3, ROWS[1], FAMILIES[0], QUALIFIERS[1]);
  assertTrue(result3.mayHaveMoreCellsInRow());

}
 

private static void assertResult(Result result, byte[] expectedValue) {
  assertFalse(result.isEmpty());
  for (Cell c : result.rawCells()) {
    assertTrue(c.toString(), Bytes.equals(ROW, CellUtil.cloneRow(c)));
    assertTrue(c.toString(), Bytes.equals(FAMILY, CellUtil.cloneFamily(c)));
    assertTrue(c.toString(), Bytes.equals(expectedValue, CellUtil.cloneValue(c)));
  }
}
 

@Override
public List<String> mapRow(Result result, int rowNum) throws Exception {
    if (result.isEmpty()) {
        return Collections.emptyList();
    }
    final Cell[] rawCells = result.rawCells();
    final List<String> agentIdList = new ArrayList<>(rawCells.length);

    for (Cell cell : rawCells) {
        final String agentId = CellUtils.qualifierToString(cell);
        agentIdList.add(agentId);
    }

    return agentIdList;
}
 

@Override
public void scan(final String tableName, final byte[] startRow, final byte[] endRow, final Collection<Column> columns, List<String> authorizations, final ResultHandler handler)
        throws IOException {

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

        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 List<AcceptApplication> mapRow(Result result, int rowNum) throws Exception {
        if (result.isEmpty()) {
            return Collections.emptyList();
        }
//       readRowKey(result.getRow());

        final List<AcceptApplication> acceptApplicationList = new ArrayList<>(result.size());
        for (Cell cell : result.rawCells()) {
            AcceptApplication acceptedApplication = createAcceptedApplication(cell);
            acceptApplicationList.add(acceptedApplication);
        }
        return acceptApplicationList;
    }
 

@Test
public static void test() throws IOException {
    String hbaseUrl = "hbase"; // use hbase-site.xml on classpath
    HConnection conn = null;
    HTableInterface table = null;
    try {
        conn = HBaseConnection.get(hbaseUrl);
        table = conn.getTable("KYLIN_II_YTYWP3CQGJ");
        ResultScanner scanner = table.getScanner(CF, QN);
        StringBuffer sb = new StringBuffer();
        while (true) {
            Result r = scanner.next();
            if (r == null)
                break;

            Cell[] cells = r.rawCells();
            Cell c = cells[0];

            k.set(c.getRowArray(), c.getRowOffset(), c.getRowLength());
            v.set(c.getValueArray(), c.getValueOffset(), c.getValueLength());

            byte[] row = k.copyBytes();
            byte[] value = v.copyBytes();
            //                byte[] row = r.getRow();
            //                byte[] value = r.getValue(CF, QN);
            //
            sb.append("row length: " + row.length + "\r\n");
            sb.append(BytesUtil.toReadableText(row) + "\r\n");
            sb.append("value length: " + value.length + "\r\n");
            sb.append(BytesUtil.toReadableText(value) + "\r\n");
        }
        System.out.println(sb.toString());
        FileUtils.writeStringToFile(new File("/Users/honma/Desktop/a3"), sb.toString());
    } catch (IOException e) {
        e.printStackTrace();
    } finally {
        if (table != null)
            table.close();
        if (conn != null)
            conn.close();
    }

}
 

@Override
public boolean advanceNextPosition() {
    String colName = null;
    try {
        // if we got error when reading data, return false to end this reading.
        if (results == null) {
            return false;
        } else if (this.currentRecordIndex >= this.results.length) {
            InetAddress localhost = InetAddress.getLocalHost();
            // Random printing
            if (System.currentTimeMillis() % SYSTEMOUT_INTERVAL == 0) {
                log.info("BATCH GET RECORD. tableName=" + split.getTableName()
                        + ", rowKey_0=" + split.getConstraint().get(0) + ", READ_DATA_TIME="
                        + (System.currentTimeMillis() - startTime) + " mill secs. recordCount=" + recordCount
                        + ", startTime=" + new Date(startTime).toString() + ", localhost=" + localhost.getHostAddress()
                        + ", specified worker ip: "
                        + (split.getAddresses().size() > 0 ? split.getAddresses().get(0).toString() : ""));
            }
            return false;
        } else {
            fields = new Object[this.columnHandles.size()];
            ordinalPositionAndFieldsIndexMap.clear();
            int fieldIndex = 0;

            Result record = this.results[this.currentRecordIndex];

            for (Cell cell : record.rawCells()) {
                colName = Bytes.toString(
                        arrayCopy(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
                HBaseColumnHandle hch = fieldIndexMap.get(colName.hashCode());
                if (hch == null) {
                    continue;
                }
                Object value = matchValue(hch.getColumnType(),
                        arrayCopy(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                fields[fieldIndex] = value;
                ordinalPositionAndFieldsIndexMap.put(hch.getOrdinalPosition(), fieldIndex);
                fieldIndex++;
            }

            // Handle the value of rowKey column
            setRowKeyValue2FieldsAry(record, fieldIndex);

            this.currentRecordIndex++;
            return true;
        }
    } catch (Exception ex) {
        log.error(ex, ex.getMessage());
        this.close();
        log.error("fieldIndexMap.size=" + fieldIndexMap.size() + ", ERROR ColName=" + colName);
        fieldIndexMap.forEach((cName, columnHandle) ->
                log.error("fieldIndexMap: key=" + cName + ", hch.toString=" + columnHandle.toString())
        );
        return false;
    }
}
 

@Test
public void testResettingCounters() throws Exception {
  HBaseTestingUtility htu = new HBaseTestingUtility();
  Configuration conf = htu.getConfiguration();
  FileSystem fs = FileSystem.get(conf);
  byte [] table = Bytes.toBytes(name.getMethodName());
  byte [][] families = new byte [][] {
      Bytes.toBytes("family1"),
      Bytes.toBytes("family2"),
      Bytes.toBytes("family3")
  };
  int numQualifiers = 10;
  byte [][] qualifiers = new byte [numQualifiers][];
  for (int i=0; i<numQualifiers; i++) qualifiers[i] = Bytes.toBytes("qf" + i);
  int numRows = 10;
  byte [][] rows = new byte [numRows][];
  for (int i=0; i<numRows; i++) rows[i] = Bytes.toBytes("r" + i);

  TableDescriptorBuilder.ModifyableTableDescriptor tableDescriptor =
    new TableDescriptorBuilder.ModifyableTableDescriptor(TableName.valueOf(table));
  for (byte[] family : families) {
    tableDescriptor.setColumnFamily(
      new ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor(family));
  }

  RegionInfo hri = RegionInfoBuilder.newBuilder(tableDescriptor.getTableName()).build();
  String testDir = htu.getDataTestDir() + "/TestResettingCounters/";
  Path path = new Path(testDir);
  if (fs.exists(path)) {
    if (!fs.delete(path, true)) {
      throw new IOException("Failed delete of " + path);
    }
  }
  HRegion region = HBaseTestingUtility.createRegionAndWAL(hri, path, conf, tableDescriptor);
  try {
    Increment odd = new Increment(rows[0]);
    odd.setDurability(Durability.SKIP_WAL);
    Increment even = new Increment(rows[0]);
    even.setDurability(Durability.SKIP_WAL);
    Increment all = new Increment(rows[0]);
    all.setDurability(Durability.SKIP_WAL);
    for (int i=0;i<numQualifiers;i++) {
      if (i % 2 == 0) even.addColumn(families[0], qualifiers[i], 1);
      else odd.addColumn(families[0], qualifiers[i], 1);
      all.addColumn(families[0], qualifiers[i], 1);
    }

    // increment odd qualifiers 5 times and flush
    for (int i=0;i<5;i++) region.increment(odd, HConstants.NO_NONCE, HConstants.NO_NONCE);
    region.flush(true);

    // increment even qualifiers 5 times
    for (int i=0;i<5;i++) region.increment(even, HConstants.NO_NONCE, HConstants.NO_NONCE);

    // increment all qualifiers, should have value=6 for all
    Result result = region.increment(all, HConstants.NO_NONCE, HConstants.NO_NONCE);
    assertEquals(numQualifiers, result.size());
    Cell[] kvs = result.rawCells();
    for (int i=0;i<kvs.length;i++) {
      System.out.println(kvs[i].toString());
      assertTrue(CellUtil.matchingQualifier(kvs[i], qualifiers[i]));
      assertEquals(6, Bytes.toLong(CellUtil.cloneValue(kvs[i])));
    }
  } finally {
    HBaseTestingUtility.closeRegionAndWAL(region);
  }
  HBaseTestingUtility.closeRegionAndWAL(region);
}
 

public void testOrderingOfCellsInPartialResults(final Scan basePartialScan) throws Exception {
  // Scan that retrieves results in pieces (partials). By setting allowPartialResults to be true
  // the results will NOT be reconstructed and instead the caller will see the partial results
  // returned by the server
  Scan partialScan = new Scan(basePartialScan);
  partialScan.setAllowPartialResults(true);
  ResultScanner partialScanner = TABLE.getScanner(partialScan);

  // Scan that retrieves all table results in single RPC request
  Scan oneShotScan = new Scan(basePartialScan);
  oneShotScan.setMaxResultSize(Long.MAX_VALUE);
  oneShotScan.setCaching(ROWS.length);
  ResultScanner oneShotScanner = TABLE.getScanner(oneShotScan);

  Result oneShotResult = oneShotScanner.next();
  Result partialResult = null;
  int iterationCount = 0;

  while (oneShotResult != null && oneShotResult.rawCells() != null) {
    List<Cell> aggregatePartialCells = new ArrayList<>();
    do {
      partialResult = partialScanner.next();
      assertTrue("Partial Result is null. iteration: " + iterationCount, partialResult != null);
      assertTrue("Partial cells are null. iteration: " + iterationCount,
          partialResult.rawCells() != null);

      for (Cell c : partialResult.rawCells()) {
        aggregatePartialCells.add(c);
      }
    } while (partialResult.mayHaveMoreCellsInRow());

    assertTrue("Number of cells differs. iteration: " + iterationCount,
        oneShotResult.rawCells().length == aggregatePartialCells.size());
    final Cell[] oneShotCells = oneShotResult.rawCells();
    for (int cell = 0; cell < oneShotCells.length; cell++) {
      Cell oneShotCell = oneShotCells[cell];
      Cell partialCell = aggregatePartialCells.get(cell);

      assertTrue("One shot cell was null", oneShotCell != null);
      assertTrue("Partial cell was null", partialCell != null);
      assertTrue("Cell differs. oneShotCell:" + oneShotCell + " partialCell:" + partialCell,
          oneShotCell.equals(partialCell));
    }

    oneShotResult = oneShotScanner.next();
    iterationCount++;
  }

  assertTrue(partialScanner.next() == null);

  partialScanner.close();
  oneShotScanner.close();
}
 

@Test (timeOut = 60_000)
public void testFilterCommitCacheNotInSnapshot() throws Throwable {
    String TEST_TABLE = "testFilterCommitCacheNotInSnapshot";
    byte[] rowName = Bytes.toBytes("row1");
    byte[] famName = Bytes.toBytes(TEST_FAMILY);

    createTableIfNotExists(TEST_TABLE, famName);
    TTable tt = new TTable(connection, TEST_TABLE);


    //add some uncommitted values
    Transaction tx1 = tm.begin();
    Put put = new Put(rowName);
    for (int i = 0; i < 200; ++i) {
        byte[] dataValue1 = Bytes.toBytes("some data");
        byte[] colName = Bytes.toBytes("col" + i);
        put.addColumn(famName, colName, dataValue1);
    }
    tt.put(tx1, put);

    //try to scan from tx
    Transaction tx = tm.begin();
    Table htable = connection.getTable(TableName.valueOf(TEST_TABLE));
    SnapshotFilterImpl snapshotFilter = spy(new SnapshotFilterImpl(new HTableAccessWrapper(htable, htable),
            tm.getCommitTableClient()));
    Filter newFilter = TransactionFilters.getVisibilityFilter(null,
            snapshotFilter, (HBaseTransaction) tx);

    Table rawTable = connection.getTable(TableName.valueOf(TEST_TABLE));

    Scan scan = new Scan();
    ResultScanner scanner = rawTable.getScanner(scan);

    for(Result row: scanner) {
        for(Cell cell: row.rawCells()) {
            newFilter.filterKeyValue(cell);
        }
    }
    verify(snapshotFilter, Mockito.times(1))
            .getTSIfInSnapshot(any(Cell.class),any(HBaseTransaction.class), any(Map.class));
    tt.close();
}
 

private Cell[] getRow(Table table, Get get) throws Exception {
  Result result = table.get(get);
  return result.rawCells();
}
 

private Cell[] getRow(Table table, Get get) throws Exception {
  Result result = table.get(get);
  return result.rawCells();
}
 

private void assertTargetDoDeletesFalse(int expectedRows, TableName sourceTableName,
    TableName targetTableName) throws Exception {
  Table sourceTable = TEST_UTIL.getConnection().getTable(sourceTableName);
  Table targetTable = TEST_UTIL.getConnection().getTable(targetTableName);

  ResultScanner sourceScanner = sourceTable.getScanner(new Scan());
  ResultScanner targetScanner = targetTable.getScanner(new Scan());
  Result targetRow = targetScanner.next();
  Result sourceRow = sourceScanner.next();
  int rowsCount = 0;
  while (targetRow != null) {
    rowsCount++;
    //only compares values for existing rows, skipping rows existing on
    //target only that were not deleted given --doDeletes=false
    if (Bytes.toInt(sourceRow.getRow()) != Bytes.toInt(targetRow.getRow())) {
      targetRow = targetScanner.next();
      continue;
    }

    LOG.debug("SOURCE row: " + (sourceRow == null ? "null"
        : Bytes.toInt(sourceRow.getRow()))
        + " cells:" + sourceRow);
    LOG.debug("TARGET row: " + (targetRow == null ? "null"
        : Bytes.toInt(targetRow.getRow()))
        + " cells:" + targetRow);

    Cell[] sourceCells = sourceRow.rawCells();
    Cell[] targetCells = targetRow.rawCells();
    int targetRowKey = Bytes.toInt(targetRow.getRow());
    if (targetRowKey >= 70 && targetRowKey < 80) {
      if (sourceCells.length == targetCells.length) {
        LOG.debug("Source cells: " + Arrays.toString(sourceCells));
        LOG.debug("Target cells: " + Arrays.toString(targetCells));
        Assert.fail("Row " + targetRowKey + " should have more cells in "
            + "target than in source");
      }

    } else {
      if (sourceCells.length != targetCells.length) {
        LOG.debug("Source cells: " + Arrays.toString(sourceCells));
        LOG.debug("Target cells: " + Arrays.toString(targetCells));
        Assert.fail("Row " + Bytes.toInt(sourceRow.getRow())
            + " has " + sourceCells.length
            + " cells in source table but " + targetCells.length
            + " cells in target table");
      }
    }
    for (int j = 0; j < sourceCells.length; j++) {
      Cell sourceCell = sourceCells[j];
      Cell targetCell = targetCells[j];
      try {
        if (!CellUtil.matchingRows(sourceCell, targetCell)) {
          Assert.fail("Rows don't match");
        }
        if (!CellUtil.matchingFamily(sourceCell, targetCell)) {
          Assert.fail("Families don't match");
        }
        if (!CellUtil.matchingQualifier(sourceCell, targetCell)) {
          Assert.fail("Qualifiers don't match");
        }
        if (targetRowKey < 80 && targetRowKey >= 90){
          if (!CellUtil.matchingTimestamp(sourceCell, targetCell)) {
            Assert.fail("Timestamps don't match");
          }
        }
        if (!CellUtil.matchingValue(sourceCell, targetCell)) {
          Assert.fail("Values don't match");
        }
      } catch (Throwable t) {
        LOG.debug("Source cell: " + sourceCell + " target cell: "
            + targetCell);
        Throwables.propagate(t);
      }
    }
    targetRow = targetScanner.next();
    sourceRow = sourceScanner.next();
  }
  assertEquals("Target expected rows does not match.",expectedRows,
      rowsCount);
  sourceScanner.close();
  targetScanner.close();
  sourceTable.close();
  targetTable.close();
}