Repository: GoogleCloudPlatform/cloud-bigtable-client
Pull request #2045 (2019-01-11 13:40:50)

Utilizing BigtableTableAdminClientWrapper in AbstractBigtableAdmin
## What changes it contains
Initial Implementation for `BigtableTableAdminClientwrapper` to adapt to v2.models classes.
This is related to the ongoing wrapping of native cloud BigTable adapters on top of Google-Cloud-java adapter.

-  `AbstractBigtableAdmin`  & `BigtableAdmin`  is now adapting to v2.models, (except i.e. **`getTable()`** & snapshot related operations ).
- Modified `BigtableAdminWrapper#dropRowRange` & `BigtableAdminWrapper#dropRowRangeAsync` to support for truncate and deleteByPrefix.
- `hbase2.xBigtableTableAdminClient` is **uneffected** by this PR as that contains snapshot related operation (this would be separate PR) .

***
### Tests performed
 - Performed simple build.
 - Performed build with `hbaseLocalMiniClusterTest` profile for both H1 & H2.
 - Performed build with `bigtableIntegrationTest ` profile for both H1 & H2 with actual bigtable.

rahulKQL: @igorbernstein2 @sduskis 

Whenever get time, Please have a look.

Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableSession.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableSession.java
index 66e45a13c..b075359c5 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableSession.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableSession.java
@@ -26,7 +26,7 @@ import com.google.cloud.bigtable.config.BulkOptions;
 import com.google.cloud.bigtable.config.CredentialOptions;
 import com.google.cloud.bigtable.config.Logger;
 import com.google.cloud.bigtable.config.RetryOptions;
-import com.google.cloud.bigtable.core.IBigtableDataClient;
+import com.google.cloud.bigtable.core.IBigtableTableAdminClient;
 import com.google.cloud.bigtable.grpc.async.AsyncExecutor;
 import com.google.cloud.bigtable.grpc.async.BulkMutation;
 import com.google.cloud.bigtable.grpc.async.BulkRead;
@@ -163,6 +163,7 @@ public class BigtableSession implements Closeable {
   private final BigtableDataClient throttlingDataClient;
 
   private BigtableTableAdminClient tableAdminClient;
+  private IBigtableTableAdminClient adminClientWrapper;
   private BigtableInstanceGrpcClient instanceAdminClient;
 
   private final BigtableOptions options;
@@ -380,6 +381,19 @@ public class BigtableSession implements Closeable {
     return tableAdminClient;
   }
 
+  /**
+   * <p>Getter for the field <code>adminClientWrapper</code>.</p>
+   *
+   * @return a {@link BigtableTableAdminClientWrapper} object.
+   * @throws java.io.IOException if any.
+   */
+  public synchronized IBigtableTableAdminClient getTableAdminClientWrapper() throws IOException {
+    if (adminClientWrapper == null) {
+      adminClientWrapper = new BigtableTableAdminClientWrapper(getTableAdminClient(), options);
+    }
+    return adminClientWrapper;
+  }
+
   /**
    * <p>Getter for the field <code>instanceAdminClient</code>.</p>
    *
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClientWrapper.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClientWrapper.java
index 923b0db64..be31e2e52 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClientWrapper.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClientWrapper.java
@@ -27,6 +27,7 @@ import com.google.cloud.bigtable.config.BigtableOptions;
 import com.google.cloud.bigtable.core.IBigtableTableAdminClient;
 import com.google.common.base.Function;
 import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
@@ -113,8 +114,7 @@ public class BigtableTableAdminClientWrapper implements IBigtableTableAdminClien
 
     ListTablesResponse response = adminClient.listTables(requestProto);
 
-    ImmutableList.Builder<String> tableIdsBuilder =
-        ImmutableList.builderWithExpectedSize(response.getTablesList().size());
+    ImmutableList.Builder<String> tableIdsBuilder = ImmutableList.builder();
     for(com.google.bigtable.admin.v2.Table tableProto : response.getTablesList()){
       tableIdsBuilder.add(instanceName.toTableId(tableProto.getName()));
     }
@@ -133,8 +133,7 @@ public class BigtableTableAdminClientWrapper implements IBigtableTableAdminClien
     return Futures.transform(response, new Function<ListTablesResponse, List<String>>() {
       @Override
       public List<String> apply(ListTablesResponse input) {
-        ImmutableList.Builder<String> tableIdsBuilder =
-            ImmutableList.builderWithExpectedSize(input.getTablesList().size());
+        ImmutableList.Builder<String> tableIdsBuilder = ImmutableList.builder();
         for(com.google.bigtable.admin.v2.Table tableProto : input.getTablesList()){
           tableIdsBuilder.add(instanceName.toTableId(tableProto.getName()));
         }
@@ -196,27 +195,37 @@ public class BigtableTableAdminClientWrapper implements IBigtableTableAdminClien
   /** {@inheritDoc} */
   @Override
   public void dropRowRange(String tableId, String rowKeyPrefix) {
-    DropRowRangeRequest requestProto = DropRowRangeRequest.newBuilder()
-        .setName(instanceName.toTableNameStr(tableId))
-        .setRowKeyPrefix(ByteString.copyFromUtf8(rowKeyPrefix))
-        .build();
 
-    adminClient.dropRowRange(requestProto);
+    adminClient.dropRowRange(buildDropRowRangeRequest(tableId, rowKeyPrefix));
   }
 
   /** {@inheritDoc} */
   @Override
   public ListenableFuture<Void> dropRowRangeAsync(String tableId, String rowKeyPrefix) {
-    DropRowRangeRequest requestProto = DropRowRangeRequest.newBuilder()
-        .setName(instanceName.toTableNameStr(tableId))
-        .setRowKeyPrefix(ByteString.copyFromUtf8(rowKeyPrefix))
-        .build();
 
-    return Futures.transform(adminClient.dropRowRangeAsync(requestProto), new Function<Empty, Void>() {
-      @Override
-      public Void apply(Empty empty) {
-          return null;
-      }
-    }, MoreExecutors.directExecutor());
+    return Futures.transform(
+        adminClient.dropRowRangeAsync(buildDropRowRangeRequest(tableId, rowKeyPrefix)),
+        new Function<Empty, Void>() {
+          @Override
+          public Void apply(Empty empty) {
+            return null;
+          }
+        }, MoreExecutors.directExecutor());
+  }
+
+  private DropRowRangeRequest buildDropRowRangeRequest(String tableId, String rowKeyPrefix) {
+    DropRowRangeRequest.Builder dropRequestProtoBuiler =
+        DropRowRangeRequest.newBuilder()
+            .setName(instanceName.toTableNameStr(tableId));
+
+    if (!Strings.isNullOrEmpty(rowKeyPrefix)) {
+      dropRequestProtoBuiler
+          .setDeleteAllDataFromTable(false)
+          .setRowKeyPrefix(ByteString.copyFromUtf8(rowKeyPrefix));
+    } else {
+      dropRequestProtoBuiler.setDeleteAllDataFromTable(true);
+    }
+
+    return dropRequestProtoBuiler.build();
   }
 }
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/CheckAndMutateUtil.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/CheckAndMutateUtil.java
index 741b17a6b..cdc9f2eaa 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/CheckAndMutateUtil.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/CheckAndMutateUtil.java
@@ -21,6 +21,7 @@ import com.google.bigtable.v2.ReadRowsRequest;
 import com.google.bigtable.v2.RowFilter;
 import com.google.cloud.bigtable.data.v2.models.ConditionalRowMutation;
 import com.google.cloud.bigtable.data.v2.models.Mutation;
+import com.google.cloud.bigtable.data.v2.models.Query;
 import com.google.cloud.bigtable.hbase.adapters.read.ReadHooks;
 import com.google.cloud.bigtable.hbase.filter.TimestampRangeFilter;
 import com.google.common.base.Function;
