diff --git a/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/BigtableDataClient.java b/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/BigtableDataClient.java
index d1061ac74d11..621c5cbf464d 100644
--- a/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/BigtableDataClient.java
+++ b/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/BigtableDataClient.java
@@ -422,7 +422,88 @@ public ApiFuture<Row> readRowAsync(String tableId, ByteString rowKey, @Nullable
     if (filter != null) {
       query = query.filter(filter);
     }
-    return readRowsCallable().first().futureCall(query);
+    return readRowCallable().futureCall(query);
+  }
+
+  /**
+   * Reads a single row. The returned callable object allows for customization of api invocation.
+   *
+   * <p>Sample code:
+   *
+   * <pre>{@code
+   * InstanceName instanceName = InstanceName.of("[PROJECT]", "[INSTANCE]");
+   * try (BigtableDataClient bigtableDataClient = BigtableDataClient.create(instanceName)) {
+   *   String tableId = "[TABLE]";
+   *
+   *   Query query = Query.create(tableId)
+   *          .rowKey("[KEY]")
+   *          .filter(FILTERS.qualifier().regex("[COLUMN PREFIX].*"));
+   *
+   *   // Synchronous invocation
+   *   try {
+   *     Row row = bigtableDataClient.readRowCallable().call(query);
+   *     if (row == null) {
+   *       System.out.println("Row not found");
+   *     }
+   *   } catch (RuntimeException e) {
+   *     e.printStackTrace();
+   *   }
+   *
+   *   // Asynchronous invocation
+   *   ApiFuture<Row> rowFuture = bigtableDataClient.readRowCallable().futureCall(query);
+   *
+   *   ApiFutures.addCallback(rowFuture, new ApiFutureCallback<Row>() {
+   *     public void onFailure(Throwable t) {
+   *       if (t instanceof NotFoundException) {
+   *         System.out.println("Tried to read a non-existent table");
+   *       } else {
+   *         t.printStackTrace();
+   *       }
+   *     }
+   *     public void onSuccess(Row row) {
+   *       if (row == null) {
+   *         System.out.println("Row not found");
+   *       }
+   *     }
+   *   }, MoreExecutors.directExecutor());
+   * }
+   * }</pre>
+   *
+   * @see UnaryCallable For call styles.
+   * @see Query For query options.
+   * @see com.google.cloud.bigtable.data.v2.models.Filters For the filter building DSL.
+   */
+  public UnaryCallable<Query, Row> readRowCallable() {
+    return stub.readRowCallable();
+  }
+
+  /**
+   * Reads a single row. 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 (BigtableDataClient bigtableDataClient = BigtableDataClient.create(instanceName)) {
+   *   String tableId = "[TABLE]";
+   *
+   *   Query query = Query.create(tableId)
+   *          .rowKey("[KEY]")
+   *          .filter(FILTERS.qualifier().regex("[COLUMN PREFIX].*"));
+   *
+   *   // Synchronous invocation
+   *   CustomRow row = bigtableDataClient.readRowCallable(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.models.Filters For the filter building DSL.
+   */
+  public <RowT> UnaryCallable<Query, RowT> readRowCallable(RowAdapter<RowT> rowAdapter) {
+    return stub.createReadRowCallable(rowAdapter);
   }
 
   /**
diff --git a/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Query.java b/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Query.java
index 41091c894bad..0041fee227fb 100644
--- a/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Query.java
+++ b/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Query.java
@@ -24,6 +24,8 @@
 import com.google.cloud.bigtable.data.v2.internal.RequestContext;
 import com.google.cloud.bigtable.data.v2.internal.RowSetUtil;
 import com.google.cloud.bigtable.data.v2.models.Range.ByteStringRange;
+import com.google.common.base.MoreObjects;
+import com.google.common.base.Objects;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableSortedSet;
 import com.google.common.collect.Lists;
@@ -264,4 +266,34 @@ private static ByteString wrapKey(String key) {
     }
     return ByteString.copyFromUtf8(key);
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    Query query = (Query) o;
+    return Objects.equal(tableId, query.tableId)
+        && Objects.equal(builder.build(), query.builder.build());
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(tableId, builder.build());
+  }
+
+  @Override
+  public String toString() {
+    ReadRowsRequest request = builder.build();
+
+    return MoreObjects.toStringHelper(this)
+        .add("tableId", tableId)
+        .add("keys", request.getRows().getRowKeysList())
+        .add("ranges", request.getRows().getRowRangesList())
+        .add("filter", request.getFilter())
+        .toString();
+  }
 }
diff --git a/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java b/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java
index 5d3c733476cc..8100d67c9234 100644
--- a/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java
+++ b/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java
@@ -74,6 +74,7 @@ public class EnhancedBigtableStub implements AutoCloseable {
   private final RequestContext requestContext;
 
   private final ServerStreamingCallable<Query, Row> readRowsCallable;
+  private final UnaryCallable<Query, Row> readRowCallable;
   private final UnaryCallable<String, List<KeyOffset>> sampleRowKeysCallable;
   private final UnaryCallable<RowMutation, Void> mutateRowCallable;
   private final UnaryCallable<BulkMutation, Void> bulkMutateRowsCallable;
@@ -151,6 +152,7 @@ public static EnhancedBigtableStub create(EnhancedBigtableStubSettings settings)
         RequestContext.create(settings.getInstanceName(), settings.getAppProfileId());
 
     readRowsCallable = createReadRowsCallable(new DefaultRowAdapter());
+    readRowCallable = createReadRowCallable(new DefaultRowAdapter());
     sampleRowKeysCallable = createSampleRowKeysCallable();
     mutateRowCallable = createMutateRowCallable();
     bulkMutateRowsCallable = createBulkMutateRowsCallable();
@@ -162,7 +164,7 @@ public static EnhancedBigtableStub create(EnhancedBigtableStubSettings settings)
   // <editor-fold desc="Callable creators">
 
   /**
-   * Creates a callable chain to handle ReadRows RPCs. The chain will:
+   * Creates a callable chain to handle streaming ReadRows RPCs. The chain will:
    *
    * <ul>
    *   <li>Convert a {@link Query} into a {@link com.google.bigtable.v2.ReadRowsRequest} and
@@ -176,6 +178,48 @@ public static EnhancedBigtableStub create(EnhancedBigtableStubSettings settings)
    */
   public <RowT> ServerStreamingCallable<Query, RowT> createReadRowsCallable(
       RowAdapter<RowT> rowAdapter) {
+    return createReadRowsCallable(settings.readRowsSettings(), rowAdapter);
+  }
+
+  /**
+   * Creates a callable chain to handle point ReadRows RPCs. The chain will:
+   *
+   * <ul>
+   *   <li>Convert a {@link Query} into a {@link com.google.bigtable.v2.ReadRowsRequest} and
+   *       dispatch the RPC.
+   *   <li>Upon receiving the response stream, it will merge the {@link
+   *       com.google.bigtable.v2.ReadRowsResponse.CellChunk}s in logical rows. The actual row
+   *       implementation can be configured in by the {@code rowAdapter} parameter.
+   *   <li>Retry/resume on failure.
+   *   <li>Filter out marker rows.
+   * </ul>
+   */
+  public <RowT> UnaryCallable<Query, RowT> createReadRowCallable(RowAdapter<RowT> rowAdapter) {
+    return createReadRowsCallable(
+            ServerStreamingCallSettings.<Query, Row>newBuilder()
+                .setRetryableCodes(settings.readRowSettings().getRetryableCodes())
+                .setRetrySettings(settings.readRowSettings().getRetrySettings())
+                .setIdleTimeout(settings.readRowSettings().getRetrySettings().getTotalTimeout())
+                .build(),
+            rowAdapter)
+        .first();
+  }
+
+  /**
+   * Creates a callable chain to handle ReadRows RPCs. The chain will:
+   *
+   * <ul>
+   *   <li>Convert a {@link Query} into a {@link com.google.bigtable.v2.ReadRowsRequest} and
+   *       dispatch the RPC.
+   *   <li>Upon receiving the response stream, it will merge the {@link
+   *       com.google.bigtable.v2.ReadRowsResponse.CellChunk}s in logical rows. The actual row
+   *       implementation can be configured in by the {@code rowAdapter} parameter.
+   *   <li>Retry/resume on failure.
+   *   <li>Filter out marker rows.
+   * </ul>
+   */
+  private <RowT> ServerStreamingCallable<Query, RowT> createReadRowsCallable(
+      ServerStreamingCallSettings<Query, Row> readRowsSettings, RowAdapter<RowT> rowAdapter) {
 
     ServerStreamingCallable<ReadRowsRequest, RowT> merging =
         new RowMergingCallable<>(stub.readRowsCallable(), rowAdapter);
@@ -185,9 +229,9 @@ public <RowT> ServerStreamingCallable<Query, RowT> createReadRowsCallable(
     ServerStreamingCallSettings<ReadRowsRequest, RowT> innerSettings =
         ServerStreamingCallSettings.<ReadRowsRequest, RowT>newBuilder()
             .setResumptionStrategy(new ReadRowsResumptionStrategy<>(rowAdapter))
-            .setRetryableCodes(settings.readRowsSettings().getRetryableCodes())
-            .setRetrySettings(settings.readRowsSettings().getRetrySettings())
-            .setIdleTimeout(settings.readRowsSettings().getIdleTimeout())
+            .setRetryableCodes(readRowsSettings.getRetryableCodes())
+            .setRetrySettings(readRowsSettings.getRetrySettings())
+            .setIdleTimeout(readRowsSettings.getIdleTimeout())
             .build();
 
     // Retry logic is split into 2 parts to workaround a rare edge case described in
@@ -356,10 +400,16 @@ private UnaryCallable<ReadModifyWriteRow, Row> createReadModifyWriteRowCallable(
   // </editor-fold>
 
   // <editor-fold desc="Callable accessors">
+  /** Returns a streaming read rows callable */
   public ServerStreamingCallable<Query, Row> readRowsCallable() {
     return readRowsCallable;
   }
 
+  /** Return a point read callable */
+  public UnaryCallable<Query, Row> readRowCallable() {
+    return readRowCallable;
+  }
+
   public UnaryCallable<String, List<KeyOffset>> sampleRowKeysCallable() {
     return sampleRowKeysCallable;
   }
diff --git a/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettings.java b/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettings.java
index 63320cfdc3d0..298c190d4df5 100644
--- a/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettings.java
+++ b/google-cloud-clients/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettings.java
@@ -112,6 +112,7 @@ public class EnhancedBigtableStubSettings extends StubSettings<EnhancedBigtableS
   private final String appProfileId;
 
   private final ServerStreamingCallSettings<Query, Row> readRowsSettings;
+  private final UnaryCallSettings<Query, Row> readRowSettings;
   private final UnaryCallSettings<String, List<KeyOffset>> sampleRowKeysSettings;
   private final UnaryCallSettings<RowMutation, Void> mutateRowSettings;
   private final BatchingCallSettings<RowMutation, Void> bulkMutateRowsSettings;
@@ -120,11 +121,22 @@ public class EnhancedBigtableStubSettings extends StubSettings<EnhancedBigtableS
 
   private EnhancedBigtableStubSettings(Builder builder) {
     super(builder);
+
+    // Since point reads & streaming reads share the same base callable that converts grpc errors
+    // into ApiExceptions, they must have the same retry codes.
+    Preconditions.checkState(
+        builder
+            .readRowSettings
+            .getRetryableCodes()
+            .equals(builder.readRowsSettings.getRetryableCodes()),
+        "Single ReadRow retry codes must match ReadRows retry codes");
+
     instanceName = builder.instanceName;
     appProfileId = builder.appProfileId;
 
     // Per method settings.
     readRowsSettings = builder.readRowsSettings.build();
+    readRowSettings = builder.readRowSettings.build();
     sampleRowKeysSettings = builder.sampleRowKeysSettings.build();
     mutateRowSettings = builder.mutateRowSettings.build();
     bulkMutateRowsSettings = builder.bulkMutateRowsSettings.build();
@@ -163,6 +175,11 @@ public UnaryCallSettings<String, List<KeyOffset>> sampleRowKeysSettings() {
     return sampleRowKeysSettings;
   }
 
+  /** Returns the object with the settings used for point reads via ReadRows. */
+  public UnaryCallSettings<Query, Row> readRowSettings() {
+    return readRowSettings;
+  }
+
   /** Returns the object with the settings used for calls to MutateRow. */
   public UnaryCallSettings<RowMutation, Void> mutateRowSettings() {
     return mutateRowSettings;
@@ -200,6 +217,7 @@ public static class Builder extends StubSettings.Builder<EnhancedBigtableStubSet
     private String appProfileId;
 
     private final ServerStreamingCallSettings.Builder<Query, Row> readRowsSettings;
+    private final UnaryCallSettings.Builder<Query, Row> readRowSettings;
     private final UnaryCallSettings.Builder<String, List<KeyOffset>> sampleRowKeysSettings;
     private final UnaryCallSettings.Builder<RowMutation, Void> mutateRowSettings;
     private final BatchingCallSettings.Builder<RowMutation, Void> bulkMutateRowsSettings;
@@ -234,18 +252,27 @@ private Builder() {
 
       // Per-method settings using baseSettings for defaults.
       readRowsSettings = ServerStreamingCallSettings.newBuilder();
-      /* TODO: copy timeouts, retryCodes & retrySettings from baseSettings.readRows once it exists in GAPIC */
       readRowsSettings
-          .setRetryableCodes(DEFAULT_RETRY_CODES)
-          .setRetrySettings(
-              DEFAULT_RETRY_SETTINGS.toBuilder().setTotalTimeout(Duration.ofHours(1)).build())
+          .setRetryableCodes(baseDefaults.readRowsSettings().getRetryableCodes())
+          .setRetrySettings(baseDefaults.readRowsSettings().getRetrySettings())
           .setIdleTimeout(Duration.ofMinutes(5));
 
+      // Point reads should use same defaults as streaming reads, but with a shorter timeout
+      readRowSettings = UnaryCallSettings.newUnaryCallSettingsBuilder();
+      readRowSettings
+          .setRetryableCodes(baseDefaults.readRowsSettings().getRetryableCodes())
+          .setRetrySettings(
+              baseDefaults
+                  .readRowsSettings()
+                  .getRetrySettings()
+                  .toBuilder()
+                  .setTotalTimeout(DEFAULT_RETRY_SETTINGS.getTotalTimeout())
+                  .build());
+
       sampleRowKeysSettings = UnaryCallSettings.newUnaryCallSettingsBuilder();
-      /* TODO: copy retryCodes & retrySettings from baseSettings.sampleRowKeysSettings once it exists in GAPIC */
       sampleRowKeysSettings
-          .setRetryableCodes(Code.DEADLINE_EXCEEDED, Code.UNAVAILABLE, Code.ABORTED)
-          .setRetrySettings(DEFAULT_RETRY_SETTINGS);
+          .setRetryableCodes(baseDefaults.sampleRowKeysSettings().getRetryableCodes())
+          .setRetrySettings(baseDefaults.sampleRowKeysSettings().getRetrySettings());
 
       mutateRowSettings = UnaryCallSettings.newUnaryCallSettingsBuilder();
       copyRetrySettings(baseDefaults.mutateRowSettings(), mutateRowSettings);
@@ -282,6 +309,7 @@ private Builder(EnhancedBigtableStubSettings settings) {
 
       // Per method settings.
       readRowsSettings = settings.readRowsSettings.toBuilder();
+      readRowSettings = settings.readRowSettings.toBuilder();
       sampleRowKeysSettings = settings.sampleRowKeysSettings.toBuilder();
       mutateRowSettings = settings.mutateRowSettings.toBuilder();
       bulkMutateRowsSettings = settings.bulkMutateRowsSettings.toBuilder();
@@ -339,6 +367,11 @@ public ServerStreamingCallSettings.Builder<Query, Row> readRowsSettings() {
       return readRowsSettings;
     }
 
+    /** Returns the builder for the settings used for point reads using readRow. */
+    public UnaryCallSettings.Builder<Query, Row> readRowSettings() {
+      return readRowSettings;
+    }
+
     /** Returns the builder for the settings used for calls to SampleRowKeysSettings. */
     public UnaryCallSettings.Builder<String, List<KeyOffset>> sampleRowKeysSettings() {
       return sampleRowKeysSettings;
diff --git a/google-cloud-clients/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/BigtableDataClientTest.java b/google-cloud-clients/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/BigtableDataClientTest.java
index be6cb2b965c5..be268b00b6fc 100644
--- a/google-cloud-clients/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/BigtableDataClientTest.java
+++ b/google-cloud-clients/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/BigtableDataClientTest.java
@@ -27,15 +27,11 @@
 import com.google.api.gax.rpc.ResponseObserver;
 import com.google.api.gax.rpc.ServerStreamingCallable;
 import com.google.api.gax.rpc.UnaryCallable;
-import com.google.bigtable.v2.ReadRowsRequest;
-import com.google.bigtable.v2.RowSet;
-import com.google.cloud.bigtable.data.v2.internal.RequestContext;
 import com.google.cloud.bigtable.data.v2.models.BulkMutation;
 import com.google.cloud.bigtable.data.v2.models.BulkMutationBatcher;
 import com.google.cloud.bigtable.data.v2.models.BulkMutationBatcher.BulkMutationFailure;
 import com.google.cloud.bigtable.data.v2.models.ConditionalRowMutation;
 import com.google.cloud.bigtable.data.v2.models.Filters.Filter;
-import com.google.cloud.bigtable.data.v2.models.InstanceName;
 import com.google.cloud.bigtable.data.v2.models.KeyOffset;
 import com.google.cloud.bigtable.data.v2.models.Mutation;
 import com.google.cloud.bigtable.data.v2.models.Query;
@@ -44,6 +40,7 @@
 import com.google.cloud.bigtable.data.v2.models.RowCell;
 import com.google.cloud.bigtable.data.v2.models.RowMutation;
 import com.google.cloud.bigtable.data.v2.stub.EnhancedBigtableStub;
+import com.google.common.collect.ImmutableList;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Empty;
 import io.grpc.Status.Code;
@@ -54,7 +51,6 @@
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Answers;
-import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
@@ -69,6 +65,7 @@ public class BigtableDataClientTest {
   @Mock(answer = Answers.RETURNS_DEEP_STUBS)
   private ServerStreamingCallable<Query, Row> mockReadRowsCallable;
 
+  @Mock private UnaryCallable<Query, Row> mockReadRowCallable;
   @Mock private UnaryCallable<String, List<KeyOffset>> mockSampleRowKeysCallable;
   @Mock private UnaryCallable<RowMutation, Void> mockMutateRowCallable;
   @Mock private UnaryCallable<ConditionalRowMutation, Boolean> mockCheckAndMutateRowCallable;
@@ -82,6 +79,7 @@ public class BigtableDataClientTest {
   public void setUp() {
     bigtableDataClient = new BigtableDataClient(mockStub);
     Mockito.when(mockStub.readRowsCallable()).thenReturn(mockReadRowsCallable);
+    Mockito.when(mockStub.readRowCallable()).thenReturn(mockReadRowCallable);
     Mockito.when(mockStub.sampleRowKeysCallable()).thenReturn(mockSampleRowKeysCallable);
     Mockito.when(mockStub.mutateRowCallable()).thenReturn(mockMutateRowCallable);
     Mockito.when(mockStub.bulkMutateRowsCallable()).thenReturn(mockBulkMutateRowsCallable);
@@ -105,38 +103,15 @@ public void proxyReadRowsCallableTest() {
   @Test
   public void proxyReadRowAsyncTest() {
     bigtableDataClient.readRowAsync("fake-table", ByteString.copyFromUtf8("fake-row-key"));
-
-    ArgumentCaptor<Query> requestCaptor = ArgumentCaptor.forClass(Query.class);
-    Mockito.verify(mockReadRowsCallable.first()).futureCall(requestCaptor.capture());
-
-    RequestContext ctx =
-        RequestContext.create(InstanceName.of("fake-project", "fake-instance"), "fake-profile");
-    // NOTE: limit(1) is added by the mocked first() call, so it's not tested here
-    assertThat(requestCaptor.getValue().toProto(ctx))
-        .isEqualTo(
-            ReadRowsRequest.newBuilder()
-                .setTableName("projects/fake-project/instances/fake-instance/tables/fake-table")
-                .setAppProfileId("fake-profile")
-                .setRows(RowSet.newBuilder().addRowKeys(ByteString.copyFromUtf8("fake-row-key")))
-                .build());
+    Mockito.verify(mockReadRowCallable)
+        .futureCall(Query.create("fake-table").rowKey("fake-row-key"));
   }
 
   @Test
   public void proxyReadRowStrAsyncTest() {
     bigtableDataClient.readRowAsync("fake-table", "fake-row-key");
-    ArgumentCaptor<Query> requestCaptor = ArgumentCaptor.forClass(Query.class);
-    Mockito.verify(mockReadRowsCallable.first()).futureCall(requestCaptor.capture());
-
-    RequestContext ctx =
-        RequestContext.create(InstanceName.of("fake-project", "fake-instance"), "fake-profile");
-    // NOTE: limit(1) is added by the mocked first() call, so it's not tested here
-    assertThat(requestCaptor.getValue().toProto(ctx))
-        .isEqualTo(
-            ReadRowsRequest.newBuilder()
-                .setTableName("projects/fake-project/instances/fake-instance/tables/fake-table")
-                .setAppProfileId("fake-profile")
-                .setRows(RowSet.newBuilder().addRowKeys(ByteString.copyFromUtf8("fake-row-key")))
-                .build());
+    Mockito.verify(mockReadRowCallable)
+        .futureCall(Query.create("fake-table").rowKey("fake-row-key"));
   }
 
   @Test
@@ -148,25 +123,9 @@ public void readRowFilterAsyncTest() {
             .filter(FILTERS.qualifier().regex("prefix.*"))
             .filter(FILTERS.limit().cellsPerRow(10));
     bigtableDataClient.readRowAsync("fake-table", ByteString.copyFromUtf8("fake-row-key"), filter);
-    ArgumentCaptor<Query> requestCaptor = ArgumentCaptor.forClass(Query.class);
-    Mockito.verify(mockReadRowsCallable.first()).futureCall(requestCaptor.capture());
-
-    RequestContext ctx =
-        RequestContext.create(InstanceName.of("fake-project", "fake-instance"), "fake-profile");
-    // NOTE: limit(1) is added by the mocked first() call, so it's not tested here
-    assertThat(requestCaptor.getValue().toProto(ctx))
-        .isEqualTo(
-            ReadRowsRequest.newBuilder()
-                .setTableName("projects/fake-project/instances/fake-instance/tables/fake-table")
-                .setAppProfileId("fake-profile")
-                .setRows(RowSet.newBuilder().addRowKeys(ByteString.copyFromUtf8("fake-row-key")))
-                .setFilter(
-                    FILTERS
-                        .chain()
-                        .filter(FILTERS.qualifier().regex("prefix.*"))
-                        .filter(FILTERS.limit().cellsPerRow(10))
-                        .toProto())
-                .build());
+
+    Mockito.verify(mockReadRowCallable)
+        .futureCall(Query.create("fake-table").rowKey("fake-row-key").filter(filter));
   }
 
   @Test
@@ -178,73 +137,34 @@ public void readRowFilterStrAsyncTest() {
             .filter(FILTERS.qualifier().regex("prefix.*"))
             .filter(FILTERS.limit().cellsPerRow(10));
     bigtableDataClient.readRowAsync("fake-table", "fake-row-key", filter);
-    ArgumentCaptor<Query> requestCaptor = ArgumentCaptor.forClass(Query.class);
-    Mockito.verify(mockReadRowsCallable.first()).futureCall(requestCaptor.capture());
-
-    RequestContext ctx =
-        RequestContext.create(InstanceName.of("fake-project", "fake-instance"), "fake-profile");
-    // NOTE: limit(1) is added by the mocked first() call, so it's not tested here
-    assertThat(requestCaptor.getValue().toProto(ctx))
-        .isEqualTo(
-            ReadRowsRequest.newBuilder()
-                .setTableName("projects/fake-project/instances/fake-instance/tables/fake-table")
-                .setAppProfileId("fake-profile")
-                .setRows(RowSet.newBuilder().addRowKeys(ByteString.copyFromUtf8("fake-row-key")))
-                .setFilter(
-                    FILTERS
-                        .chain()
-                        .filter(FILTERS.qualifier().regex("prefix.*"))
-                        .filter(FILTERS.limit().cellsPerRow(10))
-                        .toProto())
-                .build());
+
+    Mockito.verify(mockReadRowCallable)
+        .futureCall(Query.create("fake-table").rowKey("fake-row-key").filter(filter));
   }
 
   @Test
   public void readRowTest() {
-    Mockito.when(mockReadRowsCallable.first().futureCall(any(Query.class)))
-        .thenReturn(
-            ApiFutures.immediateFuture(
-                Row.create(
-                    ByteString.copyFromUtf8("fake-row-key"), Collections.<RowCell>emptyList())));
-    bigtableDataClient.readRow("fake-table", ByteString.copyFromUtf8("fake-row-key"));
-
-    ArgumentCaptor<Query> requestCaptor = ArgumentCaptor.forClass(Query.class);
-    Mockito.verify(mockReadRowsCallable.first()).futureCall(requestCaptor.capture());
-
-    RequestContext ctx =
-        RequestContext.create(InstanceName.of("fake-project", "fake-instance"), "fake-profile");
-    // NOTE: limit(1) is added by the mocked first() call, so it's not tested here
-    assertThat(requestCaptor.getValue().toProto(ctx))
-        .isEqualTo(
-            ReadRowsRequest.newBuilder()
-                .setTableName("projects/fake-project/instances/fake-instance/tables/fake-table")
-                .setAppProfileId("fake-profile")
-                .setRows(RowSet.newBuilder().addRowKeys(ByteString.copyFromUtf8("fake-row-key")))
-                .build());
+    Row expectedRow =
+        Row.create(ByteString.copyFromUtf8("fake-row-key"), ImmutableList.<RowCell>of());
+    Mockito.when(mockReadRowCallable.futureCall(Query.create("fake-table").rowKey("fake-row-key")))
+        .thenReturn(ApiFutures.immediateFuture(expectedRow));
+
+    Row actualRow =
+        bigtableDataClient.readRow("fake-table", ByteString.copyFromUtf8("fake-row-key"));
+
+    assertThat(actualRow).isEqualTo(expectedRow);
   }
 
   @Test
   public void readRowStrTest() {
-    Mockito.when(mockReadRowsCallable.first().futureCall(any(Query.class)))
-        .thenReturn(
-            ApiFutures.immediateFuture(
-                Row.create(
-                    ByteString.copyFromUtf8("fake-row-key"), Collections.<RowCell>emptyList())));
-    bigtableDataClient.readRow("fake-table", "fake-row-key");
-
-    ArgumentCaptor<Query> requestCaptor = ArgumentCaptor.forClass(Query.class);
-    Mockito.verify(mockReadRowsCallable.first()).futureCall(requestCaptor.capture());
-
-    RequestContext ctx =
-        RequestContext.create(InstanceName.of("fake-project", "fake-instance"), "fake-profile");
-    // NOTE: limit(1) is added by the mocked first() call, so it's not tested here
-    assertThat(requestCaptor.getValue().toProto(ctx))
-        .isEqualTo(
-            ReadRowsRequest.newBuilder()
-                .setTableName("projects/fake-project/instances/fake-instance/tables/fake-table")
-                .setAppProfileId("fake-profile")
-                .setRows(RowSet.newBuilder().addRowKeys(ByteString.copyFromUtf8("fake-row-key")))
-                .build());
+    Row expectedRow =
+        Row.create(ByteString.copyFromUtf8("fake-row-key"), ImmutableList.<RowCell>of());
+    Mockito.when(mockReadRowCallable.futureCall(Query.create("fake-table").rowKey("fake-row-key")))
+        .thenReturn(ApiFutures.immediateFuture(expectedRow));
+
+    Row actualRow = bigtableDataClient.readRow("fake-table", "fake-row-key");
+
+    assertThat(actualRow).isEqualTo(expectedRow);
   }
 
   @Test
@@ -255,32 +175,18 @@ public void readRowFilterTest() {
             .chain()
             .filter(FILTERS.qualifier().regex("prefix.*"))
             .filter(FILTERS.limit().cellsPerRow(10));
-    Mockito.when(mockReadRowsCallable.first().futureCall(any(Query.class)))
-        .thenReturn(
-            ApiFutures.immediateFuture(
-                Row.create(
-                    ByteString.copyFromUtf8("fake-row-key"), Collections.<RowCell>emptyList())));
-    bigtableDataClient.readRow("fake-table", ByteString.copyFromUtf8("fake-row-key"), filter);
-
-    ArgumentCaptor<Query> requestCaptor = ArgumentCaptor.forClass(Query.class);
-    Mockito.verify(mockReadRowsCallable.first()).futureCall(requestCaptor.capture());
-
-    RequestContext ctx =
-        RequestContext.create(InstanceName.of("fake-project", "fake-instance"), "fake-profile");
-    // NOTE: limit(1) is added by the mocked first() call, so it's not tested here
-    assertThat(requestCaptor.getValue().toProto(ctx))
-        .isEqualTo(
-            ReadRowsRequest.newBuilder()
-                .setTableName("projects/fake-project/instances/fake-instance/tables/fake-table")
-                .setAppProfileId("fake-profile")
-                .setRows(RowSet.newBuilder().addRowKeys(ByteString.copyFromUtf8("fake-row-key")))
-                .setFilter(
-                    FILTERS
-                        .chain()
-                        .filter(FILTERS.qualifier().regex("prefix.*"))
-                        .filter(FILTERS.limit().cellsPerRow(10))
-                        .toProto())
-                .build());
+
+    Row expectedRow =
+        Row.create(ByteString.copyFromUtf8("fake-row-key"), ImmutableList.<RowCell>of());
+    Mockito.when(
+            mockReadRowCallable.futureCall(
+                Query.create("fake-table").rowKey("fake-row-key").filter(filter)))
+        .thenReturn(ApiFutures.immediateFuture(expectedRow));
+
+    Row actualRow =
+        bigtableDataClient.readRow("fake-table", ByteString.copyFromUtf8("fake-row-key"), filter);
+
+    assertThat(actualRow).isEqualTo(expectedRow);
   }
 
   @Test
@@ -291,32 +197,16 @@ public void readRowStrFilterTest() {
             .chain()
             .filter(FILTERS.qualifier().regex("prefix.*"))
             .filter(FILTERS.limit().cellsPerRow(10));
-    Mockito.when(mockReadRowsCallable.first().futureCall(any(Query.class)))
-        .thenReturn(
-            ApiFutures.immediateFuture(
-                Row.create(
-                    ByteString.copyFromUtf8("fake-row-key"), Collections.<RowCell>emptyList())));
-    bigtableDataClient.readRow("fake-table", "fake-row-key", filter);
-
-    ArgumentCaptor<Query> requestCaptor = ArgumentCaptor.forClass(Query.class);
-    Mockito.verify(mockReadRowsCallable.first()).futureCall(requestCaptor.capture());
-
-    RequestContext ctx =
-        RequestContext.create(InstanceName.of("fake-project", "fake-instance"), "fake-profile");
-    // NOTE: limit(1) is added by the mocked first() call, so it's not tested here
-    assertThat(requestCaptor.getValue().toProto(ctx))
-        .isEqualTo(
-            ReadRowsRequest.newBuilder()
-                .setTableName("projects/fake-project/instances/fake-instance/tables/fake-table")
-                .setAppProfileId("fake-profile")
-                .setRows(RowSet.newBuilder().addRowKeys(ByteString.copyFromUtf8("fake-row-key")))
-                .setFilter(
-                    FILTERS
-                        .chain()
-                        .filter(FILTERS.qualifier().regex("prefix.*"))
-                        .filter(FILTERS.limit().cellsPerRow(10))
-                        .toProto())
-                .build());
+    Row expectedRow =
+        Row.create(ByteString.copyFromUtf8("fake-row-key"), ImmutableList.<RowCell>of());
+    Mockito.when(
+            mockReadRowCallable.futureCall(
+                Query.create("fake-table").rowKey("fake-row-key").filter(filter)))
+        .thenReturn(ApiFutures.immediateFuture(expectedRow));
+
+    Row actualRow = bigtableDataClient.readRow("fake-table", "fake-row-key", filter);
+
+    assertThat(actualRow).isEqualTo(expectedRow);
   }
 
   @Test
diff --git a/google-cloud-clients/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/it/ReadIT.java b/google-cloud-clients/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/it/ReadIT.java
index 76add465c838..203c5179190d 100644
--- a/google-cloud-clients/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/it/ReadIT.java
+++ b/google-cloud-clients/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/it/ReadIT.java
@@ -78,7 +78,7 @@ public void read() throws Throwable {
     List<Row> expectedRows = Lists.newArrayList();
     String uniqueKey = prefix + "-read";
 
-    long timestampMicros = System.nanoTime() * 1_000;
+    long timestampMicros = System.currentTimeMillis() * 1_000;
 
     for (int i = 0; i < numRows; i++) {
       testEnvRule
@@ -101,10 +101,10 @@ public void read() throws Throwable {
                       ByteString.copyFromUtf8("my-value")))));
     }
 
+    String tableId = testEnvRule.env().getTableName().getTable();
+
     // Sync
-    Query query =
-        Query.create(testEnvRule.env().getTableName().getTable())
-            .range(uniqueKey + "-0", uniqueKey + "-" + numRows);
+    Query query = Query.create(tableId).range(uniqueKey + "-0", uniqueKey + "-" + numRows);
     ArrayList<Row> actualResults =
         Lists.newArrayList(testEnvRule.env().getDataClient().readRows(query));
 
@@ -115,6 +115,16 @@ public void read() throws Throwable {
     testEnvRule.env().getDataClient().readRowsAsync(query, observer);
     observer.awaitCompletion();
     assertThat(observer.responses).containsExactlyElementsIn(expectedRows);
+
+    // Point Sync
+    Row actualRow =
+        testEnvRule.env().getDataClient().readRow(tableId, expectedRows.get(0).getKey());
+    assertThat(actualRow).isEqualTo(expectedRows.get(0));
+
+    // Point Async
+    ApiFuture<Row> actualRowFuture =
+        testEnvRule.env().getDataClient().readRowAsync(tableId, expectedRows.get(0).getKey());
+    assertThat(actualRowFuture.get()).isEqualTo(expectedRows.get(0));
   }
 
   @Test
diff --git a/google-cloud-clients/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettingsTest.java b/google-cloud-clients/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettingsTest.java
index 5e1698828ad2..def90578cad5 100644
--- a/google-cloud-clients/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettingsTest.java
+++ b/google-cloud-clients/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettingsTest.java
@@ -170,6 +170,9 @@ public void readRowsIsNotLostTest() {
         .setRetrySettings(retrySettings)
         .build();
 
+    // Point readRow settings must match streaming settings
+    builder.readRowSettings().setRetryableCodes(Code.ABORTED, Code.DEADLINE_EXCEEDED);
+
     assertThat(builder.readRowsSettings().getIdleTimeout()).isEqualTo(Duration.ofMinutes(5));
     assertThat(builder.readRowsSettings().getRetryableCodes())
         .containsAllOf(Code.ABORTED, Code.DEADLINE_EXCEEDED);
@@ -197,6 +200,84 @@ public void readRowsHasSaneDefaultsTest() {
     verifyRetrySettingAreSane(builder.getRetryableCodes(), builder.getRetrySettings());
   }
 
+  @Test
+  public void readRowIsNotLostTest() {
+    InstanceName dummyInstanceName = InstanceName.of("my-project", "my-instance");
+
+    EnhancedBigtableStubSettings.Builder builder =
+        EnhancedBigtableStubSettings.newBuilder().setInstanceName(dummyInstanceName);
+
+    RetrySettings retrySettings =
+        RetrySettings.newBuilder()
+            .setMaxAttempts(10)
+            .setTotalTimeout(Duration.ofHours(1))
+            .setInitialRpcTimeout(Duration.ofSeconds(10))
+            .setRpcTimeoutMultiplier(1)
+            .setMaxRpcTimeout(Duration.ofSeconds(10))
+            .setJittered(true)
+            .build();
+
+    builder
+        .readRowSettings()
+        .setRetryableCodes(Code.ABORTED, Code.DEADLINE_EXCEEDED)
+        .setRetrySettings(retrySettings)
+        .build();
+
+    // Streaming readRows settings must match point lookup settings.
+    builder.readRowsSettings().setRetryableCodes(Code.ABORTED, Code.DEADLINE_EXCEEDED);
+
+    assertThat(builder.readRowSettings().getRetryableCodes())
+        .containsAllOf(Code.ABORTED, Code.DEADLINE_EXCEEDED);
+    assertThat(builder.readRowSettings().getRetrySettings()).isEqualTo(retrySettings);
+
+    assertThat(builder.build().readRowSettings().getRetryableCodes())
+        .containsAllOf(Code.ABORTED, Code.DEADLINE_EXCEEDED);
+    assertThat(builder.build().readRowSettings().getRetrySettings()).isEqualTo(retrySettings);
+
+    assertThat(builder.build().toBuilder().readRowSettings().getRetryableCodes())
+        .containsAllOf(Code.ABORTED, Code.DEADLINE_EXCEEDED);
+    assertThat(builder.build().toBuilder().readRowSettings().getRetrySettings())
+        .isEqualTo(retrySettings);
+  }
+
+  @Test
+  public void readRowHasSaneDefaultsTest() {
+    UnaryCallSettings.Builder<Query, Row> builder =
+        EnhancedBigtableStubSettings.newBuilder().readRowSettings();
+
+    verifyRetrySettingAreSane(builder.getRetryableCodes(), builder.getRetrySettings());
+  }
+
+  @Test
+  public void readRowRetryCodesMustMatch() {
+    InstanceName dummyInstanceName = InstanceName.of("my-project", "my-instance");
+
+    EnhancedBigtableStubSettings.Builder builder =
+        EnhancedBigtableStubSettings.newBuilder().setInstanceName(dummyInstanceName);
+
+    builder.readRowsSettings().setRetryableCodes(Code.DEADLINE_EXCEEDED);
+
+    builder.readRowSettings().setRetryableCodes(Code.ABORTED);
+
+    Exception actualError = null;
+    try {
+      builder.build();
+    } catch (Exception e) {
+      actualError = e;
+    }
+    assertThat(actualError).isNotNull();
+
+    builder.readRowSettings().setRetryableCodes(Code.DEADLINE_EXCEEDED);
+
+    actualError = null;
+    try {
+      builder.build();
+    } catch (Exception e) {
+      actualError = e;
+    }
+    assertThat(actualError).isNull();
+  }
+
   @Test
   public void sampleRowKeysSettingsAreNotLostTest() {
     InstanceName dummyInstanceName = InstanceName.of("my-project", "my-instance");