taklwu commented on code in PR #158:
URL: https://github.com/apache/hbase-connectors/pull/158#discussion_r3669713864
##########
spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/HBaseContextSuite.scala:
##########
@@ -158,4 +159,261 @@ class HBaseContextSuite extends AnyFunSuite with
BeforeAndAfterAll with Logging
assert(rows(0).getString(0) == "value1")
assert(rows(0).get(1) == null)
}
+
+ test("bulkPut to test HBase client") {
+ val rdd = sc.parallelize(
+ Array[(Array[Byte], Array[(Array[Byte], Array[Byte], Array[Byte])])](
+ (
+ Bytes.toBytes("put1"),
+ Array((Bytes.toBytes(columnFamily), Bytes.toBytes("a"),
Bytes.toBytes("foo1")))),
+ (
+ Bytes.toBytes("put2"),
+ Array((Bytes.toBytes(columnFamily), Bytes.toBytes("b"),
Bytes.toBytes("foo2")))),
+ (
+ Bytes.toBytes("put3"),
+ Array((Bytes.toBytes(columnFamily), Bytes.toBytes("c"),
Bytes.toBytes("foo3"))))))
+
+ hbaseContext.bulkPut[(Array[Byte], Array[(Array[Byte], Array[Byte],
Array[Byte])])](
+ rdd,
+ TableName.valueOf(tableName),
+ (putRecord) => {
+ val put = new Put(putRecord._1)
+ putRecord._2.foreach((putValue) => put.addColumn(putValue._1,
putValue._2, putValue._3))
+ put
+ })
+
+ val connection =
ConnectionFactory.createConnection(TEST_UTIL.getConfiguration)
+ val table = connection.getTable(TableName.valueOf(tableName))
+ try {
+ val foo1 = Bytes.toString(
+ CellUtil.cloneValue(
+ table
+ .get(new Get(Bytes.toBytes("put1")))
+ .getColumnLatestCell(Bytes.toBytes(columnFamily),
Bytes.toBytes("a"))))
+ assert(foo1 == "foo1")
+
+ val foo2 = Bytes.toString(
+ CellUtil.cloneValue(
+ table
+ .get(new Get(Bytes.toBytes("put2")))
+ .getColumnLatestCell(Bytes.toBytes(columnFamily),
Bytes.toBytes("b"))))
+ assert(foo2 == "foo2")
+
+ val foo3 = Bytes.toString(
+ CellUtil.cloneValue(
+ table
+ .get(new Get(Bytes.toBytes("put3")))
+ .getColumnLatestCell(Bytes.toBytes(columnFamily),
Bytes.toBytes("c"))))
+ assert(foo3 == "foo3")
+ } finally {
+ table.close()
+ connection.close()
+ }
+ }
+
+ test("bulkDelete to test HBase client") {
+ val connection =
ConnectionFactory.createConnection(TEST_UTIL.getConfiguration)
+ val table = connection.getTable(TableName.valueOf(tableName))
+
+ try {
+ var put = new Put(Bytes.toBytes("delete1"))
+ put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("a"),
Bytes.toBytes("foo1"))
+ table.put(put)
+ put = new Put(Bytes.toBytes("delete2"))
+ put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("a"),
Bytes.toBytes("foo2"))
+ table.put(put)
+ put = new Put(Bytes.toBytes("delete3"))
+ put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("a"),
Bytes.toBytes("foo3"))
+ table.put(put)
+ } finally {
+ table.close()
+ connection.close()
+ }
+
+ val rdd = sc.parallelize(Array[Array[Byte]](Bytes.toBytes("delete1"),
Bytes.toBytes("delete3")))
+
+ hbaseContext.bulkDelete[Array[Byte]](
+ rdd,
+ TableName.valueOf(tableName),
+ putRecord => new Delete(putRecord),
+ 4)
+
+ val connection2 =
ConnectionFactory.createConnection(TEST_UTIL.getConfiguration)
+ val table2 = connection2.getTable(TableName.valueOf(tableName))
+ try {
+ assert(
+ table2
+ .get(new Get(Bytes.toBytes("delete1")))
+ .getColumnLatestCell(Bytes.toBytes(columnFamily),
Bytes.toBytes("a")) == null)
+ assert(
+ table2
+ .get(new Get(Bytes.toBytes("delete3")))
+ .getColumnLatestCell(Bytes.toBytes(columnFamily),
Bytes.toBytes("a")) == null)
+ assert(
+ Bytes
+ .toString(
+ CellUtil.cloneValue(table2
+ .get(new Get(Bytes.toBytes("delete2")))
+ .getColumnLatestCell(Bytes.toBytes(columnFamily),
Bytes.toBytes("a"))))
+ .equals("foo2"))
+ } finally {
+ table2.close()
+ connection2.close()
+ }
+ }
+
+ test("bulkGet to test HBase client") {
Review Comment:
nit: a failure test for `bulkGet` (bad table) in `HBaseContextSuite` of
`hbase-spark` was missing but it should be fine
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
