feat(bigtable): add simple integration test for query support

Author: mpeddada1

google/cloud/bigtable/internal/data_connection_impl.cc

@@ -713,8 +713,11 @@ StatusOr<bigtable::PreparedQuery> DataConnectionImpl::PrepareQuery(
   }
   auto operation_context = operation_context_factory_->PrepareQuery(
       instance_full_name, app_profile_id(*current));
+  auto retry = retry_policy(*current);
+  auto backoff = backoff_policy(*current);
+
   auto response = google::cloud::internal::RetryLoop(
-      retry_policy(*current), backoff_policy(*current),
+      std::move(retry), std::move(backoff),
       Idempotency::kIdempotent,
       [this, operation_context](
           grpc::ClientContext& context, Options const& options,
@@ -747,6 +750,7 @@ StatusOr<bigtable::PreparedQuery> DataConnectionImpl::PrepareQuery(
   };
   auto query_plan = QueryPlan::Create(background_->cq(), *std::move(response),
                                       std::move(refresh_fn));
+
   return bigtable::PreparedQuery(
       params.instance, std::move(params.sql_statement), std::move(query_plan));
 }

google/cloud/bigtable/table.h

@@ -1000,6 +1000,10 @@ class Table {
     ChangePolicies(std::forward<Policies>(policies)...);
   }
 
+  std::shared_ptr<DataConnection> connection() const {
+    return connection_;
+  }
+
  private:
   /**
    * Send request ReadModifyWriteRowRequest to modify the row and get it back
@@ -1043,6 +1047,7 @@ class Table {
     return idempotent_mutation_policy_->clone();
   }
 
+
   ///@{
   /// @name Helper functions to implement constructors with changed policies.
   void ChangePolicy(RPCRetryPolicy const& policy) {

google/cloud/bigtable/tests/data_integration_test.cc

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include "google/cloud/bigtable/internal/defaults.h"
+#include "google/cloud/bigtable/client.h"
 #include "google/cloud/bigtable/testing/table_integration_test.h"
 #include "google/cloud/log.h"
 #include "google/cloud/testing_util/chrono_literals.h"
@@ -186,6 +187,45 @@ TEST_P(DataIntegrationTest, TableReadRowTest) {
   CheckEqualUnordered(expected, actual);
 }
 
+TEST_P(DataIntegrationTest, ClientQueryTest) {
+  auto table = GetTable(GetParam());
+  std::string const row_key = "row-key-for-client-query-test";
+  std::string const family = kFamily4;
+  std::string const column1 = "c1";
+  std::string const column2 = "c2";
+  std::string const value1 = "v1";
+  std::string const value2 = "v2";
+
+  std::vector<Cell> created{
+      {row_key, family, column1, 0, value1},
+      {row_key, family, column2, 0, value2},
+  };
+  // CreateCells(table, created);
+  BulkApply(table, created);
+
+  auto data_connection = table.connection();
+  auto client = Client(data_connection);
+   std::vector<std::string> full_table_path =
+      absl::StrSplit(table.table_name(), "/");
+  auto table_name = full_table_path.back();
+  Project project(project_id());
+  InstanceResource instance_resource(project, instance_id());
+  auto prepared_query = client.PrepareQuery(
+      instance_resource,
+      SqlStatement("SELECT " + column1 + " FROM " + table_name + " WHERE row_key = '" + row_key + "'"));
+  ASSERT_STATUS_OK(prepared_query);
+
+  auto bound_query = prepared_query->BindParameters({});
+  auto row_stream = client.ExecuteQuery(std::move(bound_query));
+
+  std::vector<StatusOr<bigtable::QueryRow>> rows;
+  for (auto const& row : std::move(row_stream)) {
+    rows.push_back(row);
+  }
+
+  ASSERT_EQ(rows.size(), 1);
+}
+
 TEST_P(DataIntegrationTest, TableReadRowNotExistTest) {
   auto table = GetTable(GetParam());
   std::string const row_key1 = "row-key-1";