Merge pull request #202 from oceanbase/hbase_2.0_cherry-pick

[CP] Some fixes cherry-picked onto Hbase 2.0 from master

Author: WeiXinChan

src/main/java/com/alipay/oceanbase/hbase/OHTable.java

@@ -154,7 +154,7 @@ public class OHTable implements Table {
     /**
      * the buffer of put request
      */
-    private final ArrayList<Put> writeBuffer            = new ArrayList<Put>();
+    private final ArrayList<Put>  writeBuffer            = new ArrayList<Put>();
     /**
      * when the put request reach the write buffer size the do put will
      * flush commits automatically
@@ -1123,13 +1123,13 @@ public List<ResultScanner> call() throws IOException {
 
                         request = buildObTableQueryAsyncRequest(obTableQuery,
                                 getTargetTableName(tableNameString));
+                        request.setAllowDistributeScan(false);
                         String phyTableName = obTableClient.getPhyTableNameFromTableGroup(
                                 request.getObTableQueryRequest(), tableNameString);
                         List<Partition> partitions = obTableClient.getPartition(phyTableName, false);
                         for (Partition partition : partitions) {
                             request.getObTableQueryRequest().setTableQueryPartId(
                                 partition.getPartId());
-                            request.setAllowDistributeScan(false);
                             clientQueryAsyncStreamResult = (ObTableClientQueryAsyncStreamResult) obTableClient
                                     .execute(request);
                             ClientStreamScanner clientScanner = new ClientStreamScanner(
@@ -1159,6 +1159,7 @@ public List<ResultScanner> call() throws IOException {
                             String targetTableName = getTargetTableName(tableNameString, Bytes.toString(family),
                                     configuration);
                             request = buildObTableQueryAsyncRequest(obTableQuery, targetTableName);
+                            request.setAllowDistributeScan(false);
                             List<Partition> partitions = obTableClient
                                     .getPartition(targetTableName, false);
                             for (Partition partition : partitions) {
@@ -2059,8 +2060,7 @@ private QueryAndMutate buildDeleteQueryAndMutate(KeyValue kv,
                     if (!isTableGroup) {
                         filter = buildObHTableFilter(null, null, Integer.MAX_VALUE);
                     } else {
-                        filter = buildObHTableFilter(null, null, Integer.MAX_VALUE,
-                            CellUtil.cloneQualifier(kv));
+                        filter = buildObHTableFilter(null, null, Integer.MAX_VALUE);
                     }
                 } else {
                     range.setStartKey(ObRowKey.getInstance(CellUtil.cloneRow(kv), ObObj.getMin(),
@@ -2090,11 +2090,11 @@ private QueryAndMutate buildDeleteQueryAndMutate(KeyValue kv,
                         ObObj.getMax()));
                     if (!isTableGroup) {
                         filter = buildObHTableFilter(null,
-                            new TimeRange(kv.getTimestamp(), kv.getTimestamp() + 1),
+                            new TimeRange(0, kv.getTimestamp() + 1),
                             Integer.MAX_VALUE);
                     } else {
                         filter = buildObHTableFilter(null,
-                            new TimeRange(kv.getTimestamp(), kv.getTimestamp() + 1),
+                            new TimeRange(0, kv.getTimestamp() + 1),
                             Integer.MAX_VALUE, CellUtil.cloneQualifier(kv));
                     }
                 }

src/test/java/com/alipay/oceanbase/hbase/secondary/OHTableSecondaryPartDeleteTest.java

@@ -412,7 +412,7 @@ public static void testMultiCFDeleteFamilyImpl(Map.Entry<String, List<String>> e
         hTable.close();
     }
 
-    public static void testMultiCFDeleteFamilyVersionImpl(Map.Entry<String, List<String>> entry) throws Exception {
+    public static void testMultiCFDeleteFamilyWithVersionImpl(Map.Entry<String, List<String>> entry) throws Exception {
         String key = "putKey";
         String value = "value";
         String column = "putColumn";
@@ -449,6 +449,43 @@ public static void testMultiCFDeleteFamilyVersionImpl(Map.Entry<String, List<Str
         hTable.close();
     }
 
+    public static void testMultiCFDeleteFamilyVersionImpl(Map.Entry<String, List<String>> entry) throws Exception {
+        String key = "putKey";
+        String value = "value";
+        String column = "putColumn";
+
+        long ts1 = System.currentTimeMillis();
+        long ts2 = ts1 + 1000;
+        long ts3 = ts2 + 1000;
+        long ts4 = ts3 + 1000;
+        System.out.println("ts1:" + ts1 + ", ts2:" + ts2 + ", ts3:" + ts3 + ", ts4:" + ts4);
+        OHTableClient hTable = ObHTableTestUtil.newOHTableClient(entry.getKey());
+        hTable.init();
+        Put put = new Put(toBytes(key));
+        Delete delete = new Delete(toBytes(key));
+        for (String tableName : entry.getValue()) {
+            String family = getColumnFamilyName(tableName);
+            put.addColumn(toBytes(family), toBytes(column + family), ts1, toBytes(value + ts1));
+            put.addColumn(toBytes(family), toBytes(column + family), ts2, toBytes(value + ts2));
+            put.addColumn(toBytes(family), toBytes(column + family), ts3, toBytes(value + ts3));
+            put.addColumn(toBytes(family), toBytes(column + family), ts4, toBytes(value + ts4));
+            delete.addFamilyVersion(toBytes(family), ts2);
+        }
+        hTable.put(put);
+        hTable.delete(delete);
+
+
+        Get get = new Get(toBytes(key));
+        get.setMaxVersions();
+        for (String tableName : entry.getValue()) {
+            String family = getColumnFamilyName(tableName);
+            get.addColumn(toBytes(family), toBytes(column + family));
+        }
+        Result result = hTable.get(get);
+        Assert(entry.getValue(), ()->Assert.assertEquals(6, result.size()));
+        hTable.close();
+    }
+
     public static void testDeleteAllImpl(Map.Entry<String, List<String>> entry) throws Exception {
         String key = "putKey";
         String value = "value";
@@ -530,6 +567,11 @@ public void testMultiCFDeleteFamily() throws Throwable {
         FOR_EACH(group2tableNames, OHTableSecondaryPartDeleteTest::testMultiCFDeleteFamilyImpl);
     }
 
+    @Test
+    public void testMultiCFDeleteFamilyWithVersion() throws Throwable {
+        FOR_EACH(group2tableNames, OHTableSecondaryPartDeleteTest::testMultiCFDeleteFamilyWithVersionImpl);
+    }
+    
     @Test
     public void testMultiCFDeleteFamilyVersion() throws Throwable {
         FOR_EACH(group2tableNames, OHTableSecondaryPartDeleteTest::testMultiCFDeleteFamilyVersionImpl);