Repository: GoogleCloudPlatform/cloud-bigtable-client (java)

Issue:
One line batch settings
There are a bunch of good batch settings:
```
    Configuration config = new Configuration(false);

    config.set(BigtableOptionsFactory.BIGTABLE_USE_CACHED_DATA_CHANNEL_POOL, "true");

    // Beam should use a different endpoint for data operations than online traffic.
    config.set(BigtableOptionsFactory.BIGTABLE_HOST_KEY,
      BigtableOptions.BIGTABLE_BATCH_DATA_HOST_DEFAULT);

    config.set(BigtableOptionsFactory.INITIAL_ELAPSED_BACKOFF_MILLIS_KEY,
      String.valueOf(TimeUnit.SECONDS.toMillis(5)));

    config.set(BigtableOptionsFactory.MAX_ELAPSED_BACKOFF_MILLIS_KEY,
      String.valueOf(TimeUnit.MINUTES.toMillis(5)));
```
this should be a one-liner.  Also, it should encourage setting environment via the user agent.

Pull request description:
Fix #1543 One line batch settings
Fix #1543 One line batch settings

Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/BigtableOptions.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/BigtableOptions.java
index c43eea968..37b35f431 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/BigtableOptions.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/BigtableOptions.java
@@ -17,6 +17,7 @@ package com.google.cloud.bigtable.config;
 
 import java.io.Serializable;
 import java.util.Objects;
+import java.util.concurrent.TimeUnit;
 
 import com.google.cloud.bigtable.grpc.BigtableInstanceName;
 import com.google.cloud.bigtable.grpc.BigtableSession;
