diff --git a/db/batchlog_manager.cc b/db/batchlog_manager.cc
index e3fa0d6db3..afbebe5150 100644
--- a/db/batchlog_manager.cc
+++ b/db/batchlog_manager.cc
@@ -132,7 +132,7 @@ mutation db::batchlog_manager::get_batch_log_mutation_for(const std::vector<muta
     mutation m(key, schema);
     m.set_cell({}, to_bytes("version"), version, timestamp);
     m.set_cell({}, to_bytes("written_at"), now, timestamp);
-    m.set_cell({}, to_bytes("data"), std::move(data), timestamp);
+    m.set_cell({}, to_bytes("data"), data_value(std::move(data)), timestamp);
 
     return m;
 }
diff --git a/db/system_keyspace.cc b/db/system_keyspace.cc
index 5aa11cca15..632184e29d 100644
--- a/db/system_keyspace.cc
+++ b/db/system_keyspace.cc
@@ -538,7 +538,7 @@ future<> save_truncation_records(const column_family& cf, db_clock::time_point t
     out.write<db_clock::rep>(truncated_at.time_since_epoch().count());
 
     map_type_impl::native_type tmp;
-    tmp.emplace_back(cf.schema()->id(), buf);
+    tmp.emplace_back(cf.schema()->id(), data_value(buf));
     auto map_type = map_type_impl::get_instance(uuid_type, bytes_type, true);
 
     sstring req = sprint("UPDATE system.%s SET truncated_at = truncated_at + ? WHERE key = '%s'", LOCAL, LOCAL);
diff --git a/tests/frozen_mutation_test.cc b/tests/frozen_mutation_test.cc
index 334f5f9b88..0681e739d5 100644
--- a/tests/frozen_mutation_test.cc
+++ b/tests/frozen_mutation_test.cc
@@ -53,8 +53,8 @@ BOOST_AUTO_TEST_CASE(test_writing_and_reading) {
         .build();
 
     partition_key key = partition_key::from_single_value(*s, bytes("key"));
-    clustering_key ck1 = clustering_key::from_deeply_exploded(*s, {bytes("ck1_0"), bytes("ck1_1")});
-    clustering_key ck2 = clustering_key::from_deeply_exploded(*s, {bytes("ck2_0"), bytes("ck2_1")});
+    clustering_key ck1 = clustering_key::from_deeply_exploded(*s, {data_value(bytes("ck1_0")), data_value(bytes("ck1_1"))});
+    clustering_key ck2 = clustering_key::from_deeply_exploded(*s, {data_value(bytes("ck2_0")), data_value(bytes("ck2_1"))});
     auto ttl = gc_clock::duration(1);
 
     auto test_freezing = [] (const mutation& m) {
@@ -70,15 +70,15 @@ BOOST_AUTO_TEST_CASE(test_writing_and_reading) {
 
     test_freezing(m);
 
-    m.partition().apply_row_tombstone(*s, clustering_key_prefix::from_deeply_exploded(*s, {bytes("ck2_0")}), new_tombstone());
+    m.partition().apply_row_tombstone(*s, clustering_key_prefix::from_deeply_exploded(*s, {data_value(bytes("ck2_0"))}), new_tombstone());
 
     test_freezing(m);
 
-    m.set_clustered_cell(ck1, "regular_col_1", bytes("regular_col_value"), new_timestamp(), ttl);
+    m.set_clustered_cell(ck1, "regular_col_1", data_value(bytes("regular_col_value")), new_timestamp(), ttl);
 
     test_freezing(m);
 
-    m.set_clustered_cell(ck1, "regular_col_2", bytes("regular_col_value"), new_timestamp(), ttl);
+    m.set_clustered_cell(ck1, "regular_col_2", data_value(bytes("regular_col_value")), new_timestamp(), ttl);
 
     test_freezing(m);
 
@@ -86,15 +86,15 @@ BOOST_AUTO_TEST_CASE(test_writing_and_reading) {
 
     test_freezing(m);
 
-    m.set_clustered_cell(ck2, "regular_col_1", bytes("ck2_regular_col_1_value"), new_timestamp());
+    m.set_clustered_cell(ck2, "regular_col_1", data_value(bytes("ck2_regular_col_1_value")), new_timestamp());
 
     test_freezing(m);
 
-    m.set_static_cell("static_col_1", bytes("static_col_value"), new_timestamp(), ttl);
+    m.set_static_cell("static_col_1", data_value(bytes("static_col_value")), new_timestamp(), ttl);
 
     test_freezing(m);
 
-    m.set_static_cell("static_col_2", bytes("static_col_value"), new_timestamp());
+    m.set_static_cell("static_col_2", data_value(bytes("static_col_value")), new_timestamp());
 
     test_freezing(m);
 }
@@ -109,19 +109,19 @@ BOOST_AUTO_TEST_CASE(test_application_of_partition_view_has_the_same_effect_as_a
         .build();
 
     partition_key key = partition_key::from_single_value(*s, bytes("key"));
-    clustering_key ck = clustering_key::from_deeply_exploded(*s, {bytes("ck")});
+    clustering_key ck = clustering_key::from_deeply_exploded(*s, {data_value(bytes("ck"))});
 
     mutation m1(key, s);
     m1.partition().apply(new_tombstone());
-    m1.set_clustered_cell(ck, "reg_1", bytes("val1"), new_timestamp());
-    m1.set_clustered_cell(ck, "reg_2", bytes("val2"), new_timestamp());
+    m1.set_clustered_cell(ck, "reg_1", data_value(bytes("val1")), new_timestamp());
+    m1.set_clustered_cell(ck, "reg_2", data_value(bytes("val2")), new_timestamp());
     m1.partition().apply_insert(*s, ck, new_timestamp());
-    m1.set_static_cell("static_1", bytes("val3"), new_timestamp());
+    m1.set_static_cell("static_1", data_value(bytes("val3")), new_timestamp());
 
     mutation m2(key, s);
-    m2.set_clustered_cell(ck, "reg_1", bytes("val4"), new_timestamp());
+    m2.set_clustered_cell(ck, "reg_1", data_value(bytes("val4")), new_timestamp());
     m2.partition().apply_insert(*s, ck, new_timestamp());
-    m2.set_static_cell("static_1", bytes("val5"), new_timestamp());
+    m2.set_static_cell("static_1", data_value(bytes("val5")), new_timestamp());
 
     mutation m_frozen(key, s);
     m_frozen.partition().apply(*s, freeze(m1).partition());
diff --git a/tests/keys_test.cc b/tests/keys_test.cc
index cb4dc41e10..8a6bbf2222 100644
--- a/tests/keys_test.cc
+++ b/tests/keys_test.cc
@@ -117,7 +117,7 @@ BOOST_AUTO_TEST_CASE(test_legacy_ordering_for_composite_keys) {
 BOOST_AUTO_TEST_CASE(test_conversions_between_view_and_wrapper) {
     schema s({}, "", "", {{"c1", bytes_type}}, {}, {}, {}, utf8_type);
 
-    auto key = partition_key::from_deeply_exploded(s, {bytes("value")});
+    auto key = partition_key::from_deeply_exploded(s, {data_value(bytes("value"))});
     partition_key_view key_view = key;
 
     BOOST_REQUIRE(key_view.equal(s, key));
diff --git a/tests/memory_footprint.cc b/tests/memory_footprint.cc
index 9e1c12863a..750cea45e4 100644
--- a/tests/memory_footprint.cc
+++ b/tests/memory_footprint.cc
@@ -140,14 +140,14 @@ static mutation make_mutation(mutation_settings settings) {
 
     auto s = builder.build();
 
-    mutation m(partition_key::from_single_value(*s, bytes_type->decompose(random_bytes(settings.partition_key_size))), s);
+    mutation m(partition_key::from_single_value(*s, bytes_type->decompose(data_value(random_bytes(settings.partition_key_size)))), s);
 
     for (size_t i = 0; i < settings.row_count; ++i) {
-        auto ck = clustering_key::from_single_value(*s, bytes_type->decompose(random_bytes(settings.clustering_key_size)));
+        auto ck = clustering_key::from_single_value(*s, bytes_type->decompose(data_value(random_bytes(settings.clustering_key_size))));
         for (auto&& col : s->regular_columns()) {
             m.set_clustered_cell(ck, col,
                 atomic_cell::make_live(1,
-                    bytes_type->decompose(random_bytes(settings.data_size))));
+                    bytes_type->decompose(data_value(random_bytes(settings.data_size)))));
         }
     }
     return m;
diff --git a/tests/mutation_query_test.cc b/tests/mutation_query_test.cc
index b7e1f1970e..c164bfdc51 100644
--- a/tests/mutation_query_test.cc
+++ b/tests/mutation_query_test.cc
@@ -79,10 +79,10 @@ SEASTAR_TEST_CASE(test_reading_from_single_partition) {
         auto now = gc_clock::now();
 
         mutation m1(partition_key::from_single_value(*s, "key1"), s);
-        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("A")), "v1", bytes("A:v"), 1);
-        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("B")), "v1", bytes("B:v"), 1);
-        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("C")), "v1", bytes("C:v"), 1);
-        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("D")), "v1", bytes("D:v"), 1);
+        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("A")), "v1", data_value(bytes("A:v")), 1);
+        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("B")), "v1", data_value(bytes("B:v")), 1);
+        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("C")), "v1", data_value(bytes("C:v")), 1);
+        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("D")), "v1", data_value(bytes("D:v")), 1);
 
         auto src = make_source({m1});
 
