From a016e1d05df1c88ec101bb668e514a46f15484e0 Mon Sep 17 00:00:00 2001
From: Benny Halevy <bhalevy@scylladb.com>
Date: Wed, 17 Apr 2024 22:49:23 +0300
Subject: [PATCH] canonical_mutation: add make_canonical_mutation_gently

Make a canonical mutation gently using an
async serialization function.
Similar to freeze_gently, yielding is considered
only in-between range tombstones and rows.

Signed-off-by: Benny Halevy <bhalevy@scylladb.com>
---
 mutation/async_utils.cc               | 17 +++++++++++++++++
 mutation/async_utils.hh               |  1 +
 mutation/canonical_mutation.hh        |  2 ++
 test/boost/canonical_mutation_test.cc |  7 +++----
 4 files changed, 23 insertions(+), 4 deletions(-)

diff --git a/mutation/async_utils.cc b/mutation/async_utils.cc
index d07ce35eb670..66f056bb41f1 100644
--- a/mutation/async_utils.cc
+++ b/mutation/async_utils.cc
@@ -90,6 +90,23 @@ future<mutation> to_mutation_gently(const canonical_mutation& cm, schema_ptr s)
     co_return m;
 }
 
+future<canonical_mutation> make_canonical_mutation_gently(const mutation& m)
+{
+    mutation_partition_serializer part_ser(*m.schema(), m.partition());
+
+    canonical_mutation res;
+    ser::writer_of_canonical_mutation<bytes_ostream> wr(res.representation());
+    auto w = co_await std::move(wr).write_table_id(m.schema()->id())
+                 .write_schema_version(m.schema()->version())
+                 .write_key(m.key())
+                 .write_mapping(m.schema()->get_column_mapping())
+                 .partition_gently([&] (auto wr) {
+                     return part_ser.write_gently(std::move(wr));
+                 });
+    w.end_canonical_mutation();
+    co_return res;
+}
+
 future<frozen_mutation>
 freeze_gently(const mutation& m) {
     auto fm = frozen_mutation(m.key());
diff --git a/mutation/async_utils.hh b/mutation/async_utils.hh
index 3266ac69a4c7..5903e67dab57 100644
--- a/mutation/async_utils.hh
+++ b/mutation/async_utils.hh
@@ -37,6 +37,7 @@ future<> apply_gently(mutation& target, mutation&& m);
 future<> apply_gently(mutation& target, const mutation& m);
 
 future<mutation> to_mutation_gently(const canonical_mutation& cm, schema_ptr s);
+future<canonical_mutation> make_canonical_mutation_gently(const mutation& m);
 
 future<frozen_mutation> freeze_gently(const mutation& m);
 future<mutation> unfreeze_gently(const frozen_mutation& fm, schema_ptr schema);
diff --git a/mutation/canonical_mutation.hh b/mutation/canonical_mutation.hh
index 0641b4e980a3..259a1a62449b 100644
--- a/mutation/canonical_mutation.hh
+++ b/mutation/canonical_mutation.hh
@@ -19,6 +19,7 @@
 class canonical_mutation {
     bytes_ostream _data;
 public:
+    canonical_mutation() = default;
     explicit canonical_mutation(bytes_ostream);
     explicit canonical_mutation(const mutation&);
 
@@ -37,6 +38,7 @@ public:
     table_id column_family_id() const;
 
     const bytes_ostream& representation() const { return _data; }
+    bytes_ostream& representation() { return _data; }
 
     friend fmt::formatter<canonical_mutation>;
 };
diff --git a/test/boost/canonical_mutation_test.cc b/test/boost/canonical_mutation_test.cc
index cb38a4d75b91..46027f155024 100644
--- a/test/boost/canonical_mutation_test.cc
+++ b/test/boost/canonical_mutation_test.cc
@@ -19,16 +19,15 @@
 #include <seastar/core/thread.hh>
 
 SEASTAR_THREAD_TEST_CASE(test_conversion_back_and_forth) {
-    // FIXME: for (auto do_make_canonical_mutation_gently : {false, true}) {
+    for (auto do_make_canonical_mutation_gently : {false, true}) {
         for (auto do_make_mutation_gently : {false, true}) {
             for_each_mutation([&] (const mutation& m) {
-                // FIXME: for now
-                canonical_mutation cm(m);
+                auto cm = do_make_canonical_mutation_gently ? make_canonical_mutation_gently(m).get() : canonical_mutation{m};
                 auto m2 = do_make_mutation_gently ? to_mutation_gently(cm, m.schema()).get() : cm.to_mutation(m.schema());
                 assert_that(m2).is_equal_to(m);
             });
         }
-    // FIXME }
+    }
 }
 
 SEASTAR_TEST_CASE(test_reading_with_different_schemas) {