From 6bb01b6811dc57635260429e69072d3fbdb15e6b Mon Sep 17 00:00:00 2001 From: Scott Hart Date: Wed, 7 Oct 2026 17:46:58 -0400 Subject: [PATCH 1/2] feat(spanner): add support for Queues in mutations --- .../spanner/integration_tests/CMakeLists.txt | 2 +- .../queue_integration_test.cc | 270 ++++++++++++++++++ .../spanner_client_integration_tests.bzl | 1 + google/cloud/spanner/mutations.cc | 43 +++ google/cloud/spanner/mutations.h | 89 ++++++ google/cloud/spanner/mutations_test.cc | 126 ++++++++ 6 files changed, 530 insertions(+), 1 deletion(-) create mode 100644 google/cloud/spanner/integration_tests/queue_integration_test.cc diff --git a/google/cloud/spanner/integration_tests/CMakeLists.txt b/google/cloud/spanner/integration_tests/CMakeLists.txt index 0321d360797a5..6ba3ae2016701 100644 --- a/google/cloud/spanner/integration_tests/CMakeLists.txt +++ b/google/cloud/spanner/integration_tests/CMakeLists.txt @@ -24,7 +24,7 @@ function (spanner_client_define_integration_tests) set(spanner_client_integration_tests # cmake-format: sort client_stress_test.cc data_types_integration_test.cc - session_pool_integration_test.cc) + queue_integration_test.cc session_pool_integration_test.cc) set(spanner_client_session_mode_sensitive_integration_tests # cmake-format: sort diff --git a/google/cloud/spanner/integration_tests/queue_integration_test.cc b/google/cloud/spanner/integration_tests/queue_integration_test.cc new file mode 100644 index 0000000000000..52c480b6dbb17 --- /dev/null +++ b/google/cloud/spanner/integration_tests/queue_integration_test.cc @@ -0,0 +1,270 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "google/cloud/spanner/admin/database_admin_client.h" +#include "google/cloud/spanner/client.h" +#include "google/cloud/spanner/mutations.h" +#include "google/cloud/spanner/testing/cleanup_stale_databases.h" +#include "google/cloud/spanner/testing/pick_random_instance.h" +#include "google/cloud/spanner/testing/random_database_name.h" +#include "google/cloud/internal/getenv.h" +#include "google/cloud/internal/random.h" +#include "google/cloud/testing_util/integration_test.h" +#include "google/cloud/testing_util/status_matchers.h" +#include "google/spanner/admin/database/v1/spanner_database_admin.pb.h" +#include +#include +#include +#include +#include +#include +#include + +namespace google { +namespace cloud { +namespace spanner { +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN +namespace { + +using ::google::cloud::testing_util::IsOk; +using ::google::cloud::testing_util::StatusIs; +using ::testing::ElementsAre; +using ::testing::IsEmpty; +using ::testing::Not; +using ::testing::Pair; + +class QueueIntegrationTest + : public ::google::cloud::testing_util::IntegrationTest { + protected: + static void SetUpTestSuite() { + ::google::cloud::testing_util::IntegrationTest::SetUpTestSuite(); + if (internal::GetEnv("SPANNER_EMULATOR_HOST").has_value()) { + GTEST_SKIP() << "Spanner Queues is not supported on the Emulator"; + } + + std::string const project_id = + internal::GetEnv("GOOGLE_CLOUD_PROJECT").value_or(""); + ASSERT_FALSE(project_id.empty()); + + internal::DefaultPRNG generator = internal::MakeDefaultPRNG(); + StatusOr instance_id = spanner_testing::PickRandomInstance( + generator, project_id, "labels.edition:enterprise"); + if (!instance_id.ok()) { + GTEST_SKIP() << "No Enterprise edition instance available: " + << instance_id.status(); + } + + std::string const database_id = + spanner_testing::RandomDatabaseName(generator); + db_ = new Database(project_id, *instance_id, database_id); + + spanner_admin::DatabaseAdminClient admin_client( + spanner_admin::MakeDatabaseAdminConnection()); + spanner_testing::CleanupStaleDatabases( + admin_client, project_id, *instance_id, + std::chrono::system_clock::now() - std::chrono::hours(7 * 24)); + + google::spanner::admin::database::v1::CreateDatabaseRequest request; + request.set_parent(db_->instance().FullName()); + request.set_create_statement("CREATE DATABASE `" + db_->database_id() + + "`"); + request.add_extra_statements(R"sql( + CREATE TABLE tasks ( + task_id INT64 NOT NULL, + status STRING(64) NOT NULL + ) PRIMARY KEY (task_id) + )sql"); + request.add_extra_statements(R"sql( + CREATE QUEUE testqueue ( + id INT64 NOT NULL, + payload BYTES(MAX) NOT NULL, + ) PRIMARY KEY (id), + OPTIONS (receive_mode = 'PULL') + )sql"); + + StatusOr database = + admin_client.CreateDatabase(request).get(); + if (!database.ok() && + (database.status().code() == StatusCode::kInvalidArgument || + database.status().code() == StatusCode::kUnimplemented)) { + delete db_; + db_ = nullptr; + GTEST_SKIP() << "Spanner Queues DDL not enabled on target endpoint: " + << database.status(); + } + ASSERT_THAT(database, IsOk()); + } + + static void TearDownTestSuite() { + if (db_ != nullptr) { + spanner_admin::DatabaseAdminClient admin_client( + spanner_admin::MakeDatabaseAdminConnection()); + Status drop_status = admin_client.DropDatabase(db_->FullName()); + EXPECT_THAT(drop_status, IsOk()); + delete db_; + db_ = nullptr; + } + ::google::cloud::testing_util::IntegrationTest::TearDownTestSuite(); + } + + void SetUp() override { + if (db_ == nullptr) { + GTEST_SKIP() << "QueueIntegrationTest suite was skipped"; + } + } + + static Database const& GetDatabase() { return *db_; } + + private: + static Database* db_; +}; + +Database* QueueIntegrationTest::db_ = nullptr; + +TEST_F(QueueIntegrationTest, SendAndAckMutations) { + Client client(MakeConnection(GetDatabase())); + + StatusOr deliver_time = + MakeTimestamp(std::chrono::system_clock::now()); + ASSERT_THAT(deliver_time, IsOk()); + + Bytes const payload1(std::string("payload-1")); + Bytes const payload2(std::string("payload-2")); + + // 1. Send immediate and scheduled messages via MakeSendMutation & Builder. + StatusOr send_commit = client.Commit(Mutations{ + MakeSendMutation("testqueue", MakeKey(std::int64_t{1}), Value(payload1)), + SendMutationBuilder("testqueue", MakeKey(std::int64_t{2}), + Value(payload2)) + .SetDeliverTime(*deliver_time) + .Build(), + }); + ASSERT_THAT(send_commit, IsOk()); + + // Verify messages exist in testqueue. + std::vector> read_rows; + RowStream query_rows = client.ExecuteQuery( + SqlStatement("SELECT id, payload FROM testqueue ORDER BY id")); + for (auto const& row : + StreamOf>(query_rows)) { + ASSERT_THAT(row, IsOk()); + read_rows.emplace_back(std::get<0>(*row), std::get<1>(*row)); + } + EXPECT_THAT(read_rows, ElementsAre(Pair(std::int64_t{1}, payload1), + Pair(std::int64_t{2}, payload2))); + + // 2. Ack existing messages. + StatusOr ack_commit = client.Commit(Mutations{ + MakeAckMutation("testqueue", MakeKey(std::int64_t{1})), + MakeAckMutation("testqueue", MakeKey(std::int64_t{2})), + }); + ASSERT_THAT(ack_commit, IsOk()); + + std::vector remaining_ids; + RowStream remaining_stream = + client.ExecuteQuery(SqlStatement("SELECT id FROM testqueue")); + for (auto const& row : StreamOf>(remaining_stream)) { + ASSERT_THAT(row, IsOk()); + remaining_ids.push_back(std::get<0>(*row)); + } + EXPECT_THAT(remaining_ids, IsEmpty()); + + // 3. Ack missing key with ignore_not_found = true succeeds. + StatusOr ack_ignored = client.Commit(Mutations{ + MakeAckMutation("testqueue", MakeKey(std::int64_t{999}), true), + }); + EXPECT_THAT(ack_ignored, IsOk()); + + // 4. Ack missing key with default ignore_not_found = false fails with + // kNotFound. + StatusOr ack_missing = client.Commit(Mutations{ + MakeAckMutation("testqueue", MakeKey(std::int64_t{999})), + }); + EXPECT_THAT(ack_missing, StatusIs(StatusCode::kNotFound)); +} + +TEST_F(QueueIntegrationTest, EndToEndTransactionalWorkflow) { + Client client(MakeConnection(GetDatabase())); + std::int64_t constexpr kTaskId = 1001; + Bytes const expected_payload(std::string("process-order-1001")); + + // 1. Producer: atomically insert task row and send queue message. + StatusOr produce_commit = client.Commit(Mutations{ + MakeInsertMutation("tasks", {"task_id", "status"}, kTaskId, "pending"), + SendMutationBuilder("testqueue", MakeKey(kTaskId), + Value(expected_payload)) + .Build(), + }); + ASSERT_THAT(produce_commit, IsOk()); + + // 2. Consumer: receive message and lease token via receive_ TVF. + RowStream receive_stream = client.ExecuteQuery( + SqlStatement("SELECT id, payload, spanner_lease_token " + "FROM spanner.receive_testqueue(max_batch_size => 1)")); + auto receive_it = receive_stream.begin(); + ASSERT_TRUE(receive_it != receive_stream.end()); + if (!receive_it->ok() && + (receive_it->status().code() == StatusCode::kInvalidArgument || + receive_it->status().code() == StatusCode::kUnimplemented)) { + (void)client.Commit(Mutations{ + MakeAckMutation("testqueue", MakeKey(kTaskId), true), + }); + GTEST_SKIP() + << "Spanner Queue consumer TVF not enabled on target endpoint: " + << receive_it->status(); + } + ASSERT_THAT(*receive_it, IsOk()); + StatusOr> received = + (*receive_it)->get>(); + ASSERT_THAT(received, IsOk()); + EXPECT_EQ(std::get<0>(*received), kTaskId); + EXPECT_EQ(std::get<1>(*received), expected_payload); + Bytes const lease_token = std::get<2>(*received); + EXPECT_THAT(lease_token.get(), Not(IsEmpty())); + + // 3. Extend lease via renewlease_ TVF. + RowStream renew_stream = client.ExecuteQuery( + SqlStatement("SELECT * FROM spanner.renewlease_testqueue([@lease_token])", + {{"lease_token", Value(lease_token)}})); + for (auto const& row : renew_stream) { + ASSERT_THAT(row, IsOk()); + } + + // 4. Worker completion: atomically update task row and ack queue message. + StatusOr complete_commit = client.Commit(Mutations{ + MakeUpdateMutation("tasks", {"task_id", "status"}, kTaskId, "done"), + AckMutationBuilder("testqueue", MakeKey(kTaskId)).Build(), + }); + ASSERT_THAT(complete_commit, IsOk()); + + // Verify queue is empty and task status is done. + RowStream verify_queue = + client.ExecuteQuery(SqlStatement("SELECT id FROM testqueue")); + EXPECT_TRUE(verify_queue.begin() == verify_queue.end()); + + RowStream verify_task = client.ExecuteQuery( + SqlStatement("SELECT status FROM tasks WHERE task_id = @id", + {{"id", Value(kTaskId)}})); + auto task_it = verify_task.begin(); + ASSERT_TRUE(task_it != verify_task.end()); + ASSERT_THAT(*task_it, IsOk()); + EXPECT_THAT((*task_it)->get(0), IsOk()); + EXPECT_EQ(*(*task_it)->get(0), "done"); +} + +} // namespace +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END +} // namespace spanner +} // namespace cloud +} // namespace google diff --git a/google/cloud/spanner/integration_tests/spanner_client_integration_tests.bzl b/google/cloud/spanner/integration_tests/spanner_client_integration_tests.bzl index f1e518714142e..0d34beb47e337 100644 --- a/google/cloud/spanner/integration_tests/spanner_client_integration_tests.bzl +++ b/google/cloud/spanner/integration_tests/spanner_client_integration_tests.bzl @@ -19,5 +19,6 @@ spanner_client_integration_tests = [ "client_stress_test.cc", "data_types_integration_test.cc", + "queue_integration_test.cc", "session_pool_integration_test.cc", ] diff --git a/google/cloud/spanner/mutations.cc b/google/cloud/spanner/mutations.cc index 8f5dcbdb3a6f3..41fb311bd8f5e 100644 --- a/google/cloud/spanner/mutations.cc +++ b/google/cloud/spanner/mutations.cc @@ -13,11 +13,54 @@ // limitations under the License. #include "google/cloud/spanner/mutations.h" +#include #include #include +#include namespace google { namespace cloud { +namespace spanner_internal { +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN + +SendMutationBuilder::SendMutationBuilder(std::string queue, spanner::Key key, + spanner::Value payload) { + google::spanner::v1::Mutation::Send& send = *m_.proto().mutable_send(); + send.set_queue(std::move(queue)); + google::protobuf::ListValue& key_proto = *send.mutable_key(); + for (auto& k : key) { + *key_proto.add_values() = spanner_internal::ToProto(std::move(k)).second; + } + *send.mutable_payload() = + spanner_internal::ToProto(std::move(payload)).second; +} + +SendMutationBuilder& SendMutationBuilder::SetDeliverTime( + spanner::Timestamp deliver_time) & { + StatusOr ts = + deliver_time.get(); + *m_.proto().mutable_send()->mutable_deliver_time() = *std::move(ts); + return *this; +} + +AckMutationBuilder::AckMutationBuilder(std::string queue, spanner::Key key) { + google::spanner::v1::Mutation::Ack& ack = *m_.proto().mutable_ack(); + ack.set_queue(std::move(queue)); + google::protobuf::ListValue& key_proto = *ack.mutable_key(); + for (auto& k : key) { + *key_proto.add_values() = spanner_internal::ToProto(std::move(k)).second; + } +} + +AckMutationBuilder& AckMutationBuilder::SetIgnoreNotFound( + bool ignore_not_found) & { + m_.proto().mutable_ack()->set_ignore_not_found(ignore_not_found); + return *this; +} + +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END +} // namespace spanner_internal + namespace spanner { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN diff --git a/google/cloud/spanner/mutations.h b/google/cloud/spanner/mutations.h index e342406680926..3f510627f974a 100644 --- a/google/cloud/spanner/mutations.h +++ b/google/cloud/spanner/mutations.h @@ -16,9 +16,11 @@ #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_SPANNER_MUTATIONS_H #include "google/cloud/spanner/keys.h" +#include "google/cloud/spanner/timestamp.h" #include "google/cloud/spanner/value.h" #include "google/cloud/spanner/version.h" #include "google/spanner/v1/mutation.pb.h" +#include #include #include @@ -29,6 +31,8 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN template class WriteMutationBuilder; class DeleteMutationBuilder; +class SendMutationBuilder; +class AckMutationBuilder; GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace spanner_internal @@ -87,6 +91,8 @@ class Mutation { template friend class spanner_internal::WriteMutationBuilder; friend class spanner_internal::DeleteMutationBuilder; + friend class spanner_internal::SendMutationBuilder; + friend class spanner_internal::AckMutationBuilder; explicit Mutation(google::spanner::v1::Mutation m) : m_(std::move(m)) {} google::spanner::v1::Mutation m_; @@ -192,6 +198,39 @@ class DeleteMutationBuilder { spanner::Mutation m_; }; +class SendMutationBuilder { + public: + SendMutationBuilder(std::string queue, spanner::Key key, + spanner::Value payload); + + SendMutationBuilder& SetDeliverTime(spanner::Timestamp deliver_time) &; + SendMutationBuilder&& SetDeliverTime(spanner::Timestamp deliver_time) && { + return std::move(SetDeliverTime(std::move(deliver_time))); + } + + spanner::Mutation Build() const& { return m_; } + spanner::Mutation&& Build() && { return std::move(m_); } + + private: + spanner::Mutation m_; +}; + +class AckMutationBuilder { + public: + AckMutationBuilder(std::string queue, spanner::Key key); + + AckMutationBuilder& SetIgnoreNotFound(bool ignore_not_found) &; + AckMutationBuilder&& SetIgnoreNotFound(bool ignore_not_found) && { + return std::move(SetIgnoreNotFound(ignore_not_found)); + } + + spanner::Mutation Build() const& { return m_; } + spanner::Mutation&& Build() && { return std::move(m_); } + + private: + spanner::Mutation m_; +}; + GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace spanner_internal @@ -370,6 +409,56 @@ inline Mutation MakeDeleteMutation(std::string table_name, KeySet keys) { return DeleteMutationBuilder(std::move(table_name), std::move(keys)).Build(); } +/** + * A helper class to construct queue "send" mutations. + */ +using SendMutationBuilder = spanner_internal::SendMutationBuilder; + +/** + * Creates a "send" mutation to enqueue a message with @p key and @p payload + * into @p queue for immediate delivery. + */ +inline Mutation MakeSendMutation(std::string queue, Key key, Value payload) { + return SendMutationBuilder(std::move(queue), std::move(key), + std::move(payload)) + .Build(); +} + +/** + * Creates a "send" mutation to enqueue a message with @p key and @p payload + * into @p queue scheduled for @p deliver_time. + */ +inline Mutation MakeSendMutation(std::string queue, Key key, Value payload, + Timestamp deliver_time) { + return SendMutationBuilder(std::move(queue), std::move(key), + std::move(payload)) + .SetDeliverTime(std::move(deliver_time)) + .Build(); +} + +/** + * A helper class to construct queue "ack" mutations. + */ +using AckMutationBuilder = spanner_internal::AckMutationBuilder; + +/** + * Creates an "ack" mutation to acknowledge the message with @p key in @p queue. + */ +inline Mutation MakeAckMutation(std::string queue, Key key) { + return AckMutationBuilder(std::move(queue), std::move(key)).Build(); +} + +/** + * Creates an "ack" mutation to acknowledge the message with @p key in @p queue, + * optionally ignoring `NOT_FOUND` errors according to @p ignore_not_found. + */ +inline Mutation MakeAckMutation(std::string queue, Key key, + bool ignore_not_found) { + return AckMutationBuilder(std::move(queue), std::move(key)) + .SetIgnoreNotFound(ignore_not_found) + .Build(); +} + GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace spanner } // namespace cloud diff --git a/google/cloud/spanner/mutations_test.cc b/google/cloud/spanner/mutations_test.cc index ed1f57b4bd11a..1a6982e24fc63 100644 --- a/google/cloud/spanner/mutations_test.cc +++ b/google/cloud/spanner/mutations_test.cc @@ -20,6 +20,7 @@ #include "google/cloud/spanner/numeric.h" #include "google/cloud/spanner/timestamp.h" #include "google/cloud/testing_util/is_proto_equal.h" +#include "google/cloud/testing_util/status_matchers.h" #include #include #include @@ -38,6 +39,7 @@ namespace spanner { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN namespace { +using ::google::cloud::testing_util::IsOk; using ::google::cloud::testing_util::IsProtoApproximatelyEqual; using ::google::cloud::testing_util::IsProtoEqual; using ::google::protobuf::TextFormat; @@ -489,6 +491,130 @@ TEST(MutationsTest, FluentDeleteBuilder) { EXPECT_THAT(actual, IsProtoEqual(expected)); } +TEST(MutationsTest, SendDefault) { + Mutation send = MakeSendMutation("queue-name", MakeKey(std::int64_t{123}), + Value("payload-data")); + EXPECT_EQ(send, SendMutationBuilder("queue-name", MakeKey(std::int64_t{123}), + Value("payload-data")) + .Build()); + EXPECT_NE(send, MakeSendMutation("other-queue", MakeKey(std::int64_t{123}), + Value("payload-data"))); + + google::spanner::v1::Mutation actual = std::move(send).as_proto(); + char constexpr kText[] = R"pb( + send: { + queue: "queue-name" + key: { values { string_value: "123" } } + payload: { string_value: "payload-data" } + } + )pb"; + google::spanner::v1::Mutation expected; + ASSERT_TRUE(TextFormat::ParseFromString(kText, &expected)); + EXPECT_THAT(actual, IsProtoEqual(expected)); +} + +TEST(MutationsTest, SendWithDeliverTime) { + StatusOr ts = MakeTimestamp(absl::FromUnixSeconds(1700000000) + + absl::Nanoseconds(123456789)); + ASSERT_THAT(ts, IsOk()); + + Mutation send = + MakeSendMutation("queue-name", MakeKey("msg-1"), + Value(Bytes(std::string("binary-payload"))), *ts); + Mutation from_builder = + SendMutationBuilder("queue-name", MakeKey("msg-1"), + Value(Bytes(std::string("binary-payload")))) + .SetDeliverTime(*ts) + .Build(); + EXPECT_EQ(send, from_builder); + + google::spanner::v1::Mutation actual = std::move(send).as_proto(); + char constexpr kText[] = R"pb( + send: { + queue: "queue-name" + key: { values { string_value: "msg-1" } } + deliver_time: { seconds: 1700000000 nanos: 123456789 } + payload: { string_value: "YmluYXJ5LXBheWxvYWQ=" } + } + )pb"; + google::spanner::v1::Mutation expected; + ASSERT_TRUE(TextFormat::ParseFromString(kText, &expected)); + EXPECT_THAT(actual, IsProtoEqual(expected)); +} + +TEST(MutationsTest, FluentSendBuilder) { + static_assert( + std::is_rvalue_reference() + .SetDeliverTime( + std::declval()) + .Build())>::value, + "Build() should return an rvalue if called fluently on a temporary"); + + SendMutationBuilder builder("queue-name", MakeKey("k1"), Value("v1")); + Mutation m1 = builder.Build(); + Mutation m2 = std::move(builder).Build(); + EXPECT_EQ(m1, m2); +} + +TEST(MutationsTest, AckDefault) { + Mutation ack = + MakeAckMutation("queue-name", MakeKey(std::int64_t{42}, "sub")); + EXPECT_EQ(ack, + AckMutationBuilder("queue-name", MakeKey(std::int64_t{42}, "sub")) + .Build()); + EXPECT_NE(ack, + MakeAckMutation("other-queue", MakeKey(std::int64_t{42}, "sub"))); + + google::spanner::v1::Mutation actual = std::move(ack).as_proto(); + char constexpr kText[] = R"pb( + ack: { + queue: "queue-name" + key: { + values { string_value: "42" } + values { string_value: "sub" } + } + ignore_not_found: false + } + )pb"; + google::spanner::v1::Mutation expected; + ASSERT_TRUE(TextFormat::ParseFromString(kText, &expected)); + EXPECT_THAT(actual, IsProtoEqual(expected)); +} + +TEST(MutationsTest, AckWithIgnoreNotFound) { + Mutation ack = MakeAckMutation("queue-name", MakeKey("msg-1"), true); + Mutation from_builder = AckMutationBuilder("queue-name", MakeKey("msg-1")) + .SetIgnoreNotFound(true) + .Build(); + EXPECT_EQ(ack, from_builder); + EXPECT_NE(ack, MakeAckMutation("queue-name", MakeKey("msg-1"), false)); + + google::spanner::v1::Mutation actual = std::move(ack).as_proto(); + char constexpr kText[] = R"pb( + ack: { + queue: "queue-name" + key: { values { string_value: "msg-1" } } + ignore_not_found: true + } + )pb"; + google::spanner::v1::Mutation expected; + ASSERT_TRUE(TextFormat::ParseFromString(kText, &expected)); + EXPECT_THAT(actual, IsProtoEqual(expected)); +} + +TEST(MutationsTest, FluentAckBuilder) { + static_assert( + std::is_rvalue_reference() + .SetIgnoreNotFound(true) + .Build())>::value, + "Build() should return an rvalue if called fluently on a temporary"); + + AckMutationBuilder builder("queue-name", MakeKey("k1")); + Mutation m1 = builder.Build(); + Mutation m2 = std::move(builder).Build(); + EXPECT_EQ(m1, m2); +} + } // namespace GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace spanner From 91ea6c4314e2a1eb8166aef7ead1fff29145a663 Mon Sep 17 00:00:00 2001 From: Scott Hart Date: Thu, 8 Oct 2026 09:40:14 -0400 Subject: [PATCH 2/2] address review comments --- .../queue_integration_test.cc | 53 ++++++++++++------- google/cloud/spanner/mutations.cc | 11 ++-- google/cloud/spanner/mutations.h | 37 ++++++++++++- 3 files changed, 77 insertions(+), 24 deletions(-) diff --git a/google/cloud/spanner/integration_tests/queue_integration_test.cc b/google/cloud/spanner/integration_tests/queue_integration_test.cc index 52c480b6dbb17..c1321fa3968e4 100644 --- a/google/cloud/spanner/integration_tests/queue_integration_test.cc +++ b/google/cloud/spanner/integration_tests/queue_integration_test.cc @@ -22,6 +22,7 @@ #include "google/cloud/internal/random.h" #include "google/cloud/testing_util/integration_test.h" #include "google/cloud/testing_util/status_matchers.h" +#include "absl/types/optional.h" #include "google/spanner/admin/database/v1/spanner_database_admin.pb.h" #include #include @@ -38,6 +39,7 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN namespace { using ::google::cloud::testing_util::IsOk; +using ::google::cloud::testing_util::IsOkAndHolds; using ::google::cloud::testing_util::StatusIs; using ::testing::ElementsAre; using ::testing::IsEmpty; @@ -55,7 +57,7 @@ class QueueIntegrationTest std::string const project_id = internal::GetEnv("GOOGLE_CLOUD_PROJECT").value_or(""); - ASSERT_FALSE(project_id.empty()); + ASSERT_THAT(project_id, Not(IsEmpty())); internal::DefaultPRNG generator = internal::MakeDefaultPRNG(); StatusOr instance_id = spanner_testing::PickRandomInstance( @@ -67,7 +69,7 @@ class QueueIntegrationTest std::string const database_id = spanner_testing::RandomDatabaseName(generator); - db_ = new Database(project_id, *instance_id, database_id); + db_.emplace(project_id, *instance_id, database_id); spanner_admin::DatabaseAdminClient admin_client( spanner_admin::MakeDatabaseAdminConnection()); @@ -98,8 +100,7 @@ class QueueIntegrationTest if (!database.ok() && (database.status().code() == StatusCode::kInvalidArgument || database.status().code() == StatusCode::kUnimplemented)) { - delete db_; - db_ = nullptr; + db_.reset(); GTEST_SKIP() << "Spanner Queues DDL not enabled on target endpoint: " << database.status(); } @@ -107,30 +108,49 @@ class QueueIntegrationTest } static void TearDownTestSuite() { - if (db_ != nullptr) { + if (db_.has_value()) { spanner_admin::DatabaseAdminClient admin_client( spanner_admin::MakeDatabaseAdminConnection()); Status drop_status = admin_client.DropDatabase(db_->FullName()); EXPECT_THAT(drop_status, IsOk()); - delete db_; - db_ = nullptr; + db_.reset(); } ::google::cloud::testing_util::IntegrationTest::TearDownTestSuite(); } void SetUp() override { - if (db_ == nullptr) { + if (!db_.has_value()) { GTEST_SKIP() << "QueueIntegrationTest suite was skipped"; } } + void TearDown() override { + if (db_.has_value()) { + Client client(MakeConnection(*db_)); + RowStream rows = + client.ExecuteQuery(SqlStatement("SELECT id FROM testqueue")); + Mutations cleanup; + for (StatusOr> const& row : + StreamOf>(rows)) { + if (row.ok()) { + cleanup.push_back( + MakeAckMutation("testqueue", MakeKey(std::get<0>(*row)), true)); + } + } + if (!cleanup.empty()) { + (void)client.Commit(std::move(cleanup)); + } + } + ::google::cloud::testing_util::IntegrationTest::TearDown(); + } + static Database const& GetDatabase() { return *db_; } private: - static Database* db_; + static absl::optional db_; }; -Database* QueueIntegrationTest::db_ = nullptr; +absl::optional QueueIntegrationTest::db_; TEST_F(QueueIntegrationTest, SendAndAckMutations) { Client client(MakeConnection(GetDatabase())); @@ -156,7 +176,7 @@ TEST_F(QueueIntegrationTest, SendAndAckMutations) { std::vector> read_rows; RowStream query_rows = client.ExecuteQuery( SqlStatement("SELECT id, payload FROM testqueue ORDER BY id")); - for (auto const& row : + for (StatusOr> const& row : StreamOf>(query_rows)) { ASSERT_THAT(row, IsOk()); read_rows.emplace_back(std::get<0>(*row), std::get<1>(*row)); @@ -174,7 +194,8 @@ TEST_F(QueueIntegrationTest, SendAndAckMutations) { std::vector remaining_ids; RowStream remaining_stream = client.ExecuteQuery(SqlStatement("SELECT id FROM testqueue")); - for (auto const& row : StreamOf>(remaining_stream)) { + for (StatusOr> const& row : + StreamOf>(remaining_stream)) { ASSERT_THAT(row, IsOk()); remaining_ids.push_back(std::get<0>(*row)); } @@ -217,9 +238,6 @@ TEST_F(QueueIntegrationTest, EndToEndTransactionalWorkflow) { if (!receive_it->ok() && (receive_it->status().code() == StatusCode::kInvalidArgument || receive_it->status().code() == StatusCode::kUnimplemented)) { - (void)client.Commit(Mutations{ - MakeAckMutation("testqueue", MakeKey(kTaskId), true), - }); GTEST_SKIP() << "Spanner Queue consumer TVF not enabled on target endpoint: " << receive_it->status(); @@ -237,7 +255,7 @@ TEST_F(QueueIntegrationTest, EndToEndTransactionalWorkflow) { RowStream renew_stream = client.ExecuteQuery( SqlStatement("SELECT * FROM spanner.renewlease_testqueue([@lease_token])", {{"lease_token", Value(lease_token)}})); - for (auto const& row : renew_stream) { + for (StatusOr const& row : renew_stream) { ASSERT_THAT(row, IsOk()); } @@ -259,8 +277,7 @@ TEST_F(QueueIntegrationTest, EndToEndTransactionalWorkflow) { auto task_it = verify_task.begin(); ASSERT_TRUE(task_it != verify_task.end()); ASSERT_THAT(*task_it, IsOk()); - EXPECT_THAT((*task_it)->get(0), IsOk()); - EXPECT_EQ(*(*task_it)->get(0), "done"); + EXPECT_THAT((*task_it)->get(0), IsOkAndHolds("done")); } } // namespace diff --git a/google/cloud/spanner/mutations.cc b/google/cloud/spanner/mutations.cc index 41fb311bd8f5e..cb5c367b0048e 100644 --- a/google/cloud/spanner/mutations.cc +++ b/google/cloud/spanner/mutations.cc @@ -28,7 +28,8 @@ SendMutationBuilder::SendMutationBuilder(std::string queue, spanner::Key key, google::spanner::v1::Mutation::Send& send = *m_.proto().mutable_send(); send.set_queue(std::move(queue)); google::protobuf::ListValue& key_proto = *send.mutable_key(); - for (auto& k : key) { + key_proto.mutable_values()->Reserve(static_cast(key.size())); + for (spanner::Value& k : key) { *key_proto.add_values() = spanner_internal::ToProto(std::move(k)).second; } *send.mutable_payload() = @@ -37,9 +38,8 @@ SendMutationBuilder::SendMutationBuilder(std::string queue, spanner::Key key, SendMutationBuilder& SendMutationBuilder::SetDeliverTime( spanner::Timestamp deliver_time) & { - StatusOr ts = - deliver_time.get(); - *m_.proto().mutable_send()->mutable_deliver_time() = *std::move(ts); + *m_.proto().mutable_send()->mutable_deliver_time() = + deliver_time.get().value(); return *this; } @@ -47,7 +47,8 @@ AckMutationBuilder::AckMutationBuilder(std::string queue, spanner::Key key) { google::spanner::v1::Mutation::Ack& ack = *m_.proto().mutable_ack(); ack.set_queue(std::move(queue)); google::protobuf::ListValue& key_proto = *ack.mutable_key(); - for (auto& k : key) { + key_proto.mutable_values()->Reserve(static_cast(key.size())); + for (spanner::Value& k : key) { *key_proto.add_values() = spanner_internal::ToProto(std::move(k)).second; } } diff --git a/google/cloud/spanner/mutations.h b/google/cloud/spanner/mutations.h index 3f510627f974a..63c9cae0bb71c 100644 --- a/google/cloud/spanner/mutations.h +++ b/google/cloud/spanner/mutations.h @@ -20,7 +20,6 @@ #include "google/cloud/spanner/value.h" #include "google/cloud/spanner/version.h" #include "google/spanner/v1/mutation.pb.h" -#include #include #include @@ -411,12 +410,24 @@ inline Mutation MakeDeleteMutation(std::string table_name, KeySet keys) { /** * A helper class to construct queue "send" mutations. + * + * @see The Mutation class documentation for an overview of the Cloud Spanner + * mutation API + * + * @see https://cloud.google.com/spanner/docs/modify-mutation-api + * for more information about the Cloud Spanner mutation API. */ using SendMutationBuilder = spanner_internal::SendMutationBuilder; /** * Creates a "send" mutation to enqueue a message with @p key and @p payload * into @p queue for immediate delivery. + * + * @see The Mutation class documentation for an overview of the Cloud Spanner + * mutation API + * + * @see https://cloud.google.com/spanner/docs/modify-mutation-api + * for more information about the Cloud Spanner mutation API. */ inline Mutation MakeSendMutation(std::string queue, Key key, Value payload) { return SendMutationBuilder(std::move(queue), std::move(key), @@ -427,6 +438,12 @@ inline Mutation MakeSendMutation(std::string queue, Key key, Value payload) { /** * Creates a "send" mutation to enqueue a message with @p key and @p payload * into @p queue scheduled for @p deliver_time. + * + * @see The Mutation class documentation for an overview of the Cloud Spanner + * mutation API + * + * @see https://cloud.google.com/spanner/docs/modify-mutation-api + * for more information about the Cloud Spanner mutation API. */ inline Mutation MakeSendMutation(std::string queue, Key key, Value payload, Timestamp deliver_time) { @@ -438,11 +455,23 @@ inline Mutation MakeSendMutation(std::string queue, Key key, Value payload, /** * A helper class to construct queue "ack" mutations. + * + * @see The Mutation class documentation for an overview of the Cloud Spanner + * mutation API + * + * @see https://cloud.google.com/spanner/docs/modify-mutation-api + * for more information about the Cloud Spanner mutation API. */ using AckMutationBuilder = spanner_internal::AckMutationBuilder; /** * Creates an "ack" mutation to acknowledge the message with @p key in @p queue. + * + * @see The Mutation class documentation for an overview of the Cloud Spanner + * mutation API + * + * @see https://cloud.google.com/spanner/docs/modify-mutation-api + * for more information about the Cloud Spanner mutation API. */ inline Mutation MakeAckMutation(std::string queue, Key key) { return AckMutationBuilder(std::move(queue), std::move(key)).Build(); @@ -451,6 +480,12 @@ inline Mutation MakeAckMutation(std::string queue, Key key) { /** * Creates an "ack" mutation to acknowledge the message with @p key in @p queue, * optionally ignoring `NOT_FOUND` errors according to @p ignore_not_found. + * + * @see The Mutation class documentation for an overview of the Cloud Spanner + * mutation API + * + * @see https://cloud.google.com/spanner/docs/modify-mutation-api + * for more information about the Cloud Spanner mutation API. */ inline Mutation MakeAckMutation(std::string queue, Key key, bool ignore_not_found) {