@@ -97,13 +97,13 @@ SEASTAR_TEST_CASE(test_reading_from_single_partition) {
             assert_that(to_result_set(result, s, slice))
                 .has_size(2)
                 .has(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("A"))
-                    .with_column("v1", bytes("A:v")))
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("A")))
+                    .with_column("v1", data_value(bytes("A:v"))))
                 .has(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("B"))
-                    .with_column("v1", bytes("B:v")));
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("B")))
+                    .with_column("v1", data_value(bytes("B:v"))));
         }
 
         // Test slicing in the middle
@@ -117,9 +117,9 @@ SEASTAR_TEST_CASE(test_reading_from_single_partition) {
 
             assert_that(to_result_set(result, s, slice))
                 .has_only(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("B"))
-                    .with_column("v1", bytes("B:v")));
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("B")))
+                    .with_column("v1", data_value(bytes("B:v"))));
         }
     });
 }
@@ -150,9 +150,9 @@ SEASTAR_TEST_CASE(test_cells_are_expired_according_to_query_timestamp) {
 
             assert_that(to_result_set(result, s, slice))
                 .has_only(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("A"))
-                    .with_column("v1", bytes("A:v1")));
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("A")))
+                    .with_column("v1", data_value(bytes("A:v1"))));
         }
 
         // Expired
@@ -164,9 +164,9 @@ SEASTAR_TEST_CASE(test_cells_are_expired_according_to_query_timestamp) {
 
             assert_that(to_result_set(result, s, slice))
                 .has_only(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("B"))
-                    .with_column("v1", bytes("B:v1")));
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("B")))
+                    .with_column("v1", data_value(bytes("B:v1"))));
         }
     });
 }
@@ -178,11 +178,11 @@ SEASTAR_TEST_CASE(test_reverse_ordering_is_respected) {
 
         mutation m1(partition_key::from_single_value(*s, "key1"), s);
 
-        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("A")), "v1", bytes("A_v1"), 1);
-        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("B")), "v1", bytes("B_v1"), 1);
-        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("C")), "v1", bytes("C_v1"), 1);
-        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("D")), "v1", bytes("D_v1"), 1);
-        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("E")), "v1", bytes("E_v1"), 1);
+        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("A")), "v1", data_value(bytes("A_v1")), 1);
+        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("B")), "v1", data_value(bytes("B_v1")), 1);
+        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("C")), "v1", data_value(bytes("C_v1")), 1);
+        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("D")), "v1", data_value(bytes("D_v1")), 1);
+        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("E")), "v1", data_value(bytes("E_v1")), 1);
 
         auto src = make_source({m1});
 
@@ -197,17 +197,17 @@ SEASTAR_TEST_CASE(test_reverse_ordering_is_respected) {
             assert_that(to_result_set(result, s, slice))
                 .has_size(3)
                 .has(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("E"))
-                    .with_column("v1", bytes("E_v1")))
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("E")))
+                    .with_column("v1", data_value(bytes("E_v1"))))
                 .has(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("D"))
-                    .with_column("v1", bytes("D_v1")))
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("D")))
+                    .with_column("v1", data_value(bytes("D_v1"))))
                 .has(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("C"))
-                    .with_column("v1", bytes("C_v1")));
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("C")))
+                    .with_column("v1", data_value(bytes("C_v1"))));
         }
 
         {
@@ -227,17 +227,17 @@ SEASTAR_TEST_CASE(test_reverse_ordering_is_respected) {
             assert_that(to_result_set(result, s, slice))
                 .has_size(3)
                 .has(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("E"))
-                    .with_column("v1", bytes("E_v1")))
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("E")))
+                    .with_column("v1", data_value(bytes("E_v1"))))
                 .has(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("D"))
-                    .with_column("v1", bytes("D_v1")))
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("D")))
+                    .with_column("v1", data_value(bytes("D_v1"))))
                 .has(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("C"))
-                    .with_column("v1", bytes("C_v1")));
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("C")))
+                    .with_column("v1", data_value(bytes("C_v1"))));
         }
 
         {
@@ -255,17 +255,17 @@ SEASTAR_TEST_CASE(test_reverse_ordering_is_respected) {
                 assert_that(to_result_set(result, s, slice))
                     .has_size(3)
                     .has(a_row()
-                        .with_column("pk", bytes("key1"))
-                        .with_column("ck", bytes("E"))
-                        .with_column("v1", bytes("E_v1")))
+                        .with_column("pk", data_value(bytes("key1")))
+                        .with_column("ck", data_value(bytes("E")))
+                        .with_column("v1", data_value(bytes("E_v1"))))
                     .has(a_row()
-                        .with_column("pk", bytes("key1"))
-                        .with_column("ck", bytes("D"))
-                        .with_column("v1", bytes("D_v1")))
+                        .with_column("pk", data_value(bytes("key1")))
+                        .with_column("ck", data_value(bytes("D")))
+                        .with_column("v1", data_value(bytes("D_v1"))))
                     .has(a_row()
-                        .with_column("pk", bytes("key1"))
-                        .with_column("ck", bytes("C"))
-                        .with_column("v1", bytes("C_v1")));
+                        .with_column("pk", data_value(bytes("key1")))
+                        .with_column("ck", data_value(bytes("C")))
+                        .with_column("v1", data_value(bytes("C_v1"))));
             }
 
             {
@@ -275,9 +275,9 @@ SEASTAR_TEST_CASE(test_reverse_ordering_is_respected) {
                 assert_that(to_result_set(result, s, slice))
                     .has_size(1)
                     .has(a_row()
-                        .with_column("pk", bytes("key1"))
-                        .with_column("ck", bytes("E"))
-                        .with_column("v1", bytes("E_v1")));
+                        .with_column("pk", data_value(bytes("key1")))
+                        .with_column("ck", data_value(bytes("E")))
+                        .with_column("v1", data_value(bytes("E_v1"))));
             }
 
             {
@@ -287,13 +287,13 @@ SEASTAR_TEST_CASE(test_reverse_ordering_is_respected) {
                 assert_that(to_result_set(result, s, slice))
                     .has_size(2)
                     .has(a_row()
-                        .with_column("pk", bytes("key1"))
-                        .with_column("ck", bytes("E"))
-                        .with_column("v1", bytes("E_v1")))
+                        .with_column("pk", data_value(bytes("key1")))
+                        .with_column("ck", data_value(bytes("E")))
+                        .with_column("v1", data_value(bytes("E_v1"))))
                     .has(a_row()
-                        .with_column("pk", bytes("key1"))
-                        .with_column("ck", bytes("D"))
-                        .with_column("v1", bytes("D_v1")));
+                        .with_column("pk", data_value(bytes("key1")))
+                        .with_column("ck", data_value(bytes("D")))
+                        .with_column("v1", data_value(bytes("D_v1"))));
             }
         }
 
