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 <[email protected]>

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());

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);

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>;
 };

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) {