diff --git a/tests/mutation_source_test.cc b/tests/mutation_source_test.cc
index 1dac141ed0..b11270958a 100644
--- a/tests/mutation_source_test.cc
+++ b/tests/mutation_source_test.cc
@@ -30,6 +30,9 @@
 #include "flat_mutation_reader_assertions.hh"
 #include "mutation_query.hh"
 #include "mutation_rebuilder.hh"
+#include "random-utils.hh"
+#include "cql3/cql3_type.hh"
+#include <boost/algorithm/string/join.hpp>
 
 // partitions must be sorted by decorated key
 static void require_no_token_duplicates(const std::vector<mutation>& partitions) {
@@ -1700,3 +1703,317 @@ clustering_key random_mutation_generator::make_random_key() {
 std::vector<query::clustering_range> random_mutation_generator::make_random_ranges(unsigned n_ranges) {
     return _impl->make_random_ranges(n_ranges);
 }
+
+namespace tests::data_model {
+
+static constexpr const api::timestamp_type previously_removed_column_timestamp = 100;
+static constexpr const api::timestamp_type data_timestamp = 200;
+static constexpr const api::timestamp_type column_removal_timestamp = 300;
+
+class mutation_description {
+public:
+    using key = std::vector<bytes>;
+    struct collection_element {
+        bytes key;
+        bytes value;
+    };
+    using collection = std::vector<collection_element>;
+    using atomic_value = bytes;
+    using value = std::variant<atomic_value, collection>;
+    struct cell {
+        sstring column_name;
+        value data_value;
+    };
+    using row = std::vector<cell>;
+    struct clustered_row {
+        api::timestamp_type marker;
+        row cells;
+    };
+    struct range_tombstone {
+        key first;
+        key last;
+    };
+
+private:
+    key _partition_key;
+    row _static_row;
+    std::map<key, clustered_row> _clustered_rows;
+    std::vector<range_tombstone> _range_tombstones;
+
+private:
+    static void remove_column(row& r, const sstring& name) {
+        auto it = boost::range::find_if(r, [&] (const cell& c) {
+            return c.column_name == name;
+        });
+        if (it != r.end()) {
+            r.erase(it);
+        }
+    }
+
+public:
+    explicit mutation_description(key partition_key)
+        : _partition_key(std::move(partition_key))
+    { }
+
+    void add_static_cell(const sstring& column, value v) {
+        _static_row.emplace_back(cell { column, std::move(v) });
+    }
+
+    void add_clustered_cell(const key& ck, const sstring& column, value v) {
+        _clustered_rows[ck].cells.emplace_back(cell { column, std::move(v) });
+    }
+
+    void add_clustered_row_marker(const key& ck) {
+        _clustered_rows[ck].marker = data_timestamp;
+    }
+
+    void remove_static_column(const sstring& name) {
+        remove_column(_static_row, name);
+    }
+
+    void remove_regular_column(const sstring& name) {
+        for (auto& [ ckey, cr ] : _clustered_rows) {
+            (void)ckey;
+            remove_column(cr.cells, name);
+        }
+    }
+
+    void add_range_tombstone(const key& start, const key& end) {
+        _range_tombstones.emplace_back(range_tombstone { start, end });
+    }
+
+    mutation build(schema_ptr s) const {
+        auto m = mutation(s, partition_key::from_exploded(*s, _partition_key));
+        for (auto& [ column, value_or_collection ] : _static_row) {
+            auto cdef = s->get_column_definition(utf8_type->decompose(column));
+            assert(cdef);
+            std::visit(make_visitor(
+                [&] (const atomic_value& v) {
+                    assert(cdef->is_atomic());
+                    m.set_static_cell(*cdef, atomic_cell::make_live(*cdef->type, data_timestamp, v));
+                },
+                [&] (const collection& c) {
+                    assert(!cdef->is_atomic());
+                    auto ctype = static_pointer_cast<const collection_type_impl>(cdef->type);
+                    collection_type_impl::mutation mut;
+                    for (auto& [ key, value ] : c) {
+                        mut.cells.emplace_back(key, atomic_cell::make_live(*ctype->value_comparator(), data_timestamp,
+                                                                           value, atomic_cell::collection_member::yes));
+                    }
+                    m.set_static_cell(*cdef, ctype->serialize_mutation_form(std::move(mut)));
+                }
+            ), value_or_collection);
+        }
+        for (auto& [ ckey, cr ] : _clustered_rows) {
+            auto& [ marker, cells ] = cr;
+            auto ck = clustering_key::from_exploded(*s, ckey);
+            for (auto& [ column, value_or_collection ] : cells) {
+                auto cdef = s->get_column_definition(utf8_type->decompose(column));
+                assert(cdef);
+                std::visit(make_visitor(
+                [&] (const atomic_value& v) {
+                        assert(cdef->is_atomic());
+                        m.set_clustered_cell(ck, *cdef, atomic_cell::make_live(*cdef->type, data_timestamp, v));
+                    },
+                [&] (const collection& c) {
+                        assert(!cdef->is_atomic());
+                        auto ctype = static_pointer_cast<const collection_type_impl>(cdef->type);
+                        collection_type_impl::mutation mut;
+                        for (auto& [ key, value ] : c) {
+                            mut.cells.emplace_back(key, atomic_cell::make_live(*ctype->value_comparator(), data_timestamp,
+                                                                            value, atomic_cell::collection_member::yes));
+                        }
+                        m.set_clustered_cell(ck, *cdef, ctype->serialize_mutation_form(std::move(mut)));
+                    }
+                ), value_or_collection);
+            }
+            if (marker != api::missing_timestamp) {
+                m.partition().clustered_row(*s, ckey).apply(row_marker(marker));
+            }
+        }
+        clustering_key::less_compare cmp(*s);
+        for (auto& [ a, b ] : _range_tombstones) {
+            auto start = clustering_key::from_exploded(*s, a);
+            auto stop = clustering_key::from_exploded(*s, b);
+            if (cmp(stop, start)) {
+                std::swap(start, stop);
+            }
+            auto rt = ::range_tombstone(std::move(start), bound_kind::excl_start,
+                                        std::move(stop), bound_kind::excl_end,
+                                        tombstone(previously_removed_column_timestamp, gc_clock::time_point()));
+            m.partition().apply_delete(*s, std::move(rt));
+        }
+        return m;
+    }
+};
+
+class table_description {
+public:
+    using column = std::tuple<sstring, data_type>;
+    struct removed_column {
+        sstring name;
+        data_type type;
+        api::timestamp_type removal_timestamp;
+    };
+
+private:
+    std::vector<column> _partition_key;
+    std::vector<column> _clustering_key;
+    std::vector<column> _static_columns;
+    std::vector<column> _regular_columns;
+
+    std::vector<removed_column> _removed_columns;
+
+    std::vector<mutation_description> _mutations;
+
+    std::vector<sstring> _change_log;
+
+private:
+    static std::vector<column>::iterator find_column(std::vector<column>& columns, const sstring& name) {
+        return boost::range::find_if(columns, [&] (const column& c) {
+            return std::get<sstring>(c) == name;
+        });
+    }
+
+    static void add_column(std::vector<column>& columns, const sstring& name, data_type type) {
+        assert(find_column(columns, name) == columns.end());
+        columns.emplace_back(name, type);
+    }
+
+    void add_old_column(const sstring& name, data_type type) {
+        _removed_columns.emplace_back(removed_column { name, type, previously_removed_column_timestamp });
+    }
+
+    void remove_column(std::vector<column>& columns, const sstring& name) {
+        auto it = find_column(columns, name);
+        assert(it != columns.end());
+        _removed_columns.emplace_back(removed_column { name, std::get<data_type>(*it), column_removal_timestamp });
+        columns.erase(it);
+    }
+
+    static void alter_column_type(std::vector<column>& columns, const sstring& name, data_type new_type) {
+        auto it = find_column(columns, name);
+        assert(it != columns.end());
+        std::get<data_type>(*it) = new_type;
+    }
+
+    schema_ptr build_schema() const {
+        auto sb = schema_builder("ks", "cf");
+        for (auto&& [ name, type ] : _partition_key) {
+            sb.with_column(utf8_type->decompose(name), type, column_kind::partition_key);
+        }
+        for (auto&& [ name, type ] : _clustering_key) {
+            sb.with_column(utf8_type->decompose(name), type, column_kind::clustering_key);
+        }
+        for (auto&& [ name, type ] : _static_columns) {
+            sb.with_column(utf8_type->decompose(name), type, column_kind::static_column);
+        }
+        for (auto&& [ name, type ] : _regular_columns) {
+            sb.with_column(utf8_type->decompose(name), type);
+        }
+
+        for (auto&& [ name, type, timestamp ] : _removed_columns) {
+            sb.without_column(name, type, timestamp);
+        }
+
+        return sb.build();
+    }
+
+    std::vector<mutation> build_mutations(schema_ptr s) const {
+        auto ms = boost::copy_range<std::vector<mutation>>(
+            _mutations | boost::adaptors::transformed([&] (const mutation_description& md) {
+                return md.build(s);
+            })
+        );
+        boost::sort(ms, mutation_decorated_key_less_comparator());
+        return ms;
+    }
+
+public:
+    explicit table_description(std::vector<column> partition_key, std::vector<column> clustering_key)
+        : _partition_key(std::move(partition_key))
+        , _clustering_key(std::move(clustering_key))
+    { }
+
+    void add_static_column(const sstring& name, data_type type) {
+        _change_log.emplace_back(format("added static column \'{}\' of type \'{}\'", name, type->as_cql3_type()->to_string()));
+        add_column(_static_columns, name, type);
+    }
+
+    void add_regular_column(const sstring& name, data_type type) {
+        _change_log.emplace_back(format("added regular column \'{}\' of type \'{}\'", name, type->as_cql3_type()->to_string()));
+        add_column(_regular_columns, name, type);
+    }
+
+    void add_old_static_column(const sstring& name, data_type type) {
+        add_old_column(name, type);
+    }
+
+    void add_old_regular_column(const sstring& name, data_type type) {
+        add_old_column(name, type);
+    }
+
+    void remove_static_column(const sstring& name) {
+        _change_log.emplace_back(format("removed static column \'{}\'", name));
+        remove_column(_static_columns, name);
+        for (auto& m : _mutations) {
+            m.remove_static_column(name);
+        }
+    }
+
+    void remove_regular_column(const sstring& name) {
+        _change_log.emplace_back(format("removed regular column \'{}\'", name));
+        remove_column(_regular_columns, name);
+        for (auto& m : _mutations) {
+            m.remove_regular_column(name);
+        }
+    }
+
+    void alter_partition_column_type(const sstring& name, data_type new_type) {
+        _change_log.emplace_back(format("altered partition column \'{}\' type to \'{}\'", name, new_type->as_cql3_type()->to_string()));
+        alter_column_type(_partition_key, name, new_type);
+    }
+
+    void alter_clustering_column_type(const sstring& name, data_type new_type) {
+        _change_log.emplace_back(format("altered clustering column \'{}\' type to \'{}\'", name, new_type->as_cql3_type()->to_string()));
+        alter_column_type(_clustering_key, name, new_type);
+    }
+
+    void alter_static_column_type(const sstring& name, data_type new_type) {
+        _change_log.emplace_back(format("altered static column \'{}\' type to \'{}\'", name, new_type->as_cql3_type()->to_string()));
+        alter_column_type(_static_columns, name, new_type);
+    }
+
+    void alter_regular_column_type(const sstring& name, data_type new_type) {
+        _change_log.emplace_back(format("altered regular column \'{}\' type to \'{}\'", name, new_type->as_cql3_type()->to_string()));
+        alter_column_type(_regular_columns, name, new_type);
+    }
+
+    void rename_partition_column(const sstring& from, const sstring& to) {
+        _change_log.emplace_back(format("renamed partition column \'{}\' to \'{}\'", from, to));
+        auto it = find_column(_partition_key, from);
+        assert(it != _partition_key.end());
+        std::get<sstring>(*it) = to;
+    }
+    void rename_clustering_column(const sstring& from, const sstring& to) {
+        _change_log.emplace_back(format("renamed clustering column \'{}\' to \'{}\'", from, to));
+        auto it = find_column(_clustering_key, from);
+        assert(it != _clustering_key.end());
+        std::get<sstring>(*it) = to;
+    }
+
+    std::vector<mutation_description>& unordered_mutations() { return _mutations; }
+    const std::vector<mutation_description>& unordered_mutations() const { return _mutations; }
+
+    struct table {
+        sstring schema_changes_log;
+        schema_ptr schema;
+        std::vector<mutation> mutations;
+    };
+    table build() const {
+        auto s = build_schema();
+        return { boost::algorithm::join(_change_log, "\n"), s, build_mutations(s) };
+    }
+};
+
+}