@@ -314,13 +314,13 @@ SEASTAR_TEST_CASE(test_reverse_ordering_is_respected) {
             assert_that(to_result_set(result, s, slice))
                 .has_size(2)
                 .has(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("E"))
-                    .with_column("v1", bytes("E_v1")))
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("E")))
+                    .with_column("v1", data_value(bytes("E_v1"))))
                 .has(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("D"))
-                    .with_column("v1", bytes("D_v1")));
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("D")))
+                    .with_column("v1", data_value(bytes("D_v1"))));
         }
 
         {
@@ -338,13 +338,13 @@ SEASTAR_TEST_CASE(test_reverse_ordering_is_respected) {
             assert_that(to_result_set(result, s, slice))
                 .has_size(2)
                 .has(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("E"))
-                    .with_column("v1", bytes("E_v1")))
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("E")))
+                    .with_column("v1", data_value(bytes("E_v1"))))
                 .has(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("C"))
-                    .with_column("v1", bytes("C_v1")));
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("C")))
+                    .with_column("v1", data_value(bytes("C_v1"))));
         }
 
         {
@@ -359,9 +359,9 @@ SEASTAR_TEST_CASE(test_reverse_ordering_is_respected) {
 
             assert_that(to_result_set(result, s, slice))
                 .has_only(a_row()
-                    .with_column("pk", bytes("key1"))
-                    .with_column("ck", bytes("B"))
-                    .with_column("v1", bytes("B_v1")));
+                    .with_column("pk", data_value(bytes("key1")))
+                    .with_column("ck", data_value(bytes("B")))
+                    .with_column("v1", data_value(bytes("B_v1"))));
         }
     });
 }
@@ -374,7 +374,7 @@ SEASTAR_TEST_CASE(test_query_when_partition_tombstone_covers_live_cells) {
         mutation m1(partition_key::from_single_value(*s, "key1"), s);
 
         m1.partition().apply(tombstone(api::timestamp_type(1), now));
-        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("A")), "v1", bytes("A:v"), 1);
+        m1.set_clustered_cell(clustering_key::from_single_value(*s, bytes("A")), "v1", data_value(bytes("A:v")), 1);
 
         auto src = make_source({m1});
         auto slice = make_full_slice(*s);
