Bigtable: 02. Surface: Readrows

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/BigtableDataClient.java

@@ -16,8 +16,14 @@
 package com.google.cloud.bigtable.data.v2;
 
 import com.google.api.core.InternalApi;
+import com.google.api.gax.rpc.ResponseObserver;
+import com.google.api.gax.rpc.ServerStream;
+import com.google.api.gax.rpc.ServerStreamingCallable;
 import com.google.bigtable.admin.v2.InstanceName;
 import com.google.cloud.bigtable.data.v2.stub.EnhancedBigtableStub;
+import com.google.cloud.bigtable.data.v2.wrappers.Query;
+import com.google.cloud.bigtable.data.v2.wrappers.Row;
+import com.google.cloud.bigtable.data.v2.wrappers.RowAdapter;
 import java.io.IOException;
 
 /**
@@ -29,7 +35,9 @@
  * <pre>{@code
  * InstanceName instanceName = InstanceName.of("[PROJECT]", "[INSTANCE]");
  * try (BigtableDataClient bigtableDataClient = BigtableDataClient.create(instanceName)) {
- *   // TODO: add example usage
+ *   for(Row row : bigtableDataClient.readRows(Query.create("[TABLE]")) {
+ *     // Do something with row
+ *   }
  * }
  * }</pre>
  *
@@ -40,11 +48,13 @@
  * methods:
  *
  * <ol>
- *   <li>A "flattened" method. With this type of method, the fields of the request type have been
- *       converted into function parameters. It may be the case that not all fields are available as
- *       parameters, and not every API method will have a flattened method entry point.
- *   <li>A "callable" method. This type of method takes no parameters and returns an immutable API
- *       callable object, which can be used to initiate calls to the service.
+ *   <li>A "flattened" method, like `readRows()`. With this type of method, the fields of the
+ *       request type have been converted into function parameters. It may be the case that not all
+ *       fields are available as parameters, and not every API method will have a flattened method
+ *       entry point.
+ *   <li>A "callable" method, like `readRowsCallable()`. This type of method takes no parameters and
+ *       returns an immutable API callable object, which can be used to initiate calls to the
+ *       service.
  * </ol>
  *
  * <p>See the individual methods for example code.
@@ -107,6 +117,136 @@ public static BigtableDataClient create(BigtableDataSettings settings) throws IO
     this.stub = stub;
   }
 
+  /**
+   * Convenience method for synchronous streaming the results of a {@link Query}.
+   *
+   * <p>Sample code:
+   *
+   * <pre>{@code
+   * // Import the filter DSL
+   * import static com.google.cloud.bigtable.data.v2.wrappers.Filters.FILTERS;
+   *
+   * InstanceName instanceName = InstanceName.of("[PROJECT]", "[INSTANCE]");
+   * try (BigtableClient bigtableClient = BigtableClient.create(instanceName)) {
+   *   String tableId = "[TABLE]";
+   *
+   *   Query query = Query.create(tableId)
+   *          .range("[START KEY]", "[END KEY]")
+   *          .filter(FILTERS.qualifier().regex("[COLUMN PREFIX].*"));
+   *
+   *   // Iterator style
+   *   for(Row row : bigtableClient.readRows(query)) {
+   *     // Do something with row
+   *   }
+   * }
+   * }</pre>
+   *
+   * @see ServerStreamingCallable For call styles.
+   * @see Query For query options.
+   * @see com.google.cloud.bigtable.data.v2.wrappers.Filters For the filter building DSL.
+   */
+  public ServerStream<Row> readRows(Query query) {
+    return readRowsCallable().call(query);
+  }
+
+  /**
+   * Convenience method for asynchronous streaming the results of a {@link Query}.
+   *
+   * <p>Sample code:
+   *
+   * <pre>{@code
+   * InstanceName instanceName = InstanceName.of("[PROJECT]", "[INSTANCE]");
+   * try (BigtableClient bigtableClient = BigtableClient.create(instanceName)) {
+   *   String tableId = "[TABLE]";
+   *
+   *   Query query = Query.create(tableId)
+   *          .range("[START KEY]", "[END KEY]")
+   *          .filter(FILTERS.qualifier().regex("[COLUMN PREFIX].*"));
+   *
+   *   client.readRowsAsync(query, new ResponseObserver<Row>() {
+   *     public void onStart(StreamController controller) { }
+   *     public void onResponse(Row response) {
+   *       // Do something with Row
+   *     }
+   *     public void onError(Throwable t) {
+   *       // Handle error before the stream completes
+   *     }
+   *     public void onComplete() {
+   *       // Handle stream completion
+   *     }
+   *   });
+   * }
+   * }</pre>
+   */
+  public void readRowsAsync(Query query, ResponseObserver<Row> observer) {
+    readRowsCallable().call(query, observer);
+  }
+
+  /**
+   * Streams back the results of the query. The returned callable object allows for customization of
+   * api invocation.
+   *
+   * <p>Sample code:
+   *
+   * <pre>{@code
+   * InstanceName instanceName = InstanceName.of("[PROJECT]", "[INSTANCE]");
+   * try (BigtableClient bigtableClient = BigtableClient.create(instanceName)) {
+   *   String tableId = "[TABLE]";
+   *
+   *   Query query = Query.create(tableId)
+   *          .range("[START KEY]", "[END KEY]")
+   *          .filter(FILTERS.qualifier().regex("[COLUMN PREFIX].*"));
+   *
+   *   // Iterator style
+   *   for(Row row : bigtableClient.readRowsCallable().call(query)) {
+   *     // Do something with row
+   *   }
+   *
+   *   // Point look up
+   *   ApiFuture<Row> rowFuture = bigtableClient.readRowsCallable().first().futureCall(query);
+   *
+   *   // etc
+   * }
+   * }</pre>
+   *
+   * @see ServerStreamingCallable For call styles.
+   * @see Query For query options.
+   * @see com.google.cloud.bigtable.data.v2.wrappers.Filters For the filter building DSL.
+   */
+  public ServerStreamingCallable<Query, Row> readRowsCallable() {
+    return stub.readRowsCallable();
+  }
+
+  /**
+   * Streams back the results of the query. This callable allows for customization of the logical
+   * representation of a row. It's meant for advanced use cases.
+   *
+   * <p>Sample code:
+   *
+   * <pre>{@code
+   * InstanceName instanceName = InstanceName.of("[PROJECT]", "[INSTANCE]");
+   * try (BigtableClient bigtableClient = BigtableClient.create(instanceName)) {
+   *   String tableId = "[TABLE]";
+   *
+   *   Query query = Query.create(tableId)
+   *          .range("[START KEY]", "[END KEY]")
+   *          .filter(FILTERS.qualifier().regex("[COLUMN PREFIX].*"));
+   *
+   *   // Iterator style
+   *   for(CustomRow row : bigtableClient.readRowsCallable(new CustomRowAdapter()).call(query)) {
+   *     // Do something with row
+   *   }
+   * }
+   * }</pre>
+   *
+   * @see ServerStreamingCallable For call styles.
+   * @see Query For query options.
+   * @see com.google.cloud.bigtable.data.v2.wrappers.Filters For the filter building DSL.
+   */
+  public <RowT> ServerStreamingCallable<Query, RowT> readRowsCallable(RowAdapter<RowT> rowAdapter) {
+    return stub.createReadRowsCallable(rowAdapter);
+  }
+
   /** Close the clients and releases all associated resources. */
   @Override
   public void close() throws Exception {

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/BigtableDataSettings.java

@@ -16,8 +16,11 @@
 package com.google.cloud.bigtable.data.v2;
 
 import com.google.api.gax.rpc.ClientSettings;
+import com.google.api.gax.rpc.ServerStreamingCallSettings;
 import com.google.bigtable.admin.v2.InstanceName;
 import com.google.cloud.bigtable.data.v2.stub.EnhancedBigtableStubSettings;
+import com.google.cloud.bigtable.data.v2.wrappers.Query;
+import com.google.cloud.bigtable.data.v2.wrappers.Row;
 import java.io.IOException;
 import javax.annotation.Nonnull;
 
@@ -38,10 +41,13 @@
  * build() is called, the tree of builders is called to create the complete settings object.
  *
  * <pre>{@code
- * BigtableDataSettings bigtableDataSettings = BigtableDataSettings.newBuilder()
+ * BigtableDataSettings.Builder settingsBuilder = BigtableDataSettings.newBuilder()
  *   .setInstanceName(InstanceName.of("my-project", "my-instance-id"))
- *   .setAppProfileId("default")
- *   .build();
+ *   .setAppProfileId("default");
+ *
+ * settingsBuilder.readRowsSettings().setRetryableCodes(Code.DEADLINE_EXCEEDED, Code.UNAVAILABLE);
+ *
+ * BigtableDataSettings settings = builder.build();
  * }</pre>
  */
 public class BigtableDataSettings extends ClientSettings<BigtableDataSettings> {
@@ -64,6 +70,11 @@ public String getAppProfileId() {
     return getTypedStubSettings().getAppProfileId();
   }
 
+  /** Returns the object with the settings used for calls to ReadRows. */
+  public ServerStreamingCallSettings<Query, Row> readRowsSettings() {
+    return getTypedStubSettings().readRowsSettings();
+  }
+
   @SuppressWarnings("unchecked")
   EnhancedBigtableStubSettings getTypedStubSettings() {
     return (EnhancedBigtableStubSettings) getStubSettings();
@@ -124,6 +135,11 @@ public String getAppProfileId() {
       return getTypedStubSettings().getAppProfileId();
     }
 
+    /** Returns the builder for the settings used for calls to readRows. */
+    public ServerStreamingCallSettings.Builder<Query, Row> readRowsSettings() {
+      return getTypedStubSettings().readRowsSettings();
+    }
+
     @SuppressWarnings("unchecked")
     private EnhancedBigtableStubSettings.Builder getTypedStubSettings() {
       return (EnhancedBigtableStubSettings.Builder) getStubSettings();

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/RegexUtil.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2018 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigtable.data.v2.internal;
+
+import com.google.api.core.InternalApi;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.ByteString.ByteIterator;
+
+/**
+ * Contains utilities to handle RE2 flavor regular expressions. This differs from {@link
+ * java.util.regex.Pattern} in two important ways:
+ *
+ * <ol>
+ *   <li>Binary strings are supported.
+ *   <li>The syntax is a lot more restricted but allows different modifiers.
+ * </ol>
+ *
+ * <p>See <a href="https://github.com/google/re2">https://github.com/google/re2</a> for more
+ * details.
+ */
+@InternalApi
+public final class RegexUtil {
+  private static final byte[] NULL_BYTES = "\\x00".getBytes();
+
+  private RegexUtil() {}
+
+  public static String literalRegex(final String value) {
+    return literalRegex(ByteString.copyFromUtf8(value)).toStringUtf8();
+  }
+  /** Converts the value to a quoted regular expression. */
+  public static ByteString literalRegex(ByteString value) {
+    ByteString.Output output = ByteString.newOutput(value.size() * 2);
+
+    ByteIterator it = value.iterator();
+    writeLiteralRegex(it, output);
+
+    return output.toByteString();
+  }
+
+  // Extracted from: re2 QuoteMeta:
+  // https://github.com/google/re2/blob/70f66454c255080a54a8da806c52d1f618707f8a/re2/re2.cc#L456
+  private static void writeLiteralRegex(ByteIterator input, ByteString.Output output) {
+    while (input.hasNext()) {
+      byte unquoted = input.nextByte();
+
+      if ((unquoted < 'a' || unquoted > 'z')
+          && (unquoted < 'A' || unquoted > 'Z')
+          && (unquoted < '0' || unquoted > '9')
+          && unquoted != '_'
+          &&
+          // If this is the part of a UTF8 or Latin1 character, we need
+          // to copy this byte without escaping.  Experimentally this is
+          // what works correctly with the regexp library.
+          (unquoted & 128) == 0) {
+
+        if (unquoted == '\0') { // Special handling for null chars.
+          // Note that this special handling is not strictly required for RE2,
+          // but this quoting is required for other regexp libraries such as
+          // PCRE.
+          // Can't use "\\0" since the next character might be a digit.
+          output.write(NULL_BYTES, 0, NULL_BYTES.length);
+          continue;
+        }
+
+        output.write('\\');
+      }
+      output.write(unquoted);
+    }
+  }
+}

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java

@@ -16,8 +16,15 @@
 package com.google.cloud.bigtable.data.v2.stub;
 
 import com.google.api.core.InternalApi;
+import com.google.api.gax.rpc.ApiCallContext;
 import com.google.api.gax.rpc.ClientContext;
+import com.google.api.gax.rpc.ResponseObserver;
+import com.google.api.gax.rpc.ServerStreamingCallable;
 import com.google.cloud.bigtable.data.v2.internal.RequestContext;
+import com.google.cloud.bigtable.data.v2.wrappers.DefaultRowAdapter;
+import com.google.cloud.bigtable.data.v2.wrappers.Query;
+import com.google.cloud.bigtable.data.v2.wrappers.Row;
+import com.google.cloud.bigtable.data.v2.wrappers.RowAdapter;
 import java.io.IOException;
 
 /**
@@ -39,6 +46,8 @@ public class EnhancedBigtableStub implements AutoCloseable {
   private final ClientContext clientContext;
   private final RequestContext requestContext;
 
+  private final ServerStreamingCallable<Query, Row> readRowsCallable;
+
   public static EnhancedBigtableStub create(EnhancedBigtableStubSettings settings)
       throws IOException {
     // Configure the base settings
@@ -63,16 +72,31 @@ public static EnhancedBigtableStub create(EnhancedBigtableStubSettings settings)
     this.stub = stub;
     this.requestContext =
         RequestContext.create(settings.getInstanceName(), settings.getAppProfileId());
-  }
 
-  @Override
-  public void close() throws Exception {
-    stub.close();
+    readRowsCallable = createReadRowsCallable(new DefaultRowAdapter());
   }
 
   // <editor-fold desc="Callable creators">
+  public <RowT> ServerStreamingCallable<Query, RowT> createReadRowsCallable(
+      RowAdapter<RowT> rowAdapter) {
+    return new ServerStreamingCallable<Query, RowT>() {
+      @Override
+      public void call(
+          Query query, ResponseObserver<RowT> responseObserver, ApiCallContext context) {
+        throw new UnsupportedOperationException("todo");
+      }
+    };
+  }
   // </editor-fold>
 
   // <editor-fold desc="Callable accessors">
+  public ServerStreamingCallable<Query, Row> readRowsCallable() {
+    return readRowsCallable;
+  }
   // </editor-fold>
+
+  @Override
+  public void close() throws Exception {
+    stub.close();
+  }
 }