DependentColumnFilter, singleColumnValueExcludeFilter and single cf setColumnFamilyTimeRange development (#82)

* add DepentdentFilter and SingleColumnValueExcludeFilter

* add singleColumnValueExcludeFilter and DependentColumnFilter; add test cases and optimize bufferedMutator test cases

* add single cf timerange setting in Get

* single cf setColumnFamilyTimeRange in Get and Scan

* optimize code

* report error when setColumnFamilyTimeRange for a column family that is not queryed

* correct name

Author: JackShi148

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

@@ -585,6 +585,9 @@ public Result call() throws IOException {
                         // In a Get operation where the family map is greater than 1 or equal to 0,  
                         // we handle this by appending the column family to the qualifier on the client side.  
                         // The server can then use this information to filter the appropriate column families and qualifiers.
+                        if (!get.getColumnFamilyTimeRange().isEmpty()) {
+                            throw new FeatureNotSupportedException("setColumnFamilyTimeRange is only supported in single column family for now");
+                        }
                         NavigableSet<byte[]> columnFilters = new TreeSet<>(Bytes.BYTES_COMPARATOR);
                         processColumnFilters(columnFilters, get.getFamilyMap());
                         obTableQuery = buildObTableQuery(get, columnFilters);
@@ -598,6 +601,17 @@ public Result call() throws IOException {
                         for (Map.Entry<byte[], NavigableSet<byte[]>> entry : get.getFamilyMap()
                             .entrySet()) {
                             family = entry.getKey();
+                            if (!get.getColumnFamilyTimeRange().isEmpty()) {
+                                Map<byte[], TimeRange> colFamTimeRangeMap = get.getColumnFamilyTimeRange();
+                                if (colFamTimeRangeMap.size() > 1) {
+                                    throw new FeatureNotSupportedException("setColumnFamilyTimeRange is only supported in single column family for now");
+                                } else if (colFamTimeRangeMap.get(family) == null) {
+                                    throw new IllegalArgumentException("Get family is not matched in ColumnFamilyTimeRange");
+                                } else {
+                                    TimeRange tr = colFamTimeRangeMap.get(family);
+                                    get.setTimeRange(tr.getMin(), tr.getMax());
+                                }
+                            }
                             obTableQuery = buildObTableQuery(get, entry.getValue());
                             request = buildObTableQueryAsyncRequest(obTableQuery,
                                 getTargetTableName(tableNameString, Bytes.toString(family),
@@ -668,6 +682,9 @@ public ResultScanner call() throws IOException {
                         // In a Scan operation where the family map is greater than 1 or equal to 0,  
                         // we handle this by appending the column family to the qualifier on the client side.  
                         // The server can then use this information to filter the appropriate column families and qualifiers.
+                        if (!scan.getColumnFamilyTimeRange().isEmpty()) {
+                            throw new FeatureNotSupportedException("setColumnFamilyTimeRange is only supported in single column family for now");
+                        }
                         NavigableSet<byte[]> columnFilters = new TreeSet<>(Bytes.BYTES_COMPARATOR);
                         processColumnFilters(columnFilters, scan.getFamilyMap());
                         filter = buildObHTableFilter(scan.getFilter(), scan.getTimeRange(),
@@ -684,6 +701,17 @@ public ResultScanner call() throws IOException {
                         for (Map.Entry<byte[], NavigableSet<byte[]>> entry : scan.getFamilyMap()
                             .entrySet()) {
                             family = entry.getKey();
+                            if (!scan.getColumnFamilyTimeRange().isEmpty()) {
+                                Map<byte[], TimeRange> colFamTimeRangeMap = scan.getColumnFamilyTimeRange();
+                                if (colFamTimeRangeMap.size() > 1) {
+                                    throw new FeatureNotSupportedException("setColumnFamilyTimeRange is only supported in single column family for now");
+                                } else if (colFamTimeRangeMap.get(family) == null) {
+                                    throw new IllegalArgumentException("Scan family is not matched in ColumnFamilyTimeRange");
+                                } else {
+                                    TimeRange tr = colFamTimeRangeMap.get(family);
+                                    scan.setTimeRange(tr.getMin(), tr.getMax());
+                                }
+                            }
                             filter = buildObHTableFilter(scan.getFilter(), scan.getTimeRange(),
                                 scan.getMaxVersions(), entry.getValue());
                             obTableQuery = buildObTableQuery(filter, scan);

src/main/java/com/alipay/oceanbase/hbase/filter/HBaseFilterUtils.java

@@ -42,9 +42,13 @@ private static void toParseableByteArray(ByteArrayOutputStream byteStream, Filte
                                                                                              throws IOException {
         if (filter == null) {
             throw new IllegalArgumentException("Filter is null");
+        } else if (filter instanceof DependentColumnFilter) {
+            toParseableByteArray(byteStream, (DependentColumnFilter) filter);
         } else if (filter instanceof CompareFilter) {
             // RowFilter, ValueFilter, QualifierFilter
             toParseableByteArray(byteStream, (CompareFilter) filter);
+        } else if (filter instanceof SingleColumnValueExcludeFilter) {
+            toParseableByteArray(byteStream, (SingleColumnValueExcludeFilter) filter);
         } else if (filter instanceof SingleColumnValueFilter) {
             toParseableByteArray(byteStream, (SingleColumnValueFilter) filter);
         } else if (filter instanceof PageFilter) {
@@ -163,6 +167,25 @@ private static void toParseableByteArray(ByteArrayOutputStream byteStream,
         byteStream.write(')');
     }
 
+    // SingleColumnValueExcludeFilter('cf1','col1',=,'binary:123',true,true)
+    private static void toParseableByteArray(ByteArrayOutputStream byteStream,
+                                             SingleColumnValueExcludeFilter filter) throws IOException {
+        byteStream.write(filter.getClass().getSimpleName().getBytes());
+        byteStream.write("('".getBytes());
+        writeBytesWithEscape(byteStream, filter.getFamily());
+        byteStream.write("','".getBytes());
+        writeBytesWithEscape(byteStream, filter.getQualifier());
+        byteStream.write("',".getBytes());
+        byteStream.write(toParseableByteArray(filter.getOperator()));
+        byteStream.write(',');
+        toParseableByteArray(byteStream, filter.getComparator());
+        byteStream.write(',');
+        byteStream.write(Boolean.toString(filter.getFilterIfMissing()).getBytes());
+        byteStream.write(',');
+        byteStream.write(Boolean.toString(filter.getLatestVersionOnly()).getBytes());
+        byteStream.write(')');
+    }
+
     // PageFilter(100);
     private static void toParseableByteArray(ByteArrayOutputStream byteStream, PageFilter filter)
                                                                                                  throws IOException {
@@ -181,6 +204,34 @@ private static void toParseableByteArray(ByteArrayOutputStream byteStream,
         byteStream.write(')');
     }
 
+    private static void toParseableByteArray(ByteArrayOutputStream byteStream,
+                                             DependentColumnFilter filter) throws IOException {
+        // DependentColumnFilter '(' family ',' qualifier ',' BOOL_VALUE ')'
+        if (filter.getComparator() == null) {
+            byteStream.write(filter.getClass().getSimpleName().getBytes());
+            byteStream.write("('".getBytes());
+            writeBytesWithEscape(byteStream, filter.getFamily());
+            byteStream.write("','".getBytes());
+            writeBytesWithEscape(byteStream, filter.getQualifier());
+            byteStream.write("',".getBytes());
+            byteStream.write(Boolean.toString(filter.getDropDependentColumn()).getBytes());
+            byteStream.write(')');
+        } else { // DependentColumnFilter '(' family ',' qualifier ',' BOOL_VALUE ',' compare_op ',' comparator ')'
+            byteStream.write(filter.getClass().getSimpleName().getBytes());
+            byteStream.write("('".getBytes());
+            writeBytesWithEscape(byteStream, filter.getFamily());
+            byteStream.write("','".getBytes());
+            writeBytesWithEscape(byteStream, filter.getQualifier());
+            byteStream.write("',".getBytes());
+            byteStream.write(Boolean.toString(filter.getDropDependentColumn()).getBytes());
+            byteStream.write(',');
+            byteStream.write(toParseableByteArray(filter.getOperator()));
+            byteStream.write(',');
+            toParseableByteArray(byteStream, filter.getComparator());
+            byteStream.write(')');
+        }
+    }
+
     private static void toParseableByteArray(ByteArrayOutputStream byteStream,
                                              ColumnPaginationFilter filter) throws IOException {
         byteStream.write(filter.getClass().getSimpleName().getBytes());