diff --git a/tests/mutation_reader_test.cc b/tests/mutation_reader_test.cc
index deeb235d82..b0f0fb91a9 100644
--- a/tests/mutation_reader_test.cc
+++ b/tests/mutation_reader_test.cc
@@ -44,10 +44,10 @@ SEASTAR_TEST_CASE(test_combining_two_readers_with_the_same_row) {
         auto s = make_schema();
 
         mutation m1(partition_key::from_single_value(*s, "key1"), s);
-        m1.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v1"), 1);
+        m1.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v1")), 1);
 
         mutation m2(partition_key::from_single_value(*s, "key1"), s);
-        m2.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v2"), 2);
+        m2.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v2")), 2);
 
         assert_that(make_combined_reader(make_reader_returning(m1), make_reader_returning(m2)))
             .produces(m2)
@@ -60,10 +60,10 @@ SEASTAR_TEST_CASE(test_combining_two_non_overlapping_readers) {
         auto s = make_schema();
 
         mutation m1(partition_key::from_single_value(*s, "keyB"), s);
-        m1.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v1"), 1);
+        m1.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v1")), 1);
 
         mutation m2(partition_key::from_single_value(*s, "keyA"), s);
-        m2.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v2"), 2);
+        m2.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v2")), 2);
 
         auto cr = make_combined_reader(make_reader_returning(m1), make_reader_returning(m2));
         assert_that(std::move(cr))
@@ -78,13 +78,13 @@ SEASTAR_TEST_CASE(test_combining_two_partially_overlapping_readers) {
         auto s = make_schema();
 
         mutation m1(partition_key::from_single_value(*s, "keyA"), s);
-        m1.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v1"), 1);
+        m1.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v1")), 1);
 
         mutation m2(partition_key::from_single_value(*s, "keyB"), s);
-        m2.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v2"), 1);
+        m2.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v2")), 1);
 
         mutation m3(partition_key::from_single_value(*s, "keyC"), s);
-        m3.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v3"), 1);
+        m3.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v3")), 1);
 
         assert_that(make_combined_reader(make_reader_returning_many({m1, m2}), make_reader_returning_many({m2, m3})))
             .produces(m1)
@@ -99,13 +99,13 @@ SEASTAR_TEST_CASE(test_combining_one_reader_with_many_partitions) {
         auto s = make_schema();
 
         mutation m1(partition_key::from_single_value(*s, "keyA"), s);
-        m1.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v1"), 1);
+        m1.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v1")), 1);
 
         mutation m2(partition_key::from_single_value(*s, "keyB"), s);
-        m2.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v2"), 1);
+        m2.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v2")), 1);
 
         mutation m3(partition_key::from_single_value(*s, "keyC"), s);
-        m3.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v3"), 1);
+        m3.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v3")), 1);
 
         std::vector<mutation_reader> v;
         v.push_back(make_reader_returning_many({m1, m2, m3}));
@@ -119,7 +119,7 @@ SEASTAR_TEST_CASE(test_combining_one_reader_with_many_partitions) {
 
 static mutation make_mutation_with_key(schema_ptr s, sstring key) {
     mutation m(partition_key::from_single_value(*s, to_bytes(key)), s);
-    m.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v1"), 1);
+    m.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v1")), 1);
     return m;
 }
 
@@ -194,7 +194,7 @@ SEASTAR_TEST_CASE(test_combining_two_readers_with_one_reader_empty) {
     return seastar::async([] {
         auto s = make_schema();
         mutation m1(partition_key::from_single_value(*s, "key1"), s);
-        m1.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v1"), 1);
+        m1.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v1")), 1);
 
         assert_that(make_combined_reader(make_reader_returning(m1), make_empty_reader()))
             .produces(m1)
@@ -223,10 +223,10 @@ SEASTAR_TEST_CASE(test_joining_reader) {
         auto s = make_schema();
 
         mutation m1(partition_key::from_single_value(*s, "keyB"), s);
-        m1.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v1"), 1);
+        m1.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v1")), 1);
 
         mutation m2(partition_key::from_single_value(*s, "keyA"), s);
-        m2.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v2"), 2);
+        m2.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v2")), 2);
 
         std::vector<mutation_reader> v;
         v.push_back(make_reader_returning(m1));
diff --git a/tests/mutation_source_test.cc b/tests/mutation_source_test.cc
index e65cd1e9cf..ac3c6f3375 100644
--- a/tests/mutation_source_test.cc
+++ b/tests/mutation_source_test.cc
@@ -45,7 +45,7 @@ static void test_range_queries(populate_fn populate) {
 
     auto make_partition_mutation = [s] (bytes key) -> mutation {
         mutation m(partition_key::from_single_value(*s, key), s);
-        m.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v1"), 1);
+        m.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v1")), 1);
         return m;
     };
 
diff --git a/tests/mutation_test.cc b/tests/mutation_test.cc
index e5aacdae66..2b77c598e0 100644
--- a/tests/mutation_test.cc
+++ b/tests/mutation_test.cc
@@ -332,7 +332,7 @@ SEASTAR_TEST_CASE(test_flush_in_the_middle_of_a_scan) {
             };
             auto make_mutation = [&] {
                 mutation m(new_key(), s);
-                m.set_clustered_cell(clustering_key::make_empty(*s), "v", to_bytes("value"), 1);
+                m.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(to_bytes("value")), 1);
                 return m;
             };
 
@@ -544,12 +544,12 @@ SEASTAR_TEST_CASE(test_querying_of_mutation) {
         };
 
         mutation m(partition_key::from_single_value(*s, "key1"), s);
-        m.set_clustered_cell(clustering_key::make_empty(*s), "v", bytes("v1"), 1);
+        m.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(bytes("v1")), 1);
 
         assert_that(resultify(m))
             .has_only(a_row()
-                .with_column("pk", bytes("key1"))
-                .with_column("v", bytes("v1")));
+                .with_column("pk", data_value(bytes("key1")))
+                .with_column("v", data_value(bytes("v1"))));
 
         m.partition().apply(tombstone(2, gc_clock::now()));
 
@@ -570,7 +570,7 @@ SEASTAR_TEST_CASE(test_partition_with_no_live_data_is_absent_in_data_query_resul
         m.partition().apply(tombstone(1, gc_clock::now()));
         m.partition().static_row().apply(*s->get_column_definition("sc1"),
             atomic_cell::make_dead(2, gc_clock::now()));
-        m.set_clustered_cell(clustering_key::from_single_value(*s, bytes_type->decompose(bytes("A"))),
+        m.set_clustered_cell(clustering_key::from_single_value(*s, bytes_type->decompose(data_value(bytes("A")))),
             *s->get_column_definition("v"), atomic_cell::make_dead(2, gc_clock::now()));
 
         auto slice = make_full_slice(*s);
@@ -591,7 +591,7 @@ SEASTAR_TEST_CASE(test_partition_with_live_data_in_static_row_is_present_in_the_
 
         mutation m(partition_key::from_single_value(*s, "key1"), s);
         m.partition().static_row().apply(*s->get_column_definition("sc1"),
-            atomic_cell::make_live(2, bytes_type->decompose(bytes("sc1:value"))));
+            atomic_cell::make_live(2, bytes_type->decompose(data_value(bytes("sc1:value")))));
 
         auto slice = partition_slice_builder(*s)
             .with_no_static_columns()
@@ -600,7 +600,7 @@ SEASTAR_TEST_CASE(test_partition_with_live_data_in_static_row_is_present_in_the_
 
         assert_that(query::result_set::from_raw_result(s, slice, m.query(slice)))
             .has_only(a_row()
-                .with_column("pk", bytes("key1"))
+                .with_column("pk", data_value(bytes("key1")))
                 .with_column("v", data_value::make_null(bytes_type)));
     });
 }
@@ -617,15 +617,15 @@ SEASTAR_TEST_CASE(test_query_result_with_one_regular_column_missing) {
         mutation m(partition_key::from_single_value(*s, "key1"), s);
         m.set_clustered_cell(clustering_key::from_single_value(*s, bytes("ck:A")),
             *s->get_column_definition("v1"),
-            atomic_cell::make_live(2, bytes_type->decompose(bytes("v1:value"))));
+            atomic_cell::make_live(2, bytes_type->decompose(data_value(bytes("v1:value")))));
 
         auto slice = partition_slice_builder(*s).build();
 
         assert_that(query::result_set::from_raw_result(s, slice, m.query(slice)))
             .has_only(a_row()
-                .with_column("pk", bytes("key1"))
-                .with_column("ck", bytes("ck:A"))
-                .with_column("v1", bytes("v1:value"))
+                .with_column("pk", data_value(bytes("key1")))
+                .with_column("ck", data_value(bytes("ck:A")))
+                .with_column("v1", data_value(bytes("v1:value")))
                 .with_column("v2", data_value::make_null(bytes_type)));
     });
 }
@@ -645,15 +645,15 @@ SEASTAR_TEST_CASE(test_row_counting) {
 
         BOOST_REQUIRE_EQUAL(0, m.live_row_count());
 
-        auto ckey1 = clustering_key::from_single_value(*s, bytes_type->decompose(bytes("A")));
-        auto ckey2 = clustering_key::from_single_value(*s, bytes_type->decompose(bytes("B")));
+        auto ckey1 = clustering_key::from_single_value(*s, bytes_type->decompose(data_value(bytes("A"))));
+        auto ckey2 = clustering_key::from_single_value(*s, bytes_type->decompose(data_value(bytes("B"))));
 
-        m.set_clustered_cell(ckey1, col_v, atomic_cell::make_live(2, bytes_type->decompose(bytes("v:value"))));
+        m.set_clustered_cell(ckey1, col_v, atomic_cell::make_live(2, bytes_type->decompose(data_value(bytes("v:value")))));
 
         BOOST_REQUIRE_EQUAL(1, m.live_row_count());
 
         m.partition().static_row().apply(*s->get_column_definition("sc1"),
-            atomic_cell::make_live(2, bytes_type->decompose(bytes("sc1:value"))));
+            atomic_cell::make_live(2, bytes_type->decompose(data_value(bytes("sc1:value")))));
 
         BOOST_REQUIRE_EQUAL(1, m.live_row_count());
 
@@ -674,8 +674,8 @@ SEASTAR_TEST_CASE(test_row_counting) {
 
         BOOST_REQUIRE_EQUAL(0, m.live_row_count());
 
-        m.set_clustered_cell(ckey1, col_v, atomic_cell::make_live(4, bytes_type->decompose(bytes("v:value"))));
-        m.set_clustered_cell(ckey2, col_v, atomic_cell::make_live(4, bytes_type->decompose(bytes("v:value"))));
+        m.set_clustered_cell(ckey1, col_v, atomic_cell::make_live(4, bytes_type->decompose(data_value(bytes("v:value")))));
+        m.set_clustered_cell(ckey2, col_v, atomic_cell::make_live(4, bytes_type->decompose(data_value(bytes("v:value")))));
 
         BOOST_REQUIRE_EQUAL(2, m.live_row_count());
     });
@@ -693,8 +693,8 @@ SEASTAR_TEST_CASE(test_mutation_diff) {
             .with_column("v3", my_set_type, column_kind::regular_column)
             .build();
 
-        auto ckey1 = clustering_key::from_single_value(*s, bytes_type->decompose(bytes("A")));
-        auto ckey2 = clustering_key::from_single_value(*s, bytes_type->decompose(bytes("B")));
+        auto ckey1 = clustering_key::from_single_value(*s, bytes_type->decompose(data_value(bytes("A"))));
+        auto ckey2 = clustering_key::from_single_value(*s, bytes_type->decompose(data_value(bytes("B"))));
 
         mutation m1(partition_key::from_single_value(*s, "key1"), s);
         m1.set_static_cell(*s->get_column_definition("sc1"),
@@ -702,39 +702,39 @@ SEASTAR_TEST_CASE(test_mutation_diff) {
 
         m1.partition().apply(tombstone { 1, gc_clock::now() });
         m1.set_clustered_cell(ckey1, *s->get_column_definition("v1"),
-            atomic_cell::make_live(2, bytes_type->decompose(bytes("v1:value1"))));
+            atomic_cell::make_live(2, bytes_type->decompose(data_value(bytes("v1:value1")))));
         m1.set_clustered_cell(ckey1, *s->get_column_definition("v2"),
-            atomic_cell::make_live(2, bytes_type->decompose(bytes("v2:value2"))));
+            atomic_cell::make_live(2, bytes_type->decompose(data_value(bytes("v2:value2")))));
 
         m1.partition().clustered_row(ckey2).apply(row_marker(3));
         m1.set_clustered_cell(ckey2, *s->get_column_definition("v2"),
-            atomic_cell::make_live(2, bytes_type->decompose(bytes("v2:value4"))));
+            atomic_cell::make_live(2, bytes_type->decompose(data_value(bytes("v2:value4")))));
         map_type_impl::mutation mset1 {{}, {{int32_type->decompose(1), make_atomic_cell({})}, {int32_type->decompose(2), make_atomic_cell({})}}};
         m1.set_clustered_cell(ckey2, *s->get_column_definition("v3"),
             my_set_type->serialize_mutation_form(mset1));
 
         mutation m2(partition_key::from_single_value(*s, "key1"), s);
         m2.set_clustered_cell(ckey1, *s->get_column_definition("v1"),
-            atomic_cell::make_live(1, bytes_type->decompose(bytes("v1:value1a"))));
+            atomic_cell::make_live(1, bytes_type->decompose(data_value(bytes("v1:value1a")))));
         m2.set_clustered_cell(ckey1, *s->get_column_definition("v2"),
-            atomic_cell::make_live(2, bytes_type->decompose(bytes("v2:value2"))));
+            atomic_cell::make_live(2, bytes_type->decompose(data_value(bytes("v2:value2")))));
 
         m2.set_clustered_cell(ckey2, *s->get_column_definition("v1"),
-            atomic_cell::make_live(2, bytes_type->decompose(bytes("v1:value3"))));
+            atomic_cell::make_live(2, bytes_type->decompose(data_value(bytes("v1:value3")))));
         m2.set_clustered_cell(ckey2, *s->get_column_definition("v2"),
-            atomic_cell::make_live(3, bytes_type->decompose(bytes("v2:value4a"))));
+            atomic_cell::make_live(3, bytes_type->decompose(data_value(bytes("v2:value4a")))));
         map_type_impl::mutation mset2 {{}, {{int32_type->decompose(1), make_atomic_cell({})}, {int32_type->decompose(3), make_atomic_cell({})}}};
         m2.set_clustered_cell(ckey2, *s->get_column_definition("v3"),
             my_set_type->serialize_mutation_form(mset2));
 
         mutation m3(partition_key::from_single_value(*s, "key1"), s);
         m3.set_clustered_cell(ckey1, *s->get_column_definition("v1"),
-            atomic_cell::make_live(2, bytes_type->decompose(bytes("v1:value1"))));
+            atomic_cell::make_live(2, bytes_type->decompose(data_value(bytes("v1:value1")))));
 
         m3.set_clustered_cell(ckey2, *s->get_column_definition("v1"),
-            atomic_cell::make_live(2, bytes_type->decompose(bytes("v1:value3"))));
+            atomic_cell::make_live(2, bytes_type->decompose(data_value(bytes("v1:value3")))));
         m3.set_clustered_cell(ckey2, *s->get_column_definition("v2"),
-            atomic_cell::make_live(3, bytes_type->decompose(bytes("v2:value4a"))));
+            atomic_cell::make_live(3, bytes_type->decompose(data_value(bytes("v2:value4a")))));
         map_type_impl::mutation mset3 {{}, {{int32_type->decompose(1), make_atomic_cell({})}}};
         m3.set_clustered_cell(ckey2, *s->get_column_definition("v3"),
             my_set_type->serialize_mutation_form(mset3));