@@ -102,6 +103,7 @@ public class BigtableOptions implements Serializable, Cloneable {
       // the Google Compute Engine metadata service or gcloud configuration in other environments. A
       // user can also override the default behavior with P12 or JSON configuration.
       options.credentialOptions = CredentialOptions.defaultCredentials();
+      options.useBatch = false;
     }
 
     private Builder(BigtableOptions options) {
@@ -190,6 +192,11 @@ public class BigtableOptions implements Serializable, Cloneable {
       return this;
     }
 
+    public Builder setUseBatch(boolean useBatch) {
+      options.useBatch = useBatch;
+      return this;
+    }
+
     /**
      * Apply emulator settings from the relevant environment variable, if set.
      */
@@ -253,6 +260,14 @@ public class BigtableOptions implements Serializable, Cloneable {
         options.instanceName = null;
       }
 
+      if(options.useBatch) {
+        options.useCachedDataPool = true;
+        options.dataHost = BIGTABLE_BATCH_DATA_HOST_DEFAULT;
+        RetryOptions.Builder retryOptionsBuilder = options.retryOptions.toBuilder();
+        retryOptionsBuilder.setInitialBackoffMillis((int) TimeUnit.SECONDS.toMillis(5));
+        retryOptionsBuilder.setMaxElapsedBackoffMillis((int) TimeUnit.SECONDS.toMillis(5));
+        options.retryOptions = retryOptionsBuilder.build();
+      }
       LOG.debug("Connection Configuration: projectId: %s, instanceId: %s, data host %s, "
               + "admin host %s.",
           options.projectId,
@@ -281,6 +296,7 @@ public class BigtableOptions implements Serializable, Cloneable {
   private CallOptionsConfig callOptionsConfig;
   private CredentialOptions credentialOptions;
   private RetryOptions retryOptions;
+  private boolean useBatch;
 
   @VisibleForTesting
   BigtableOptions() {
@@ -436,7 +452,8 @@ public class BigtableOptions implements Serializable, Cloneable {
         && Objects.equals(credentialOptions, other.credentialOptions)
         && Objects.equals(retryOptions, other.retryOptions)
         && Objects.equals(bulkOptions, other.bulkOptions)
-        && Objects.equals(callOptionsConfig, other.callOptionsConfig);
+        && Objects.equals(callOptionsConfig, other.callOptionsConfig)
+        && Objects.equals(useBatch, other.useBatch);
   }
 
   /** {@inheritDoc} */
@@ -458,6 +475,7 @@ public class BigtableOptions implements Serializable, Cloneable {
         .add("callOptionsConfig", callOptionsConfig)
         .add("usePlaintextNegotiation", usePlaintextNegotiation)
         .add("useCachedDataPool", useCachedDataPool)
+        .add("useBatch", useBatch)
         .toString();
   }
 
@@ -479,6 +497,10 @@ public class BigtableOptions implements Serializable, Cloneable {
     return useCachedDataPool;
   }
 
+  public boolean useBatch() {
+    return useBatch;
+  }
+
   protected BigtableOptions clone() {
     try {
       return (BigtableOptions) super.clone();
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/RetryOptions.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/RetryOptions.java
index 13cb983bc..2d8515f25 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/RetryOptions.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/RetryOptions.java
@@ -33,7 +33,7 @@ import com.google.common.collect.ImmutableSet;
  * @author sduskis
  * @version $Id: $Id
  */
-public class RetryOptions implements Serializable {
+public class RetryOptions implements Serializable, Cloneable {
 
   private static final long serialVersionUID = 1L;
 
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/ThreadUtil.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/ThreadUtil.java
index 2c428acd5..63a051dca 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/ThreadUtil.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/ThreadUtil.java
@@ -22,7 +22,7 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import java.util.concurrent.ThreadFactory;
 
 /**
- * Creates a {@link ThreadFactory} that's safe to use in AppEngine.
+ * Creates a {@link ThreadFactory} that's safe to use in App Engine.
  * <p>
  * This class copies code that originates in {@link io.grpc.internal.GrpcUtil#getThreadFactory(String, boolean)}.
  */
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java
index d100126d6..89fdbf1c4 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java
@@ -199,6 +199,10 @@ public class BigtableOptionsFactory {
   public static final String BIGTABLE_USE_BULK_API =
       "google.bigtable.use.bulk.api";
 
+  /** Constant <code>BIGTABLE_USE_BATCH="google.bigtable.use.batch"</code> */
+  public static final String BIGTABLE_USE_BATCH =
+      "google.bigtable.use.batch";
+
   /** Constant <code>BIGTABLE_BULK_MAX_REQUEST_SIZE_BYTES="google.bigtable.bulk.max.request.size.b"{trunked}</code> */
   public static final String BIGTABLE_BULK_MAX_REQUEST_SIZE_BYTES =
       "google.bigtable.bulk.max.request.size.bytes";
@@ -316,6 +320,7 @@ public class BigtableOptionsFactory {
     if (emulatorHost != null) {
       bigtableOptionsBuilder.enableEmulator(emulatorHost);
     }
+    bigtableOptionsBuilder.setUseBatch(configuration.getBoolean(BIGTABLE_USE_BATCH, false));
 
     return bigtableOptionsBuilder.build();
   }
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/filters/ValueFilterAdapter.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/filters/ValueFilterAdapter.java
index a4b1f46ac..86ffd918a 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/filters/ValueFilterAdapter.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/filters/ValueFilterAdapter.java
@@ -20,6 +20,7 @@ import static com.google.cloud.bigtable.data.v2.models.Filters.FILTERS;
 import com.google.bigtable.v2.RowFilter;
 import com.google.cloud.bigtable.data.v2.models.Filters.Filter;
 import com.google.cloud.bigtable.data.v2.models.Filters.ValueRangeFilter;
+import com.google.cloud.bigtable.hbase.adapters.read.ReaderExpressionHelper;
 import com.google.protobuf.ByteString;
 
 import org.apache.hadoop.hbase.filter.BinaryComparator;
@@ -88,10 +89,15 @@ public class ValueFilterAdapter extends TypedFilterAdapterBase<ValueFilter> {
           return range().startClosed(value).endClosed(value);
         }
       case NOT_EQUAL:
-        // This strictly less than + strictly greater than:
-        return FILTERS.interleave()
-            .filter(range().endOpen(value))
-            .filter(range().startOpen(value));
+        if(comparator.getValue().length == 0) {
+          //Special case for NOT_EQUAL to EMPTY_STRING
+          return FILTERS.value().regex(ReaderExpressionHelper.ANY_BYTES);
+        } else {
+          // This strictly less than + strictly greater than:
+          return FILTERS.interleave()
+              .filter(range().endOpen(value))
+              .filter(range().startOpen(value));
+        }
       case GREATER_OR_EQUAL:
         return range().startClosed(value);
       case GREATER:
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ReaderExpressionHelper.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ReaderExpressionHelper.java
index 94344c1bd..225083eba 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ReaderExpressionHelper.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ReaderExpressionHelper.java
@@ -32,7 +32,9 @@ import java.io.OutputStream;
 public class ReaderExpressionHelper {
   /** Constant <code>ANY_BYTE="\\C"</code> */
   public static final String ANY_BYTE = "\\C";
-  /** Constant <code>ANY_BYTE="\\C"</code> */
+  /** Constant <code>ANY_BYTES="\\C+"</code> */
+  public static final String ANY_BYTES = "\\C+";
+  /** Constant <code>ALL_BYTES="\\C*"</code> */
   public static final String ALL_BYTES = "\\C*";
   /** Constant <code>ANY_BYTE_BYTES=Bytes.toBytes(ANY_BYTE)</code> */
   public static final byte[] ANY_BYTE_BYTES = Bytes.toBytes(ANY_BYTE);
diff --git a/bigtable-dataflow-parent/bigtable-hbase-beam/src/main/java/com/google/cloud/bigtable/beam/CloudBigtableConfiguration.java b/bigtable-dataflow-parent/bigtable-hbase-beam/src/main/java/com/google/cloud/bigtable/beam/CloudBigtableConfiguration.java
index 460ca8996..15792807a 100644
--- a/bigtable-dataflow-parent/bigtable-hbase-beam/src/main/java/com/google/cloud/bigtable/beam/CloudBigtableConfiguration.java
+++ b/bigtable-dataflow-parent/bigtable-hbase-beam/src/main/java/com/google/cloud/bigtable/beam/CloudBigtableConfiguration.java
@@ -218,17 +218,14 @@ public class CloudBigtableConfiguration implements Serializable {
   public Configuration toHBaseConfig() {
     Configuration config = new Configuration(false);
 
-    config.set(BigtableOptionsFactory.BIGTABLE_USE_CACHED_DATA_CHANNEL_POOL, "true");
-
-    // Beam should use a different endpoint for data operations than online traffic.
-    config.set(BigtableOptionsFactory.BIGTABLE_HOST_KEY,
-      BigtableOptions.BIGTABLE_BATCH_DATA_HOST_DEFAULT);
-
-    config.set(BigtableOptionsFactory.INITIAL_ELAPSED_BACKOFF_MILLIS_KEY,
-      String.valueOf(TimeUnit.SECONDS.toMillis(5)));
-
-    config.set(BigtableOptionsFactory.MAX_ELAPSED_BACKOFF_MILLIS_KEY,
-      String.valueOf(TimeUnit.MINUTES.toMillis(5)));
+    /**
+     * Sets below setting for batch
+     * BIGTABLE_USE_CACHED_DATA_CHANNEL_POOL = true
+     * BigtableOptionsFactory.BIGTABLE_HOST_KEY = batch-bigtable.googleapis.com
+     * BigtableOptionsFactory.INITIAL_ELAPSED_BACKOFF_MILLIS_KEY = 5ms
+     * BigtableOptionsFactory.MAX_ELAPSED_BACKOFF_MILLIS_KEY = 5ms
+     */
+    config.set(BigtableOptionsFactory.BIGTABLE_USE_BATCH, "true");
 
     // This setting can potentially decrease performance for large scale writes. However, this
     // setting prevents problems that occur when streaming Sources, such as PubSub, are used.


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/config/TestBigtableOptions.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/config/TestBigtableOptions.java
index cdce6a893..8deb928b6 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/config/TestBigtableOptions.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/config/TestBigtableOptions.java
@@ -114,6 +114,15 @@ public class TestBigtableOptions {
     Assert.assertEquals(CredentialOptions.defaultCredentials(), options.getCredentialOptions());
   }
 
+  @Test
+  public void testUseBatch() {
+    BigtableOptions options = BigtableOptions.builder().setUseBatch(true).build();
+    Assert.assertTrue(options.useCachedChannel());
+    Assert.assertEquals(BigtableOptions.BIGTABLE_BATCH_DATA_HOST_DEFAULT, options.getDataHost());
+    Assert.assertEquals(5000, options.getRetryOptions().getInitialBackoffMillis());
+    Assert.assertEquals(5000, options.getRetryOptions().getMaxElapsedBackoffMillis());
+  }
+
   /**
    * This is a dirty way to override the environment that is accessible to a test.
    * It only modifies the JVM's view of the environment, not the environment itself.
diff --git a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/AbstractTestFilters.java b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/AbstractTestFilters.java
index 61b4fcf5a..2c375d1da 100644
--- a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/AbstractTestFilters.java
+++ b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/AbstractTestFilters.java
@@ -1547,6 +1547,53 @@ public abstract class AbstractTestFilters extends AbstractTest {
     Assert.assertEquals(0, results.length);
   }
 
+  @Test
+  public void testSingleColumnValueFilterNotEqualEmptyString() throws IOException {
+    // Set up:
+    // Row 1: f:qualifier1 = value1
+    // Row 2: f:qualifier1 = ""
+
+    // Cases to test:
+    // a: Filter NOT_EQUAL to EMPTY_STRING
+
+    byte[] rowKey1 = dataHelper.randomData("scvfrk1");
+    byte[] rowKey2 = dataHelper.randomData("scvfrk2");
+    byte[] qualifier1 = dataHelper.randomData("scvfq1");
+    byte[] qualifier2 = dataHelper.randomData("scvfq2");
+    byte[] nonEmptyStringValue = dataHelper.randomData("val1.1");
+    byte[] emptyStringValue = "".getBytes();
+
+
+    Table table = getDefaultTable();
+    Put put = new Put(rowKey1);
+    put.addColumn(COLUMN_FAMILY, qualifier1, nonEmptyStringValue);
+    put.addColumn(COLUMN_FAMILY, qualifier2, nonEmptyStringValue);
+    table.put(put);
+
+    put = new Put(rowKey2);
+    put.addColumn(COLUMN_FAMILY, qualifier1, nonEmptyStringValue);
+    put.addColumn(COLUMN_FAMILY, qualifier2, emptyStringValue);
+    table.put(put);
+
+    Result[] results;
+    Scan scan = new Scan();
+    scan.addColumn(COLUMN_FAMILY, qualifier1);
+    scan.setRowPrefixFilter(Bytes.toBytes("scvfrk"));
+
+    // This is not intuitive. In order to get filter.setLatestVersionOnly to have an effect,
+    // we must enable the scanner to see more versions:
+    scanAddVersion(scan, 3);
+    ByteArrayComparable emptyStringValueComparable = new BinaryComparator(emptyStringValue);
+    SingleColumnValueFilter filter =
+        new SingleColumnValueFilter(COLUMN_FAMILY, qualifier2, CompareOp.NOT_EQUAL, emptyStringValueComparable);
+
+    // a: Qualifier exists in the row and the value NOT_EQUAL to EMPTY_STRING (row1)
+    scan.setFilter(filter);
+    results = table.getScanner(scan).next(10);
+    Assert.assertEquals(1, results.length);
+    Assert.assertArrayEquals(rowKey1, results[0].getRow());
+  }
+
   @Test
   public void testRandomRowFilter() throws IOException {
     byte[][] rowKeys = dataHelper.randomData("trandA", 100);
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/filters/TestValueFilterAdapter.java b/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/filters/TestValueFilterAdapter.java
index 27db2e060..94d6b57d3 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/filters/TestValueFilterAdapter.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/filters/TestValueFilterAdapter.java
@@ -30,6 +30,7 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
+import com.google.cloud.bigtable.hbase.adapters.read.ReaderExpressionHelper;
 import com.google.bigtable.v2.RowFilter;
 import com.google.bigtable.v2.RowFilter.Interleave;
 import com.google.bigtable.v2.ValueRange;
@@ -88,6 +89,12 @@ public class TestValueFilterAdapter {
       toRowFilter(ValueRange.newBuilder().setStartValueClosed(FOO_BYTESTRING)));
   }
 
+  @Test
+  public void testNotEqualEmptyStringValueFilter() throws IOException {
+    assertAdaptedForm(new BinaryComparator("".getBytes()), CompareOp.NOT_EQUAL,
+        RowFilter.newBuilder().setValueRegexFilter(ByteString.copyFrom(ReaderExpressionHelper.ANY_BYTES.getBytes())).build());
+  }
+
   @Test
   public void testNotEqualValueFilter() throws IOException {
     assertAdaptedForm(FOO_BINARY_COMPARATOR, CompareOp.NOT_EQUAL,