Bugfix/sql and conflict fixes (#69)

* add sql for multi-cf test; fix conflict error

* fix buffered mutation family_violation check

* Remove multi-family checks for buffered mutations

* fix check family empty

Author: maochongxin

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

@@ -512,7 +512,7 @@ public Result get(final Get get) throws IOException {
         if (get.getFamilyMap().keySet() == null || get.getFamilyMap().keySet().isEmpty()) {
             // check nothing, use table group;
         } else {
-            checkFamilyViolation(get.getFamilyMap().keySet());
+            checkFamilyViolation(get.getFamilyMap().keySet(), false);
         }
 
         ServerCallable<Result> serverCallable = new ServerCallable<Result>(configuration,
@@ -590,7 +590,7 @@ public ResultScanner getScanner(final Scan scan) throws IOException {
         if (scan.getFamilyMap().keySet().isEmpty()) {
             // check nothing, use table group;
         } else {
-            checkFamilyViolation(scan.getFamilyMap().keySet());
+            checkFamilyViolation(scan.getFamilyMap().keySet(), false);
         }
 
         //be careful about the packet size ,may the packet exceed the max result size ,leading to error
@@ -681,7 +681,7 @@ private void doPut(List<Put> puts) throws IOException {
         int n = 0;
         for (Put put : puts) {
             validatePut(put);
-            checkFamilyViolation(put.getFamilyMap().keySet());
+            checkFamilyViolation(put.getFamilyMap().keySet(), true);
             writeBuffer.add(put);
             currentWriteBufferSize += put.heapSize();
 
@@ -778,7 +778,7 @@ private void innerDelete(Delete delete) throws IOException {
         BatchOperationResult results = null;
 
         try {
-            checkFamilyViolation(delete.getFamilyMap().keySet());
+            checkFamilyViolation(delete.getFamilyMap().keySet(), false);
             if (delete.getFamilyMap().isEmpty()) {
                 // For a Delete operation without any qualifiers, we construct a DeleteFamily request.  
                 // The server then performs the operation on all column families.
@@ -801,7 +801,7 @@ private void innerDelete(Delete delete) throws IOException {
                 } else {
                     for (Map.Entry<byte[], List<KeyValue>> entry : delete.getFamilyMap().entrySet()) {
                         BatchOperation batch = buildBatchOperation(
-                                getTargetTableName(tableNameString, Bytes.toString(entry.getKey())),
+                                getTargetTableName(tableNameString, Bytes.toString(entry.getKey()), configuration),
                                 entry.getValue(), false, null);
                         results = batch.execute();
                     }
@@ -830,7 +830,7 @@ private void innerDelete(Delete delete) throws IOException {
 
     @Override
     public void delete(Delete delete) throws IOException {
-        checkFamilyViolation(delete.getFamilyMap().keySet());
+        checkFamilyViolation(delete.getFamilyMap().keySet(), false);
         innerDelete(delete);
     }
 
@@ -1419,7 +1419,7 @@ private ObHTableFilter buildObHTableFilter(Filter filter, TimeRange timeRange, i
         ObHTableFilter obHTableFilter = new ObHTableFilter();
 
         if (filter != null) {
-            obHTableFilter.setFilterString(HBaseFilterUtils.toParseableString(filter).getBytes());
+            obHTableFilter.setFilterString(HBaseFilterUtils.toParseableString(filter));
         }
 
         if (timeRange != null) {
@@ -1459,7 +1459,7 @@ private ObHTableFilter buildObHTableFilter(String filterString, TimeRange timeRa
         ObHTableFilter obHTableFilter = new ObHTableFilter();
 
         if (filterString != null) {
-            obHTableFilter.setFilterString(filterString.getBytes());
+            obHTableFilter.setFilterString(filterString);
         }
 
         if (timeRange != null) {
@@ -1753,9 +1753,15 @@ private ObTableQueryAndMutateRequest buildObTableQueryAndMutateRequest(ObTableQu
         return request;
     }
 
-    public static void checkFamilyViolation(Collection<byte[]> families) {
+    public static void checkFamilyViolation(Collection<byte[]> families, boolean check_empty_family) {
+        if (check_empty_family && (families == null || families.isEmpty())) {
+            throw new FeatureNotSupportedException("family is empty");
+        }
+
         for (byte[] family : families) {
-            if (isBlank(Bytes.toString(family))) {
+            if (family == null || family.length == 0) {
+                throw new IllegalArgumentException("family is empty");
+            } else if (isBlank(Bytes.toString(family))) {
                 throw new IllegalArgumentException("family is blank");
             }
         }
@@ -1765,7 +1771,7 @@ public static void checkFamilyViolation(Collection<byte[]> families) {
     // This method is currently only used for append and increment operations.  
     // It restricts these two methods to use multi-column family operations.  
     // Note: After completing operations on multiple column families, they are deleted using the method described above.
-    private void checkFamilyViolationForOneFamily(Collection<byte[]> families) {
+    public static void checkFamilyViolationForOneFamily(Collection<byte[]> families) {
         if (families == null || families.size() == 0) {
             throw new FeatureNotSupportedException("family is empty.");
         }
@@ -1774,7 +1780,9 @@ private void checkFamilyViolationForOneFamily(Collection<byte[]> families) {
             throw new FeatureNotSupportedException("multi family is not supported yet.");
         }
         for (byte[] family : families) {
-            if (isBlank(Bytes.toString(family))) {
+            if (family == null || family.length == 0) {
+                throw new IllegalArgumentException("family is empty");
+            } else if (isBlank(Bytes.toString(family))) {
                 throw new IllegalArgumentException("family is blank");
             }
         }

src/main/java/com/alipay/oceanbase/hbase/util/OHBufferedMutatorImpl.java

@@ -142,7 +142,7 @@ public void mutate(List<? extends Mutation> mutations) throws IOException {
         // check if every mutation's family is the same
         // check if mutations are the same type
         for (Mutation m : mutations) {
-            OHTable.checkFamilyViolation(m.getFamilyMap().keySet());
+            OHTable.checkFamilyViolation(m.getFamilyMap().keySet(), true);
             validateInsUpAndDelete(m);
             Class<?> curType = m.getClass();
             // set the type of this BufferedMutator

src/test/java/unit_test_db.sql

@@ -162,19 +162,29 @@ CREATE TABLE `test_t$family_with_local_index` (
     PRIMARY KEY (`K`, `Q`, `T`)
 );
 
-CREATE TABLE `test$family'1` (
+
+CREATE TABLEGROUP test_multi_cf SHARDING = 'ADAPTIVE';
+
+CREATE TABLE `test_multi_cf$family_with_group1` (
     `K` varbinary(1024) NOT NULL,
     `Q` varbinary(256) NOT NULL,
     `T` bigint(20) NOT NULL,
     `V` varbinary(1024) DEFAULT NULL,
-    PRIMARY KEY (`K`, `Q`, `T`)
-) TABLEGROUP = test;
+    PRIMARY KEY (`K`, `Q`, `T`) 
+) TABLEGROUP = test_multi_cf PARTITION BY KEY(`K`) PARTITIONS 3;
 
-CREATE TABLE `test_t$family'1` (
+CREATE TABLE `test_multi_cf$family_with_group2` (
     `K` varbinary(1024) NOT NULL,
     `Q` varbinary(256) NOT NULL,
     `T` bigint(20) NOT NULL,
     `V` varbinary(1024) DEFAULT NULL,
-    PRIMARY KEY (`K`, `Q`, `T`)
-) TABLEGROUP = test_t;
+    PRIMARY KEY (`K`, `Q`, `T`) 
+) TABLEGROUP = test_multi_cf PARTITION BY KEY(`K`) PARTITIONS 3;
 
+CREATE TABLE `test_multi_cf$family_with_group3` (
+    `K` varbinary(1024) NOT NULL,
+    `Q` varbinary(256) NOT NULL,
+    `T` bigint(20) NOT NULL,
+    `V` varbinary(1024) DEFAULT NULL,
+    PRIMARY KEY (`K`, `Q`, `T`) 
+) TABLEGROUP = test_multi_cf PARTITION BY KEY(`K`) PARTITIONS 3;