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

private Result scanAfterBulkLoad(ResultScanner scanner, Result result, String expctedVal)
    throws IOException {
  while (result != null) {
    List<Cell> cells = result.getColumnCells(Bytes.toBytes("col"), Bytes.toBytes("q"));
    for (Cell _c : cells) {
      if (Bytes.toString(_c.getRowArray(), _c.getRowOffset(), _c.getRowLength())
          .equals("row1")) {
        System.out
            .println(Bytes.toString(_c.getRowArray(), _c.getRowOffset(), _c.getRowLength()));
        System.out.println(Bytes.toString(_c.getQualifierArray(), _c.getQualifierOffset(),
          _c.getQualifierLength()));
        System.out.println(
          Bytes.toString(_c.getValueArray(), _c.getValueOffset(), _c.getValueLength()));
        Assert.assertEquals(expctedVal,
          Bytes.toString(_c.getValueArray(), _c.getValueOffset(), _c.getValueLength()));
      }
    }
    result = scanner.next();
  }
  return result;
}
 

private List<String> verifyRunStatusFromResultTable(Connection conn, Long scn, String indexTable, int totalRows, List<String> expectedStatus) throws SQLException, IOException {
    Table hIndexToolTable = conn.unwrap(PhoenixConnection.class).getQueryServices()
            .getTable(RESULT_TABLE_NAME_BYTES);
    Assert.assertEquals(totalRows, TestUtil.getRowCount(hIndexToolTable, false));
    List<String> output = new ArrayList<>();
    Scan s = new Scan();
    s.setRowPrefixFilter(Bytes.toBytes(String.format("%s%s%s", scn, ROW_KEY_SEPARATOR, indexTable)));
    ResultScanner rs = hIndexToolTable.getScanner(s);
    int count =0;
    for(Result r : rs) {
        Assert.assertTrue(r != null);
        List<Cell> cells = r.getColumnCells(RESULT_TABLE_COLUMN_FAMILY, INDEX_TOOL_RUN_STATUS_BYTES);
        Assert.assertEquals(cells.size(), 1);
        Assert.assertTrue(Bytes.toString(CellUtil.cloneRow(cells.get(0))).startsWith(String.valueOf(scn)));
        output.add(Bytes.toString(CellUtil.cloneValue(cells.get(0))));
        count++;
    }
    //for each region
    Assert.assertEquals(3, count);
    for(int i=0; i< count; i++) {
        Assert.assertEquals(expectedStatus.get(i), output.get(i));
    }
    return output;
}
 

/**
 * @return Count of replicas found reading hbase:meta Region row or zk if
 *   asking about the hbase:meta table itself..
 */
private int getReplicaCountInMeta(Connection connection, int regionReplicaCount,
    List<RegionInfo> regionsOfTable) throws IOException {
  Result r = MetaTableAccessor.getCatalogFamilyRow(connection, regionsOfTable.get(0));
  int replicasFound = 0;
  for (int i = 1; i < regionReplicaCount; i++) {
    // Since we have already added the entries to the META we will be getting only that here
    List<Cell> columnCells =
        r.getColumnCells(HConstants.CATALOG_FAMILY, CatalogFamilyFormat.getServerColumn(i));
    if (!columnCells.isEmpty()) {
      replicasFound++;
    }
  }
  return replicasFound;
}
 

private void checkResult(Result r, byte[] fam, byte[] col, byte[] ... vals) {
  assertEquals(r.size(), vals.length);
  List<Cell> kvs = r.getColumnCells(fam, col);
  assertEquals(kvs.size(), vals.length);
  for (int i=0;i<vals.length;i++) {
    assertArrayEquals(CellUtil.cloneValue(kvs.get(i)), vals[i]);
  }
}
 

private void checkResult(Result r, byte[] col, byte[] ... vals) {
  assertEquals(vals.length, r.size());
  List<Cell> kvs = r.getColumnCells(col, col);
  assertEquals(kvs.size(), vals.length);
  for (int i=0;i<vals.length;i++) {
    String expected = Bytes.toString(vals[i]);
    String actual = Bytes.toString(CellUtil.cloneValue(kvs.get(i)));
    assertTrue(expected + " was expected but doesn't match " + actual,
        CellUtil.matchingValue(kvs.get(i), vals[i]));
  }
}
 