@@ -46,13 +47,13 @@ public class CheckAndMutateUtil {
   // them (which they shouldn't since we built the filter), throw an exception.
   private static final ReadHooks UNSUPPORTED_READ_HOOKS = new ReadHooks() {
     @Override
-    public void composePreSendHook(Function<ReadRowsRequest, ReadRowsRequest> newHook) {
+    public void composePreSendHook(Function<Query, Query> newHook) {
       throw new IllegalStateException(
           "We built a bad Filter for conditional mutation.");
     }
 
     @Override
-    public ReadRowsRequest applyPreSendHook(ReadRowsRequest readRowsRequest) {
+    public void applyPreSendHook(Query query) {
       throw new UnsupportedOperationException(
           "We built a bad Filter for conditional mutation.");
     }
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/HBaseRequestAdapter.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/HBaseRequestAdapter.java
index a0b9a62b2..65e8f50d6 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/HBaseRequestAdapter.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/HBaseRequestAdapter.java
@@ -20,6 +20,7 @@ import com.google.cloud.bigtable.data.v2.internal.RequestContext;
 import com.google.cloud.bigtable.data.v2.models.InstanceName;
 import com.google.cloud.bigtable.data.v2.models.Mutation;
 import com.google.cloud.bigtable.data.v2.models.MutationApi;
+import com.google.cloud.bigtable.data.v2.models.Query;
 import com.google.cloud.bigtable.data.v2.models.ReadModifyWriteRow;
 import com.google.cloud.bigtable.data.v2.models.RowMutation;
 import com.google.common.annotations.VisibleForTesting;
@@ -169,27 +170,29 @@ public class HBaseRequestAdapter {
   /**
    * <p>adapt.</p>
    *
-   * @param get a {@link org.apache.hadoop.hbase.client.Get} object.
-   * @return a {@link com.google.bigtable.v2.ReadRowsRequest} object.
+   * @param get a {@link Get} object.
+   * @return a {@link ReadRowsRequest} object.
    */
   public ReadRowsRequest adapt(Get get) {
     ReadHooks readHooks = new DefaultReadHooks();
-    ReadRowsRequest.Builder builder = Adapters.GET_ADAPTER.adapt(get, readHooks);
-    builder.setTableName(getTableNameString());
-    return readHooks.applyPreSendHook(builder.build());
+    Query query = Query.create(bigtableTableName.getTableId());
+    Adapters.GET_ADAPTER.adapt(get, readHooks, query);
+    readHooks.applyPreSendHook(query);
+    return query.toProto(requestContext);
   }
 
   /**
    * <p>adapt.</p>
    *
-   * @param scan a {@link org.apache.hadoop.hbase.client.Scan} object.
-   * @return a {@link com.google.bigtable.v2.ReadRowsRequest} object.
+   * @param scan a {@link Scan} object.
+   * @return a {@link ReadRowsRequest} object.
    */
   public ReadRowsRequest adapt(Scan scan) {
     ReadHooks readHooks = new DefaultReadHooks();
-    ReadRowsRequest.Builder builder = Adapters.SCAN_ADAPTER.adapt(scan, readHooks);
-    builder.setTableName(getTableNameString());
-    return readHooks.applyPreSendHook(builder.build());
+    Query query = Query.create(bigtableTableName.getTableId());
+    Adapters.SCAN_ADAPTER.adapt(scan, readHooks, query);
+    readHooks.applyPreSendHook(query);
+    return query.toProto(requestContext);
   }
 
   /**
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/filters/PageFilterAdapter.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/filters/PageFilterAdapter.java
index eca156649..be25e97b7 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/filters/PageFilterAdapter.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/filters/PageFilterAdapter.java
@@ -15,8 +15,8 @@
  */
 package com.google.cloud.bigtable.hbase.adapters.filters;
 
-import com.google.bigtable.v2.ReadRowsRequest;
 import com.google.cloud.bigtable.data.v2.models.Filters.Filter;
+import com.google.cloud.bigtable.data.v2.models.Query;
 import com.google.common.base.Function;
 import com.google.common.base.Optional;
 
@@ -43,10 +43,10 @@ public class PageFilterAdapter extends TypedFilterAdapterBase<PageFilter> {
   @Override
   public Filter adapt(FilterAdapterContext context, PageFilter filter) throws IOException {
     final long pageSize = filter.getPageSize();
-    context.getReadHooks().composePreSendHook(new Function<ReadRowsRequest, ReadRowsRequest>() {
+    context.getReadHooks().composePreSendHook(new Function<Query, Query>() {
       @Override
-      public ReadRowsRequest apply(ReadRowsRequest request) {
-        return request.toBuilder().setRowsLimit(pageSize).build();
+      public Query apply(Query query) {
+        return query.limit(pageSize);
       }
     });
     // This filter cannot be translated to a RowFilter, all logic is done as a read hook.
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/DefaultReadHooks.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/DefaultReadHooks.java
index 28883f09e..7290b8e2a 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/DefaultReadHooks.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/DefaultReadHooks.java
@@ -15,27 +15,27 @@
  */
 package com.google.cloud.bigtable.hbase.adapters.read;
 
-import com.google.bigtable.v2.ReadRowsRequest;
+import com.google.cloud.bigtable.data.v2.models.Query;
 import com.google.common.base.Function;
 import com.google.common.base.Functions;
 
 /**
- * Default implementation of {@link com.google.cloud.bigtable.hbase.adapters.read.ReadHooks}.
+ * Default implementation of {@link ReadHooks}.
  *
  * @author sduskis
  * @version $Id: $Id
  */
 public class DefaultReadHooks implements ReadHooks {
-  private Function<ReadRowsRequest, ReadRowsRequest> preSendHook = Functions.identity();
+  private Function<Query, Query> preSendHook = Functions.identity();
   /** {@inheritDoc} */
   @Override
-  public void composePreSendHook(Function<ReadRowsRequest, ReadRowsRequest> newHook) {
+  public void composePreSendHook(Function<Query, Query> newHook) {
     preSendHook = Functions.compose(newHook, preSendHook);
   }
 
   /** {@inheritDoc} */
   @Override
-  public ReadRowsRequest applyPreSendHook(ReadRowsRequest readRowsRequest) {
-    return preSendHook.apply(readRowsRequest);
+  public void applyPreSendHook(Query query) {
+    preSendHook.apply(query);
   }
 }
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/GetAdapter.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/GetAdapter.java
index df8daba2a..07a2069d6 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/GetAdapter.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/GetAdapter.java
@@ -15,8 +15,7 @@
  */
 package com.google.cloud.bigtable.hbase.adapters.read;
 
-import com.google.bigtable.v2.ReadRowsRequest;
-import com.google.bigtable.v2.RowSet;
+import com.google.cloud.bigtable.data.v2.models.Query;
 import com.google.protobuf.ByteString;
 
 import org.apache.hadoop.hbase.client.Get;
@@ -25,7 +24,7 @@ import org.apache.hadoop.hbase.filter.FilterList;
 import org.apache.hadoop.hbase.filter.KeyOnlyFilter;
 
 /**
- * A Get adapter that transform the Get into a ReadRowsRequest using the proto-based
+ * A {@link Get} adapter that transform the Get into a {@link Query} using the proto-based
  * filter language.
  *
  * @author sduskis
@@ -47,7 +46,7 @@ public class GetAdapter implements ReadOperationAdapter<Get> {
   /**
    * <p>Constructor for GetAdapter.</p>
    *
-   * @param scanAdapter a {@link com.google.cloud.bigtable.hbase.adapters.read.ScanAdapter} object.
+   * @param scanAdapter a {@link ScanAdapter} object.
    */
   public GetAdapter(ScanAdapter scanAdapter) {
     this.scanAdapter = scanAdapter;
@@ -55,12 +54,12 @@ public class GetAdapter implements ReadOperationAdapter<Get> {
 
   /** {@inheritDoc} */
   @Override
-  public ReadRowsRequest.Builder adapt(Get operation, ReadHooks readHooks) {
+  public void adapt(Get operation, ReadHooks readHooks, Query query) {
     Scan operationAsScan = new Scan(addKeyOnlyFilter(operation));
     scanAdapter.throwIfUnsupportedScan(operationAsScan);
-    return ReadRowsRequest.newBuilder()
-        .setFilter(scanAdapter.buildFilter(operationAsScan, readHooks).toProto())
-        .setRows(RowSet.newBuilder().addRowKeys(ByteString.copyFrom(operation.getRow())));
+
+    query.filter(scanAdapter.buildFilter(operationAsScan, readHooks))
+        .rowKey(ByteString.copyFrom(operation.getRow()));
   }
 
   private Get addKeyOnlyFilter(Get get) {
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ReadHooks.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ReadHooks.java
index c1df2ee0b..8d164b889 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ReadHooks.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ReadHooks.java
@@ -15,11 +15,11 @@
  */
 package com.google.cloud.bigtable.hbase.adapters.read;
 
-import com.google.bigtable.v2.ReadRowsRequest;
+import com.google.cloud.bigtable.data.v2.models.Query;
 import com.google.common.base.Function;
 
 /**
- * Hooks for modifying a ReadRowsRequest before being sent to Cloud Bigtable.
+ * Hooks for modifying a {@link Query} before being sent to Cloud Bigtable.
  *
  * Note that it is expected that this will be extended to include post-read
  * hooks to transform Rows when appropriate.
@@ -30,17 +30,16 @@ import com.google.common.base.Function;
 public interface ReadHooks {
 
   /**
-   * Add a Function that will modify the ReadRowsRequest before it is sent to Cloud Bigtable.
+   * Add a {@link Function} that will modify the {@link Query} before it is sent to Cloud Bigtable.
    *
-   * @param newHook a {@link com.google.common.base.Function} object.
+   * @param newHook a {@link Function} object.
    */
-  void composePreSendHook(Function<ReadRowsRequest, ReadRowsRequest> newHook);
+  void composePreSendHook(Function<Query, Query> newHook);
 
   /**
    * Apply all pre-send hooks to the given request.
    *
-   * @param readRowsRequest a {@link com.google.bigtable.v2.ReadRowsRequest} object.
-   * @return a {@link com.google.bigtable.v2.ReadRowsRequest} object.
+   * @param query a {@link Query} object.
    */
-  ReadRowsRequest applyPreSendHook(ReadRowsRequest readRowsRequest);
+  void applyPreSendHook(Query query);
 }
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ReadOperationAdapter.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ReadOperationAdapter.java
index a0dc32c29..0dbdde2dc 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ReadOperationAdapter.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ReadOperationAdapter.java
@@ -15,8 +15,7 @@
  */
 package com.google.cloud.bigtable.hbase.adapters.read;
 
-import com.google.bigtable.v2.ReadRowsRequest;
-
+import com.google.cloud.bigtable.data.v2.models.Query;
 import org.apache.hadoop.hbase.client.Operation;
 
 /**
@@ -30,8 +29,8 @@ public interface ReadOperationAdapter<T extends Operation> {
    * <p>adapt.</p>
    *
    * @param request a T object.
-   * @param readHooks a {@link com.google.cloud.bigtable.hbase.adapters.read.ReadHooks} object.
-   * @return a {@link com.google.bigtable.v2.ReadRowsRequest.Builder} object.
+   * @param readHooks a {@link ReadHooks} object.
+   * @param query a  {@link Query} object.
    */
-  ReadRowsRequest.Builder adapt(T request, ReadHooks readHooks);
+  void adapt(T request, ReadHooks readHooks, Query query);
 }
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/RowRangeAdapter.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/RowRangeAdapter.java
index 295acb2e9..0324d608c 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/RowRangeAdapter.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/RowRangeAdapter.java
@@ -17,7 +17,10 @@ package com.google.cloud.bigtable.hbase.adapters.read;
 
 import com.google.bigtable.v2.RowRange;
 import com.google.bigtable.v2.RowSet;
+import com.google.cloud.bigtable.data.v2.models.Query;
+import com.google.cloud.bigtable.data.v2.models.Range.ByteStringRange;
 import com.google.cloud.bigtable.util.RowKeyWrapper;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.BoundType;
 import com.google.common.collect.Range;
 import com.google.common.collect.RangeSet;
@@ -25,15 +28,16 @@ import com.google.common.collect.TreeRangeSet;
 import com.google.protobuf.ByteString;
 
 /**
- * Adapter to convert between a Bigtable RowSet and guava's RangeSet.
+ * Adapter to convert between a Bigtable {@link RowSet} and guava's {@link RangeSet}.
  */
 public class RowRangeAdapter {
 
   /**
-   * Convert bigtable's RowSet -> Guava's RangeSet.
+   * Convert bigtable's {@link RowSet} -> Guava's {@link RangeSet}.
    * Note that this will normalize the ranges, such that: overlapping keys and ranges will be merged
    * and empty range keys will be converted in boundless ranges.
    */
+  @VisibleForTesting
   RangeSet<RowKeyWrapper> rowSetToRangeSet(RowSet btRowSet) {
     RangeSet<RowKeyWrapper> rangeSet = TreeRangeSet.create();
     for (RowRange btRowRange : btRowSet.getRowRangesList()) {
@@ -46,7 +50,7 @@ public class RowRangeAdapter {
   }
 
   /**
-   * Convert Bigtable's RowRange -> guava Range.
+   * Convert Bigtable's {@link RowRange} -> guava {@link Range}.
    */
   private Range<RowKeyWrapper> rowRangeToRange(RowRange btRange) {
     final BoundType startBound;
@@ -69,10 +73,6 @@ public class RowRangeAdapter {
         throw new IllegalArgumentException("Unexpected start key case: " +
             btRange.getStartKeyCase());
     }
-    // Bigtable doesn't allow empty row keys, so an empty start row which is open or closed is
-    // considered unbounded. ie. all row keys are bigger than the empty key (no need to
-    // differentiate between open/closed)
-    final boolean startUnbounded = startKey.isEmpty();
 
     final BoundType endBound;
     final ByteString endKey;
@@ -92,6 +92,21 @@ public class RowRangeAdapter {
       default:
         throw new IllegalArgumentException("Unexpected end key case: " + btRange.getEndKeyCase());
     }
+
+    return boundedRange(startBound, startKey, endBound, endKey);
+  }
+
+  /**
+   * To determine {@link Range<RowKeyWrapper>} based start/end key & {@link BoundType}.
+   */
+  @VisibleForTesting
+  Range<RowKeyWrapper> boundedRange(BoundType startBound, ByteString startKey, BoundType endBound,
+      ByteString endKey) {
+    // Bigtable doesn't allow empty row keys, so an empty start row which is open or closed is
+    // considered unbounded. ie. all row keys are bigger than the empty key (no need to
+    // differentiate between open/closed)
+    final boolean startUnbounded = startKey.isEmpty();
+
     // Bigtable doesn't allow empty row keys, so an empty end row which is open or closed is
     // considered unbounded. ie. all row keys are smaller than the empty end key (no need to
     // differentiate between open/closed)
@@ -110,11 +125,11 @@ public class RowRangeAdapter {
   }
 
   /**
-   * Convert guava's RangeSet to Bigtable's RowSet. Please note that this will convert
+   * Convert guava's {@link RangeSet} to Bigtable's {@link ByteStringRange}. Please note that this will convert
    * boundless ranges into unset key cases.
    */
-  RowSet rangeSetToRowSet(RangeSet<RowKeyWrapper> guavaRangeSet) {
-    RowSet.Builder rowSet = RowSet.newBuilder();
+  @VisibleForTesting
+  void rangeSetToByteStringRange(RangeSet<RowKeyWrapper> guavaRangeSet, Query query) {
 
     for (Range<RowKeyWrapper> guavaRange : guavaRangeSet.asRanges()) {
       // Is it a point?
@@ -122,18 +137,18 @@ public class RowRangeAdapter {
           && guavaRange.hasUpperBound() && guavaRange.upperBoundType() == BoundType.CLOSED
           && guavaRange.lowerEndpoint().equals(guavaRange.upperEndpoint())) {
 
-        rowSet.addRowKeys(guavaRange.lowerEndpoint().getKey());
+        query.rowKey(guavaRange.lowerEndpoint().getKey());
       } else {
-        RowRange.Builder btRange = RowRange.newBuilder();
+        ByteStringRange byteRange = ByteStringRange.unbounded();
 
         // Handle start key
         if (guavaRange.hasLowerBound()) {
           switch (guavaRange.lowerBoundType()) {
             case CLOSED:
-              btRange.setStartKeyClosed(guavaRange.lowerEndpoint().getKey());
+              byteRange.startClosed(guavaRange.lowerEndpoint().getKey());
               break;
             case OPEN:
-              btRange.setStartKeyOpen(guavaRange.lowerEndpoint().getKey());
+              byteRange.startOpen(guavaRange.lowerEndpoint().getKey());
               break;
             default:
               throw new IllegalArgumentException("Unexpected lower bound type: "
@@ -145,19 +160,18 @@ public class RowRangeAdapter {
         if (guavaRange.hasUpperBound()) {
           switch (guavaRange.upperBoundType()) {
             case CLOSED:
-              btRange.setEndKeyClosed(guavaRange.upperEndpoint().getKey());
+              byteRange.endClosed(guavaRange.upperEndpoint().getKey());
               break;
             case OPEN:
-              btRange.setEndKeyOpen(guavaRange.upperEndpoint().getKey());
+              byteRange.endOpen(guavaRange.upperEndpoint().getKey());
               break;
             default:
               throw new IllegalArgumentException("Unexpected upper bound type: " +
                   guavaRange.upperBoundType());
           }
         }
-        rowSet.addRowRanges(btRange);
+        query.range(byteRange);
       }
     }
-    return rowSet.build();
   }
 }
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ScanAdapter.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ScanAdapter.java
index 94305b1ce..6cf238991 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ScanAdapter.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/ScanAdapter.java
@@ -17,9 +17,9 @@ package com.google.cloud.bigtable.hbase.adapters.read;
 
 import static com.google.cloud.bigtable.data.v2.models.Filters.FILTERS;
 
+import com.google.cloud.bigtable.data.v2.models.Query;
+import com.google.common.collect.BoundType;
 import com.google.common.collect.Range;
-import com.google.bigtable.v2.ReadRowsRequest;
-import com.google.bigtable.v2.RowRange;
 import com.google.bigtable.v2.RowSet;
 import com.google.cloud.bigtable.data.v2.models.Filters;
 import com.google.cloud.bigtable.data.v2.models.Filters.ChainFilter;
@@ -32,6 +32,7 @@ import com.google.cloud.bigtable.hbase.adapters.filters.FilterAdapterContext;
 import com.google.cloud.bigtable.util.RowKeyWrapper;
 import com.google.common.base.Optional;
 import com.google.common.collect.RangeSet;
+import com.google.common.collect.TreeRangeSet;
 import com.google.protobuf.ByteString;
 
 import org.apache.hadoop.hbase.client.Scan;
@@ -43,7 +44,7 @@ import java.util.Map;
 import java.util.NavigableSet;
 
 /**
- * An adapter for Scan operation that makes use of the proto filter language.
+ * An adapter for {@link Scan} operation that makes use of the proto filter language.
  *
  * @author sduskis
  * @version $Id: $Id
@@ -82,7 +83,7 @@ public class ScanAdapter implements ReadOperationAdapter<Scan> {
   /**
    * <p>Constructor for ScanAdapter.</p>
    *
-   * @param filterAdapter a {@link com.google.cloud.bigtable.hbase.adapters.filters.FilterAdapter} object.
+   * @param filterAdapter a {@link FilterAdapter} object.
    */
   public ScanAdapter(FilterAdapter filterAdapter, RowRangeAdapter rowRangeAdapter) {
     this.filterAdapter = filterAdapter;
@@ -92,7 +93,7 @@ public class ScanAdapter implements ReadOperationAdapter<Scan> {
   /**
    * <p>throwIfUnsupportedScan.</p>
    *
-   * @param scan a {@link org.apache.hadoop.hbase.client.Scan} object.
+   * @param scan a {@link Scan} object.
    */
   public void throwIfUnsupportedScan(Scan scan) {
     if (scan.getFilter() != null) {
@@ -106,11 +107,11 @@ public class ScanAdapter implements ReadOperationAdapter<Scan> {
   }
 
   /**
-   * Given a Scan, build a RowFilter that include matching columns
+   * Given a {@link Scan}, build a {@link Filters.Filter} that include matching columns
    *
-   * @param scan a {@link org.apache.hadoop.hbase.client.Scan} object.
-   * @param hooks a {@link com.google.cloud.bigtable.hbase.adapters.read.ReadHooks} object.
-   * @return a {@link com.google.bigtable.v2.RowFilter} object.
+   * @param scan a {@link Scan} object.
+   * @param hooks a {@link ReadHooks} object.
+   * @return a {@link Filters.Filter} object.
    */
   public Filters.Filter buildFilter(Scan scan, ReadHooks hooks) {
     ChainFilter chain = FILTERS.chain();
@@ -137,60 +138,46 @@ public class ScanAdapter implements ReadOperationAdapter<Scan> {
 
   /** {@inheritDoc} */
   @Override
-  public ReadRowsRequest.Builder adapt(Scan scan, ReadHooks readHooks) {
+  public void adapt(Scan scan, ReadHooks readHooks, Query query) {
     throwIfUnsupportedScan(scan);
 
-    ReadRowsRequest.Builder requestBuilder = ReadRowsRequest.newBuilder()
-        .setRows(toRowSet(scan))
-        .setFilter(buildFilter(scan, readHooks).toProto());
+    toByteStringRange(scan, query);
+    query.filter(buildFilter(scan, readHooks));
 
     if (LIMIT_AVAILABLE && scan.getLimit() > 0) {
-      requestBuilder.setRowsLimit(scan.getLimit());
+      query.limit(scan.getLimit());
     }
-
-    return requestBuilder;
   }
 
-  private RowSet toRowSet(Scan scan) {
-    RowSet rowSet = getRowSet(scan);
-    return narrowRowSet(rowSet, scan.getFilter());
+  private void toByteStringRange(Scan scan, Query query) {
+    RangeSet<RowKeyWrapper> rangeSet = narrowRange(getRangeSet(scan), scan.getFilter());
+    rowRangeAdapter.rangeSetToByteStringRange(rangeSet, query);
   }
 
-  private RowSet getRowSet(Scan scan) {
+  private RangeSet<RowKeyWrapper> getRangeSet(Scan scan) {
     if (scan instanceof BigtableExtendedScan) {
-      return ((BigtableExtendedScan) scan).getRowSet();
+      RowSet rowSet = ((BigtableExtendedScan) scan).getRowSet();
+      return rowRangeAdapter.rowSetToRangeSet(rowSet);
     } else {
-      RowSet.Builder rowSetBuilder = RowSet.newBuilder();
-      ByteString startRow = ByteString.copyFrom(scan.getStartRow());
+      RangeSet<RowKeyWrapper> rangeSet = TreeRangeSet.create();
+      final ByteString startRow = ByteString.copyFrom(scan.getStartRow());
+      final ByteString stopRow = ByteString.copyFrom(scan.getStopRow());
+
       if (scan.isGetScan()) {
-        rowSetBuilder.addRowKeys(startRow);
+        rangeSet.add(Range.singleton(new RowKeyWrapper(startRow)));
       } else {
-        RowRange.Builder range =  RowRange.newBuilder();
-        if (!startRow.isEmpty()) {
-          if (!OPEN_CLOSED_AVAILABLE || scan.includeStartRow()) {
-            // the default for start is closed
-            range.setStartKeyClosed(startRow);
-          } else {
-            range.setStartKeyOpen(startRow);
-          }
-        }
+        final BoundType startBound =
+            (!OPEN_CLOSED_AVAILABLE || scan.includeStartRow()) ? BoundType.CLOSED : BoundType.OPEN;
+        final BoundType endBound =
+            (!OPEN_CLOSED_AVAILABLE || !scan.includeStopRow()) ? BoundType.OPEN : BoundType.CLOSED;
 
-        ByteString stopRow = ByteString.copyFrom(scan.getStopRow());
-        if (!stopRow.isEmpty()) {
-          if (!OPEN_CLOSED_AVAILABLE || !scan.includeStopRow()) {
-            // the default for stop is open
-            range.setEndKeyOpen(stopRow);
-          } else {
-            range.setEndKeyClosed(stopRow);
-          }
-        }
-        rowSetBuilder.addRowRanges(range);
+        rangeSet.add(rowRangeAdapter.boundedRange(startBound, startRow, endBound, stopRow));
       }
-      return rowSetBuilder.build();
+      return rangeSet;
     }
   }
 
-  private static ByteString quoteRegex(byte[] unquoted)  {
+  private static ByteString quoteRegex(byte[] unquoted) {
     try {
       return ReaderExpressionHelper.quoteRegularExpression(unquoted);
     } catch (IOException e) {
@@ -204,25 +191,22 @@ public class ScanAdapter implements ReadOperationAdapter<Scan> {
       return Optional.absent();
     }
     try {
-      return filterAdapter
-          .adaptFilter(new FilterAdapterContext(scan, hooks), scan.getFilter());
+      return filterAdapter.adaptFilter(new FilterAdapterContext(scan, hooks), scan.getFilter());
     } catch (IOException ioe) {
       throw new RuntimeException("Failed to adapt filter", ioe);
     }
   }
 
-  private RowSet narrowRowSet(RowSet rowSet, Filter filter) {
+  private RangeSet<RowKeyWrapper> narrowRange(RangeSet<RowKeyWrapper> rangeSet, Filter filter) {
     if (filter == null) {
-      return rowSet;
+      return rangeSet;
     }
     RangeSet<RowKeyWrapper> filterRangeSet = filterAdapter.getIndexScanHint(filter);
     if (filterRangeSet.encloses(Range.<RowKeyWrapper>all())) {
-      return rowSet;
+      return rangeSet;
     }
-    RangeSet<RowKeyWrapper> scanRangeSet = rowRangeAdapter.rowSetToRangeSet(rowSet);
-    // intersection of scan ranges & filter ranges
-    scanRangeSet.removeAll(filterRangeSet.complement());
-    return rowRangeAdapter.rangeSetToRowSet(scanRangeSet);
+    rangeSet.removeAll(filterRangeSet.complement());
+    return rangeSet;
   }
 
   private Filters.Filter createColumnQualifierFilter(byte[] unquotedQualifier) {
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/org/apache/hadoop/hbase/client/AbstractBigtableAdmin.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/org/apache/hadoop/hbase/client/AbstractBigtableAdmin.java
index 762049d54..7b8e6728c 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/org/apache/hadoop/hbase/client/AbstractBigtableAdmin.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/org/apache/hadoop/hbase/client/AbstractBigtableAdmin.java
@@ -19,18 +19,14 @@ import static com.google.cloud.bigtable.hbase.util.ModifyTableBuilder.buildModif
 
 import com.google.bigtable.admin.v2.CreateTableFromSnapshotRequest;
 import com.google.bigtable.admin.v2.DeleteSnapshotRequest;
-import com.google.bigtable.admin.v2.DeleteTableRequest;
-import com.google.bigtable.admin.v2.DeleteTableRequest.Builder;
-import com.google.bigtable.admin.v2.DropRowRangeRequest;
 import com.google.bigtable.admin.v2.GetTableRequest;
-import com.google.bigtable.admin.v2.ListTablesRequest;
-import com.google.bigtable.admin.v2.ListTablesResponse;
 import com.google.bigtable.admin.v2.SnapshotTableRequest;
-import com.google.bigtable.admin.v2.Table;
 import com.google.cloud.bigtable.admin.v2.models.CreateTableRequest;
 import com.google.cloud.bigtable.admin.v2.models.ModifyColumnFamiliesRequest;
+import com.google.cloud.bigtable.admin.v2.models.Table;
 import com.google.cloud.bigtable.config.BigtableOptions;
 import com.google.cloud.bigtable.config.Logger;
+import com.google.cloud.bigtable.core.IBigtableTableAdminClient;
 import com.google.cloud.bigtable.grpc.BigtableClusterName;
 import com.google.cloud.bigtable.grpc.BigtableInstanceName;
 import com.google.cloud.bigtable.grpc.BigtableTableAdminClient;
@@ -44,7 +40,6 @@ import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 import com.google.longrunning.Operation;
-import com.google.protobuf.ByteString;
 import com.google.protobuf.Duration;
 import io.grpc.Status;
 import org.apache.hadoop.conf.Configuration;
@@ -104,6 +99,7 @@ public abstract class AbstractBigtableAdmin implements Admin {
   private final BigtableOptions options;
   protected final CommonConnection connection;
   protected final BigtableTableAdminClient bigtableTableAdminClient;
+  protected final IBigtableTableAdminClient tableAdminClientWrapper;
   protected final BigtableInstanceName bigtableInstanceName;
   private BigtableClusterName bigtableSnapshotClusterName;
   protected final TableAdapter tableAdapter;
@@ -124,6 +120,7 @@ public abstract class AbstractBigtableAdmin implements Admin {
     disabledTables = connection.getDisabledTables();
     bigtableInstanceName = options.getInstanceName();
     tableAdapter = new TableAdapter(bigtableInstanceName);
+    tableAdminClientWrapper = connection.getSession().getTableAdminClientWrapper();
 
     String clusterId = configuration.get(BigtableOptionsFactory.BIGTABLE_SNAPSHOT_CLUSTER_ID_KEY, null);
     if (clusterId != null) {
@@ -137,7 +134,6 @@ public abstract class AbstractBigtableAdmin implements Admin {
     return (Connection) connection;
   }
 
-  /** {@inheritDoc} */
   /** {@inheritDoc} */
   @Override
   public boolean tableExists(TableName tableName) throws IOException {
@@ -245,36 +241,12 @@ public abstract class AbstractBigtableAdmin implements Admin {
    * Lists all table names for the cluster provided in the configuration.
    */
   public TableName[] listTableNames() throws IOException {
-    return asTableNames(requestTableList().getTablesList());
-  }
+    //tablesList contains list of tableId.
+    List<String> tablesList = tableAdminClientWrapper.listTables();
 
-  /**
-   * Request a list of Tables for the cluster.  The {@link Table}s in the response will only
-   * contain fully qualified Bigtable table names, and not column family information.
-   */
-  private ListTablesResponse requestTableList() throws IOException {
-    try {
-      ListTablesRequest.Builder builder = ListTablesRequest.newBuilder();
-      builder.setParent(bigtableInstanceName.toString());
-      return bigtableTableAdminClient.listTables(builder.build());
-    } catch (Throwable throwable) {
-      throw new IOException("Failed to listTables", throwable);
-    }
-  }
-
-  /**
-   * Convert a list of Bigtable {@link Table}s to hbase {@link TableName}.
-   */
-  private TableName[] asTableNames(List<Table> tablesList) {
     TableName[] result = new TableName[tablesList.size()];
     for (int i = 0; i < tablesList.size(); i++) {
-      // This will contain things like project, zone and cluster.
-      String bigtableFullTableName = tablesList.get(i).getName();
-
-      // Strip out the Bigtable info.
-      String name = bigtableInstanceName.toTableId(bigtableFullTableName);
-
-      result[i] = TableName.valueOf(name);
+      result[i] = TableName.valueOf(tablesList.get(i));
     }
     return result;
   }
@@ -362,8 +334,7 @@ public abstract class AbstractBigtableAdmin implements Admin {
   protected void createTable(TableName tableName, CreateTableRequest request)
           throws IOException {
     try {
-      bigtableTableAdminClient.createTable(
-              request.toProto(options.getProjectId(), options.getInstanceId()));
+      tableAdminClientWrapper.createTable(request);
     } catch (Throwable throwable) {
       throw convertToTableExistsException(tableName, throwable);
     }
@@ -384,8 +355,7 @@ public abstract class AbstractBigtableAdmin implements Admin {
   protected ListenableFuture<Table> createTableAsync(final TableName tableName,
             CreateTableRequest request) throws IOException {
     ListenableFuture<Table> future =
-            bigtableTableAdminClient.createTableAsync(
-                    request.toProto(options.getProjectId(), options.getInstanceId()));
+        tableAdminClientWrapper.createTableAsync(request);
     final SettableFuture<Table> settableFuture = SettableFuture.create();
     Futures.addCallback(future, new FutureCallback<Table>() {
       @Override public void onSuccess(@Nullable Table result) {
@@ -410,10 +380,8 @@ public abstract class AbstractBigtableAdmin implements Admin {
   /** {@inheritDoc} */
   @Override
   public void deleteTable(TableName tableName) throws IOException {
-    Builder deleteBuilder = DeleteTableRequest.newBuilder();
-    deleteBuilder.setName(toBigtableName(tableName));
     try {
-      bigtableTableAdminClient.deleteTable(deleteBuilder.build());
+      tableAdminClientWrapper.deleteTable(tableName.getNameAsString());
     } catch (Throwable throwable) {
       throw new IOException(
           String.format(
@@ -611,8 +579,7 @@ public abstract class AbstractBigtableAdmin implements Admin {
       try {
         ModifyColumnFamiliesRequest request =
             buildModifications(newDecriptor, getTableDescriptor(tableName)).build();
-        bigtableTableAdminClient.modifyColumnFamily(
-                request.toProto(options.getProjectId(), options.getInstanceId()));
+        tableAdminClientWrapper.modifyFamilies(request);
       } catch (Throwable throwable) {
         throw new IOException(
             String.format("Failed to modify table '%s'", tableName.getNameAsString()), throwable);
@@ -633,10 +600,8 @@ public abstract class AbstractBigtableAdmin implements Admin {
    */
   protected Void modifyColumns(TableName tableName, String columnName,
       String modificationType, ModifyTableBuilder builder) throws IOException {
-    ModifyColumnFamiliesRequest request = builder.build();
     try {
-      bigtableTableAdminClient.modifyColumnFamily(
-              request.toProto(options.getProjectId(), options.getInstanceId()));
+      tableAdminClientWrapper.modifyFamilies(builder.build());
       return null;
     } catch (Throwable throwable) {
       throw new IOException(
@@ -778,7 +743,7 @@ public abstract class AbstractBigtableAdmin implements Admin {
     if (!preserveSplits) {
       LOG.info("truncate will preserveSplits. The passed in variable is ignored.");
     }
-    issueBulkDelete(tableName, DropRowRangeRequest.newBuilder().setDeleteAllDataFromTable(true));
+    issueBulkDelete(tableName, "");
     disabledTables.remove(tableName);
   }
 
@@ -790,18 +755,14 @@ public abstract class AbstractBigtableAdmin implements Admin {
    * @throws java.io.IOException if any.
    */
   public void deleteRowRangeByPrefix(TableName tableName, byte[] prefix) throws IOException {
-    issueBulkDelete(
-        tableName,
-        DropRowRangeRequest.newBuilder()
-            .setDeleteAllDataFromTable(false)
-            .setRowKeyPrefix(ByteString.copyFrom(prefix)));
+    issueBulkDelete(tableName, Bytes.toString(prefix));
   }
 
-  private void issueBulkDelete(TableName tableName, DropRowRangeRequest.Builder deleteRequest)
+  private void issueBulkDelete(TableName tableName, String rowKeyPrefix)
       throws IOException {
     try {
-      bigtableTableAdminClient
-          .dropRowRange(deleteRequest.setName(toBigtableName(tableName)).build());
+      tableAdminClientWrapper
+          .dropRowRange(tableName.getNameAsString(), rowKeyPrefix);
     } catch (Throwable throwable) {
       throw new IOException(
           String.format("Failed to truncate table '%s'", tableName.getNameAsString()), throwable);
diff --git a/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/TemplateUtils.java b/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/TemplateUtils.java
index 1e1c628f2..f2198b4fa 100644
--- a/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/TemplateUtils.java
+++ b/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/TemplateUtils.java
@@ -15,7 +15,14 @@
  */
 package com.google.cloud.bigtable.beam;
 
+import static com.google.cloud.bigtable.beam.CloudBigtableScanConfiguration.PLACEHOLDER_APP_PROFILE_ID;
+import static com.google.cloud.bigtable.beam.CloudBigtableScanConfiguration.PLACEHOLDER_INSTANCE_ID;
+import static com.google.cloud.bigtable.beam.CloudBigtableScanConfiguration.PLACEHOLDER_PROJECT_ID;
+import static com.google.cloud.bigtable.beam.CloudBigtableScanConfiguration.PLACEHOLDER_TABLE_ID;
+
 import com.google.bigtable.repackaged.com.google.bigtable.v2.ReadRowsRequest;
+import com.google.bigtable.repackaged.com.google.cloud.bigtable.data.v2.internal.RequestContext;
+import com.google.bigtable.repackaged.com.google.cloud.bigtable.data.v2.models.Query;
 import com.google.cloud.bigtable.beam.sequencefiles.ExportJob.ExportOptions;
 import com.google.cloud.bigtable.beam.sequencefiles.ImportJob.ImportOptions;
 import com.google.cloud.bigtable.hbase.adapters.Adapters;
@@ -95,8 +102,14 @@ public class TemplateUtils {
         }
 
         ReadHooks readHooks = new DefaultReadHooks();
-        ReadRowsRequest.Builder builder = Adapters.SCAN_ADAPTER.adapt(scan, readHooks);
-        cachedRequest = readHooks.applyPreSendHook(builder.build());
+        Query query = Query.create(PLACEHOLDER_TABLE_ID);
+        Adapters.SCAN_ADAPTER.adapt(scan, readHooks, query);
+        readHooks.applyPreSendHook(query);
+        RequestContext requestContext = RequestContext.create(PLACEHOLDER_PROJECT_ID,
+            PLACEHOLDER_INSTANCE_ID, PLACEHOLDER_APP_PROFILE_ID);
+
+        cachedRequest =
+            query.toProto(requestContext).toBuilder().setTableName("").setAppProfileId("").build();
       }
       return cachedRequest;
     }
diff --git a/bigtable-dataflow-parent/bigtable-hbase-beam/src/main/java/com/google/cloud/bigtable/beam/CloudBigtableScanConfiguration.java b/bigtable-dataflow-parent/bigtable-hbase-beam/src/main/java/com/google/cloud/bigtable/beam/CloudBigtableScanConfiguration.java
index ff2cbfa2d..2a1e65bc0 100644
--- a/bigtable-dataflow-parent/bigtable-hbase-beam/src/main/java/com/google/cloud/bigtable/beam/CloudBigtableScanConfiguration.java
+++ b/bigtable-dataflow-parent/bigtable-hbase-beam/src/main/java/com/google/cloud/bigtable/beam/CloudBigtableScanConfiguration.java
@@ -46,6 +46,10 @@ import com.google.cloud.bigtable.hbase.adapters.read.ReadHooks;
 public class CloudBigtableScanConfiguration extends CloudBigtableTableConfiguration {
 
   private static final long serialVersionUID = 2435897354284600685L;
+  protected static final String PLACEHOLDER_TABLE_ID = "PLACEHOLDER_TABLE_ID";
+  protected static final String PLACEHOLDER_PROJECT_ID = "PLACEHOLDER_PROJECT_ID";
+  protected static final String PLACEHOLDER_INSTANCE_ID = "PLACEHOLDER_INSTANCE_ID";
+  protected static final String PLACEHOLDER_APP_PROFILE_ID = "PLACEHOLDER_APP_PROFILE_ID";
 
   /**
    * Converts a {@link CloudBigtableTableConfiguration} object to a
@@ -79,15 +83,19 @@ public class CloudBigtableScanConfiguration extends CloudBigtableTableConfigurat
      */
     public Builder withScan(Scan scan) {
       Preconditions.checkArgument(scan != null, "Scan cannot be null");
+      Query query = Query.create(PLACEHOLDER_TABLE_ID);
       ReadHooks readHooks = new DefaultReadHooks();
-      ReadRowsRequest.Builder builder = Adapters.SCAN_ADAPTER.adapt(scan, readHooks);
-      withRequest(readHooks.applyPreSendHook(builder.build()));
+      Adapters.SCAN_ADAPTER.adapt(scan, readHooks, query);
+      readHooks.applyPreSendHook(query);
+      withQuery(query);
       return this;
     }
 
     Builder withQuery(Query query) {
-      RequestContext dummyContext = RequestContext.create("Dummy Project", "Dummy Instance", "");
-      return this.withRequest(query.toProto(dummyContext).toBuilder().setTableName("").build());
+      RequestContext dummyContext = RequestContext
+          .create(PLACEHOLDER_PROJECT_ID, PLACEHOLDER_INSTANCE_ID, PLACEHOLDER_APP_PROFILE_ID);
+      return this.withRequest(
+          query.toProto(dummyContext).toBuilder().setTableName("").setAppProfileId("").build());
     }
 
     /**
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java
index f3627d47d..e7718b008 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java
@@ -15,9 +15,6 @@
  */
 package com.google.cloud.bigtable.hbase2_x;
 
-import static com.google.cloud.bigtable.hbase.adapters.admin.ColumnDescriptorAdapter.buildGarbageCollectionRule;
-import static com.google.cloud.bigtable.hbase.util.ModifyTableBuilder.buildModifications;
-
 import com.google.cloud.bigtable.hbase.util.ModifyTableBuilder;
 import java.io.IOException;
 import java.util.ArrayList;
@@ -65,12 +62,9 @@ import org.apache.hadoop.hbase.snapshot.SnapshotCreationException;
 import org.apache.hadoop.hbase.snapshot.UnknownSnapshotException;
 import org.apache.hadoop.hbase.util.Bytes;
 
-import com.google.bigtable.admin.v2.DeleteTableRequest;
-import com.google.bigtable.admin.v2.DropRowRangeRequest;
 import com.google.bigtable.admin.v2.ListSnapshotsRequest;
 import com.google.bigtable.admin.v2.ListSnapshotsResponse;
 import com.google.bigtable.admin.v2.Snapshot;
-import com.google.cloud.bigtable.admin.v2.models.ModifyColumnFamiliesRequest;
 import com.google.cloud.bigtable.hbase2_x.adapters.admin.TableAdapter2x;
 import com.google.common.util.concurrent.Futures;
 
@@ -244,11 +238,8 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
   }
 
   protected CompletableFuture<Void> deleteTableAsyncInternal(TableName tableName) {
-    DeleteTableRequest deleteTableRequest = DeleteTableRequest.newBuilder()
-        .setName(toBigtableName(tableName))
-        .build();
     return FutureUtils.toCompletableFuture(
-        bigtableTableAdminClient.deleteTableAsync(deleteTableRequest))
+        tableAdminClientWrapper.deleteTableAsync(tableName.getNameAsString()))
         .thenApply(r -> null);
   }
 
@@ -353,10 +344,10 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
    if (!preserveSplits) {
       LOG.info("truncate will preserveSplits. The passed in variable is ignored.");
    }
-   DropRowRangeRequest.Builder deleteRequest = DropRowRangeRequest.newBuilder().setDeleteAllDataFromTable(true);
+   // rowKeyPrefix set to Empty String which truncates data from specified table.
    return FutureUtils.toCompletableFuture(
-        bigtableTableAdminClient
-          .dropRowRangeAsync(deleteRequest.setName(toBigtableName(tableName)).build()))
+        tableAdminClientWrapper
+          .dropRowRangeAsync(tableName.getNameAsString(), ""))
         .thenApply(r -> null);
   }
   /* ******* Unsupported methods *********** */


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminClientWrapper.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminClientWrapper.java
index adc4a759c..f7f33c807 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminClientWrapper.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminClientWrapper.java
@@ -216,9 +216,10 @@ public class TestBigtableTableAdminClientWrapper {
   }
 
   @Test
-  public void dropRowRange(){
+  public void dropRowRangeForDeleteByPrefix(){
     DropRowRangeRequest request = DropRowRangeRequest.newBuilder()
         .setName(TABLE_NAME)
+        .setDeleteAllDataFromTable(false)
         .setRowKeyPrefix(ByteString.copyFromUtf8(ROW_KEY_PREFIX))
         .build();
 
@@ -229,9 +230,23 @@ public class TestBigtableTableAdminClientWrapper {
   }
 
   @Test
-  public void dropRowRangeAsync(){
+  public void dropRowRangeForTruncate(){
     DropRowRangeRequest request = DropRowRangeRequest.newBuilder()
         .setName(TABLE_NAME)
+        .setDeleteAllDataFromTable(true)
+        .build();
+
+    doNothing().when(mockAdminClient).dropRowRange(request);
+    clientWrapper.dropRowRange(TABLE_ID, null);
+
+    verify(mockAdminClient).dropRowRange(request);
+  }
+
+  @Test
+  public void dropRowRangeAsyncForDeleteByPrefix(){
+    DropRowRangeRequest request = DropRowRangeRequest.newBuilder()
+        .setName(TABLE_NAME)
+        .setDeleteAllDataFromTable(false)
         .setRowKeyPrefix(ByteString.copyFromUtf8(ROW_KEY_PREFIX))
         .build();
 
@@ -242,6 +257,20 @@ public class TestBigtableTableAdminClientWrapper {
     verify(mockAdminClient).dropRowRangeAsync(request);
   }
 
+  @Test
+  public void dropRowRangeAsyncForTruncate(){
+    DropRowRangeRequest request = DropRowRangeRequest.newBuilder()
+        .setName(TABLE_NAME)
+        .setDeleteAllDataFromTable(true)
+        .build();
+
+    when(mockAdminClient.dropRowRangeAsync(request))
+        .thenReturn(immediateFuture(Empty.newBuilder().build()));
+    clientWrapper.dropRowRangeAsync(TABLE_ID, null);
+
+    verify(mockAdminClient).dropRowRangeAsync(request);
+  }
+
   private static com.google.bigtable.admin.v2.Table createTableData(){
     GCRules.GCRule gcRule = GCRULES.maxVersions(1);
     ColumnFamily columnFamily = ColumnFamily.newBuilder()
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/filters/TestPageFilterAdapter.java b/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/filters/TestPageFilterAdapter.java
index 94dba37a2..1c4a17eb8 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/filters/TestPageFilterAdapter.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/filters/TestPageFilterAdapter.java
@@ -15,8 +15,9 @@
  */
 package com.google.cloud.bigtable.hbase.adapters.filters;
 
-import com.google.bigtable.v2.ReadRowsRequest;
+import com.google.cloud.bigtable.data.v2.internal.RequestContext;
 import com.google.cloud.bigtable.data.v2.models.Filters;
+import com.google.cloud.bigtable.data.v2.models.Query;
 import com.google.cloud.bigtable.hbase.adapters.filters.FilterAdapterContext.ContextCloseable;
 import com.google.cloud.bigtable.hbase.adapters.read.DefaultReadHooks;
 import com.google.cloud.bigtable.hbase.adapters.read.ReadHooks;
@@ -110,14 +111,17 @@ public class TestPageFilterAdapter {
 
   @Test
   public void pageFilterIsAppliedToReadRowsRequest() throws IOException {
+    final String TABLE_ID = "tableId";
+    final RequestContext requestContext =
+        RequestContext.create("ProjectId", "InstanceId", "AppProfile");
     ReadHooks hooks = new DefaultReadHooks();
     FilterAdapterContext context = new FilterAdapterContext(new Scan(), hooks);
     PageFilter pageFilter = new PageFilter(20);
     Filters.Filter adaptedFilter = pageFilterAdapter.adapt(context, pageFilter);
     Assert.assertNull("PageFilterAdapter should not return a Filters.Filter.", adaptedFilter);
 
-    ReadRowsRequest request = ReadRowsRequest.newBuilder().setRowsLimit(100).build();
-    ReadRowsRequest postHookRequest = hooks.applyPreSendHook(request);
-    Assert.assertEquals(20, postHookRequest.getRowsLimit());
+    Query query = Query.create(TABLE_ID).limit(100);
+    hooks.applyPreSendHook(query);
+    Assert.assertEquals(20, query.toProto(requestContext).getRowsLimit());
   }
 }
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/read/TestGetAdapter.java b/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/read/TestGetAdapter.java
index 2cedb762e..997a83f1a 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/read/TestGetAdapter.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/read/TestGetAdapter.java
@@ -17,8 +17,9 @@ package com.google.cloud.bigtable.hbase.adapters.read;
 
 import static com.google.cloud.bigtable.data.v2.models.Filters.FILTERS;
 
-import com.google.bigtable.v2.ReadRowsRequest;
+import com.google.cloud.bigtable.data.v2.internal.RequestContext;
 import com.google.cloud.bigtable.data.v2.models.Filters;
+import com.google.cloud.bigtable.data.v2.models.Query;
 import com.google.cloud.bigtable.hbase.DataGenerationHelper;
 import com.google.cloud.bigtable.hbase.adapters.filters.FilterAdapter;
 import com.google.common.base.Function;
@@ -44,17 +45,21 @@ public class TestGetAdapter {
   public static final String PREFIX_DATA = "rk1";
   public static final String FAMILY_ID = "f1";
   public static final String QUALIFIER_ID = "q1";
+
+  private final RequestContext requestContext =
+      RequestContext.create("ProjectId", "InstanceId", "AppProfile");
+  private final Query query = Query.create("tableId");
   private GetAdapter getAdapter =
       new GetAdapter(new ScanAdapter(FilterAdapter.buildAdapter(), new RowRangeAdapter()));
   private DataGenerationHelper dataHelper = new DataGenerationHelper();
   private ReadHooks throwingReadHooks = new ReadHooks() {
     @Override
-    public void composePreSendHook(Function<ReadRowsRequest, ReadRowsRequest> newHook) {
+    public void composePreSendHook(Function<Query, Query> newHook) {
       throw new IllegalStateException("Read hooks not supported in tests.");
     }
 
     @Override
-    public ReadRowsRequest applyPreSendHook(ReadRowsRequest readRowsRequest) {
+    public void applyPreSendHook(Query query) {
       throw new IllegalStateException("Read hooks not supported in tests.");
     }
   };
@@ -68,8 +73,9 @@ public class TestGetAdapter {
   @Test
   public void rowKeyIsSetInRequest() throws IOException {
     Get get = makeValidGet(dataHelper.randomData(PREFIX_DATA));
-    ReadRowsRequest.Builder rowRequestBuilder = getAdapter.adapt(get, throwingReadHooks);
-    ByteString adaptedRowKey = rowRequestBuilder.getRows().getRowKeys(0);
+    getAdapter.adapt(get, throwingReadHooks, query);
+
+    ByteString adaptedRowKey = query.toProto(requestContext).getRows().getRowKeys(0);
     Assert.assertEquals(
         new String(get.getRow(), StandardCharsets.UTF_8),
         adaptedRowKey.toStringUtf8());
@@ -79,33 +85,33 @@ public class TestGetAdapter {
   public void maxVersionsIsSet() throws IOException {
     Get get = makeValidGet(dataHelper.randomData(PREFIX_DATA));
     get.setMaxVersions(10);
-    ReadRowsRequest.Builder rowRequestBuilder = getAdapter.adapt(get, throwingReadHooks);
+    getAdapter.adapt(get, throwingReadHooks, query);
     Assert.assertEquals(
         FILTERS.limit().cellsPerColumn(10).toProto(),
-        rowRequestBuilder.getFilter());
+        query.toProto(requestContext).getFilter());
   }
 
   @Test
   public void columnFamilyIsSet() throws IOException {
     Get get = makeValidGet(dataHelper.randomData(PREFIX_DATA));
     get.addFamily(Bytes.toBytes(FAMILY_ID));
-    ReadRowsRequest.Builder rowRequestBuilder = getAdapter.adapt(get, throwingReadHooks);
+    getAdapter.adapt(get, throwingReadHooks, query);
     Assert.assertEquals(
         FILTERS.family().exactMatch(FAMILY_ID).toProto(),
-        rowRequestBuilder.getFilter());
+        query.toProto(requestContext).getFilter());
   }
 
   @Test
   public void columnQualifierIsSet() throws IOException {
     Get get = makeValidGet(dataHelper.randomData(PREFIX_DATA));
     get.addColumn(Bytes.toBytes(FAMILY_ID), Bytes.toBytes(QUALIFIER_ID));
-    ReadRowsRequest.Builder rowRequestBuilder = getAdapter.adapt(get, throwingReadHooks);
+    getAdapter.adapt(get, throwingReadHooks, query);
     Assert.assertEquals(
         FILTERS.chain()
             .filter(FILTERS.family().regex(FAMILY_ID))
             .filter(FILTERS.qualifier().regex(QUALIFIER_ID))
             .toProto(),
-        rowRequestBuilder.getFilter());
+            query.toProto(requestContext).getFilter());
   }
 
   @Test
@@ -113,7 +119,7 @@ public class TestGetAdapter {
     Get get = makeValidGet(dataHelper.randomData(PREFIX_DATA));
     get.addColumn(Bytes.toBytes(FAMILY_ID), Bytes.toBytes(QUALIFIER_ID));
     get.addColumn(Bytes.toBytes(FAMILY_ID), Bytes.toBytes("q2"));
-    ReadRowsRequest.Builder rowRequestBuilder = getAdapter.adapt(get, throwingReadHooks);
+    getAdapter.adapt(get, throwingReadHooks, query);
     Assert.assertEquals(
         FILTERS.chain()
             .filter(FILTERS.family().regex(FAMILY_ID))
@@ -121,7 +127,7 @@ public class TestGetAdapter {
                 .filter(FILTERS.qualifier().regex(QUALIFIER_ID))
                 .filter(FILTERS.qualifier().regex("q2")))
             .toProto(),
-        rowRequestBuilder.getFilter());
+        query.toProto(requestContext).getFilter());
   }
 
   @Test
@@ -141,12 +147,12 @@ public class TestGetAdapter {
   public void testAddKeyOnlyFilterWithoutFilter() throws IOException {
     Get get = makeValidGet(dataHelper.randomData(PREFIX_DATA));
     get.setCheckExistenceOnly(true);
-    ReadRowsRequest.Builder rowRequestBuilder = getAdapter.adapt(get, throwingReadHooks);
+    getAdapter.adapt(get, throwingReadHooks, query);
     Assert.assertEquals(
         FILTERS.chain().filter(FILTERS.limit().cellsPerColumn(1))
             .filter(FILTERS.value().strip())
             .toProto(),
-        rowRequestBuilder.getFilter()
+        query.toProto(requestContext).getFilter()
     );
   }
 
@@ -155,7 +161,7 @@ public class TestGetAdapter {
     Get get = makeValidGet(dataHelper.randomData(PREFIX_DATA));
     get.setCheckExistenceOnly(true);
     get.setFilter(new KeyOnlyFilter());
-    ReadRowsRequest.Builder rowRequestBuilder = getAdapter.adapt(get, throwingReadHooks);
+    getAdapter.adapt(get, throwingReadHooks, query);
     Filters.Filter filterOne = FILTERS.chain()
         .filter(FILTERS.limit().cellsPerColumn(1))
         .filter(FILTERS.value().strip());
@@ -164,6 +170,6 @@ public class TestGetAdapter {
         .filter(FILTERS.value().strip());
     Assert.assertEquals(
         FILTERS.chain().filter(filterOne).filter(filterTwo).toProto(),
-        rowRequestBuilder.getFilter());
+        query.toProto(requestContext).getFilter());
   }
 }
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/read/TestRowRangeAdapter.java b/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/read/TestRowRangeAdapter.java
index 415c38f58..6fe1d5d2b 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/read/TestRowRangeAdapter.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/read/TestRowRangeAdapter.java
@@ -19,6 +19,8 @@ import static org.junit.Assert.assertEquals;
 
 import com.google.bigtable.v2.RowRange;
 import com.google.bigtable.v2.RowSet;
+import com.google.cloud.bigtable.data.v2.internal.RequestContext;
+import com.google.cloud.bigtable.data.v2.models.Query;
 import com.google.cloud.bigtable.util.RowKeyWrapper;
 import com.google.common.collect.ImmutableRangeSet;
 import com.google.common.collect.Range;
@@ -32,8 +34,13 @@ import org.junit.runners.JUnit4;
 @RunWith(JUnit4.class)
 public class TestRowRangeAdapter {
 
+  private final RequestContext requestContext =
+      RequestContext.create("ProjectId", "InstanceId", "AppProfile");
+
   private RowRangeAdapter adapter;
 
+  private final Query query = Query.create("tableId");
+
   @Before
   public void setup() {
     this.adapter = new RowRangeAdapter();
@@ -45,7 +52,8 @@ public class TestRowRangeAdapter {
     RangeSet<RowKeyWrapper> out = adapter.rowSetToRangeSet(in);
 
     assertEquals(ImmutableRangeSet.builder().build(), out);
-    assertEquals(in, adapter.rangeSetToRowSet(out));
+    adapter.rangeSetToByteStringRange(out, query);
+    assertEquals(in, query.toProto(requestContext).getRows());
   }
 
   @Test
@@ -61,7 +69,8 @@ public class TestRowRangeAdapter {
         .build();
 
     assertEquals(expected, out);
-    assertEquals(in, adapter.rangeSetToRowSet(out));
+    adapter.rangeSetToByteStringRange(out, query);
+    assertEquals(in, query.toProto(requestContext).getRows());
   }
 
   @Test
@@ -83,7 +92,8 @@ public class TestRowRangeAdapter {
         .build();
 
     assertEquals(expected, out);
-    assertEquals(in, adapter.rangeSetToRowSet(out));
+    adapter.rangeSetToByteStringRange(out, query);
+    assertEquals(in, query.toProto(requestContext).getRows());
   }
 
   @Test
@@ -105,7 +115,8 @@ public class TestRowRangeAdapter {
         .build();
 
     assertEquals(expected, out);
-    assertEquals(in, adapter.rangeSetToRowSet(out));
+    adapter.rangeSetToByteStringRange(out, query);
+    assertEquals(in, query.toProto(requestContext).getRows());
   }
 
   @Test
@@ -127,7 +138,8 @@ public class TestRowRangeAdapter {
         .build();
 
     assertEquals(expected, out);
-    assertEquals(in, adapter.rangeSetToRowSet(out));
+    adapter.rangeSetToByteStringRange(out, query);
+    assertEquals(in, query.toProto(requestContext).getRows());
   }
 
   @Test
@@ -149,7 +161,8 @@ public class TestRowRangeAdapter {
         .build();
 
     assertEquals(expected, out);
-    assertEquals(in, adapter.rangeSetToRowSet(out));
+    adapter.rangeSetToByteStringRange(out, query);
+    assertEquals(in, query.toProto(requestContext).getRows());
   }
 
   @Test
@@ -165,7 +178,8 @@ public class TestRowRangeAdapter {
         .build();
 
     assertEquals(expected, out);
-    assertEquals(in, adapter.rangeSetToRowSet(out));
+    adapter.rangeSetToByteStringRange(out, query);
+    assertEquals(in, query.toProto(requestContext).getRows());
   }
 
   @Test
@@ -177,7 +191,6 @@ public class TestRowRangeAdapter {
                 .setEndKeyOpen(ByteString.EMPTY)
         )
         .build();
-
     RangeSet<RowKeyWrapper> out = adapter.rowSetToRangeSet(in);
 
     RangeSet<RowKeyWrapper> expected = ImmutableRangeSet.<RowKeyWrapper>builder()
@@ -206,7 +219,8 @@ public class TestRowRangeAdapter {
         .build();
 
     assertEquals(expected, out);
-    assertEquals(in, adapter.rangeSetToRowSet(out));
+    adapter.rangeSetToByteStringRange(out, query);
+    assertEquals(in, query.toProto(requestContext).getRows());
   }
 
   @Test
@@ -227,7 +241,8 @@ public class TestRowRangeAdapter {
         .build();
 
     assertEquals(expected, out);
-    assertEquals(in, adapter.rangeSetToRowSet(out));
+    adapter.rangeSetToByteStringRange(out, query);
+    assertEquals(in, query.toProto(requestContext).getRows());
   }
 
   @Test
@@ -248,7 +263,8 @@ public class TestRowRangeAdapter {
         .build();
 
     assertEquals(expected, out);
-    assertEquals(in, adapter.rangeSetToRowSet(out));
+    adapter.rangeSetToByteStringRange(out, query);
+    assertEquals(in, query.toProto(requestContext).getRows());
   }
 
   @Test
@@ -269,7 +285,8 @@ public class TestRowRangeAdapter {
         .build();
 
     assertEquals(expected, out);
-    assertEquals(in, adapter.rangeSetToRowSet(out));
+    adapter.rangeSetToByteStringRange(out, query);
+    assertEquals(in, query.toProto(requestContext).getRows());
   }
 
   @Test
@@ -308,6 +325,7 @@ public class TestRowRangeAdapter {
         .build();
 
     assertEquals(expected, out);
-    assertEquals(in, adapter.rangeSetToRowSet(out));
+    adapter.rangeSetToByteStringRange(out, query);
+    assertEquals(in, query.toProto(requestContext).getRows());
   }
 }
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/read/TestScanAdapter.java b/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/read/TestScanAdapter.java
index efdae154b..d25eec515 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/read/TestScanAdapter.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/read/TestScanAdapter.java
@@ -19,11 +19,11 @@ import static com.google.cloud.bigtable.data.v2.models.Filters.FILTERS;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.eq;
 
-import com.google.bigtable.v2.ReadRowsRequest;
-import com.google.bigtable.v2.ReadRowsRequest.Builder;
 import com.google.bigtable.v2.RowRange;
 import com.google.bigtable.v2.RowSet;
+import com.google.cloud.bigtable.data.v2.internal.RequestContext;
 import com.google.cloud.bigtable.data.v2.models.Filters;
+import com.google.cloud.bigtable.data.v2.models.Query;
 import com.google.cloud.bigtable.hbase.BigtableExtendedScan;
 import com.google.cloud.bigtable.hbase.adapters.filters.FilterAdapter;
 import com.google.cloud.bigtable.hbase.adapters.filters.FilterAdapterContext;
@@ -50,7 +50,7 @@ import org.junit.runners.JUnit4;
 import org.mockito.Mockito;
 
 /**
- * Lightweight tests for the ScanAdapter. Many of the methods, such as filter building are
+ * Lightweight tests for the {@link ScanAdapter}. Many of the methods, such as filter building are
  * already tested in {@link TestGetAdapter}.
  */
 @RunWith(JUnit4.class)
@@ -61,15 +61,16 @@ public class TestScanAdapter {
   );
   private final static ReadHooks throwingReadHooks = new ReadHooks() {
     @Override
-    public void composePreSendHook(Function<ReadRowsRequest, ReadRowsRequest> newHook) {
+    public void composePreSendHook(Function<Query, Query> newHook) {
       throw new IllegalStateException("Read hooks not supported in TestScanAdapter.");
     }
 
     @Override
-    public ReadRowsRequest applyPreSendHook(ReadRowsRequest readRowsRequest) {
+    public void applyPreSendHook(Query query) {
       throw new IllegalStateException("Read hooks not supported in TestScanAdapter.");
     }
   };
+
   public static final String START_KEY = "startKey";
   public static final String STOP_KEY = "stopKey";
 
@@ -89,72 +90,76 @@ public class TestScanAdapter {
     return prefixEnd;
   }
 
+  private final RequestContext requestContext =
+      RequestContext.create("ProjectId", "InstanceId", "AppProfile");
+  private final Query query = Query.create("tableId");
+
   @Test
   public void testNewScan() {
     Scan scan = new Scan();
-    ReadRowsRequest.Builder request = scanAdapter.adapt(scan, throwingReadHooks);
+    scanAdapter.adapt(scan, throwingReadHooks, query);
     Assert.assertEquals(toRowSet(RowRange.getDefaultInstance()),
-      request.getRows());
+            query.toProto(requestContext).getRows());
   }
 
   @Test
   public void testStartDefault() {
     byte[] startKey = Bytes.toBytes(START_KEY);
     Scan scan = new Scan().withStartRow(startKey);
-    ReadRowsRequest.Builder request = scanAdapter.adapt(scan, throwingReadHooks);
+    scanAdapter.adapt(scan, throwingReadHooks, query);
     RowSet expected = toRowSet(
         RowRange.newBuilder().setStartKeyClosed(ByteString.copyFrom(startKey)).build());
-    Assert.assertEquals(expected, request.getRows());
+    Assert.assertEquals(expected, query.toProto(requestContext).getRows());
   }
 
   @Test
   public void testStartInclusive() {
     byte[] startKey = Bytes.toBytes(START_KEY);
     Scan scan = new Scan().withStartRow(startKey, true);
-    ReadRowsRequest.Builder request = scanAdapter.adapt(scan, throwingReadHooks);
+    scanAdapter.adapt(scan, throwingReadHooks, query);
     RowSet expected = toRowSet(
         RowRange.newBuilder().setStartKeyClosed(ByteString.copyFrom(startKey)).build());
-    Assert.assertEquals(expected, request.getRows());
+    Assert.assertEquals(expected, query.toProto(requestContext).getRows());
   }
 
   @Test
   public void testStartExclusive() {
     byte[] startKey = Bytes.toBytes(START_KEY);
     Scan scan = new Scan().withStartRow(startKey, false);
-    ReadRowsRequest.Builder request = scanAdapter.adapt(scan, throwingReadHooks);
+    scanAdapter.adapt(scan, throwingReadHooks, query);
     RowSet expected = toRowSet(
         RowRange.newBuilder().setStartKeyOpen(ByteString.copyFrom(startKey)).build());
-    Assert.assertEquals(expected, request.getRows());
+    Assert.assertEquals(expected, query.toProto(requestContext).getRows());
   }
 
   @Test
   public void testStopDefault() {
     byte[] stopKey = Bytes.toBytes(STOP_KEY);
     Scan scan = new Scan().withStopRow(stopKey);
-    ReadRowsRequest.Builder request = scanAdapter.adapt(scan, throwingReadHooks);
+    scanAdapter.adapt(scan, throwingReadHooks, query);
     RowSet expected = toRowSet(
         RowRange.newBuilder().setEndKeyOpen(ByteString.copyFrom(stopKey)).build());
-    Assert.assertEquals(expected, request.getRows());
+    Assert.assertEquals(expected, query.toProto(requestContext).getRows());
   }
 
   @Test
   public void testStopInclusive() {
     byte[] stopKey = Bytes.toBytes(STOP_KEY);
     Scan scan = new Scan().withStopRow(stopKey, true);
-    ReadRowsRequest.Builder request = scanAdapter.adapt(scan, throwingReadHooks);
+    scanAdapter.adapt(scan, throwingReadHooks, query);
     RowSet expected = toRowSet(
         RowRange.newBuilder().setEndKeyClosed(ByteString.copyFrom(stopKey)).build());
-    Assert.assertEquals(expected, request.getRows());
+    Assert.assertEquals(expected, query.toProto(requestContext).getRows());
   }
 
   @Test
   public void testStopExclusive() {
     byte[] stopKey = Bytes.toBytes(STOP_KEY);
     Scan scan = new Scan().withStopRow(stopKey, false);
-    ReadRowsRequest.Builder request = scanAdapter.adapt(scan, throwingReadHooks);
+    scanAdapter.adapt(scan, throwingReadHooks, query);
     RowSet expected = toRowSet(
         RowRange.newBuilder().setEndKeyOpen(ByteString.copyFrom(stopKey)).build());
-    Assert.assertEquals(expected, request.getRows());
+    Assert.assertEquals(expected, query.toProto(requestContext).getRows());
   }
 
   @Test
@@ -164,8 +169,9 @@ public class TestScanAdapter {
     Scan scan = new Scan()
         .withStartRow(startKey)
         .withStopRow(stopKey);
-    ReadRowsRequest.Builder request = scanAdapter.adapt(scan, throwingReadHooks);
-    Assert.assertEquals(toRowSet(toRange(startKey, stopKey)), request.getRows());
+    scanAdapter.adapt(scan, throwingReadHooks, query);
+    Assert.assertEquals(toRowSet(toRange(startKey, stopKey)),
+        query.toProto(requestContext).getRows());
   }
 
   @Test
@@ -175,10 +181,11 @@ public class TestScanAdapter {
     Scan scan = new Scan()
         .withStartRow(startKey, false)
         .withStopRow(stopKey, true);
-    ReadRowsRequest.Builder request = scanAdapter.adapt(scan, throwingReadHooks);
+    scanAdapter.adapt(scan, throwingReadHooks, query);
     Assert.assertEquals(toRowSet(
         RowRange.newBuilder().setStartKeyOpen(ByteStringer.wrap(startKey))
-            .setEndKeyClosed(ByteStringer.wrap(stopKey)).build()), request.getRows());
+            .setEndKeyClosed(ByteStringer.wrap(stopKey)).build()),
+        query.toProto(requestContext).getRows());
   }
 
   @Test
@@ -187,18 +194,19 @@ public class TestScanAdapter {
     byte[] prefixEnd = calculatePrefixEnd(prefix);
     Scan scan = new Scan();
     scan.setRowPrefixFilter(prefix);
-    ReadRowsRequest.Builder request = scanAdapter.adapt(scan, throwingReadHooks);
-    Assert.assertEquals(toRowSet(toRange(prefix, prefixEnd)), request.getRows());
+    scanAdapter.adapt(scan, throwingReadHooks, query);
+    Assert.assertEquals(toRowSet(toRange(prefix, prefixEnd)),
+        query.toProto(requestContext).getRows());
   }
 
   @Test
   public void maxVersionsIsSet() {
     Scan scan = new Scan();
     scan.setMaxVersions(10);
-    ReadRowsRequest.Builder rowRequestBuilder = scanAdapter.adapt(scan, throwingReadHooks);
+    scanAdapter.adapt(scan, throwingReadHooks, query);
     Assert.assertEquals(
         FILTERS.limit().cellsPerColumn(10).toProto(),
-        rowRequestBuilder.getFilter());
+        query.toProto(requestContext).getFilter());
   }
 
   @Test
@@ -217,15 +225,16 @@ public class TestScanAdapter {
     scan.addRowKey(row2);
     scan.addRange(startRow, stopRow);
     scan.addRangeWithPrefix(prefix);
+    scanAdapter.adapt(scan, throwingReadHooks, query);
 
     RowSet expected = RowSet.newBuilder()
         .addRowKeys(ByteStringer.wrap(row1))
         .addRowKeys(ByteStringer.wrap(row2))
-        .addRowRanges(toRange(startRow, stopRow))
         .addRowRanges(toRange(prefix, prefixEnd))
+        .addRowRanges(toRange(startRow, stopRow))
         .build();
 
-    Assert.assertEquals(expected, scanAdapter.adapt(scan, throwingReadHooks).getRows());
+    Assert.assertEquals(expected, query.toProto(requestContext).getRows());
   }
 
   @Test
@@ -255,7 +264,7 @@ public class TestScanAdapter {
         .withStopRow("z".getBytes())
         .setFilter(fakeFilter);
 
-    Builder adapted = scanAdapter.adapt(scan, throwingReadHooks);
+    scanAdapter.adapt(scan, throwingReadHooks, query);
 
     Assert.assertEquals(
         RowSet.newBuilder()
@@ -265,7 +274,7 @@ public class TestScanAdapter {
                     .setEndKeyOpen(ByteString.copyFromUtf8("d"))
             )
             .build(),
-        adapted.getRows()
+        query.toProto(requestContext).getRows()
     );
   }
 
@@ -276,7 +285,8 @@ public class TestScanAdapter {
         .withStartRow("a".getBytes())
         .withStopRow("z".getBytes());
 
-    RowSet result = scanAdapter.adapt(scan, throwingReadHooks).build().getRows();
+    scanAdapter.adapt(scan, throwingReadHooks, query);
+    RowSet result = query.toProto(requestContext).getRows();
     RowSet expected = RowSet.newBuilder()
         .addRowRanges(
             RowRange.newBuilder()
@@ -295,7 +305,8 @@ public class TestScanAdapter {
         .withStartRow("a".getBytes())
         .withStopRow("z".getBytes());
 
-    RowSet result = scanAdapter.adapt(scan, throwingReadHooks).build().getRows();
+    scanAdapter.adapt(scan, throwingReadHooks, query);
+    RowSet result = query.toProto(requestContext).getRows();
     RowSet expected = RowSet.newBuilder()
         .addRowRanges(
             RowRange.newBuilder()
@@ -310,7 +321,8 @@ public class TestScanAdapter {
   @Test
   public void testLimit() {
     Scan scan = new Scan().setLimit(10);
-    int adaptedLimit = (int) scanAdapter.adapt(scan, throwingReadHooks).build().getRowsLimit();
+    scanAdapter.adapt(scan, throwingReadHooks, query);
+    int adaptedLimit = (int) query.toProto(requestContext).getRowsLimit();
     Assert.assertEquals(scan.getLimit(), adaptedLimit);
   }
 
@@ -320,7 +332,8 @@ public class TestScanAdapter {
     Get get = new Get(key);
     get.setMaxVersions(Integer.MAX_VALUE);
     Scan scan = new Scan(get);
-    RowSet actual = scanAdapter.adapt(scan, throwingReadHooks).build().getRows();
+    scanAdapter.adapt(scan, throwingReadHooks, query);
+    RowSet actual = query.toProto(requestContext).getRows();
     RowSet expected = RowSet.newBuilder()
         .addRowKeys(ByteString.copyFrom(key))
         .build();