diff --git a/tests/perf_row_cache_update.cc b/tests/perf_row_cache_update.cc
index b00969bef8..d752a61d38 100644
--- a/tests/perf_row_cache_update.cc
+++ b/tests/perf_row_cache_update.cc
@@ -80,7 +80,7 @@ int main(int argc, char** argv) {
             for (unsigned i = 0; i < partitions; ++i) {
                 mutation m(new_key(s), s);
                 for (size_t j = 0; j < row_count; j++) {
-                    m.set_clustered_cell(new_ckey(s), "v", bytes(bytes::initialized_later(), cell_size), 2);
+                    m.set_clustered_cell(new_ckey(s), "v", data_value(bytes(bytes::initialized_later(), cell_size)), 2);
                 }
                 mutations.emplace_back(std::move(m));
             }
diff --git a/tests/range_test.cc b/tests/range_test.cc
index 5b5de55ebc..485929d602 100644
--- a/tests/range_test.cc
+++ b/tests/range_test.cc
@@ -49,10 +49,10 @@ BOOST_AUTO_TEST_CASE(test_range_with_positions_within_the_same_token) {
     dht::token tok = dht::global_partitioner().get_random_token();
 
     auto key1 = dht::decorated_key{tok,
-                                   partition_key::from_single_value(*s, bytes_type->decompose(bytes("key1")))};
+                                   partition_key::from_single_value(*s, bytes_type->decompose(data_value(bytes("key1"))))};
 
     auto key2 = dht::decorated_key{tok,
-                                   partition_key::from_single_value(*s, bytes_type->decompose(bytes("key2")))};
+                                   partition_key::from_single_value(*s, bytes_type->decompose(data_value(bytes("key2"))))};
 
     {
         auto r = query::partition_range::make(
@@ -174,10 +174,10 @@ BOOST_AUTO_TEST_CASE(test_range_with_equal_value_but_opposite_inclusiveness_is_a
     dht::token tok = dht::global_partitioner().get_random_token();
 
     auto key1 = dht::decorated_key{
-        tok, partition_key::from_single_value(*s, bytes_type->decompose(bytes("key1")))};
+        tok, partition_key::from_single_value(*s, bytes_type->decompose(data_value(bytes("key1"))))};
 
     auto key2 = dht::decorated_key{
-        tok, partition_key::from_single_value(*s, bytes_type->decompose(bytes("key2")))};
+        tok, partition_key::from_single_value(*s, bytes_type->decompose(data_value(bytes("key2"))))};
 
     {
         auto r = query::partition_range::make(
diff --git a/tests/row_cache_alloc_stress.cc b/tests/row_cache_alloc_stress.cc
index a65d02bd60..62fde01319 100644
--- a/tests/row_cache_alloc_stress.cc
+++ b/tests/row_cache_alloc_stress.cc
@@ -77,14 +77,14 @@ int main(int argc, char** argv) {
 
             auto make_small_mutation = [&] {
                 mutation m(new_key(s), s);
-                m.set_clustered_cell(new_ckey(s), "v", bytes(bytes::initialized_later(), cell_size), 1);
+                m.set_clustered_cell(new_ckey(s), "v", data_value(bytes(bytes::initialized_later(), cell_size)), 1);
                 return m;
             };
 
             auto make_large_mutation = [&] {
                 mutation m(new_key(s), s);
                 for (size_t j = 0; j < row_count; j++) {
-                    m.set_clustered_cell(new_ckey(s), "v", bytes(bytes::initialized_later(), cell_size), 2);
+                    m.set_clustered_cell(new_ckey(s), "v", data_value(bytes(bytes::initialized_later(), cell_size)), 2);
                 }
                 return m;
             };
@@ -93,13 +93,13 @@ int main(int argc, char** argv) {
                 auto key = dht::global_partitioner().decorate_key(*s, new_key(s));
 
                 mutation m1(key, s);
-                m1.set_clustered_cell(new_ckey(s), "v", bytes(bytes::initialized_later(), cell_size), 1);
+                m1.set_clustered_cell(new_ckey(s), "v", data_value(bytes(bytes::initialized_later(), cell_size)), 1);
                 cache.populate(m1);
 
                 // Putting large mutations into the memtable. Should take about row_count*cell_size each.
                 mutation m2(key, s);
                 for (size_t j = 0; j < row_count; j++) {
-                    m2.set_clustered_cell(new_ckey(s), "v", bytes(bytes::initialized_later(), cell_size), 2);
+                    m2.set_clustered_cell(new_ckey(s), "v", data_value(bytes(bytes::initialized_later(), cell_size)), 2);
                 }
 
                 mt->apply(m2);
diff --git a/tests/row_cache_test.cc b/tests/row_cache_test.cc
index d5e872adb0..668f3df381 100644
--- a/tests/row_cache_test.cc
+++ b/tests/row_cache_test.cc
@@ -45,7 +45,7 @@ mutation make_new_mutation(schema_ptr s, partition_key key) {
     mutation m(key, s);
     static thread_local int next_value = 1;
     static thread_local api::timestamp_type next_timestamp = 1;
-    m.set_clustered_cell(clustering_key::make_empty(*s), "v", to_bytes(sprint("v%d", next_value++)), next_timestamp++);
+    m.set_clustered_cell(clustering_key::make_empty(*s), "v", data_value(to_bytes(sprint("v%d", next_value++))), next_timestamp++);
     return m;
 }
 
diff --git a/tests/sstable_datafile_test.cc b/tests/sstable_datafile_test.cc
index 11371e48a6..b69ade0bae 100644
--- a/tests/sstable_datafile_test.cc
+++ b/tests/sstable_datafile_test.cc
@@ -1326,7 +1326,7 @@ SEASTAR_TEST_CASE(datafile_generation_37) {
         auto c_key = exploded_clustering_prefix({to_bytes("cl1") });
         const column_definition& cl2 = *s->get_column_definition("cl2");
 
-        m.set_clustered_cell(c_key, cl2, make_atomic_cell(bytes_type->decompose(to_bytes("cl2"))));
+        m.set_clustered_cell(c_key, cl2, make_atomic_cell(bytes_type->decompose(data_value(to_bytes("cl2")))));
         mtp->apply(std::move(m));
 
         auto sst = make_lw_shared<sstable>("ks", "cf", "tests/sstables/tests-temporary", 37, la, big);
@@ -1340,7 +1340,7 @@ SEASTAR_TEST_CASE(datafile_generation_37) {
                         auto clustering = clustering_key::from_clustering_prefix(*s, exploded);
 
                         auto row = mp.clustered_row(clustering);
-                        match_live_cell(row.cells(), *s, "cl2", to_bytes("cl2"));
+                        match_live_cell(row.cells(), *s, "cl2", data_value(to_bytes("cl2")));
                         return make_ready_future<>();
                     });
                 });
@@ -1362,7 +1362,7 @@ SEASTAR_TEST_CASE(datafile_generation_38) {
         auto c_key = clustering_key::from_clustering_prefix(*s, exploded);
 
         const column_definition& cl3 = *s->get_column_definition("cl3");
-        m.set_clustered_cell(c_key, cl3, make_atomic_cell(bytes_type->decompose(to_bytes("cl3"))));
+        m.set_clustered_cell(c_key, cl3, make_atomic_cell(bytes_type->decompose(data_value(to_bytes("cl3")))));
         mtp->apply(std::move(m));
 
         auto sst = make_lw_shared<sstable>("ks", "cf", "tests/sstables/tests-temporary", 38, la, big);
@@ -1375,7 +1375,7 @@ SEASTAR_TEST_CASE(datafile_generation_38) {
                         auto clustering = clustering_key::from_clustering_prefix(*s, exploded);
 
                         auto row = mp.clustered_row(clustering);
-                        match_live_cell(row.cells(), *s, "cl3", to_bytes("cl3"));
+                        match_live_cell(row.cells(), *s, "cl3", data_value(to_bytes("cl3")));
                         return make_ready_future<>();
                     });
                 });
@@ -1396,9 +1396,9 @@ SEASTAR_TEST_CASE(datafile_generation_39) {
         auto c_key = clustering_key::make_empty(*s);
 
         const column_definition& cl1 = *s->get_column_definition("cl1");
-        m.set_clustered_cell(c_key, cl1, make_atomic_cell(bytes_type->decompose(to_bytes("cl1"))));
+        m.set_clustered_cell(c_key, cl1, make_atomic_cell(bytes_type->decompose(data_value(to_bytes("cl1")))));
         const column_definition& cl2 = *s->get_column_definition("cl2");
-        m.set_clustered_cell(c_key, cl2, make_atomic_cell(bytes_type->decompose(to_bytes("cl2"))));
+        m.set_clustered_cell(c_key, cl2, make_atomic_cell(bytes_type->decompose(data_value(to_bytes("cl2")))));
         mtp->apply(std::move(m));
 
         auto sst = make_lw_shared<sstable>("ks", "cf", "tests/sstables/tests-temporary", 39, la, big);
@@ -1408,8 +1408,8 @@ SEASTAR_TEST_CASE(datafile_generation_39) {
                     return sstp->read_row(s, key).then([sstp, s] (auto mutation) {
                         auto& mp = mutation->partition();
                         auto row = mp.clustered_row(clustering_key::make_empty(*s));
-                        match_live_cell(row.cells(), *s, "cl1", to_bytes("cl1"));
-                        match_live_cell(row.cells(), *s, "cl2", to_bytes("cl2"));
+                        match_live_cell(row.cells(), *s, "cl1", data_value(data_value(to_bytes("cl1"))));
+                        match_live_cell(row.cells(), *s, "cl2", data_value(data_value(to_bytes("cl2"))));
                         return make_ready_future<>();
                     });
                 });
diff --git a/tests/sstable_mutation_test.cc b/tests/sstable_mutation_test.cc
index 5a653d6b49..f16d17e33c 100644
--- a/tests/sstable_mutation_test.cc
+++ b/tests/sstable_mutation_test.cc
@@ -138,10 +138,10 @@ SEASTAR_TEST_CASE(complex_sst1_k1) {
         auto s = complex_schema();
 
         auto sr = mutation.partition().static_row();
-        match_live_cell(sr, *s, "static_obj", to_bytes("static_value"));
+        match_live_cell(sr, *s, "static_obj", data_value(to_bytes("static_value")));
 
         auto row1 = clustered_row(mutation, *s, {"cl1.1", "cl2.1"});
-        match_live_cell(row1.cells(), *s, "reg", to_bytes("v1"));
+        match_live_cell(row1.cells(), *s, "reg", data_value(to_bytes("v1")));
         match_absent(row1.cells(), *s, "reg_list");
         match_absent(row1.cells(), *s, "reg_map");
         match_absent(row1.cells(), *s, "reg_fset");
@@ -150,7 +150,7 @@ SEASTAR_TEST_CASE(complex_sst1_k1) {
         match_collection_element<status::live>(reg_set.cells[1], to_bytes("2"), bytes_opt{});
 
         auto row2 = clustered_row(mutation, *s, {"cl1.2", "cl2.2"});
-        match_live_cell(row2.cells(), *s, "reg", to_bytes("v2"));
+        match_live_cell(row2.cells(), *s, "reg", data_value(to_bytes("v2")));
         match_absent(row2.cells(), *s, "reg_set");
         match_absent(row2.cells(), *s, "reg_map");
         match_absent(row2.cells(), *s, "reg_fset");
@@ -167,7 +167,7 @@ SEASTAR_TEST_CASE(complex_sst1_k2) {
         auto s = complex_schema();
 
         auto sr = mutation.partition().static_row();
-        match_live_cell(sr, *s, "static_obj", to_bytes("static_value"));
+        match_live_cell(sr, *s, "static_obj", data_value(to_bytes("static_value")));
         auto static_set = match_collection(sr, *s, "static_collection", tombstone(deletion_time{1431451390, 1431451390225257l}));
         match_collection_element<status::live>(static_set.cells[0], to_bytes("1"), bytes_opt{});
         match_collection_element<status::live>(static_set.cells[1], to_bytes("2"), bytes_opt{});
@@ -175,7 +175,7 @@ SEASTAR_TEST_CASE(complex_sst1_k2) {
         match_collection_element<status::live>(static_set.cells[3], to_bytes("4"), bytes_opt{});
 
         auto row1 = clustered_row(mutation, *s, {"kcl1.1", "kcl2.1"});
-        match_live_cell(row1.cells(), *s, "reg", to_bytes("v3"));
+        match_live_cell(row1.cells(), *s, "reg", data_value(to_bytes("v3")));
         match_absent(row1.cells(), *s, "reg_list");
         match_absent(row1.cells(), *s, "reg_set");
         match_absent(row1.cells(), *s, "reg_fset");
@@ -184,7 +184,7 @@ SEASTAR_TEST_CASE(complex_sst1_k2) {
         match_collection_element<status::live>(reg_map.cells[1], to_bytes("4"), to_bytes("2"));
 
         auto row2 = clustered_row(mutation, *s, {"kcl1.2", "kcl2.2"});
-        match_live_cell(row2.cells(), *s, "reg", to_bytes("v4"));
+        match_live_cell(row2.cells(), *s, "reg", data_value(to_bytes("v4")));
         match_absent(row2.cells(), *s, "reg_set");
         match_absent(row2.cells(), *s, "reg_map");
         match_absent(row2.cells(), *s, "reg_list");
@@ -245,11 +245,11 @@ SEASTAR_TEST_CASE(complex_sst2_k3) {
         auto s = complex_schema();
 
         auto sr = mutation.partition().static_row();
-        match_expiring_cell(sr, *s, "static_obj", to_bytes("static_value_3"), 1431451394597062l, 1431537794);
+        match_expiring_cell(sr, *s, "static_obj", data_value(to_bytes("static_value_3")), 1431451394597062l, 1431537794);
 
         auto row1 = clustered_row(mutation, *s, {"tcl1.1", "tcl2.1"});
         BOOST_REQUIRE(row1.created_at() == 1431451394597062l);
-        match_expiring_cell(row1.cells(), *s, "reg", to_bytes("v5"), 1431451394597062l, 1431537794);
+        match_expiring_cell(row1.cells(), *s, "reg", data_value(to_bytes("v5")), 1431451394597062l, 1431537794);
         match_absent(row1.cells(), *s, "reg_list");
         match_absent(row1.cells(), *s, "reg_set");
         match_absent(row1.cells(), *s, "reg_map");
@@ -283,7 +283,7 @@ SEASTAR_TEST_CASE(complex_sst3_k2) {
         auto s = complex_schema();
 
         auto sr = mutation.partition().static_row();
-        match_live_cell(sr, *s, "static_obj", to_bytes("final_static"));
+        match_live_cell(sr, *s, "static_obj", data_value(to_bytes("final_static")));
 
         auto row = clustered_row(mutation, *s, {"kcl1.1", "kcl2.1"});
         auto reg_map = match_collection(row.cells(), *s, "reg_map", tombstone(deletion_time{0, api::missing_timestamp}));
@@ -385,8 +385,8 @@ SEASTAR_TEST_CASE(compact_storage_sparse_read) {
             return sstp->read_row(s, key).then([sstp, s, &key] (auto mutation) {
                 auto& mp = mutation->partition();
                 auto row = mp.clustered_row(clustering_key::make_empty(*s));
-                match_live_cell(row.cells(), *s, "cl1", to_bytes("cl1"));
-                match_live_cell(row.cells(), *s, "cl2", to_bytes("cl2"));
+                match_live_cell(row.cells(), *s, "cl1", data_value(to_bytes("cl1")));
+                match_live_cell(row.cells(), *s, "cl2", data_value(to_bytes("cl2")));
                 return make_ready_future<>();
             });
         });
@@ -404,7 +404,7 @@ SEASTAR_TEST_CASE(compact_storage_simple_dense_read) {
                 auto clustering = clustering_key::from_clustering_prefix(*s, exploded);
 
                 auto row = mp.clustered_row(clustering);
-                match_live_cell(row.cells(), *s, "cl2", to_bytes("cl2"));
+                match_live_cell(row.cells(), *s, "cl2", data_value(to_bytes("cl2")));
                 return make_ready_future<>();
             });
         });
@@ -422,7 +422,7 @@ SEASTAR_TEST_CASE(compact_storage_dense_read) {
                 auto clustering = clustering_key::from_clustering_prefix(*s, exploded);
 
                 auto row = mp.clustered_row(clustering);
-                match_live_cell(row.cells(), *s, "cl3", to_bytes("cl3"));
+                match_live_cell(row.cells(), *s, "cl3", data_value(to_bytes("cl3")));
                 return make_ready_future<>();
             });
         });
diff --git a/tests/sstable_test.cc b/tests/sstable_test.cc
index ea158b3ed8..c9bebb2c96 100644
--- a/tests/sstable_test.cc
+++ b/tests/sstable_test.cc
@@ -675,7 +675,7 @@ SEASTAR_TEST_CASE(find_key_map) {
         auto b2 = to_bytes("2");
 
         auto map_type = map_type_impl::get_instance(bytes_type, bytes_type, true);
-        auto map_element = std::make_pair<data_value, data_value>(b1, b2);
+        auto map_element = std::make_pair<data_value, data_value>(data_value(b1), data_value(b2));
         std::vector<std::pair<data_value, data_value>> map;
         map.push_back(map_element);
 
@@ -697,8 +697,8 @@ SEASTAR_TEST_CASE(find_key_set) {
         bytes b1("1");
         bytes b2("2");
 
-        set.push_back(b1);
-        set.push_back(b2);
+        set.push_back(data_value(b1));
+        set.push_back(data_value(b2));
         auto set_type = set_type_impl::get_instance(bytes_type, true);
         kk.push_back(make_set_value(set_type, set));
 
@@ -717,8 +717,8 @@ SEASTAR_TEST_CASE(find_key_list) {
 
         bytes b1("1");
         bytes b2("2");
-        list.push_back(b1);
-        list.push_back(b2);
+        list.push_back(data_value(b1));
+        list.push_back(data_value(b2));
 
         auto list_type = list_type_impl::get_instance(bytes_type, true);
         kk.push_back(make_list_value(list_type, list));
@@ -738,8 +738,8 @@ SEASTAR_TEST_CASE(find_key_composite) {
         auto b1 = bytes("HCG8Ee7ENWqfCXipk4-Ygi2hzrbfHC8pTtH3tEmV3d9p2w8gJPuMN_-wp1ejLRf4kNEPEgtgdHXa6NoFE7qUig==");
         auto b2 = bytes("VJizqYxC35YpLaPEJNt_4vhbmKJxAg54xbiF1UkL_9KQkqghVvq34rZ6Lm8eRTi7JNJCXcH6-WtNUSFJXCOfdg==");
 
-        kk.push_back(b1);
-        kk.push_back(b2);
+        kk.push_back(data_value(b1));
+        kk.push_back(data_value(b2));
 
         auto key = sstables::key::from_deeply_exploded(*s, kk);
         BOOST_REQUIRE(sstables::test(sstp).binary_search(summary.entries, key) == 0);
@@ -779,8 +779,8 @@ SEASTAR_TEST_CASE(not_find_key_composite_bucket0) {
         auto b1 = bytes("ZEunFCoqAidHOrPiU3U6UAvUU01IYGvT3kYtYItJ1ODTk7FOsEAD-dqmzmFNfTDYvngzkZwKrLxthB7ItLZ4HQ==");
         auto b2 = bytes("K-GpWx-QtyzLb12z5oNS0C03d3OzNyBKdYJh1XjHiC53KudoqdoFutHUMFLe6H9Emqv_fhwIJEKEb5Csn72f9A==");
 
-        kk.push_back(b1);
-        kk.push_back(b2);
+        kk.push_back(data_value(b1));
+        kk.push_back(data_value(b2));
 
         auto key = sstables::key::from_deeply_exploded(*s, kk);
         // (result + 1) * -1 -1 = 0
diff --git a/tests/storage_proxy_test.cc b/tests/storage_proxy_test.cc
index 160c1ca482..1f74d7f8d4 100644
--- a/tests/storage_proxy_test.cc
+++ b/tests/storage_proxy_test.cc
@@ -69,11 +69,11 @@ SEASTAR_TEST_CASE(test_make_local_reader) {
                 auto reader = service::get_storage_proxy().local().make_local_reader(s->id(), query::full_partition_range);
                 assert_that(to_result_set(s, reader))
                     .has_size(5)
-                    .has(a_row().with_column(bytes("k"), bytes("\01")))
-                    .has(a_row().with_column(bytes("k"), bytes("\02")))
-                    .has(a_row().with_column(bytes("k"), bytes("\03")))
-                    .has(a_row().with_column(bytes("k"), bytes("\04")))
-                    .has(a_row().with_column(bytes("k"), bytes("\05")));
+                    .has(a_row().with_column(bytes("k"), data_value(bytes("\01"))))
+                    .has(a_row().with_column(bytes("k"), data_value(bytes("\02"))))
+                    .has(a_row().with_column(bytes("k"), data_value(bytes("\03"))))
+                    .has(a_row().with_column(bytes("k"), data_value(bytes("\04"))))
+                    .has(a_row().with_column(bytes("k"), data_value(bytes("\05"))));
             }
 
             {
@@ -83,11 +83,11 @@ SEASTAR_TEST_CASE(test_make_local_reader) {
                         {dht::ring_position(dht::maximum_token(), dht::ring_position::token_bound::end)}));
                 assert_that(to_result_set(s, reader))
                     .has_size(5)
-                    .has(a_row().with_column(bytes("k"), bytes("\01")))
-                    .has(a_row().with_column(bytes("k"), bytes("\02")))
-                    .has(a_row().with_column(bytes("k"), bytes("\03")))
-                    .has(a_row().with_column(bytes("k"), bytes("\04")))
-                    .has(a_row().with_column(bytes("k"), bytes("\05")));
+                    .has(a_row().with_column(bytes("k"), data_value(bytes("\01"))))
+                    .has(a_row().with_column(bytes("k"), data_value(bytes("\02"))))
+                    .has(a_row().with_column(bytes("k"), data_value(bytes("\03"))))
+                    .has(a_row().with_column(bytes("k"), data_value(bytes("\04"))))
+                    .has(a_row().with_column(bytes("k"), data_value(bytes("\05"))));
             }
 
             {
diff --git a/tests/types_test.cc b/tests/types_test.cc
index e0bc40aaef..2a521b8e6d 100644
--- a/tests/types_test.cc
+++ b/tests/types_test.cc
@@ -46,7 +46,7 @@ void test_parsing_fails(const shared_ptr<const abstract_type>& type, sstring str
 }
 
 BOOST_AUTO_TEST_CASE(test_bytes_type_string_conversions) {
-    BOOST_REQUIRE(bytes_type->equal(bytes_type->from_string("616263646566"), bytes_type->decompose(bytes{"abcdef"})));
+    BOOST_REQUIRE(bytes_type->equal(bytes_type->from_string("616263646566"), bytes_type->decompose(data_value(bytes{"abcdef"}))));
 }
 
 BOOST_AUTO_TEST_CASE(test_int32_type_string_conversions) {
@@ -515,8 +515,8 @@ BOOST_AUTO_TEST_CASE(test_create_reversed_type) {
     BOOST_REQUIRE(ri->is_reversed());
     BOOST_REQUIRE(ri->is_value_compatible_with(*bytes_type));
     BOOST_REQUIRE(!ri->is_compatible_with(*bytes_type));
-    auto val_lt = bytes_type->decompose(bytes("a"));
-    auto val_gt = bytes_type->decompose(bytes("b"));
+    auto val_lt = bytes_type->decompose(data_value(bytes("a")));
+    auto val_gt = bytes_type->decompose(data_value(bytes("b")));
     auto straight_comp = bytes_type->compare(bytes_view(val_lt), bytes_view(val_gt));
     auto reverse_comp = ri->compare(bytes_view(val_lt), bytes_view(val_gt));
     BOOST_REQUIRE(straight_comp == -reverse_comp);
@@ -532,13 +532,13 @@ BOOST_AUTO_TEST_CASE(test_create_reverse_collection_type) {
     std::vector<data_value> first_set;
     bytes b1("1");
     bytes b2("2");
-    first_set.push_back(b1);
-    first_set.push_back(b2);
+    first_set.push_back(data_value(b1));
+    first_set.push_back(data_value(b2));
 
     std::vector<data_value> second_set;
     bytes b3("2");
-    second_set.push_back(b1);
-    second_set.push_back(b3);
+    second_set.push_back(data_value(b1));
+    second_set.push_back(data_value(b3));
 
     auto bv1 = my_set_type->decompose(make_set_value(my_set_type, first_set));
     auto bv2 = my_set_type->decompose(make_set_value(my_set_type, second_set));
diff --git a/types.hh b/types.hh
index 108fade2ed..b3a13249df 100644
--- a/types.hh
+++ b/types.hh
@@ -214,7 +214,7 @@ public:
     }
     // common conversions from C++ types to database types
     // note: somewhat dangerous, consider a factory function instead
-    data_value(bytes);
+    explicit data_value(bytes);
     data_value(sstring);
     data_value(int32_t);
     data_value(int64_t);
@@ -225,6 +225,7 @@ public:
     data_value(db_clock::time_point);
     data_value(boost::multiprecision::cpp_int);
     data_value(big_decimal);
+    explicit data_value(std::experimental::optional<bytes>);
     template <typename NativeType>
     data_value(std::experimental::optional<NativeType>);
 
@@ -1316,6 +1317,11 @@ data_value make_user_value(data_type tuple_type, user_type_impl::native_type val
 using user_type = shared_ptr<const user_type_impl>;
 using tuple_type = shared_ptr<const tuple_type_impl>;
 
+inline
+data_value::data_value(std::experimental::optional<bytes> v)
+        : data_value(v ? data_value(*v) : data_value::make_null(data_type_for<bytes>())) {
+}
+
 template <typename NativeType>
 data_value::data_value(std::experimental::optional<NativeType> v)
         : data_value(v ? data_value(*v) : data_value::make_null(data_type_for<NativeType>())) {