mutation: Make mutation equality comparable

mutation.cc

@@ -88,3 +88,11 @@ void mutation::update_column(row& row, const column_definition& def, atomic_cell
         merge_column(def, i->second, value);
     }
 }
+
+bool mutation::operator==(const mutation& m) const {
+    return _dk.equal(*_schema, m._dk) && _p.equal(*_schema, m._p);
+}
+
+bool mutation::operator!=(const mutation& m) const {
+    return !(*this == m);
+}

mutation.hh

@@ -35,6 +35,8 @@ public:
     const mutation_partition& partition() const { return _p; }
     mutation_partition& partition() { return _p; }
     const utils::UUID& column_family_id() const { return _schema->id(); }
+    bool operator==(const mutation&) const;
+    bool operator!=(const mutation&) const;
 private:
     static void update_column(row& row, const column_definition& def, atomic_cell_or_collection&& value);
     friend std::ostream& operator<<(std::ostream& os, const mutation& m);

mutation_partition.cc

@@ -346,3 +346,50 @@ operator<<(std::ostream& os, const mutation_partition& mp) {
                   mp._tombstone, ::join(", ", mp._row_tombstones), mp._static_row,
                   ::join(", ", mp._rows));
 }
+
+static bool
+rows_equal(const schema& s, const row& r1, const row& r2) {
+    return std::equal(r1.begin(), r1.end(), r2.begin(), r2.end(),
+        [] (const row::value_type& c1, const row::value_type& c2) {
+            return c1.first == c2.first && c1.second.serialize() == c2.second.serialize();
+        });
+}
+
+bool
+deletable_row::equal(const schema& s, const deletable_row& other) const {
+    if (t != other.t || created_at != other.created_at) {
+        return false;
+    }
+    return rows_equal(s, cells, other.cells);
+}
+
+bool
+rows_entry::equal(const schema& s, const rows_entry& other) const {
+    return key().equal(s, other.key()) && row().equal(s, other.row());
+}
+
+bool
+row_tombstones_entry::equal(const schema& s, const row_tombstones_entry& other) const {
+    return prefix().equal(s, other.prefix()) && t() == other.t();
+}
+
+bool mutation_partition::equal(const schema& s, const mutation_partition& p) const {
+    if (_tombstone != p._tombstone) {
+        return false;
+    }
+
+    if (!std::equal(_rows.begin(), _rows.end(), p._rows.begin(), p._rows.end(),
+        [&s] (const rows_entry& e1, const rows_entry& e2) { return e1.equal(s, e2); }
+    )) {
+        return false;
+    }
+
+    if (!std::equal(_row_tombstones.begin(), _row_tombstones.end(),
+        p._row_tombstones.begin(), p._row_tombstones.end(),
+        [&s] (const row_tombstones_entry& e1, const row_tombstones_entry& e2) { return e1.equal(s, e2); }
+    )) {
+        return false;
+    }
+
+    return rows_equal(s, _static_row, p._static_row);
+}

mutation_partition.hh

@@ -14,11 +14,13 @@
 #include "atomic_cell.hh"
 #include "query-result-writer.hh"
 
+// FIXME: Encapsulate
 using row = std::map<column_id, atomic_cell_or_collection>;
 
 std::ostream& operator<<(std::ostream& os, const row::value_type& rv);
 std::ostream& operator<<(std::ostream& os, const row& r);
 
+// FIXME: Encapsulate
 struct deletable_row final {
     tombstone t;
     api::timestamp_type created_at = api::missing_timestamp;
@@ -29,6 +31,7 @@ struct deletable_row final {
     }
 
     friend std::ostream& operator<<(std::ostream& os, const deletable_row& dr);
+    bool equal(const schema& s, const deletable_row& other) const;
 };
 
 class row_tombstones_entry : public boost::intrusive::set_base_hook<> {
@@ -86,6 +89,7 @@ public:
     }
 
     friend std::ostream& operator<<(std::ostream& os, const row_tombstones_entry& rte);
+    bool equal(const schema& s, const row_tombstones_entry& other) const;
 };
 
 class rows_entry : public boost::intrusive::set_base_hook<> {
@@ -154,6 +158,7 @@ public:
         return delegating_compare<Comparator>(std::move(c));
     }
     friend std::ostream& operator<<(std::ostream& os, const rows_entry& re);
+    bool equal(const schema& s, const rows_entry& other) const;
 };
 
 namespace db {
@@ -209,4 +214,6 @@ public:
     friend std::ostream& operator<<(std::ostream& os, const mutation_partition& mp);
     boost::iterator_range<rows_type::const_iterator> range(const schema& schema, const query::range<clustering_key_prefix>& r) const;
     void query(const schema& s, const query::partition_slice& slice, uint32_t limit, query::result::partition_writer& pw) const;
+public:
+    bool equal(const schema& s, const mutation_partition&) const;
 };

tests/urchin/serializer_test.cc

@@ -143,42 +143,6 @@ static atomic_cell make_atomic_cell(bytes value) {
     return atomic_cell::make_live(0, ttl_opt{}, std::move(value));
 }
 
-inline bool operator==(const deletable_row& r1, const deletable_row& r2) {
-    return r1.t == r2.t && r1.cells == r2.cells;
-}
-
-inline bool operator==(const partition_key& m1, const partition_key& m2) {
-    const bytes_view& b1 = m1;
-    const bytes_view& b2 = m2;
-    return b1 == b2;
-}
-
-inline bool operator==(const rows_entry& r1, const rows_entry& r2) {
-    return r1.row() == r2.row();
-}
-
-inline bool operator!=(const rows_entry& r1, const rows_entry& r2) {
-    return !(r1 == r2);
-}
-
-// TODO: not complete... meh...
-inline bool operator==(const mutation_partition& cp1, const mutation_partition& cp2) {
-    static schema dummy({}, "", "", {}, {}, {}, {}, utf8_type);
-    auto& p1 = const_cast<mutation_partition&>(cp1);
-    auto& p2 = const_cast<mutation_partition&>(cp2);
-    return p1.static_row() == p2.static_row()
-            && p1.range(dummy, query::range<clustering_key_prefix>::make_open_ended_both_sides())
-            == p2.range(dummy, query::range<clustering_key_prefix>::make_open_ended_both_sides())
-            ;
-}
-
-inline bool operator==(const mutation& m1, const mutation& m2) {
-    return m1.schema().get() == m2.schema().get()
-            && m1.key() == m2.key()
-            && m1.partition() == m2.partition()
-            ;
-}
-
 SEASTAR_TEST_CASE(test_mutation){
     auto s = make_lw_shared(schema({}, some_keyspace, some_column_family,
         {{"p1", utf8_type}}, {{"c1", int32_type}}, {{"r1", int32_type}}, {}, utf8_type));