private byte[] getMergedStats(
    final Result result,
    final boolean clientsideStatsMerge,
    final byte[] columnFamily,
    final byte[] columnQualifier) {
  final List<Cell> columnCells = result.getColumnCells(columnFamily, columnQualifier);
  if ((columnCells.size() == 1)) {
    return CellUtil.cloneValue(columnCells.get(0));
  }

  return URLClassloaderUtils.toBinary(HBaseUtils.getMergedStats(columnCells));
}
 

private void testFlushFile(TableDescriptorBuilder.ModifyableTableDescriptor tableDescriptor)
    throws Exception {
  Table table = null;
  try {
    table = TEST_UTIL.createTable(tableDescriptor, null);

    //put data
    Put put0 = new Put(row1);
    put0.addColumn(family, qf1, 1, value1);
    table.put(put0);

    //put more data
    Put put1 = new Put(row2);
    put1.addColumn(family, qf2, 1, value2);
    table.put(put1);

    //flush
    TEST_UTIL.flush(tableDescriptor.getTableName());

    //Scan
    Scan scan = new Scan();
    scan.addColumn(family, qf1);
    scan.readVersions(4);
    ResultScanner scanner = table.getScanner(scan);

    //Compare
    int size = 0;
    for (Result result : scanner) {
      size++;
      List<Cell> cells = result.getColumnCells(family, qf1);
      // Verify the cell size
      Assert.assertEquals(1, cells.size());
      // Verify the value
      Assert.assertArrayEquals(value1, CellUtil.cloneValue(cells.get(0)));
    }
    scanner.close();
    Assert.assertEquals(1, size);
  } finally {
    table.close();
  }
}
 

@Test
public void testBulkLoad() throws Exception {
  final TableName tableName = TableName.valueOf(name.getMethodName());
  long l = System.currentTimeMillis();
  Admin admin = TEST_UTIL.getAdmin();
  createTable(admin, tableName);
  Scan scan = createScan();
  final Table table = init(admin, l, scan, tableName);
  // use bulkload
  final Path hfilePath = writeToHFile(l, "/temp/testBulkLoad/", "/temp/testBulkLoad/col/file",
    false);
  Configuration conf = TEST_UTIL.getConfiguration();
  conf.setBoolean("hbase.mapreduce.bulkload.assign.sequenceNumbers", true);
  BulkLoadHFiles.create(conf).bulkLoad(tableName, hfilePath);
  ResultScanner scanner = table.getScanner(scan);
  Result result = scanner.next();
  result = scanAfterBulkLoad(scanner, result, "version2");
  Put put0 = new Put(Bytes.toBytes("row1"));
  put0.add(new KeyValue(Bytes.toBytes("row1"), Bytes.toBytes("col"), Bytes.toBytes("q"), l, Bytes
      .toBytes("version3")));
  table.put(put0);
  admin.flush(tableName);
  scanner = table.getScanner(scan);
  result = scanner.next();
  while (result != null) {
    List<Cell> cells = result.getColumnCells(Bytes.toBytes("col"), Bytes.toBytes("q"));
    for (Cell _c : cells) {
      if (Bytes.toString(_c.getRowArray(), _c.getRowOffset(), _c.getRowLength())
          .equals("row1")) {
        System.out
            .println(Bytes.toString(_c.getRowArray(), _c.getRowOffset(), _c.getRowLength()));
        System.out.println(Bytes.toString(_c.getQualifierArray(), _c.getQualifierOffset(),
          _c.getQualifierLength()));
        System.out.println(
          Bytes.toString(_c.getValueArray(), _c.getValueOffset(), _c.getValueLength()));
        Assert.assertEquals("version3",
          Bytes.toString(_c.getValueArray(), _c.getValueOffset(), _c.getValueLength()));
      }
    }
    result = scanner.next();
  }
  scanner.close();
  table.close();
}
 

