move empty metadata check to prepare query

Author: mpeddada1

google/cloud/bigtable/internal/data_connection_impl.cc

@@ -757,7 +757,13 @@ StatusOr<bigtable::PreparedQuery> DataConnectionImpl::PrepareQuery(
   if (!response) {
     return std::move(response).status();
   }
-  for (auto const& column : response->metadata().proto_schema().columns()) {
+  auto const& proto_schema = response->metadata().proto_schema();
+  auto const columns_size = proto_schema.columns_size();
+  if (columns_size == 0) {
+    return internal::InternalError("ResultSetMetadata columns cannot be empty",
+                                   GCP_ERROR_INFO());
+  }
+  for (auto const& column : proto_schema.columns()) {
     if (!column.has_type() ||
         column.type().kind_case() == google::bigtable::v2::Type::KIND_NOT_SET) {
       return internal::InternalError("Column type cannot be empty",
@@ -847,8 +853,13 @@ future<StatusOr<bigtable::PreparedQuery>> DataConnectionImpl::AsyncPrepareQuery(
         if (!response) {
           return std::move(response).status();
         }
-        for (auto const& column :
-             response->metadata().proto_schema().columns()) {
+        auto const& proto_schema = response->metadata().proto_schema();
+        auto const columns_size = proto_schema.columns_size();
+        if (columns_size == 0) {
+          return internal::InternalError(
+              "ResultSetMetadata columns cannot be empty", GCP_ERROR_INFO());
+        }
+        for (auto const& column : proto_schema.columns()) {
           if (!column.has_type() ||
               column.type().kind_case() ==
                   google::bigtable::v2::Type::KIND_NOT_SET) {

google/cloud/bigtable/internal/data_connection_impl_test.cc

@@ -2841,16 +2841,31 @@ TEST_F(DataConnectionTest, PrepareQuerySuccess) {
   auto factory = std::make_unique<SimpleOperationContextFactory>();
 #endif
   auto mock = std::make_shared<MockBigtableStub>();
+  v2::PrepareQueryResponse response;
+
+  auto constexpr kResultMetadataText = R"pb(
+    proto_schema {
+      columns {
+        name: "row_key"
+        type { string_type {} }
+      }
+      columns {
+        name: "value"
+        type { string_type {} }
+      }
+    }
+  )pb";
+  *response.mutable_valid_until() = internal::ToProtoTimestamp(
+      std::chrono::system_clock::now() + std::chrono::seconds(3600));
+  ASSERT_TRUE(google::protobuf::TextFormat::ParseFromString(
+      kResultMetadataText, response.mutable_metadata()));
   EXPECT_CALL(*mock, PrepareQuery)
-      .WillOnce([](grpc::ClientContext&, Options const&,
-                   v2::PrepareQueryRequest const& request) {
+      .WillOnce([&](grpc::ClientContext&, Options const&,
+                    v2::PrepareQueryRequest const& request) {
         EXPECT_EQ(kAppProfile, request.app_profile_id());
         EXPECT_EQ("projects/the-project/instances/the-instance",
                   request.instance_name());
         EXPECT_EQ("SELECT * FROM the-table", request.query());
-        v2::PrepareQueryResponse response;
-        *response.mutable_valid_until() = internal::ToProtoTimestamp(
-            std::chrono::system_clock::now() + std::chrono::seconds(3600));
         return response;
       });
 
@@ -2922,14 +2937,29 @@ TEST_F(DataConnectionTest, AsyncPrepareQuerySuccess) {
   auto factory = std::make_unique<SimpleOperationContextFactory>();
 #endif
   auto mock = std::make_shared<MockBigtableStub>();
+  auto constexpr kResultMetadataText = R"pb(
+    proto_schema {
+      columns {
+        name: "row_key"
+        type { string_type {} }
+      }
+      columns {
+        name: "value"
+        type { string_type {} }
+      }
+    }
+  )pb";
+  v2::PrepareQueryResponse response;
+  ASSERT_TRUE(google::protobuf::TextFormat::ParseFromString(
+      kResultMetadataText, response.mutable_metadata()));
   EXPECT_CALL(*mock, AsyncPrepareQuery)
-      .WillOnce([](CompletionQueue const&, auto, auto,
-                   v2::PrepareQueryRequest const& request) {
+      .WillOnce([&](CompletionQueue const&, auto, auto,
+                    v2::PrepareQueryRequest const& request) {
         EXPECT_EQ(kAppProfile, request.app_profile_id());
         EXPECT_EQ("projects/the-project/instances/the-instance",
                   request.instance_name());
         EXPECT_EQ("SELECT * FROM the-table", request.query());
-        return make_ready_future(make_status_or(v2::PrepareQueryResponse{}));
+        return make_ready_future(make_status_or(response));
       });
 
   auto fake_cq_impl = std::make_shared<FakeCompletionQueueImpl>();

google/cloud/bigtable/internal/partial_result_set_source.cc

@@ -210,11 +210,6 @@ Status PartialResultSetSource::BufferProtoRows() {
     auto const columns_size = proto_schema.columns_size();
     auto const& proto_values = proto_rows_.values();
 
-    if (columns_size == 0 && !proto_values.empty()) {
-      return internal::InternalError(
-          "ProtoRows has values but ResultSetMetadata columns cannot be empty",
-          GCP_ERROR_INFO());
-    }
     if (proto_values.size() % columns_size != 0) {
       state_ = State::kFinished;
       return internal::InternalError(