diff --git a/database.cc b/database.cc
index d27c5bb29bef..6b233af4c0da 100644
--- a/database.cc
+++ b/database.cc
@@ -601,21 +601,6 @@ compare_atomic_cell_for_merge(atomic_cell_view left, atomic_cell_view right) {
     return 0;
 }
 
-void
-merge_column(const column_definition& def,
-             atomic_cell_or_collection& old,
-             const atomic_cell_or_collection& neww) {
-    if (def.is_atomic()) {
-        if (compare_atomic_cell_for_merge(old.as_atomic_cell(), neww.as_atomic_cell()) < 0) {
-            // FIXME: move()?
-            old = neww;
-        }
-    } else {
-        auto ct = static_pointer_cast<const collection_type_impl>(def.type);
-        old = ct->merge(old.as_collection_mutation(), neww.as_collection_mutation());
-    }
-}
-
 future<lw_shared_ptr<query::result>>
 column_family::query(const query::read_command& cmd) const {
     query::result::builder builder(cmd.slice);
diff --git a/mutation.cc b/mutation.cc
index 5273c638e56c..b1686007eb3b 100644
--- a/mutation.cc
+++ b/mutation.cc
@@ -17,7 +17,7 @@ mutation::mutation(partition_key key_, schema_ptr schema)
 { }
 
 void mutation::set_static_cell(const column_definition& def, atomic_cell_or_collection value) {
-    update_column(_p.static_row(), def, std::move(value));
+    _p.static_row().apply(def, std::move(value));
 }
 
 void mutation::set_static_cell(const bytes& name, const boost::any& value, api::timestamp_type timestamp, ttl_opt ttl) {
@@ -28,12 +28,12 @@ void mutation::set_static_cell(const bytes& name, const boost::any& value, api::
     if (!column_def->is_static()) {
         throw std::runtime_error(sprint("column '%s' is not static", name));
     }
-    update_column(_p.static_row(), *column_def, atomic_cell::make_live(timestamp, column_def->type->decompose(value), ttl));
+    _p.static_row().apply(*column_def, atomic_cell::make_live(timestamp, column_def->type->decompose(value), ttl));
 }
 
 void mutation::set_clustered_cell(const exploded_clustering_prefix& prefix, const column_definition& def, atomic_cell_or_collection value) {
     auto& row = _p.clustered_row(clustering_key::from_clustering_prefix(*_schema, prefix)).cells;
-    update_column(row, def, std::move(value));
+    row.apply(def, std::move(value));
 }
 
 void mutation::set_clustered_cell(const clustering_key& key, const bytes& name, const boost::any& value,
@@ -47,7 +47,7 @@ void mutation::set_clustered_cell(const clustering_key& key, const bytes& name,
 
 void mutation::set_clustered_cell(const clustering_key& key, const column_definition& def, atomic_cell_or_collection value) {
     auto& row = _p.clustered_row(key).cells;
-    update_column(row, def, std::move(value));
+    row.apply(def, std::move(value));
 }
 
 void mutation::set_cell(const exploded_clustering_prefix& prefix, const bytes& name, const boost::any& value,
@@ -71,32 +71,19 @@ void mutation::set_cell(const exploded_clustering_prefix& prefix, const column_d
 
 std::experimental::optional<atomic_cell_or_collection>
 mutation::get_cell(const clustering_key& rkey, const column_definition& def) const {
-    auto find_cell = [&def] (const row& r) {
-        auto i = r.find(def.id);
-        if (i == r.end()) {
-            return std::experimental::optional<atomic_cell_or_collection>{};
-        }
-        return std::experimental::optional<atomic_cell_or_collection>{i->second};
-    };
     if (def.is_static()) {
-        return find_cell(_p.static_row());
+        const atomic_cell_or_collection* cell = _p.static_row().find_cell(def.id);
+        if (!cell) {
+            return {};
+        }
+        return { *cell };
     } else {
-        auto r = _p.find_row(rkey);
+        const row* r = _p.find_row(rkey);
         if (!r) {
             return {};
         }
-        return find_cell(*r);
-    }
-}
-
-void mutation::update_column(row& row, const column_definition& def, atomic_cell_or_collection&& value) {
-    // our mutations are not yet immutable
-    auto id = def.id;
-    auto i = row.lower_bound(id);
-    if (i == row.end() || i->first != id) {
-        row.emplace_hint(i, id, std::move(value));
-    } else {
-        merge_column(def, i->second, value);
+        const atomic_cell_or_collection* cell = r->find_cell(def.id);
+        return { *cell };
     }
 }
 
diff --git a/mutation.hh b/mutation.hh
index 7b844fac2623..fa0a8d765438 100644
--- a/mutation.hh
+++ b/mutation.hh
@@ -39,6 +39,5 @@ public:
     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);
 };
diff --git a/mutation_partition.cc b/mutation_partition.cc
index 02c5dacab85a..89d34f19992e 100644
--- a/mutation_partition.cc
+++ b/mutation_partition.cc
@@ -37,15 +37,7 @@ mutation_partition::apply(const schema& schema, const mutation_partition& p) {
 
     static auto merge_cells = [] (row& old_row, const row& new_row, auto&& find_column_def) {
         for (auto&& new_column : new_row) {
-            auto col = new_column.first;
-            auto i = old_row.find(col);
-            if (i == old_row.end()) {
-                old_row.emplace_hint(i, new_column);
-            } else {
-                auto& old_column = *i;
-                auto& def = find_column_def(col);
-                merge_column(def, old_column.second, new_column.second);
-            }
+            old_row.apply(new_column.first, new_column.second, find_column_def);
         }
     };
 
@@ -228,22 +220,22 @@ template <typename ColumnDefResolver>
 static void get_row_slice(const row& cells, const std::vector<column_id>& columns, tombstone tomb,
         ColumnDefResolver&& id_to_def, query::result::row_writer& writer) {
     for (auto id : columns) {
-        auto i = cells.find(id);
-        if (i == cells.end()) {
+        const atomic_cell_or_collection* cell = cells.find_cell(id);
+        if (!cell) {
             writer.add_empty();
         } else {
             auto&& def = id_to_def(id);
             if (def.is_atomic()) {
-                auto c = i->second.as_atomic_cell();
+                auto c = cell->as_atomic_cell();
                 if (!c.is_live(tomb)) {
                     writer.add_empty();
                 } else {
-                    writer.add(i->second.as_atomic_cell());
+                    writer.add(cell->as_atomic_cell());
                 }
             } else {
-                auto&& cell = i->second.as_collection_mutation();
+                auto&& mut = cell->as_collection_mutation();
                 auto&& ctype = static_pointer_cast<const collection_type_impl>(def.type);
-                auto m_view = ctype->deserialize_mutation_form(cell);
+                auto m_view = ctype->deserialize_mutation_form(mut);
                 m_view.tomb.apply(tomb);
                 auto m_ser = ctype->serialize_mutation_form_only_live(m_view);
                 if (ctype->is_empty(m_ser)) {
@@ -414,3 +406,44 @@ bool mutation_partition::equal(const schema& s, const mutation_partition& p) con
 
     return rows_equal(s, _static_row, p._static_row);
 }
+
+void
+merge_column(const column_definition& def,
+             atomic_cell_or_collection& old,
+             const atomic_cell_or_collection& neww) {
+    if (def.is_atomic()) {
+        if (compare_atomic_cell_for_merge(old.as_atomic_cell(), neww.as_atomic_cell()) < 0) {
+            // FIXME: move()?
+            old = neww;
+        }
+    } else {
+        auto ct = static_pointer_cast<const collection_type_impl>(def.type);
+        old = ct->merge(old.as_collection_mutation(), neww.as_collection_mutation());
+    }
+}
+
+void
+row::apply(const column_definition& column, atomic_cell_or_collection value) {
+    // our mutations are not yet immutable
+    auto id = column.id;
+    auto i = _cells.lower_bound(id);
+    if (i == _cells.end() || i->first != id) {
+        _cells.emplace_hint(i, id, std::move(value));
+    } else {
+        merge_column(column, i->second, value);
+    }
+}
+
+void
+row::append_cell(column_id id, atomic_cell_or_collection value) {
+    _cells.emplace_hint(_cells.end(), id, std::move(value));
+}
+
+const atomic_cell_or_collection*
+row::find_cell(column_id id) const {
+    auto i = _cells.find(id);
+    if (i == _cells.end()) {
+        return nullptr;
+    }
+    return &i->second;
+}
diff --git a/mutation_partition.hh b/mutation_partition.hh
index 56410ad2766d..79cfa413571d 100644
--- a/mutation_partition.hh
+++ b/mutation_partition.hh
@@ -15,8 +15,47 @@
 #include "query-result-writer.hh"
 #include "mutation_partition_view.hh"
 
-// FIXME: Encapsulate
-using row = std::map<column_id, atomic_cell_or_collection>;
+// Container for cells of a row. Cells are identified by column_id.
+//
+// Can be used as a range of std::pair<column_id, atomic_cell_or_collection>.
+//
+class row {
+    using map_type = std::map<column_id, atomic_cell_or_collection>;
+    map_type _cells;
+public:
+    using value_type = map_type::value_type;
+    using iterator = map_type::iterator;
+    using const_iterator = map_type::const_iterator;
+public:
+    iterator begin() { return _cells.begin(); }
+    iterator end() { return _cells.end(); }
+    const_iterator begin() const { return _cells.begin(); }
+    const_iterator end() const { return _cells.end(); }
+    size_t size() const { return _cells.size(); }
+
+    // Returns a reference to cell's value or throws std::out_of_range
+    const atomic_cell_or_collection& cell_at(column_id id) const { return _cells.at(id); }
+
+    // Returns a pointer to cell's value or nullptr if column is not set.
+    const atomic_cell_or_collection* find_cell(column_id id) const;
+public:
+    // Merges cell's value into the row.
+    void apply(const column_definition& column, atomic_cell_or_collection cell);
+
+    // Adds cell to the row. The column must not be already set.
+    void append_cell(column_id id, atomic_cell_or_collection cell);
+
+    // Merges given cell into the row.
+    template <typename ColumnDefinitionResolver>
+    void apply(column_id id, atomic_cell_or_collection cell, ColumnDefinitionResolver&& resolver) {
+        auto i = _cells.lower_bound(id);
+        if (i == _cells.end() || i->first != id) {
+            _cells.emplace_hint(i, id, std::move(cell));
+        } else {
+            merge_column(resolver(id), i->second, std::move(cell));
+        }
+    }
+};
 
 std::ostream& operator<<(std::ostream& os, const row::value_type& rv);
 std::ostream& operator<<(std::ostream& os, const row& r);
diff --git a/mutation_partition_applier.hh b/mutation_partition_applier.hh
index 2933020048df..e12051a42714 100644
--- a/mutation_partition_applier.hh
+++ b/mutation_partition_applier.hh
@@ -13,16 +13,6 @@ class mutation_partition_applier : public mutation_partition_visitor {
     const schema& _schema;
     mutation_partition& _p;
     deletable_row* _current_row;
-private:
-    template <typename ColumnIdResolver>
-    void apply_cell(row& r, column_id id, atomic_cell_or_collection c, ColumnIdResolver&& resolver) {
-        auto i = r.lower_bound(id);
-        if (i == r.end() || i->first != id) {
-            r.emplace_hint(i, id, std::move(c));
-        } else {
-            merge_column(resolver(id), i->second, std::move(c));
-        }
-    }
 public:
     mutation_partition_applier(const schema& s, mutation_partition& target)
         : _schema(s), _p(target) { }
@@ -32,12 +22,12 @@ public:
     }
 
     virtual void accept_static_cell(column_id id, atomic_cell_view cell) override {
-        apply_cell(_p._static_row, id, atomic_cell_or_collection(cell),
+        _p._static_row.apply(id, atomic_cell_or_collection(cell),
             [this](column_id id) -> const column_definition& { return _schema.static_column_at(id); });
     }
 
     virtual void accept_static_cell(column_id id, collection_mutation::view collection) override {
-        apply_cell(_p._static_row, id, atomic_cell_or_collection(collection),
+        _p._static_row.apply(id, atomic_cell_or_collection(collection),
             [this](column_id id) -> const column_definition& { return _schema.static_column_at(id); });
     }
 
@@ -53,12 +43,12 @@ public:
     }
 
     virtual void accept_row_cell(column_id id, atomic_cell_view cell) override {
-        apply_cell(_current_row->cells, id, atomic_cell_or_collection(cell),
+        _current_row->cells.apply(id, atomic_cell_or_collection(cell),
             [this](column_id id) -> const column_definition& { return _schema.regular_column_at(id); });
     }
 
     virtual void accept_row_cell(column_id id, collection_mutation::view collection) override {
-        apply_cell(_current_row->cells, id, atomic_cell_or_collection(collection),
+        _current_row->cells.apply(id, atomic_cell_or_collection(collection),
             [this](column_id id) -> const column_definition& { return _schema.regular_column_at(id); });
     }
 };
diff --git a/partition_builder.hh b/partition_builder.hh
index acc51f74814b..5bb6261c628a 100644
--- a/partition_builder.hh
+++ b/partition_builder.hh
@@ -27,12 +27,12 @@ public:
 
     virtual void accept_static_cell(column_id id, atomic_cell_view cell) override {
         row& r = _partition.static_row();
-        r.emplace_hint(r.end(), id, atomic_cell_or_collection(cell));
+        r.append_cell(id, atomic_cell_or_collection(cell));
     }
 
     virtual void accept_static_cell(column_id id, collection_mutation::view collection) override {
         row& r = _partition.static_row();
-        r.emplace_hint(r.end(), id, atomic_cell_or_collection(collection));
+        r.append_cell(id, atomic_cell_or_collection(collection));
     }
 
     virtual void accept_row_tombstone(clustering_key_prefix_view prefix, tombstone t) override {
@@ -48,11 +48,11 @@ public:
 
     virtual void accept_row_cell(column_id id, atomic_cell_view cell) override {
         row& r = _current_row->cells;
-        r.emplace_hint(r.end(), id, atomic_cell_or_collection(cell));
+        r.append_cell(id, atomic_cell_or_collection(cell));
     }
 
     virtual void accept_row_cell(column_id id, collection_mutation::view collection) override {
         row& r = _current_row->cells;
-        r.emplace_hint(r.end(), id, atomic_cell_or_collection(collection));
+        r.append_cell(id, atomic_cell_or_collection(collection));
     }
 };
diff --git a/tests/urchin/cql_test_env.cc b/tests/urchin/cql_test_env.cc
index e72b574508fe..fd31746342b5 100644
--- a/tests/urchin/cql_test_env.cc
+++ b/tests/urchin/cql_test_env.cc
@@ -136,19 +136,19 @@ class in_memory_cql_env : public cql_test_env {
             assert(row != nullptr);
             auto col_def = schema->get_column_definition(utf8_type->decompose(column_name));
             assert(col_def != nullptr);
-            auto i = row->find(col_def->id);
-            if (i == row->end()) {
+            const atomic_cell_or_collection* cell = row->find_cell(col_def->id);
+            if (!cell) {
                 assert(((void)"column not set", 0));
             }
             bytes actual;
             if (!col_def->type->is_multi_cell()) {
-                auto cell = i->second.as_atomic_cell();
-                assert(cell.is_live());
-                actual = { cell.value().begin(), cell.value().end() };
+                auto c = cell->as_atomic_cell();
+                assert(c.is_live());
+                actual = { c.value().begin(), c.value().end() };
             } else {
-                auto cell = i->second.as_collection_mutation();
+                auto c = cell->as_collection_mutation();
                 auto type = dynamic_pointer_cast<const collection_type_impl>(col_def->type);
-                actual = type->to_value(type->deserialize_mutation_form(cell),
+                actual = type->to_value(type->deserialize_mutation_form(c),
                                         serialization_format::internal());
             }
             assert(col_def->type->equal(actual, col_def->type->decompose(expected)));
diff --git a/tests/urchin/mutation_test.cc b/tests/urchin/mutation_test.cc
index bab9128100d7..78a29edfa509 100644
--- a/tests/urchin/mutation_test.cc
+++ b/tests/urchin/mutation_test.cc
@@ -33,9 +33,9 @@ BOOST_AUTO_TEST_CASE(test_mutation_is_applied) {
     cf.apply(std::move(m));
 
     row& r = cf.find_or_create_row_slow(key, c_key);
-    auto i = r.find(r1_col.id);
-    BOOST_REQUIRE(i != r.end());
-    auto cell = i->second.as_atomic_cell();
+    auto i = r.find_cell(r1_col.id);
+    BOOST_REQUIRE(i);
+    auto cell = i->as_atomic_cell();
     BOOST_REQUIRE(cell.is_live());
     BOOST_REQUIRE(int32_type->equal(cell.value(), int32_type->decompose(3)));
 }
@@ -124,9 +124,9 @@ BOOST_AUTO_TEST_CASE(test_map_mutations) {
     cf.apply(m2o);
 
     row& r = cf.find_or_create_partition_slow(key).static_row();
-    auto i = r.find(column.id);
-    BOOST_REQUIRE(i != r.end());
-    auto cell = i->second.as_collection_mutation();
+    auto i = r.find_cell(column.id);
+    BOOST_REQUIRE(i);
+    auto cell = i->as_collection_mutation();
     auto muts = my_map_type->deserialize_mutation_form(cell);
     BOOST_REQUIRE(muts.cells.size() == 3);
     // FIXME: more strict tests
@@ -157,9 +157,9 @@ BOOST_AUTO_TEST_CASE(test_set_mutations) {
     cf.apply(m2o);
 
     row& r = cf.find_or_create_partition_slow(key).static_row();
-    auto i = r.find(column.id);
-    BOOST_REQUIRE(i != r.end());
-    auto cell = i->second.as_collection_mutation();
+    auto i = r.find_cell(column.id);
+    BOOST_REQUIRE(i);
+    auto cell = i->as_collection_mutation();
     auto muts = my_set_type->deserialize_mutation_form(cell);
     BOOST_REQUIRE(muts.cells.size() == 3);
     // FIXME: more strict tests
@@ -191,9 +191,9 @@ BOOST_AUTO_TEST_CASE(test_list_mutations) {
     cf.apply(m2o);
 
     row& r = cf.find_or_create_partition_slow(key).static_row();
-    auto i = r.find(column.id);
-    BOOST_REQUIRE(i != r.end());
-    auto cell = i->second.as_collection_mutation();
+    auto i = r.find_cell(column.id);
+    BOOST_REQUIRE(i);
+    auto cell = i->as_collection_mutation();
     auto muts = my_list_type->deserialize_mutation_form(cell);
     BOOST_REQUIRE(muts.cells.size() == 4);
     // FIXME: more strict tests
@@ -223,9 +223,9 @@ BOOST_AUTO_TEST_CASE(test_multiple_memtables_one_partition) {
         auto c_key = clustering_key::from_exploded(*s, {int32_type->decompose(c1)});
         auto r = cf.find_row(dht::global_partitioner().decorate_key(*s, key), c_key);
         BOOST_REQUIRE(r);
-        auto i = r->find(r1_col.id);
-        BOOST_REQUIRE(i != r->end());
-        auto cell = i->second.as_atomic_cell();
+        auto i = r->find_cell(r1_col.id);
+        BOOST_REQUIRE(i);
+        auto cell = i->as_atomic_cell();
         BOOST_REQUIRE(cell.is_live());
         BOOST_REQUIRE(int32_type->equal(cell.value(), int32_type->decompose(r1)));
     };
@@ -267,9 +267,9 @@ BOOST_AUTO_TEST_CASE(test_multiple_memtables_multiple_partitions) {
         auto p1 = boost::any_cast<int32_t>(int32_type->deserialize(pk._key.explode(*s)[0]));
         for (const rows_entry& re : mp.range(*s, query::range<clustering_key_prefix>())) {
             auto c1 = boost::any_cast<int32_t>(int32_type->deserialize(re.key().explode(*s)[0]));
-            auto i = re.row().cells.find(r1_col.id);
-            if (i != re.row().cells.end()) {
-                result[p1][c1] = boost::any_cast<int32_t>(int32_type->deserialize(i->second.as_atomic_cell().value()));
+            auto cell = re.row().cells.find_cell(r1_col.id);
+            if (cell) {
+                result[p1][c1] = boost::any_cast<int32_t>(int32_type->deserialize(cell->as_atomic_cell().value()));
             }
         }
         return true;
diff --git a/thrift/handler.cc b/thrift/handler.cc
index 593ab5ecb01d..974af2bcd5fe 100644
--- a/thrift/handler.cc
+++ b/thrift/handler.cc
@@ -148,7 +148,7 @@ class CassandraAsyncHandler : public CassandraCobSvIf {
                     // FIXME: force limit count?
                     while (beg != end && count--) {
                         const column_definition& def = range.reversed ? *--end : *beg++;
-                        atomic_cell_view cell = (*rw).at(def.id).as_atomic_cell();
+                        atomic_cell_view cell = (*rw).cell_at(def.id).as_atomic_cell();
                         if (def.is_atomic()) {
                             if (cell.is_live()) { // FIXME: we should actually use tombstone information from all levels
                                 Column col;