diff --git a/sstables/partition.cc b/sstables/partition.cc
index de271ddcab..847ba18a7b 100644
--- a/sstables/partition.cc
+++ b/sstables/partition.cc
@@ -137,11 +137,11 @@ public:
         }
     }
 
-    virtual void consume_row_start(bytes_view key, sstables::deletion_time deltime) override {
+    virtual void consume_row_start(sstables::key_view key, sstables::deletion_time deltime) override {
         // FIXME: We should be doing more than that: We need to check the deletion time and propagate the tombstone information
-        auto k = bytes_view(_key);
+        key_view k(_key);
         if (key != k) {
-            throw malformed_sstable_exception(sprint("Key mismatch. Got %s while processing %s", to_hex(key).c_str(), to_hex(k).c_str()));
+            throw malformed_sstable_exception(sprint("Key mismatch. Got %s while processing %s", to_hex(bytes_view(key)).c_str(), to_hex(bytes_view(k)).c_str()));
         }
     }
 
diff --git a/sstables/row.hh b/sstables/row.hh
index c5496185f9..ac1434a9a5 100644
--- a/sstables/row.hh
+++ b/sstables/row.hh
@@ -6,6 +6,7 @@
 #pragma once
 
 #include "bytes.hh"
+#include "key.hh"
 #include "core/temporary_buffer.hh"
 
 // sstables::data_consume_row feeds the contents of a single row into a
@@ -33,7 +34,7 @@ public:
     // (according to the schema) before use.
     // As explained above, the key object is only valid during this call, and
     // if the implementation wishes to save it, it must copy the *contents*.
-    virtual void consume_row_start(bytes_view key, sstables::deletion_time deltime) = 0;
+    virtual void consume_row_start(sstables::key_view key, sstables::deletion_time deltime) = 0;
 
     // Consume one cell (column name and value). Both are serialized, and need
     // to be deserialized according to the schema.
diff --git a/tests/urchin/sstable_test.cc b/tests/urchin/sstable_test.cc
index b49b2adcaf..7d33c23d64 100644
--- a/tests/urchin/sstable_test.cc
+++ b/tests/urchin/sstable_test.cc
@@ -321,8 +321,8 @@ public:
     int count_deleted_cell = 0;
     int count_range_tombstone = 0;
     int count_row_end = 0;
-    virtual void consume_row_start(bytes_view key, sstables::deletion_time deltime) override {
-        BOOST_REQUIRE(key == as_bytes("vinna"));
+    virtual void consume_row_start(sstables::key_view key, sstables::deletion_time deltime) override {
+        BOOST_REQUIRE(bytes_view(key) == as_bytes("vinna"));
         BOOST_REQUIRE(deltime.local_deletion_time == (uint32_t)std::numeric_limits<int32_t>::max());
         BOOST_REQUIRE(deltime.marked_for_delete_at == (uint64_t)std::numeric_limits<int64_t>::min());
         count_row_start++;
@@ -443,7 +443,7 @@ public:
     int count_deleted_cell = 0;
     int count_row_end = 0;
     int count_range_tombstone = 0;
-    virtual void consume_row_start(bytes_view key, sstables::deletion_time deltime) override {
+    virtual void consume_row_start(sstables::key_view key, sstables::deletion_time deltime) override {
         count_row_start++;
     }
     virtual void consume_cell(bytes_view col_name, bytes_view value,
@@ -531,9 +531,9 @@ class ttl_row_consumer : public count_row_consumer {
 public:
     const uint64_t desired_timestamp;
     ttl_row_consumer(uint64_t t) : desired_timestamp(t) { }
-    virtual void consume_row_start(bytes_view key, sstables::deletion_time deltime) override {
+    virtual void consume_row_start(sstables::key_view key, sstables::deletion_time deltime) override {
         count_row_consumer::consume_row_start(key, deltime);
-        BOOST_REQUIRE(key == as_bytes("nadav"));
+        BOOST_REQUIRE(bytes_view(key) == as_bytes("nadav"));
         BOOST_REQUIRE(deltime.local_deletion_time == (uint32_t)std::numeric_limits<int32_t>::max());
         BOOST_REQUIRE(deltime.marked_for_delete_at == (uint64_t)std::numeric_limits<int64_t>::min());
     }
@@ -582,9 +582,9 @@ SEASTAR_TEST_CASE(ttl_read) {
 
 class deleted_cell_row_consumer : public count_row_consumer {
 public:
-    virtual void consume_row_start(bytes_view key, sstables::deletion_time deltime) override {
+    virtual void consume_row_start(sstables::key_view key, sstables::deletion_time deltime) override {
         count_row_consumer::consume_row_start(key, deltime);
-        BOOST_REQUIRE(key == as_bytes("nadav"));
+        BOOST_REQUIRE(bytes_view(key) == as_bytes("nadav"));
         BOOST_REQUIRE(deltime.local_deletion_time == (uint32_t)std::numeric_limits<int32_t>::max());
         BOOST_REQUIRE(deltime.marked_for_delete_at == (uint64_t)std::numeric_limits<int64_t>::min());
     }