@Test
public void testBulkLoadNativeHFile() throws Exception {
  final TableName tableName = TableName.valueOf(name.getMethodName());
  long l = System.currentTimeMillis();
  Admin admin = TEST_UTIL.getAdmin();
  createTable(admin, tableName);
  Scan scan = createScan();
  final Table table = init(admin, l, scan, tableName);
  // use bulkload
  final Path hfilePath = writeToHFile(l, "/temp/testBulkLoadNativeHFile/",
    "/temp/testBulkLoadNativeHFile/col/file", true);
  Configuration conf = TEST_UTIL.getConfiguration();
  conf.setBoolean("hbase.mapreduce.bulkload.assign.sequenceNumbers", true);
  BulkLoadHFiles.create(conf).bulkLoad(tableName, hfilePath);
  ResultScanner scanner = table.getScanner(scan);
  Result result = scanner.next();
  // We had 'version0', 'version1' for 'row1,col:q' in the table.
  // Bulk load added 'version2'  scanner should be able to see 'version2'
  result = scanAfterBulkLoad(scanner, result, "version2");
  Put put0 = new Put(Bytes.toBytes("row1"));
  put0.add(new KeyValue(Bytes.toBytes("row1"), Bytes.toBytes("col"), Bytes.toBytes("q"), l, Bytes
      .toBytes("version3")));
  table.put(put0);
  admin.flush(tableName);
  scanner = table.getScanner(scan);
  result = scanner.next();
  while (result != null) {
    List<Cell> cells = result.getColumnCells(Bytes.toBytes("col"), Bytes.toBytes("q"));
    for (Cell _c : cells) {
      if (Bytes.toString(_c.getRowArray(), _c.getRowOffset(), _c.getRowLength())
          .equals("row1")) {
        System.out
            .println(Bytes.toString(_c.getRowArray(), _c.getRowOffset(), _c.getRowLength()));
        System.out.println(Bytes.toString(_c.getQualifierArray(), _c.getQualifierOffset(),
          _c.getQualifierLength()));
        System.out.println(
          Bytes.toString(_c.getValueArray(), _c.getValueOffset(), _c.getValueLength()));
        Assert.assertEquals("version3",
          Bytes.toString(_c.getValueArray(), _c.getValueOffset(), _c.getValueLength()));
      }
    }
    result = scanner.next();
  }
  scanner.close();
  table.close();
}
 

@Test
public void testMappingHbaseTableToPhoenixTable() throws Exception {
    String mtest = generateUniqueName();
    final TableName tableName = TableName.valueOf(mtest);
    Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
    PhoenixConnection conn = DriverManager.getConnection(getUrl(), props).unwrap(PhoenixConnection.class);
    
    Admin admin = conn.getQueryServices().getAdmin();
    try {
        // Create table then get the single region for our new table.
        TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName);
        builder.addColumnFamily(ColumnFamilyDescriptorBuilder.of(Bytes.toBytes("cf1")))
                .addColumnFamily(ColumnFamilyDescriptorBuilder.of(Bytes.toBytes("cf2")));
        admin.createTable(builder.build());
        Table t = conn.getQueryServices().getTable(Bytes.toBytes(mtest));
        insertData(tableName.getName(), admin, t);
        t.close();
        // create phoenix table that maps to existing HBase table
        createPhoenixTable(mtest);
        
        String selectSql = "SELECT * FROM " + mtest;
        ResultSet rs = conn.createStatement().executeQuery(selectSql);
        ResultSetMetaData rsMetaData = rs.getMetaData();
        assertTrue("Expected single row", rs.next());
        // verify values from cf2 is not returned
        assertEquals("Number of columns", 2, rsMetaData.getColumnCount());
        assertEquals("Column Value", "value1", rs.getString(2));
        assertFalse("Expected single row ", rs.next());
        
        // delete the row
        String deleteSql = "DELETE FROM " + mtest + " WHERE id = 'row'";
        conn.createStatement().executeUpdate(deleteSql);
        conn.commit();
        
        // verify that no rows are returned when querying through phoenix
        rs = conn.createStatement().executeQuery(selectSql);
        assertFalse("Expected no row` ", rs.next());
        
        // verify that row with value for cf2 still exists when using hbase apis
        Scan scan = new Scan();
        ResultScanner results = t.getScanner(scan);
        Result result = results.next();
        assertNotNull("Expected single row", result);
        List<Cell> kvs = result.getColumnCells(Bytes.toBytes("cf2"), Bytes.toBytes("q2"));
        assertEquals("Expected single value ", 1, kvs.size());
        assertEquals("Column Value", "value2", Bytes.toString(kvs.get(0).getValueArray(), kvs.get(0).getValueOffset(), kvs.get(0).getValueLength()));
        assertNull("Expected single row", results.next());
    } finally {
        admin.close();
    }
}