diff --git a/tests/cql_query_test.cc b/tests/cql_query_test.cc
index 5c2abea770..6b8cc54ba1 100644
--- a/tests/cql_query_test.cc
+++ b/tests/cql_query_test.cc
@@ -1012,6 +1012,15 @@ SEASTAR_TEST_CASE(test_tuples) {
                 .with_rows({{
                      {tt->decompose(tuple_type_impl::native_type({int32_t(1001), int64_t(2001), sstring("abc1")}))},
                 }});
+            return e.execute_cql("create table cf2 (p1 int PRIMARY KEY, r1 tuple<int, bigint, text>)").discard_result();
+        }).then([&e] {
+            return e.execute_cql("insert into cf2 (p1, r1) values (1, (1, 2, 'abc'));").discard_result();
+        }).then([&e] {
+            return e.execute_cql("select * from cf2 where p1 = 1;");
+        }).then([&e, tt] (auto msg) {
+            assert_that(msg).is_rows().with_rows({
+                { int32_type->decompose(int32_t(1)), tt->decompose(tuple_type_impl::native_type({int32_t(1), int64_t(2), sstring("abc")})) }
+            });
         });
     });
 }
diff --git a/transport/server.cc b/transport/server.cc
index 438df763fa..ce034fe244 100644
--- a/transport/server.cc
+++ b/transport/server.cc
@@ -1201,6 +1201,15 @@ public:
         if (type->is_reversed()) {
             fail(unimplemented::cause::REVERSED);
         }
+        if (type->is_tuple()) {
+            r.write_short(uint16_t(type_id::TUPLE));
+            auto ttype = static_pointer_cast<const tuple_type_impl>(type);
+            r.write_short(ttype->size());
+            for (auto&& t : ttype->all_types()) {
+                encode(r, t);
+            }
+            return;
+        }
         if (type->is_collection()) {
             auto&& ctype = static_cast<const collection_type_impl*>(type.get());
             if (&ctype->_kind == &collection_type_impl::kind::map) {
diff --git a/types.cc b/types.cc
index 279e365c09..070fb6ee74 100644
--- a/types.cc
+++ b/types.cc
@@ -1913,7 +1913,10 @@ tuple_type_impl::tuple_type_impl(sstring name, std::vector<data_type> types)
 }
 
 tuple_type_impl::tuple_type_impl(std::vector<data_type> types)
-        : tuple_type_impl(make_name(types), std::move(types)) {
+        : abstract_type(make_name(types)), _types(std::move(types)) {
+    for (auto& t : _types) {
+        t = t->freeze();
+    }
 }
 
 shared_ptr<tuple_type_impl>
@@ -2053,7 +2056,7 @@ tuple_type_impl::as_cql3_type() const {
 
 sstring
 tuple_type_impl::make_name(const std::vector<data_type>& types) {
-    return sprint("tuple<%s>", ::join(", ", types | boost::adaptors::transformed(std::mem_fn(&abstract_type::name))));
+    return sprint("org.apache.cassandra.db.marshal.TupleType(%s)", ::join(", ", types | boost::adaptors::transformed(std::mem_fn(&abstract_type::name))));
 }
 
 sstring
diff --git a/types.hh b/types.hh
index 1a9f583b01..1c85001c92 100644
--- a/types.hh
+++ b/types.hh
@@ -303,6 +303,7 @@ public:
     virtual bool is_collection() const { return false; }
     virtual bool is_multi_cell() const { return false; }
     virtual bool is_reversed() const { return false; }
+    virtual bool is_tuple() const { return false; }
     virtual ::shared_ptr<cql3::cql3_type> as_cql3_type() const = 0;
     virtual shared_ptr<const abstract_type> freeze() const { return shared_from_this(); }
     friend class list_type_impl;
@@ -1061,6 +1062,7 @@ public:
     virtual bool is_compatible_with(const abstract_type& previous) const override;
     virtual bool is_value_compatible_with_internal(const abstract_type& previous) const override;
     virtual shared_ptr<cql3::cql3_type> as_cql3_type() const override;
+    virtual bool is_tuple() const override { return true; }
 private:
     bool check_compatibility(const abstract_type& previous, bool (abstract_type::*predicate)(const abstract_type&) const) const;
     static sstring make_name(const std::vector<data_type>& types);