diff --git a/cql3/abstract_marker.cc b/cql3/abstract_marker.cc
index 1d2c41f8e2..33140a4095 100644
--- a/cql3/abstract_marker.cc
+++ b/cql3/abstract_marker.cc
@@ -50,7 +50,7 @@
 
 namespace cql3 {
 
-abstract_marker::abstract_marker(int32_t bind_index, ::shared_ptr<column_specification>&& receiver)
+abstract_marker::abstract_marker(int32_t bind_index, lw_shared_ptr<column_specification>&& receiver)
     : _bind_index{bind_index}
     , _receiver{std::move(receiver)}
 { }
@@ -67,7 +67,7 @@ abstract_marker::raw::raw(int32_t bind_index)
     : _bind_index{bind_index}
 { }
 
-::shared_ptr<term> abstract_marker::raw::prepare(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const
+::shared_ptr<term> abstract_marker::raw::prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const
 {
     if (receiver->type->is_collection()) {
         if (receiver->type->get_kind() == abstract_type::kind::list) {
@@ -87,7 +87,7 @@ abstract_marker::raw::raw(int32_t bind_index)
     return ::make_shared<constants::marker>(_bind_index, receiver);
 }
 
-assignment_testable::test_result abstract_marker::raw::test_assignment(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const {
+assignment_testable::test_result abstract_marker::raw::test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const {
     return assignment_testable::test_result::WEAKLY_ASSIGNABLE;
 }
 
@@ -99,12 +99,12 @@ abstract_marker::in_raw::in_raw(int32_t bind_index)
     : raw{bind_index}
 { }
 
-::shared_ptr<column_specification> abstract_marker::in_raw::make_in_receiver(const column_specification& receiver) {
+lw_shared_ptr<column_specification> abstract_marker::in_raw::make_in_receiver(const column_specification& receiver) {
     auto in_name = ::make_shared<column_identifier>(sstring("in(") + receiver.name->to_string() + sstring(")"), true);
-    return ::make_shared<column_specification>(receiver.ks_name, receiver.cf_name, in_name, list_type_impl::get_instance(receiver.type, false));
+    return make_lw_shared<column_specification>(receiver.ks_name, receiver.cf_name, in_name, list_type_impl::get_instance(receiver.type, false));
 }
 
-::shared_ptr<term> abstract_marker::in_raw::prepare(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const {
+::shared_ptr<term> abstract_marker::in_raw::prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const {
     return ::make_shared<lists::marker>(_bind_index, make_in_receiver(*receiver));
 }
 
diff --git a/cql3/abstract_marker.hh b/cql3/abstract_marker.hh
index c9fcb2f495..cfa3b6c629 100644
--- a/cql3/abstract_marker.hh
+++ b/cql3/abstract_marker.hh
@@ -53,9 +53,9 @@ namespace cql3 {
 class abstract_marker : public non_terminal {
 protected:
     const int32_t _bind_index;
-    const ::shared_ptr<column_specification> _receiver;
+    const lw_shared_ptr<column_specification> _receiver;
 public:
-    abstract_marker(int32_t bind_index, ::shared_ptr<column_specification>&& receiver);
+    abstract_marker(int32_t bind_index, lw_shared_ptr<column_specification>&& receiver);
 
     virtual void collect_marker_specification(variable_specifications& bound_names) const override;
 
@@ -70,9 +70,9 @@ public:
     public:
         raw(int32_t bind_index);
 
-        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const override;
+        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override;
 
-        virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const override;
+        virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override;
 
         virtual sstring to_string() const override;
     };
@@ -87,9 +87,9 @@ public:
     public:
         in_raw(int32_t bind_index);
     private:
-        static ::shared_ptr<column_specification> make_in_receiver(const column_specification& receiver);
+        static lw_shared_ptr<column_specification> make_in_receiver(const column_specification& receiver);
     public:
-        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const override;
+        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override;
     };
 };
 
diff --git a/cql3/assignment_testable.hh b/cql3/assignment_testable.hh
index 42216a7df0..b7292ac0d5 100644
--- a/cql3/assignment_testable.hh
+++ b/cql3/assignment_testable.hh
@@ -70,7 +70,7 @@ public:
     // Test all elements of toTest for assignment. If all are exact match, return exact match. If any is not assignable,
     // return not assignable. Otherwise, return weakly assignable.
     template <typename AssignmentTestablePtrRange>
-    static test_result test_all(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver,
+    static test_result test_all(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver,
                 AssignmentTestablePtrRange&& to_test) {
         test_result res = test_result::EXACT_MATCH;
         for (auto&& rt : to_test) {
@@ -99,7 +99,7 @@ public:
      * Most caller should just call the isAssignable() method on the result, though functions have a use for
      * testing "strong" equality to decide the most precise overload to pick when multiple could match.
      */
-    virtual test_result test_assignment(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const = 0;
+    virtual test_result test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const = 0;
 
     // for error reporting
     virtual sstring assignment_testable_source_context() const = 0;
diff --git a/cql3/attributes.cc b/cql3/attributes.cc
index 0a5b0c1dca..91dd17b4c6 100644
--- a/cql3/attributes.cc
+++ b/cql3/attributes.cc
@@ -135,12 +135,12 @@ std::unique_ptr<attributes> attributes::raw::prepare(database& db, const sstring
     return std::unique_ptr<attributes>{new attributes{std::move(ts), std::move(ttl)}};
 }
 
-::shared_ptr<column_specification> attributes::raw::timestamp_receiver(const sstring& ks_name, const sstring& cf_name) const {
-    return ::make_shared<column_specification>(ks_name, cf_name, ::make_shared<column_identifier>("[timestamp]", true), data_type_for<int64_t>());
+lw_shared_ptr<column_specification> attributes::raw::timestamp_receiver(const sstring& ks_name, const sstring& cf_name) const {
+    return make_lw_shared<column_specification>(ks_name, cf_name, ::make_shared<column_identifier>("[timestamp]", true), data_type_for<int64_t>());
 }
 
-::shared_ptr<column_specification> attributes::raw::time_to_live_receiver(const sstring& ks_name, const sstring& cf_name) const {
-    return ::make_shared<column_specification>(ks_name, cf_name, ::make_shared<column_identifier>("[ttl]", true), data_type_for<int32_t>());
+lw_shared_ptr<column_specification> attributes::raw::time_to_live_receiver(const sstring& ks_name, const sstring& cf_name) const {
+    return make_lw_shared<column_specification>(ks_name, cf_name, ::make_shared<column_identifier>("[ttl]", true), data_type_for<int32_t>());
 }
 
 }
diff --git a/cql3/attributes.hh b/cql3/attributes.hh
index d5a5d27258..8b416f808b 100644
--- a/cql3/attributes.hh
+++ b/cql3/attributes.hh
@@ -78,9 +78,9 @@ public:
 
         std::unique_ptr<attributes> prepare(database& db, const sstring& ks_name, const sstring& cf_name) const;
     private:
-        ::shared_ptr<column_specification> timestamp_receiver(const sstring& ks_name, const sstring& cf_name) const;
+        lw_shared_ptr<column_specification> timestamp_receiver(const sstring& ks_name, const sstring& cf_name) const;
 
-        ::shared_ptr<column_specification> time_to_live_receiver(const sstring& ks_name, const sstring& cf_name) const;
+        lw_shared_ptr<column_specification> time_to_live_receiver(const sstring& ks_name, const sstring& cf_name) const;
     };
 };
 
diff --git a/cql3/column_condition.cc b/cql3/column_condition.cc
index 6b2b210f8b..0e422fc9c2 100644
--- a/cql3/column_condition.cc
+++ b/cql3/column_condition.cc
@@ -297,7 +297,7 @@ column_condition::raw::prepare(database& db, const sstring& keyspace, const colu
         throw exceptions::invalid_request_exception("Conditions on counters are not supported");
     }
     shared_ptr<term> collection_element_term;
-    shared_ptr<column_specification> value_spec = receiver.column_specification;
+    lw_shared_ptr<column_specification> value_spec = receiver.column_specification;
 
     if (_collection_element) {
         if (!receiver.type->is_collection()) {
@@ -306,7 +306,7 @@ column_condition::raw::prepare(database& db, const sstring& keyspace, const colu
         }
         // Pass  a correct type specification to the collection_element->prepare(), so that it can
         // later be used to validate the parameter type is compatible with receiver type.
-        shared_ptr<column_specification> element_spec;
+        lw_shared_ptr<column_specification> element_spec;
         auto ctype = static_cast<const collection_type_impl*>(receiver.type.get());
         const column_specification& recv_column_spec = *receiver.column_specification;
         if (ctype->get_kind() == abstract_type::kind::list) {
diff --git a/cql3/column_specification.cc b/cql3/column_specification.cc
index 78472ba1fc..d77da5211a 100644
--- a/cql3/column_specification.cc
+++ b/cql3/column_specification.cc
@@ -51,7 +51,7 @@ column_specification::column_specification(std::string_view ks_name_, std::strin
     { }
 
 
-bool column_specification::all_in_same_table(const std::vector<::shared_ptr<column_specification>>& names)
+bool column_specification::all_in_same_table(const std::vector<lw_shared_ptr<column_specification>>& names)
 {
     assert(!names.empty());
 
diff --git a/cql3/column_specification.hh b/cql3/column_specification.hh
index 65dbb71cd7..d7e0370819 100644
--- a/cql3/column_specification.hh
+++ b/cql3/column_specification.hh
@@ -45,7 +45,6 @@
 
 namespace cql3 {
 
-class column_specification;
 class column_identifier;
 
 class column_specification final {
@@ -63,15 +62,15 @@ public:
      * @param alias the column alias
      * @return a new <code>ColumnSpecification</code> for the same column but with the specified alias.
      */
-    ::shared_ptr<column_specification> with_alias(::shared_ptr<column_identifier> alias) {
-        return ::make_shared<column_specification>(ks_name, cf_name, alias, type);
+    lw_shared_ptr<column_specification> with_alias(::shared_ptr<column_identifier> alias) {
+        return make_lw_shared<column_specification>(ks_name, cf_name, alias, type);
     }
     
     bool is_reversed_type() const {
         return ::dynamic_pointer_cast<const reversed_type_impl>(type) != nullptr;
     }
 
-    static bool all_in_same_table(const std::vector<::shared_ptr<column_specification>>& names);
+    static bool all_in_same_table(const std::vector<lw_shared_ptr<column_specification>>& names);
 };
 
 }
diff --git a/cql3/constants.cc b/cql3/constants.cc
index d34720e178..261d2e63d3 100644
--- a/cql3/constants.cc
+++ b/cql3/constants.cc
@@ -82,7 +82,7 @@ constants::literal::parsed_value(data_type validator) const
 }
 
 assignment_testable::test_result
-constants::literal::test_assignment(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const
+constants::literal::test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const
 {
     auto receiver_type = receiver->type->as_cql3_type();
     if (receiver_type.is_collection() || receiver_type.is_user_type()) {
@@ -155,7 +155,7 @@ constants::literal::test_assignment(database& db, const sstring& keyspace, ::sha
 }
 
 ::shared_ptr<term>
-constants::literal::prepare(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const
+constants::literal::prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const
 {
     if (!is_assignable(test_assignment(db, keyspace, receiver))) {
         throw exceptions::invalid_request_exception(format("Invalid {} constant ({}) for \"{}\" of type {}",
diff --git a/cql3/constants.hh b/cql3/constants.hh
index 9e7081c672..02725c5023 100644
--- a/cql3/constants.hh
+++ b/cql3/constants.hh
@@ -87,7 +87,7 @@ public:
         };
     public:
         static thread_local const ::shared_ptr<terminal> NULL_VALUE;
-        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const override {
+        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override {
             if (!is_assignable(test_assignment(db, keyspace, receiver))) {
                 throw exceptions::invalid_request_exception("Invalid null value for counter increment/decrement");
             }
@@ -96,7 +96,7 @@ public:
 
         virtual assignment_testable::test_result test_assignment(database& db,
             const sstring& keyspace,
-            ::shared_ptr<column_specification> receiver) const override {
+            lw_shared_ptr<column_specification> receiver) const override {
                 return receiver->type->is_counter()
                     ? assignment_testable::test_result::NOT_ASSIGNABLE
                     : assignment_testable::test_result::WEAKLY_ASSIGNABLE;
@@ -153,7 +153,7 @@ public:
             return ::make_shared<literal>(type::DURATION, text);
         }
 
-        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const override;
+        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override;
     private:
         bytes parsed_value(data_type validator) const;
     public:
@@ -161,7 +161,7 @@ public:
             return _text;
         }
 
-        virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const;
+        virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const;
 
         virtual sstring to_string() const override {
             return _type == type::STRING ? sstring(format("'{}'", _text)) : _text;
@@ -170,7 +170,7 @@ public:
 
     class marker : public abstract_marker {
     public:
-        marker(int32_t bind_index, ::shared_ptr<column_specification> receiver)
+        marker(int32_t bind_index, lw_shared_ptr<column_specification> receiver)
             : abstract_marker{bind_index, std::move(receiver)}
         {
             assert(!_receiver->type->is_collection() && !_receiver->type->is_user_type());
diff --git a/cql3/functions/function_call.hh b/cql3/functions/function_call.hh
index e04d24af67..aa5118a3df 100644
--- a/cql3/functions/function_call.hh
+++ b/cql3/functions/function_call.hh
@@ -74,12 +74,12 @@ public:
         raw(function_name name, std::vector<shared_ptr<term::raw>> terms)
             : _name(std::move(name)), _terms(std::move(terms)) {
         }
-        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const override;
+        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override;
     private:
         // All parameters must be terminal
         static bytes_opt execute(scalar_function& fun, std::vector<shared_ptr<term>> parameters);
     public:
-        virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const override;
+        virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override;
         virtual sstring to_string() const override;
     };
 };
diff --git a/cql3/functions/functions.cc b/cql3/functions/functions.cc
index 1af619b85a..2bd8e24eae 100644
--- a/cql3/functions/functions.cc
+++ b/cql3/functions/functions.cc
@@ -141,12 +141,12 @@ void functions::remove_function(const function_name& name, const std::vector<dat
     with_udf_iter(name, arg_types, [] (functions::declared_t::iterator i) { _declared.erase(i); });
 }
 
-shared_ptr<column_specification>
+lw_shared_ptr<column_specification>
 functions::make_arg_spec(const sstring& receiver_ks, const sstring& receiver_cf,
         const function& fun, size_t i) {
     auto&& name = boost::lexical_cast<std::string>(fun.name());
     std::transform(name.begin(), name.end(), name.begin(), ::tolower);
-    return ::make_shared<column_specification>(receiver_ks,
+    return make_lw_shared<column_specification>(receiver_ks,
                                    receiver_cf,
                                    ::make_shared<column_identifier>(format("arg{:d}({})", i, name), true),
                                    fun.arg_types()[i]);
@@ -187,7 +187,7 @@ functions::get(database& db,
         const std::vector<shared_ptr<assignment_testable>>& provided_args,
         const sstring& receiver_ks,
         const sstring& receiver_cf,
-        shared_ptr<column_specification> receiver) {
+        lw_shared_ptr<column_specification> receiver) {
 
     static const function_name TOKEN_FUNCTION_NAME = function_name::native_function("token");
     static const function_name TO_JSON_FUNCTION_NAME = function_name::native_function("tojson");
@@ -507,7 +507,7 @@ function_call::make_terminal(shared_ptr<function> fun, cql3::raw_value result, c
 }
 
 ::shared_ptr<term>
-function_call::raw::prepare(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const {
+function_call::raw::prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const {
     std::vector<shared_ptr<assignment_testable>> args;
     args.reserve(_terms.size());
     std::transform(_terms.begin(), _terms.end(), std::back_inserter(args),
@@ -572,7 +572,7 @@ function_call::raw::execute(scalar_function& fun, std::vector<shared_ptr<term>>
 }
 
 assignment_testable::test_result
-function_call::raw::test_assignment(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const {
+function_call::raw::test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const {
     // Note: Functions.get() will return null if the function doesn't exist, or throw is no function matching
     // the arguments can be found. We may get one of those if an undefined/wrong function is used as argument
     // of another, existing, function. In that case, we return true here because we'll throw a proper exception
diff --git a/cql3/functions/functions.hh b/cql3/functions/functions.hh
index b27f396a07..8eaf3d9edb 100644
--- a/cql3/functions/functions.hh
+++ b/cql3/functions/functions.hh
@@ -67,7 +67,7 @@ class functions {
 private:
     static std::unordered_multimap<function_name, shared_ptr<function>> init();
 public:
-    static shared_ptr<column_specification> make_arg_spec(const sstring& receiver_ks, const sstring& receiver_cf,
+    static lw_shared_ptr<column_specification> make_arg_spec(const sstring& receiver_ks, const sstring& receiver_cf,
             const function& fun, size_t i);
     static int get_overload_count(const function_name& name);
 public:
@@ -77,7 +77,7 @@ public:
                                     const std::vector<shared_ptr<assignment_testable>>& provided_args,
                                     const sstring& receiver_ks,
                                     const sstring& receiver_cf,
-                                    ::shared_ptr<column_specification> receiver = nullptr);
+                                    lw_shared_ptr<column_specification> receiver = nullptr);
     template <typename AssignmentTestablePtrRange>
     static shared_ptr<function> get(database& db,
                                     const sstring& keyspace,
@@ -85,7 +85,7 @@ public:
                                     AssignmentTestablePtrRange&& provided_args,
                                     const sstring& receiver_ks,
                                     const sstring& receiver_cf,
-                                    ::shared_ptr<column_specification> receiver = nullptr) {
+                                    lw_shared_ptr<column_specification> receiver = nullptr) {
         const std::vector<shared_ptr<assignment_testable>> args(std::begin(provided_args), std::end(provided_args));
         return get(db, keyspace, name, args, receiver_ks, receiver_cf, receiver);
     }
diff --git a/cql3/lists.cc b/cql3/lists.cc
index fd37d4947c..9eeb7880df 100644
--- a/cql3/lists.cc
+++ b/cql3/lists.cc
@@ -30,28 +30,28 @@
 
 namespace cql3 {
 
-shared_ptr<column_specification>
+lw_shared_ptr<column_specification>
 lists::index_spec_of(const column_specification& column) {
-    return ::make_shared<column_specification>(column.ks_name, column.cf_name,
+    return make_lw_shared<column_specification>(column.ks_name, column.cf_name,
             ::make_shared<column_identifier>(format("idx({})", *column.name), true), int32_type);
 }
 
-shared_ptr<column_specification>
+lw_shared_ptr<column_specification>
 lists::value_spec_of(const column_specification& column) {
-    return ::make_shared<column_specification>(column.ks_name, column.cf_name,
+    return make_lw_shared<column_specification>(column.ks_name, column.cf_name,
             ::make_shared<column_identifier>(format("value({})", *column.name), true),
                 dynamic_pointer_cast<const list_type_impl>(column.type)->get_elements_type());
 }
 
-shared_ptr<column_specification>
+lw_shared_ptr<column_specification>
 lists::uuid_index_spec_of(const column_specification& column) {
-    return ::make_shared<column_specification>(column.ks_name, column.cf_name,
+    return make_lw_shared<column_specification>(column.ks_name, column.cf_name,
             ::make_shared<column_identifier>(format("uuid_idx({})", *column.name), true), uuid_type);
 }
 
 
 shared_ptr<term>
-lists::literal::prepare(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const {
+lists::literal::prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const {
     validate_assignable_to(db, keyspace, *receiver);
 
     // In Cassandra, an empty (unfrozen) map/set/list is equivalent to the column being null. In
@@ -101,7 +101,7 @@ lists::literal::validate_assignable_to(database& db, const sstring keyspace, con
 }
 
 assignment_testable::test_result
-lists::literal::test_assignment(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const {
+lists::literal::test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const {
     if (!dynamic_pointer_cast<const list_type_impl>(receiver->type)) {
         return assignment_testable::test_result::NOT_ASSIGNABLE;
     }
diff --git a/cql3/lists.hh b/cql3/lists.hh
index 4280d929de..c430aa12ad 100644
--- a/cql3/lists.hh
+++ b/cql3/lists.hh
@@ -54,9 +54,9 @@ namespace cql3 {
 class lists {
     lists() = delete;
 public:
-    static shared_ptr<column_specification> index_spec_of(const column_specification&);
-    static shared_ptr<column_specification> value_spec_of(const column_specification&);
-    static shared_ptr<column_specification> uuid_index_spec_of(const column_specification&);
+    static lw_shared_ptr<column_specification> index_spec_of(const column_specification&);
+    static lw_shared_ptr<column_specification> value_spec_of(const column_specification&);
+    static lw_shared_ptr<column_specification> uuid_index_spec_of(const column_specification&);
 
     class literal : public term::raw {
         const std::vector<shared_ptr<term::raw>> _elements;
@@ -64,11 +64,11 @@ public:
         explicit literal(std::vector<shared_ptr<term::raw>> elements)
             : _elements(std::move(elements)) {
         }
-        virtual shared_ptr<term> prepare(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const override;
+        virtual shared_ptr<term> prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override;
     private:
         void validate_assignable_to(database& db, const sstring keyspace, const column_specification& receiver) const;
     public:
-        virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const override;
+        virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override;
         virtual sstring to_string() const override;
     };
 
@@ -113,7 +113,7 @@ public:
      */
     class marker : public abstract_marker {
     public:
-        marker(int32_t bind_index, ::shared_ptr<column_specification> receiver)
+        marker(int32_t bind_index, lw_shared_ptr<column_specification> receiver)
             : abstract_marker{bind_index, std::move(receiver)}
         { }
         virtual ::shared_ptr<terminal> bind(const query_options& options) override;
diff --git a/cql3/maps.cc b/cql3/maps.cc
index 6b71812101..9bb04780b8 100644
--- a/cql3/maps.cc
+++ b/cql3/maps.cc
@@ -51,22 +51,22 @@
 
 namespace cql3 {
 
-shared_ptr<column_specification>
+lw_shared_ptr<column_specification>
 maps::key_spec_of(const column_specification& column) {
-    return ::make_shared<column_specification>(column.ks_name, column.cf_name,
+    return make_lw_shared<column_specification>(column.ks_name, column.cf_name,
                 ::make_shared<column_identifier>(format("key({})", *column.name), true),
                  dynamic_pointer_cast<const map_type_impl>(column.type)->get_keys_type());
 }
 
-shared_ptr<column_specification>
+lw_shared_ptr<column_specification>
 maps::value_spec_of(const column_specification& column) {
-    return ::make_shared<column_specification>(column.ks_name, column.cf_name,
+    return make_lw_shared<column_specification>(column.ks_name, column.cf_name,
                 ::make_shared<column_identifier>(format("value({})", *column.name), true),
                  dynamic_pointer_cast<const map_type_impl>(column.type)->get_values_type());
 }
 
 ::shared_ptr<term>
-maps::literal::prepare(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const {
+maps::literal::prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const {
     validate_assignable_to(db, keyspace, *receiver);
 
     auto key_spec = maps::key_spec_of(*receiver);
@@ -114,7 +114,7 @@ maps::literal::validate_assignable_to(database& db, const sstring& keyspace, con
 }
 
 assignment_testable::test_result
-maps::literal::test_assignment(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const {
+maps::literal::test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const {
     if (!dynamic_pointer_cast<const map_type_impl>(receiver->type)) {
         return assignment_testable::test_result::NOT_ASSIGNABLE;
     }
diff --git a/cql3/maps.hh b/cql3/maps.hh
index 19d4e6bc5c..8268af45fe 100644
--- a/cql3/maps.hh
+++ b/cql3/maps.hh
@@ -56,8 +56,8 @@ class maps {
 private:
     maps() = delete;
 public:
-    static shared_ptr<column_specification> key_spec_of(const column_specification& column);
-    static shared_ptr<column_specification> value_spec_of(const column_specification& column);
+    static lw_shared_ptr<column_specification> key_spec_of(const column_specification& column);
+    static lw_shared_ptr<column_specification> value_spec_of(const column_specification& column);
 
     class literal : public term::raw {
     public:
@@ -66,11 +66,11 @@ public:
         literal(const std::vector<std::pair<::shared_ptr<term::raw>, ::shared_ptr<term::raw>>>& entries_)
             : entries{entries_}
         { }
-        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const override;
+        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override;
     private:
         void validate_assignable_to(database& db, const sstring& keyspace, const column_specification& receiver) const;
     public:
-        virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const override;
+        virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override;
         virtual sstring to_string() const override;
     };
 
@@ -104,7 +104,7 @@ public:
 
     class marker : public abstract_marker {
     public:
-        marker(int32_t bind_index, ::shared_ptr<column_specification> receiver)
+        marker(int32_t bind_index, lw_shared_ptr<column_specification> receiver)
             : abstract_marker{bind_index, std::move(receiver)}
         { }
         virtual ::shared_ptr<terminal> bind(const query_options& options) override;
diff --git a/cql3/multi_column_relation.hh b/cql3/multi_column_relation.hh
index 24157046a3..77961a1a65 100644
--- a/cql3/multi_column_relation.hh
+++ b/cql3/multi_column_relation.hh
@@ -140,7 +140,7 @@ protected:
     virtual shared_ptr<restrictions::restriction> new_EQ_restriction(database& db, schema_ptr schema,
                                                                      variable_specifications& bound_names) override {
         auto rs = receivers(db, *schema);
-        std::vector<::shared_ptr<column_specification>> col_specs(rs.size());
+        std::vector<lw_shared_ptr<column_specification>> col_specs(rs.size());
         std::transform(rs.begin(), rs.end(), col_specs.begin(), [] (auto cs) {
             return cs->column_specification;
         });
@@ -151,7 +151,7 @@ protected:
     virtual shared_ptr<restrictions::restriction> new_IN_restriction(database& db, schema_ptr schema,
                                                                      variable_specifications& bound_names) override {
         auto rs = receivers(db, *schema);
-        std::vector<::shared_ptr<column_specification>> col_specs(rs.size());
+        std::vector<lw_shared_ptr<column_specification>> col_specs(rs.size());
         std::transform(rs.begin(), rs.end(), col_specs.begin(), [] (auto cs) {
             return cs->column_specification;
         });
@@ -175,7 +175,7 @@ protected:
                                                                         variable_specifications& bound_names,
                                                                         statements::bound bound, bool inclusive) override {
         auto rs = receivers(db, *schema);
-        std::vector<::shared_ptr<column_specification>> col_specs(rs.size());
+        std::vector<lw_shared_ptr<column_specification>> col_specs(rs.size());
         std::transform(rs.begin(), rs.end(), col_specs.begin(), [] (auto cs) {
             return cs->column_specification;
         });
@@ -200,7 +200,7 @@ protected:
         return ::make_shared(multi_column_relation(std::move(new_entities), _relation_type, _values_or_marker, _in_values, _in_marker));
     }
 
-    virtual shared_ptr<term> to_term(const std::vector<shared_ptr<column_specification>>& receivers,
+    virtual shared_ptr<term> to_term(const std::vector<lw_shared_ptr<column_specification>>& receivers,
                                      const term::raw& raw, database& db, const sstring& keyspace,
                                      variable_specifications& bound_names) const override {
         const auto& as_multi_column_raw = dynamic_cast<const term::multi_column_raw&>(raw);
diff --git a/cql3/operation.cc b/cql3/operation.cc
index 378b06930a..2a85ad3d6f 100644
--- a/cql3/operation.cc
+++ b/cql3/operation.cc
@@ -216,7 +216,7 @@ operation::subtraction::prepare(database& db, const sstring& keyspace, const col
     } else if (ctype->get_kind() == abstract_type::kind::map) {
         auto&& mtype = dynamic_pointer_cast<const map_type_impl>(ctype);
         // The value for a map subtraction is actually a set
-        auto&& vr = ::make_shared<column_specification>(
+        auto&& vr = make_lw_shared<column_specification>(
                 receiver.column_specification->ks_name,
                 receiver.column_specification->cf_name,
                 receiver.column_specification->name,
@@ -294,7 +294,7 @@ operation::set_counter_value_from_tuple_list::prepare(database& db, const sstrin
 
     // We need to fake a column of list<tuple<...>> to prepare the value term
     auto & os = receiver.column_specification;
-    auto spec = ::make_shared<cql3::column_specification>(os->ks_name, os->cf_name, os->name, counter_tuple_list_type);
+    auto spec = make_lw_shared<cql3::column_specification>(os->ks_name, os->cf_name, os->name, counter_tuple_list_type);
     auto v = _value->prepare(db, keyspace, spec);
 
     // Will not be used elsewhere, so make it local.
diff --git a/cql3/query_options.cc b/cql3/query_options.cc
index 694edb281b..e18f42c5e4 100644
--- a/cql3/query_options.cc
+++ b/cql3/query_options.cc
@@ -189,7 +189,7 @@ bytes_view query_options::linearize(fragmented_temporary_buffer::view view) cons
     }
 }
 
-void query_options::prepare(const std::vector<::shared_ptr<column_specification>>& specs)
+void query_options::prepare(const std::vector<lw_shared_ptr<column_specification>>& specs)
 {
     if (!_names) {
         return;
diff --git a/cql3/query_options.hh b/cql3/query_options.hh
index 121605eba5..ddf4c42074 100644
--- a/cql3/query_options.hh
+++ b/cql3/query_options.hh
@@ -245,7 +245,7 @@ public:
         return _cql_config;
     }
 
-    void prepare(const std::vector<::shared_ptr<column_specification>>& specs);
+    void prepare(const std::vector<lw_shared_ptr<column_specification>>& specs);
 private:
     void fill_value_views();
 };
diff --git a/cql3/relation.hh b/cql3/relation.hh
index a4585c0530..a9e1743b9a 100644
--- a/cql3/relation.hh
+++ b/cql3/relation.hh
@@ -249,7 +249,7 @@ protected:
      * @return the <code>Term</code> corresponding to the specified <code>Raw</code>
      * @throws InvalidRequestException if the <code>Raw</code> term is not valid
      */
-    virtual ::shared_ptr<term> to_term(const std::vector<::shared_ptr<column_specification>>& receivers,
+    virtual ::shared_ptr<term> to_term(const std::vector<lw_shared_ptr<column_specification>>& receivers,
                                        const term::raw& raw,
                                        database& db,
                                        const sstring& keyspace,
@@ -265,7 +265,7 @@ protected:
      * @return the <code>Term</code>s corresponding to the specified <code>Raw</code> terms
      * @throws InvalidRequestException if the <code>Raw</code> terms are not valid
      */
-    std::vector<::shared_ptr<term>> to_terms(const std::vector<::shared_ptr<column_specification>>& receivers,
+    std::vector<::shared_ptr<term>> to_terms(const std::vector<lw_shared_ptr<column_specification>>& receivers,
                                              const std::vector<::shared_ptr<term::raw>>& raws,
                                              database& db,
                                              const sstring& keyspace,
diff --git a/cql3/result_set.cc b/cql3/result_set.cc
index c9307963e8..4c9c556222 100644
--- a/cql3/result_set.cc
+++ b/cql3/result_set.cc
@@ -43,12 +43,12 @@
 
 namespace cql3 {
 
-metadata::metadata(std::vector<::shared_ptr<column_specification>> names_)
+metadata::metadata(std::vector<lw_shared_ptr<column_specification>> names_)
         : _flags(flag_enum_set())
         , _column_info(make_lw_shared<column_info>(std::move(names_)))
 { }
 
-metadata::metadata(flag_enum_set flags, std::vector<::shared_ptr<column_specification>> names_, uint32_t column_count,
+metadata::metadata(flag_enum_set flags, std::vector<lw_shared_ptr<column_specification>> names_, uint32_t column_count,
         lw_shared_ptr<const service::pager::paging_state> paging_state)
     : _flags(flags)
     , _column_info(make_lw_shared<column_info>(std::move(names_), column_count))
@@ -60,7 +60,7 @@ uint32_t metadata::value_count() const {
     return _flags.contains<flag::NO_METADATA>() ? _column_info->_column_count : _column_info->_names.size();
 }
 
-void metadata::add_non_serialized_column(::shared_ptr<column_specification> name) {
+void metadata::add_non_serialized_column(lw_shared_ptr<column_specification> name) {
     // See comment above. Because columnCount doesn't account the newly added name, it
     // won't be serialized.
     _column_info->_names.emplace_back(std::move(name));
@@ -101,7 +101,7 @@ lw_shared_ptr<const service::pager::paging_state> metadata::paging_state() const
     return _paging_state;
 }
 
-prepared_metadata::prepared_metadata(const std::vector<::shared_ptr<column_specification>>& names,
+prepared_metadata::prepared_metadata(const std::vector<lw_shared_ptr<column_specification>>& names,
                                      const std::vector<uint16_t>& partition_key_bind_indices)
     : _names{names}
     , _partition_key_bind_indices{partition_key_bind_indices}
@@ -115,7 +115,7 @@ prepared_metadata::flag_enum_set prepared_metadata::flags() const {
     return _flags;
 }
 
-const std::vector<::shared_ptr<column_specification>>& prepared_metadata::names() const {
+const std::vector<lw_shared_ptr<column_specification>>& prepared_metadata::names() const {
     return _names;
 }
 
@@ -123,7 +123,7 @@ const std::vector<uint16_t>& prepared_metadata::partition_key_bind_indices() con
     return _partition_key_bind_indices;
 }
 
-result_set::result_set(std::vector<::shared_ptr<column_specification>> metadata_)
+result_set::result_set(std::vector<lw_shared_ptr<column_specification>> metadata_)
     : _metadata(::make_shared<metadata>(std::move(metadata_)))
 { }
 
@@ -179,7 +179,7 @@ const std::deque<std::vector<bytes_opt>>& result_set::rows() const {
 shared_ptr<const cql3::metadata>
 make_empty_metadata() {
     static thread_local shared_ptr<const metadata> empty_metadata_cache = [] {
-        auto result = ::make_shared<metadata>(std::vector<::shared_ptr<cql3::column_specification>>{});
+        auto result = ::make_shared<metadata>(std::vector<lw_shared_ptr<cql3::column_specification>>{});
         result->set_skip_metadata();
         return result;
     }();
diff --git a/cql3/result_set.hh b/cql3/result_set.hh
index 0b398a92ff..47b3a5d821 100644
--- a/cql3/result_set.hh
+++ b/cql3/result_set.hh
@@ -74,15 +74,15 @@ public:
     // used to include columns in the resultSet that we need to do post-query re-orderings
     // (SelectStatement.orderResults) but that shouldn't be sent to the user as they haven't been requested
     // (CASSANDRA-4911). So the serialization code will exclude any columns in name whose index is >= columnCount.
-        std::vector<::shared_ptr<column_specification>> _names;
+        std::vector<lw_shared_ptr<column_specification>> _names;
         uint32_t _column_count;
 
-        column_info(std::vector<::shared_ptr<column_specification>> names, uint32_t column_count)
+        column_info(std::vector<lw_shared_ptr<column_specification>> names, uint32_t column_count)
             : _names(std::move(names))
             , _column_count(column_count)
         { }
 
-        explicit column_info(std::vector<::shared_ptr<column_specification>> names)
+        explicit column_info(std::vector<lw_shared_ptr<column_specification>> names)
             : _names(std::move(names))
             , _column_count(_names.size())
         { }
@@ -95,15 +95,15 @@ private:
     lw_shared_ptr<const service::pager::paging_state> _paging_state;
 
 public:
-    metadata(std::vector<::shared_ptr<column_specification>> names_);
+    metadata(std::vector<lw_shared_ptr<column_specification>> names_);
 
-    metadata(flag_enum_set flags, std::vector<::shared_ptr<column_specification>> names_, uint32_t column_count,
+    metadata(flag_enum_set flags, std::vector<lw_shared_ptr<column_specification>> names_, uint32_t column_count,
             lw_shared_ptr<const service::pager::paging_state> paging_state);
 
     // The maximum number of values that the ResultSet can hold. This can be bigger than columnCount due to CASSANDRA-4911
     uint32_t value_count() const;
 
-    void add_non_serialized_column(::shared_ptr<column_specification> name);
+    void add_non_serialized_column(lw_shared_ptr<column_specification> name);
 
 private:
     bool all_in_same_cf() const;
@@ -120,7 +120,7 @@ public:
 
     lw_shared_ptr<const service::pager::paging_state> paging_state() const;
 
-    const std::vector<::shared_ptr<column_specification>>& get_names() const {
+    const std::vector<lw_shared_ptr<column_specification>>& get_names() const {
         return _column_info->_names;
     }
 };
@@ -139,14 +139,14 @@ public:
     using flag_enum_set = enum_set<flag_enum>;
 private:
     flag_enum_set _flags;
-    std::vector<::shared_ptr<column_specification>> _names;
+    std::vector<lw_shared_ptr<column_specification>> _names;
     std::vector<uint16_t> _partition_key_bind_indices;
 public:
-    prepared_metadata(const std::vector<::shared_ptr<column_specification>>& names,
+    prepared_metadata(const std::vector<lw_shared_ptr<column_specification>>& names,
                       const std::vector<uint16_t>& partition_key_bind_indices);
 
     flag_enum_set flags() const;
-    const std::vector<::shared_ptr<column_specification>>& names() const;
+    const std::vector<lw_shared_ptr<column_specification>>& names() const;
     const std::vector<uint16_t>& partition_key_bind_indices() const;
 };
 
@@ -167,7 +167,7 @@ class result_set {
 
     friend class result;
 public:
-    result_set(std::vector<::shared_ptr<column_specification>> metadata_);
+    result_set(std::vector<lw_shared_ptr<column_specification>> metadata_);
 
     result_set(::shared_ptr<metadata> metadata);
 
diff --git a/cql3/selection/selection.cc b/cql3/selection/selection.cc
index ff314014f1..470fed17b8 100644
--- a/cql3/selection/selection.cc
+++ b/cql3/selection/selection.cc
@@ -56,7 +56,7 @@ namespace selection {
 
 selection::selection(schema_ptr schema,
     std::vector<const column_definition*> columns,
-    std::vector<::shared_ptr<column_specification>> metadata_,
+    std::vector<lw_shared_ptr<column_specification>> metadata_,
     bool collect_timestamps,
     bool collect_TTLs,
     trivial is_trivial)
@@ -92,7 +92,7 @@ private:
     const bool _is_wildcard;
 public:
     static ::shared_ptr<simple_selection> make(schema_ptr schema, std::vector<const column_definition*> columns, bool is_wildcard) {
-        std::vector<::shared_ptr<column_specification>> metadata;
+        std::vector<lw_shared_ptr<column_specification>> metadata;
         metadata.reserve(columns.size());
         for (auto&& col : columns) {
             metadata.emplace_back(col->column_specification);
@@ -106,7 +106,7 @@ public:
      * get much duplicate in practice, it's more efficient not to bother.
      */
     simple_selection(schema_ptr schema, std::vector<const column_definition*> columns,
-        std::vector<::shared_ptr<column_specification>> metadata, bool is_wildcard)
+        std::vector<lw_shared_ptr<column_specification>> metadata, bool is_wildcard)
             : selection(schema, std::move(columns), std::move(metadata), false, false, trivial::yes)
             , _is_wildcard(is_wildcard)
     { }
@@ -155,7 +155,7 @@ private:
     ::shared_ptr<selector_factories> _factories;
 public:
     selection_with_processing(schema_ptr schema, std::vector<const column_definition*> columns,
-            std::vector<::shared_ptr<column_specification>> metadata, ::shared_ptr<selector_factories> factories)
+            std::vector<lw_shared_ptr<column_specification>> metadata, ::shared_ptr<selector_factories> factories)
         : selection(schema, std::move(columns), std::move(metadata),
             factories->contains_write_time_selector_factory(),
             factories->contains_ttl_selector_factory())
@@ -264,14 +264,14 @@ uint32_t selection::add_column_for_post_processing(const column_definition& c) {
     }
 }
 
-std::vector<::shared_ptr<column_specification>>
+std::vector<lw_shared_ptr<column_specification>>
 selection::collect_metadata(const schema& schema, const std::vector<::shared_ptr<raw_selector>>& raw_selectors,
         const selector_factories& factories) {
-    std::vector<::shared_ptr<column_specification>> r;
+    std::vector<lw_shared_ptr<column_specification>> r;
     r.reserve(raw_selectors.size());
     auto i = raw_selectors.begin();
     for (auto&& factory : factories) {
-        ::shared_ptr<column_specification> col_spec = factory->get_column_specification(schema);
+        lw_shared_ptr<column_specification> col_spec = factory->get_column_specification(schema);
         ::shared_ptr<column_identifier> alias = (*i++)->alias;
         r.push_back(alias ? col_spec->with_alias(alias) : col_spec);
     }
diff --git a/cql3/selection/selection.hh b/cql3/selection/selection.hh
index 4cd2a40cd2..69aa58bd6f 100644
--- a/cql3/selection/selection.hh
+++ b/cql3/selection/selection.hh
@@ -99,7 +99,7 @@ protected:
 
     selection(schema_ptr schema,
         std::vector<const column_definition*> columns,
-        std::vector<::shared_ptr<column_specification>> metadata_,
+        std::vector<lw_shared_ptr<column_specification>> metadata_,
         bool collect_timestamps,
         bool collect_TTLs, trivial is_trivial = trivial::no);
 
@@ -197,7 +197,7 @@ private:
             [] (auto&& s) { return s->processes_selection(); });
     }
 
-    static std::vector<::shared_ptr<column_specification>> collect_metadata(const schema& schema,
+    static std::vector<lw_shared_ptr<column_specification>> collect_metadata(const schema& schema,
         const std::vector<::shared_ptr<raw_selector>>& raw_selectors, const selector_factories& factories);
 public:
     static ::shared_ptr<selection> from_selectors(database& db, schema_ptr schema, const std::vector<::shared_ptr<raw_selector>>& raw_selectors);
diff --git a/cql3/selection/selector.cc b/cql3/selection/selector.cc
index 37ec9d0c82..507cfee67c 100644
--- a/cql3/selection/selector.cc
+++ b/cql3/selection/selector.cc
@@ -26,9 +26,9 @@ namespace cql3 {
 
 namespace selection {
 
-::shared_ptr<column_specification>
+lw_shared_ptr<column_specification>
 selector::factory::get_column_specification(const schema& schema) const {
-    return ::make_shared<column_specification>(schema.ks_name(),
+    return make_lw_shared<column_specification>(schema.ks_name(),
         schema.cf_name(),
         ::make_shared<column_identifier>(column_name(), true),
         get_return_type());
diff --git a/cql3/selection/selector.hh b/cql3/selection/selector.hh
index fb85f8afb5..adceb0c8f3 100644
--- a/cql3/selection/selector.hh
+++ b/cql3/selection/selector.hh
@@ -107,7 +107,7 @@ public:
      */
     virtual void reset() = 0;
 
-    virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const override {
+    virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override {
         auto t1 = receiver->type->underlying_type();
         auto t2 = get_type()->underlying_type();
         // We want columns of `counter_type' to be served by underlying type's overloads
@@ -142,7 +142,7 @@ public:
      * @param schema the column family schema
      * @return a column specification
      */
-    ::shared_ptr<column_specification> get_column_specification(const schema& schema) const;
+    lw_shared_ptr<column_specification> get_column_specification(const schema& schema) const;
 
     /**
      * Creates a new <code>selector</code> instance.
diff --git a/cql3/sets.cc b/cql3/sets.cc
index a49b80aab7..8c8c32533a 100644
--- a/cql3/sets.cc
+++ b/cql3/sets.cc
@@ -27,15 +27,15 @@
 
 namespace cql3 {
 
-shared_ptr<column_specification>
+lw_shared_ptr<column_specification>
 sets::value_spec_of(const column_specification& column) {
-    return ::make_shared<column_specification>(column.ks_name, column.cf_name,
+    return make_lw_shared<column_specification>(column.ks_name, column.cf_name,
             ::make_shared<column_identifier>(format("value({})", *column.name), true),
             dynamic_pointer_cast<const set_type_impl>(column.type)->get_elements_type());
 }
 
 shared_ptr<term>
-sets::literal::prepare(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const {
+sets::literal::prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const {
     validate_assignable_to(db, keyspace, *receiver);
 
     if (_elements.empty()) {
@@ -105,7 +105,7 @@ sets::literal::validate_assignable_to(database& db, const sstring& keyspace, con
 }
 
 assignment_testable::test_result
-sets::literal::test_assignment(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const {
+sets::literal::test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const {
     if (!dynamic_pointer_cast<const set_type_impl>(receiver->type)) {
         // We've parsed empty maps as a set literal to break the ambiguity so handle that case now
         if (dynamic_pointer_cast<const map_type_impl>(receiver->type) && _elements.empty()) {
@@ -224,7 +224,7 @@ sets::delayed_value::bind(const query_options& options) {
 }
 
 
-sets::marker::marker(int32_t bind_index, ::shared_ptr<column_specification> receiver)
+sets::marker::marker(int32_t bind_index, lw_shared_ptr<column_specification> receiver)
     : abstract_marker{bind_index, std::move(receiver)} {
         assert(dynamic_cast<const set_type_impl*>(_receiver->type.get()));
     }
diff --git a/cql3/sets.hh b/cql3/sets.hh
index 8325d4538b..5057b8c9eb 100644
--- a/cql3/sets.hh
+++ b/cql3/sets.hh
@@ -56,7 +56,7 @@ namespace cql3 {
 class sets {
     sets() = delete;
 public:
-    static shared_ptr<column_specification> value_spec_of(const column_specification& column);
+    static lw_shared_ptr<column_specification> value_spec_of(const column_specification& column);
 
     class literal : public term::raw {
         std::vector<shared_ptr<term::raw>> _elements;
@@ -64,10 +64,10 @@ public:
         explicit literal(std::vector<shared_ptr<term::raw>> elements)
                 : _elements(std::move(elements)) {
         }
-        virtual shared_ptr<term> prepare(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const override;
+        virtual shared_ptr<term> prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override;
         void validate_assignable_to(database& db, const sstring& keyspace, const column_specification& receiver) const;
         assignment_testable::test_result
-        test_assignment(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const;
+        test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const;
         virtual sstring to_string() const override;
     };
 
@@ -100,7 +100,7 @@ public:
 
     class marker : public abstract_marker {
     public:
-        marker(int32_t bind_index, ::shared_ptr<column_specification> receiver);
+        marker(int32_t bind_index, lw_shared_ptr<column_specification> receiver);
         virtual ::shared_ptr<terminal> bind(const query_options& options) override;
     };
 
diff --git a/cql3/single_column_relation.cc b/cql3/single_column_relation.cc
index a4ffa55f9a..6ad419b87e 100644
--- a/cql3/single_column_relation.cc
+++ b/cql3/single_column_relation.cc
@@ -53,7 +53,7 @@ using namespace cql3::restrictions;
 namespace cql3 {
 
 ::shared_ptr<term>
-single_column_relation::to_term(const std::vector<::shared_ptr<column_specification>>& receivers,
+single_column_relation::to_term(const std::vector<lw_shared_ptr<column_specification>>& receivers,
                                 const term::raw& raw,
                                 database& db,
                                 const sstring& keyspace,
@@ -107,7 +107,7 @@ single_column_relation::new_LIKE_restriction(
     return ::make_shared<single_column_restriction::LIKE>(column_def, std::move(term));
 }
 
-std::vector<::shared_ptr<column_specification>>
+std::vector<lw_shared_ptr<column_specification>>
 single_column_relation::to_receivers(const schema& schema, const column_definition& column_def) const
 {
     using namespace statements::request_validations;
diff --git a/cql3/single_column_relation.hh b/cql3/single_column_relation.hh
index b1cdef7b1d..4cae14c557 100644
--- a/cql3/single_column_relation.hh
+++ b/cql3/single_column_relation.hh
@@ -117,7 +117,7 @@ public:
     }
 
 protected:
-    virtual ::shared_ptr<term> to_term(const std::vector<::shared_ptr<column_specification>>& receivers,
+    virtual ::shared_ptr<term> to_term(const std::vector<lw_shared_ptr<column_specification>>& receivers,
                           const term::raw& raw, database& db, const sstring& keyspace,
                           variable_specifications& bound_names) const override;
 
@@ -202,9 +202,9 @@ private:
      * @return the receivers for the specified relation.
      * @throws exceptions::invalid_request_exception if the relation is invalid
      */
-    std::vector<::shared_ptr<column_specification>> to_receivers(const schema& schema, const column_definition& column_def) const;
+    std::vector<lw_shared_ptr<column_specification>> to_receivers(const schema& schema, const column_definition& column_def) const;
 
-    static shared_ptr<column_specification> make_collection_receiver(shared_ptr<column_specification> receiver, bool for_key) {
+    static lw_shared_ptr<column_specification> make_collection_receiver(lw_shared_ptr<column_specification> receiver, bool for_key) {
         return static_cast<const collection_type_impl*>(receiver->type.get())->make_collection_receiver(*receiver, for_key);
     }
 
diff --git a/cql3/statements/batch_statement.cc b/cql3/statements/batch_statement.cc
index fe20e0e90e..97e81c52fb 100644
--- a/cql3/statements/batch_statement.cc
+++ b/cql3/statements/batch_statement.cc
@@ -400,9 +400,9 @@ void batch_statement::build_cas_result_set_metadata() {
     _columns_of_cas_result_set.resize(schema.all_columns_count());
 
     // Add the mandatory [applied] column to result set metadata
-    std::vector<shared_ptr<column_specification>> columns;
+    std::vector<lw_shared_ptr<column_specification>> columns;
 
-    auto applied = ::make_shared<cql3::column_specification>(schema.ks_name(), schema.cf_name(),
+    auto applied = make_lw_shared<cql3::column_specification>(schema.ks_name(), schema.cf_name(),
             ::make_shared<cql3::column_identifier>("[applied]", false), boolean_type);
     columns.push_back(applied);
 
diff --git a/cql3/statements/list_permissions_statement.cc b/cql3/statements/list_permissions_statement.cc
index fc447d6b01..d4f986ee57 100644
--- a/cql3/statements/list_permissions_statement.cc
+++ b/cql3/statements/list_permissions_statement.cc
@@ -110,14 +110,14 @@ cql3::statements::list_permissions_statement::execute(
         service::query_state& state,
         const query_options& options) const {
     static auto make_column = [](sstring name) {
-        return ::make_shared<column_specification>(
+        return make_lw_shared<column_specification>(
                 auth::meta::AUTH_KS,
                 "permissions",
                 ::make_shared<column_identifier>(std::move(name), true),
                 utf8_type);
     };
 
-    static thread_local const std::vector<::shared_ptr<column_specification>> metadata({
+    static thread_local const std::vector<lw_shared_ptr<column_specification>> metadata({
         make_column("role"), make_column("username"), make_column("resource"), make_column("permission")
     });
 
diff --git a/cql3/statements/list_users_statement.cc b/cql3/statements/list_users_statement.cc
index 9f6cba6d69..1a073be094 100644
--- a/cql3/statements/list_users_statement.cc
+++ b/cql3/statements/list_users_statement.cc
@@ -63,7 +63,7 @@ cql3::statements::list_users_statement::execute(service::storage_proxy& proxy, s
     static const sstring virtual_table_name("users");
 
     static const auto make_column_spec = [](const sstring& name, const ::shared_ptr<const abstract_type>& ty) {
-        return ::make_shared<column_specification>(
+        return make_lw_shared<column_specification>(
             auth::meta::AUTH_KS,
             virtual_table_name,
             ::make_shared<column_identifier>(name, true),
@@ -71,7 +71,7 @@ cql3::statements::list_users_statement::execute(service::storage_proxy& proxy, s
     };
 
     static thread_local const auto metadata = ::make_shared<cql3::metadata>(
-        std::vector<::shared_ptr<column_specification>>{
+        std::vector<lw_shared_ptr<column_specification>>{
                 make_column_spec("name", utf8_type),
                 make_column_spec("super", boolean_type)});
 
diff --git a/cql3/statements/modification_statement.cc b/cql3/statements/modification_statement.cc
index 37bf088b79..028faff56a 100644
--- a/cql3/statements/modification_statement.cc
+++ b/cql3/statements/modification_statement.cc
@@ -408,9 +408,9 @@ modification_statement::build_cas_result_set(seastar::shared_ptr<cql3::metadata>
 
 void modification_statement::build_cas_result_set_metadata() {
 
-    std::vector<shared_ptr<column_specification>> columns;
+    std::vector<lw_shared_ptr<column_specification>> columns;
     // Add the mandatory [applied] column to result set metadata
-    auto applied = seastar::make_shared<cql3::column_specification>(s->ks_name(), s->cf_name(),
+    auto applied = make_lw_shared<cql3::column_specification>(s->ks_name(), s->cf_name(),
             make_shared<cql3::column_identifier>("[applied]", false), boolean_type);
 
     columns.push_back(applied);
diff --git a/cql3/statements/prepared_statement.hh b/cql3/statements/prepared_statement.hh
index 926ffab093..42d36e35c8 100644
--- a/cql3/statements/prepared_statement.hh
+++ b/cql3/statements/prepared_statement.hh
@@ -69,10 +69,10 @@ public:
 
 public:
     const ::shared_ptr<cql_statement> statement;
-    const std::vector<::shared_ptr<column_specification>> bound_names;
+    const std::vector<lw_shared_ptr<column_specification>> bound_names;
     std::vector<uint16_t> partition_key_bind_indices;
 
-    prepared_statement(::shared_ptr<cql_statement> statement_, std::vector<::shared_ptr<column_specification>> bound_names_, std::vector<uint16_t> partition_key_bind_indices);
+    prepared_statement(::shared_ptr<cql_statement> statement_, std::vector<lw_shared_ptr<column_specification>> bound_names_, std::vector<uint16_t> partition_key_bind_indices);
 
     prepared_statement(::shared_ptr<cql_statement> statement_, const variable_specifications& names, const std::vector<uint16_t>& partition_key_bind_indices);
 
diff --git a/cql3/statements/raw/parsed_statement.cc b/cql3/statements/raw/parsed_statement.cc
index 7e33559ead..889c2b261b 100644
--- a/cql3/statements/raw/parsed_statement.cc
+++ b/cql3/statements/raw/parsed_statement.cc
@@ -71,7 +71,7 @@ bool parsed_statement::uses_function(const sstring& ks_name, const sstring& func
 
 }
 
-prepared_statement::prepared_statement(::shared_ptr<cql_statement> statement_, std::vector<::shared_ptr<column_specification>> bound_names_, std::vector<uint16_t> partition_key_bind_indices)
+prepared_statement::prepared_statement(::shared_ptr<cql_statement> statement_, std::vector<lw_shared_ptr<column_specification>> bound_names_, std::vector<uint16_t> partition_key_bind_indices)
     : statement(std::move(statement_))
     , bound_names(std::move(bound_names_))
     , partition_key_bind_indices(std::move(partition_key_bind_indices))
@@ -86,7 +86,7 @@ prepared_statement::prepared_statement(::shared_ptr<cql_statement> statement_, v
 { }
 
 prepared_statement::prepared_statement(::shared_ptr<cql_statement>&& statement_)
-    : prepared_statement(statement_, std::vector<::shared_ptr<column_specification>>(), std::vector<uint16_t>())
+    : prepared_statement(statement_, std::vector<lw_shared_ptr<column_specification>>(), std::vector<uint16_t>())
 { }
 
 }
diff --git a/cql3/statements/raw/select_statement.hh b/cql3/statements/raw/select_statement.hh
index 738c9d66c5..adf6e6eac3 100644
--- a/cql3/statements/raw/select_statement.hh
+++ b/cql3/statements/raw/select_statement.hh
@@ -157,7 +157,7 @@ private:
 
     bool contains_alias(const column_identifier& name) const;
 
-    ::shared_ptr<column_specification> limit_receiver(bool per_partition = false);
+    lw_shared_ptr<column_specification> limit_receiver(bool per_partition = false);
 
 #if 0
     public:
diff --git a/cql3/statements/role-management-statements.cc b/cql3/statements/role-management-statements.cc
index 6d8d0d311f..b3b9ea951a 100644
--- a/cql3/statements/role-management-statements.cc
+++ b/cql3/statements/role-management-statements.cc
@@ -337,7 +337,7 @@ list_roles_statement::execute(service::storage_proxy&, service::query_state& sta
     static const sstring virtual_table_name("roles");
 
     static const auto make_column_spec = [](const sstring& name, const ::shared_ptr<const abstract_type>& ty) {
-        return ::make_shared<column_specification>(
+        return make_lw_shared<column_specification>(
                 auth::meta::AUTH_KS,
                 virtual_table_name,
                 ::make_shared<column_identifier>(name, true),
@@ -347,7 +347,7 @@ list_roles_statement::execute(service::storage_proxy&, service::query_state& sta
     static const thread_local auto custom_options_type = map_type_impl::get_instance(utf8_type, utf8_type, true);
 
     static const thread_local auto metadata = ::make_shared<cql3::metadata>(
-            std::vector<::shared_ptr<column_specification>>{
+            std::vector<lw_shared_ptr<column_specification>>{
                     make_column_spec("role", utf8_type),
                     make_column_spec("super", boolean_type),
                     make_column_spec("login", boolean_type),
diff --git a/cql3/statements/select_statement.cc b/cql3/statements/select_statement.cc
index bf16042b8d..b99a8cc9f2 100644
--- a/cql3/statements/select_statement.cc
+++ b/cql3/statements/select_statement.cc
@@ -1584,9 +1584,9 @@ bool select_statement::contains_alias(const column_identifier& name) const {
     });
 }
 
-::shared_ptr<column_specification> select_statement::limit_receiver(bool per_partition) {
+lw_shared_ptr<column_specification> select_statement::limit_receiver(bool per_partition) {
     sstring name = per_partition ? "[per_partition_limit]" : "[limit]";
-    return ::make_shared<column_specification>(keyspace(), column_family(), ::make_shared<column_identifier>(name, true),
+    return make_lw_shared<column_specification>(keyspace(), column_family(), ::make_shared<column_identifier>(name, true),
         int32_type);
 }
 
diff --git a/cql3/statements/update_statement.cc b/cql3/statements/update_statement.cc
index e07aa61660..3bdf3d4373 100644
--- a/cql3/statements/update_statement.cc
+++ b/cql3/statements/update_statement.cc
@@ -370,7 +370,7 @@ insert_json_statement::prepare_internal(database& db, schema_ptr schema,
 {
     assert(dynamic_pointer_cast<constants::literal>(_json_value) || dynamic_pointer_cast<abstract_marker::raw>(_json_value));
     auto json_column_placeholder = ::make_shared<column_identifier>("", true);
-    auto prepared_json_value = _json_value->prepare(db, "", ::make_shared<column_specification>("", "", json_column_placeholder, utf8_type));
+    auto prepared_json_value = _json_value->prepare(db, "", make_lw_shared<column_specification>("", "", json_column_placeholder, utf8_type));
     prepared_json_value->collect_marker_specification(bound_names);
     auto stmt = ::make_shared<cql3::statements::insert_prepared_json_statement>(bound_names.size(), schema, std::move(attrs), stats, std::move(prepared_json_value), _default_unset);
     prepare_conditions(db, *schema, bound_names, *stmt);
diff --git a/cql3/term.hh b/cql3/term.hh
index 19f6c533a2..8625e5fecd 100644
--- a/cql3/term.hh
+++ b/cql3/term.hh
@@ -132,7 +132,7 @@ public:
          * case this RawTerm describe a list index or a map key, etc...
          * @return the prepared term.
          */
-        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const = 0;
+        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const = 0;
 
         virtual sstring to_string() const = 0;
 
@@ -147,7 +147,7 @@ public:
 
     class multi_column_raw : public virtual raw {
     public:
-        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, const std::vector<shared_ptr<column_specification>>& receiver) const = 0;
+        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, const std::vector<lw_shared_ptr<column_specification>>& receiver) const = 0;
     };
 };
 
diff --git a/cql3/token_relation.cc b/cql3/token_relation.cc
index b01e90940d..353186847a 100644
--- a/cql3/token_relation.cc
+++ b/cql3/token_relation.cc
@@ -53,7 +53,7 @@ std::vector<const column_definition*> cql3::token_relation::get_column_definitio
     return res;
 }
 
-std::vector<::shared_ptr<cql3::column_specification>> cql3::token_relation::to_receivers(
+std::vector<lw_shared_ptr<cql3::column_specification>> cql3::token_relation::to_receivers(
         const schema& schema,
         const std::vector<const column_definition*>& column_defs) const {
     auto pk = schema.partition_key_columns();
@@ -74,7 +74,7 @@ std::vector<::shared_ptr<cql3::column_specification>> cql3::token_relation::to_r
                         std::to_string(column_defs)));
     }
     //auto* c = column_defs.front();
-    return {::make_shared<column_specification>(schema.ks_name(), schema.cf_name(),
+    return {make_lw_shared<column_specification>(schema.ks_name(), schema.cf_name(),
                 ::make_shared<column_identifier>("partition key token", true),
                 dht::token::get_token_validator())};
 }
@@ -126,7 +126,7 @@ sstring cql3::token_relation::to_string() const {
 }
 
 ::shared_ptr<cql3::term> cql3::token_relation::to_term(
-        const std::vector<::shared_ptr<column_specification>>& receivers,
+        const std::vector<lw_shared_ptr<column_specification>>& receivers,
         const term::raw& raw, database& db, const sstring& keyspace,
         variable_specifications& bound_names) const {
     auto term = raw.prepare(db, keyspace, receivers.front());
diff --git a/cql3/token_relation.hh b/cql3/token_relation.hh
index 279b2bcfa0..1c0fe319fa 100644
--- a/cql3/token_relation.hh
+++ b/cql3/token_relation.hh
@@ -82,7 +82,7 @@ private:
      * @return the receivers for the specified relation.
      * @throws InvalidRequestException if the relation is invalid
      */
-    std::vector<::shared_ptr<column_specification>> to_receivers(const schema& schema, const std::vector<const column_definition*>& column_defs) const;
+    std::vector<lw_shared_ptr<column_specification>> to_receivers(const schema& schema, const std::vector<const column_definition*>& column_defs) const;
 
 public:
     token_relation(std::vector<::shared_ptr<column_identifier::raw>> entities,
@@ -122,7 +122,7 @@ public:
     sstring to_string() const override;
 
 protected:
-    ::shared_ptr<term> to_term(const std::vector<::shared_ptr<column_specification>>& receivers,
+    ::shared_ptr<term> to_term(const std::vector<lw_shared_ptr<column_specification>>& receivers,
                                        const term::raw& raw,
                                        database& db,
                                        const sstring& keyspace,
diff --git a/cql3/tuples.cc b/cql3/tuples.cc
index 7223122a79..342321d64d 100644
--- a/cql3/tuples.cc
+++ b/cql3/tuples.cc
@@ -26,9 +26,9 @@
 
 namespace cql3 {
 
-shared_ptr<column_specification>
+lw_shared_ptr<column_specification>
 tuples::component_spec_of(const column_specification& column, size_t component) {
-    return ::make_shared<column_specification>(
+    return make_lw_shared<column_specification>(
             column.ks_name,
             column.cf_name,
             ::make_shared<column_identifier>(format("{}[{:d}]", column.name, component), true),
@@ -36,7 +36,7 @@ tuples::component_spec_of(const column_specification& column, size_t component)
 }
 
 shared_ptr<term>
-tuples::literal::prepare(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const {
+tuples::literal::prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const {
     validate_assignable_to(db, keyspace, *receiver);
     std::vector<shared_ptr<term>> values;
     bool all_terminal = true;
@@ -56,7 +56,7 @@ tuples::literal::prepare(database& db, const sstring& keyspace, shared_ptr<colum
 }
 
 shared_ptr<term>
-tuples::literal::prepare(database& db, const sstring& keyspace, const std::vector<shared_ptr<column_specification>>& receivers) const {
+tuples::literal::prepare(database& db, const sstring& keyspace, const std::vector<lw_shared_ptr<column_specification>>& receivers) const {
     if (_elements.size() != receivers.size()) {
         throw exceptions::invalid_request_exception(format("Expected {:d} elements in value tuple, but got {:d}: {}", receivers.size(), _elements.size(), *this));
     }
@@ -102,8 +102,8 @@ tuples::in_value::from_serialized(const fragmented_temporary_buffer::view& value
     }
 }
 
-::shared_ptr<column_specification>
-tuples::in_raw::make_in_receiver(const std::vector<shared_ptr<column_specification>>& receivers) {
+lw_shared_ptr<column_specification>
+tuples::in_raw::make_in_receiver(const std::vector<lw_shared_ptr<column_specification>>& receivers) {
     std::vector<data_type> types;
     types.reserve(receivers.size());
     sstring in_name = "in(";
@@ -123,10 +123,10 @@ tuples::in_raw::make_in_receiver(const std::vector<shared_ptr<column_specificati
 
     auto identifier = ::make_shared<column_identifier>(in_name, true);
     auto type = tuple_type_impl::get_instance(types);
-    return ::make_shared<column_specification>(receivers.front()->ks_name, receivers.front()->cf_name, identifier, list_type_impl::get_instance(type, false));
+    return make_lw_shared<column_specification>(receivers.front()->ks_name, receivers.front()->cf_name, identifier, list_type_impl::get_instance(type, false));
 }
 
-tuples::in_marker::in_marker(int32_t bind_index, ::shared_ptr<column_specification> receiver)
+tuples::in_marker::in_marker(int32_t bind_index, lw_shared_ptr<column_specification> receiver)
     : abstract_marker(bind_index, std::move(receiver))
 {
     assert(dynamic_pointer_cast<const list_type_impl>(_receiver->type));
diff --git a/cql3/tuples.hh b/cql3/tuples.hh
index bdb1b3139d..b897820f43 100644
--- a/cql3/tuples.hh
+++ b/cql3/tuples.hh
@@ -52,7 +52,7 @@ namespace cql3 {
  */
 class tuples {
 public:
-    static shared_ptr<column_specification> component_spec_of(const column_specification& column, size_t component);
+    static lw_shared_ptr<column_specification> component_spec_of(const column_specification& column, size_t component);
 
     /**
      * A raw, literal tuple.  When prepared, this will become a Tuples.Value or Tuples.DelayedValue, depending
@@ -64,9 +64,9 @@ public:
         literal(std::vector<shared_ptr<raw>> elements)
                 : _elements(std::move(elements)) {
         }
-        virtual shared_ptr<term> prepare(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const override;
+        virtual shared_ptr<term> prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override;
 
-        virtual shared_ptr<term> prepare(database& db, const sstring& keyspace, const std::vector<shared_ptr<column_specification>>& receivers) const override;
+        virtual shared_ptr<term> prepare(database& db, const sstring& keyspace, const std::vector<lw_shared_ptr<column_specification>>& receivers) const override;
 
     private:
         void validate_assignable_to(database& db, const sstring& keyspace, const column_specification& receiver) const {
@@ -88,7 +88,7 @@ public:
             }
         }
     public:
-        virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const override {
+        virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override {
             try {
                 validate_assignable_to(db, keyspace, *receiver);
                 return assignment_testable::test_result::WEAKLY_ASSIGNABLE;
@@ -234,11 +234,11 @@ public:
     public:
         using abstract_marker::raw::raw;
 
-        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, const std::vector<shared_ptr<column_specification>>& receivers) const override {
+        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, const std::vector<lw_shared_ptr<column_specification>>& receivers) const override {
             return make_shared<tuples::marker>(_bind_index, make_receiver(receivers));
         }
 
-        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const override {
+        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override {
             throw std::runtime_error("Tuples.Raw.prepare() requires a list of receivers");
         }
 
@@ -250,7 +250,7 @@ public:
             return abstract_marker::raw::to_string();
         }
     private:
-        static ::shared_ptr<column_specification> make_receiver(const std::vector<shared_ptr<column_specification>>& receivers) {
+        static lw_shared_ptr<column_specification> make_receiver(const std::vector<lw_shared_ptr<column_specification>>& receivers) {
             std::vector<data_type> types;
             types.reserve(receivers.size());
             sstring in_name = "(";
@@ -265,7 +265,7 @@ public:
 
             auto identifier = ::make_shared<column_identifier>(in_name, true);
             auto type = tuple_type_impl::get_instance(types);
-            return ::make_shared<column_specification>(receivers.front()->ks_name, receivers.front()->cf_name, identifier, type);
+            return make_lw_shared<column_specification>(receivers.front()->ks_name, receivers.front()->cf_name, identifier, type);
         }
     };
 
@@ -276,11 +276,11 @@ public:
     public:
         using abstract_marker::raw::raw;
 
-        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, const std::vector<shared_ptr<column_specification>>& receivers) const override {
+        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, const std::vector<lw_shared_ptr<column_specification>>& receivers) const override {
             return make_shared<tuples::in_marker>(_bind_index, make_in_receiver(receivers));
         }
 
-        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, ::shared_ptr<column_specification> receiver) const override {
+        virtual ::shared_ptr<term> prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override {
             throw std::runtime_error("Tuples.INRaw.prepare() requires a list of receivers");
         }
 
@@ -292,7 +292,7 @@ public:
             return abstract_marker::raw::to_string();
         }
     private:
-        static ::shared_ptr<column_specification> make_in_receiver(const std::vector<shared_ptr<column_specification>>& receivers);
+        static lw_shared_ptr<column_specification> make_in_receiver(const std::vector<lw_shared_ptr<column_specification>>& receivers);
     };
 
     /**
@@ -300,7 +300,7 @@ public:
      */
     class marker : public abstract_marker {
     public:
-        marker(int32_t bind_index, ::shared_ptr<column_specification> receiver)
+        marker(int32_t bind_index, lw_shared_ptr<column_specification> receiver)
             : abstract_marker(bind_index, std::move(receiver))
         { }
 
@@ -329,7 +329,7 @@ public:
      */
     class in_marker : public abstract_marker {
     public:
-        in_marker(int32_t bind_index, ::shared_ptr<column_specification> receiver);
+        in_marker(int32_t bind_index, lw_shared_ptr<column_specification> receiver);
 
         virtual shared_ptr<terminal> bind(const query_options& options) override;
     };
diff --git a/cql3/type_cast.hh b/cql3/type_cast.hh
index 03f3f4a160..b1cea6c86a 100644
--- a/cql3/type_cast.hh
+++ b/cql3/type_cast.hh
@@ -52,7 +52,7 @@ public:
             : _type(std::move(type)), _term(std::move(term)) {
     }
 
-    virtual shared_ptr<term> prepare(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const override {
+    virtual shared_ptr<term> prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override {
         if (!is_assignable(_term->test_assignment(db, keyspace, casted_spec_of(db, keyspace, *receiver)))) {
             throw exceptions::invalid_request_exception(format("Cannot cast value {} to type {}", _term, _type));
         }
@@ -62,12 +62,12 @@ public:
         return _term->prepare(db, keyspace, receiver);
     }
 private:
-    shared_ptr<column_specification> casted_spec_of(database& db, const sstring& keyspace, const column_specification& receiver) const {
-        return ::make_shared<column_specification>(receiver.ks_name, receiver.cf_name,
+    lw_shared_ptr<column_specification> casted_spec_of(database& db, const sstring& keyspace, const column_specification& receiver) const {
+        return make_lw_shared<column_specification>(receiver.ks_name, receiver.cf_name,
                 ::make_shared<column_identifier>(to_string(), true), _type->prepare(db, keyspace).get_type());
     }
 public:
-    virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const override {
+    virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override {
         try {
             auto&& casted_type = _type->prepare(db, keyspace).get_type();
             if (receiver->type == casted_type) {
diff --git a/cql3/untyped_result_set.cc b/cql3/untyped_result_set.cc
index 4202beb579..4dd5b1ec96 100644
--- a/cql3/untyped_result_set.cc
+++ b/cql3/untyped_result_set.cc
@@ -51,11 +51,11 @@ cql3::untyped_result_set_row::untyped_result_set_row(const std::unordered_map<ss
     : _data(data)
 {}
 
-cql3::untyped_result_set_row::untyped_result_set_row(const std::vector<::shared_ptr<column_specification>>& columns, std::vector<bytes_opt> data)
+cql3::untyped_result_set_row::untyped_result_set_row(const std::vector<lw_shared_ptr<column_specification>>& columns, std::vector<bytes_opt> data)
 : _columns(columns)
 , _data([&columns, data = std::move(data)] () mutable {
     std::unordered_map<sstring, bytes_opt> tmp;
-    std::transform(columns.begin(), columns.end(), data.begin(), std::inserter(tmp, tmp.end()), [](::shared_ptr<column_specification> c, bytes_opt& d) {
+    std::transform(columns.begin(), columns.end(), data.begin(), std::inserter(tmp, tmp.end()), [](lw_shared_ptr<column_specification> c, bytes_opt& d) {
        return std::make_pair<sstring, bytes_opt>(c->name->to_string(), std::move(d));
     });
     return tmp;
diff --git a/cql3/untyped_result_set.hh b/cql3/untyped_result_set.hh
index ef20710154..665458521d 100644
--- a/cql3/untyped_result_set.hh
+++ b/cql3/untyped_result_set.hh
@@ -54,11 +54,11 @@ namespace cql3 {
 
 class untyped_result_set_row {
 private:
-    const std::vector<::shared_ptr<column_specification>> _columns;
+    const std::vector<lw_shared_ptr<column_specification>> _columns;
     const std::unordered_map<sstring, bytes_opt> _data;
 public:
     untyped_result_set_row(const std::unordered_map<sstring, bytes_opt>&);
-    untyped_result_set_row(const std::vector<::shared_ptr<column_specification>>&, std::vector<bytes_opt>);
+    untyped_result_set_row(const std::vector<lw_shared_ptr<column_specification>>&, std::vector<bytes_opt>);
     untyped_result_set_row(untyped_result_set_row&&) = default;
     untyped_result_set_row(const untyped_result_set_row&) = delete;
 
@@ -144,7 +144,7 @@ public:
         get_set_data<V>(name, std::inserter(res, res.end()), valtype);
         return res;
     }
-    const std::vector<::shared_ptr<column_specification>>& get_columns() const {
+    const std::vector<lw_shared_ptr<column_specification>>& get_columns() const {
         return _columns;
     }
 };
diff --git a/cql3/user_types.cc b/cql3/user_types.cc
index bd41f75205..756afd5f11 100644
--- a/cql3/user_types.cc
+++ b/cql3/user_types.cc
@@ -51,11 +51,11 @@
 
 namespace cql3 {
 
-shared_ptr<column_specification> user_types::field_spec_of(const column_specification& column, size_t field) {
+lw_shared_ptr<column_specification> user_types::field_spec_of(const column_specification& column, size_t field) {
     auto&& ut = static_pointer_cast<const user_type_impl>(column.type);
     auto&& name = ut->field_name(field);
     auto&& sname = sstring(reinterpret_cast<const char*>(name.data()), name.size());
-    return ::make_shared<column_specification>(
+    return make_lw_shared<column_specification>(
                                    column.ks_name,
                                    column.cf_name,
                                    ::make_shared<column_identifier>(column.name->to_string() + "." + sname, true),
@@ -66,7 +66,7 @@ user_types::literal::literal(elements_map_type entries)
         : _entries(std::move(entries)) {
 }
 
-shared_ptr<term> user_types::literal::prepare(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const {
+shared_ptr<term> user_types::literal::prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const {
     validate_assignable_to(db, keyspace, *receiver);
     auto&& ut = static_pointer_cast<const user_type_impl>(receiver->type);
     bool all_terminal = true;
@@ -128,7 +128,7 @@ void user_types::literal::validate_assignable_to(database& db, const sstring& ke
     }
 }
 
-assignment_testable::test_result user_types::literal::test_assignment(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const {
+assignment_testable::test_result user_types::literal::test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const {
     try {
         validate_assignable_to(db, keyspace, *receiver);
         return assignment_testable::test_result::WEAKLY_ASSIGNABLE;
diff --git a/cql3/user_types.hh b/cql3/user_types.hh
index d01b952051..6052889d0a 100644
--- a/cql3/user_types.hh
+++ b/cql3/user_types.hh
@@ -55,7 +55,7 @@ namespace cql3 {
 class user_types {
     user_types() = delete;
 public:
-    static shared_ptr<column_specification> field_spec_of(const column_specification& column, size_t field);
+    static lw_shared_ptr<column_specification> field_spec_of(const column_specification& column, size_t field);
 
     class literal : public term::raw {
     public:
@@ -63,11 +63,11 @@ public:
         elements_map_type _entries;
 
         literal(elements_map_type entries);
-        virtual shared_ptr<term> prepare(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const override;
+        virtual shared_ptr<term> prepare(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override;
     private:
         void validate_assignable_to(database& db, const sstring& keyspace, const column_specification& receiver) const;
     public:
-        virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, shared_ptr<column_specification> receiver) const override;
+        virtual assignment_testable::test_result test_assignment(database& db, const sstring& keyspace, lw_shared_ptr<column_specification> receiver) const override;
         virtual sstring assignment_testable_source_context() const override;
         virtual sstring to_string() const override;
     };
@@ -103,7 +103,7 @@ public:
 
     class marker : public abstract_marker {
     public:
-        marker(int32_t bind_index, ::shared_ptr<column_specification> receiver)
+        marker(int32_t bind_index, lw_shared_ptr<column_specification> receiver)
             : abstract_marker{bind_index, std::move(receiver)}
         {
             assert(_receiver->type->is_user_type());
diff --git a/cql3/variable_specifications.cc b/cql3/variable_specifications.cc
index d758fda9c6..302121b1bf 100644
--- a/cql3/variable_specifications.cc
+++ b/cql3/variable_specifications.cc
@@ -55,11 +55,11 @@ size_t variable_specifications::size() const {
     return _variable_names.size();
 }
 
-std::vector<::shared_ptr<column_specification>> variable_specifications::get_specifications() const & {
-    return std::vector<::shared_ptr<column_specification>>(_specs.begin(), _specs.end());
+std::vector<lw_shared_ptr<column_specification>> variable_specifications::get_specifications() const & {
+    return std::vector<lw_shared_ptr<column_specification>>(_specs.begin(), _specs.end());
 }
 
-std::vector<::shared_ptr<column_specification>> variable_specifications::get_specifications() && {
+std::vector<lw_shared_ptr<column_specification>> variable_specifications::get_specifications() && {
     return std::move(_specs);
 }
 
@@ -83,12 +83,12 @@ std::vector<uint16_t> variable_specifications::get_partition_key_bind_indexes(co
     return partition_key_positions;
 }
 
-void variable_specifications::add(int32_t bind_index, ::shared_ptr<column_specification> spec) {
+void variable_specifications::add(int32_t bind_index, lw_shared_ptr<column_specification> spec) {
     _target_columns[bind_index] = spec;
     auto name = _variable_names[bind_index];
     // Use the user name, if there is one
     if (name) {
-        spec = ::make_shared<column_specification>(spec->ks_name, spec->cf_name, name, spec->type);
+        spec = make_lw_shared<column_specification>(spec->ks_name, spec->cf_name, name, spec->type);
     }
     _specs[bind_index] = spec;
 }
diff --git a/cql3/variable_specifications.hh b/cql3/variable_specifications.hh
index 66dbb53fed..0f0c18c7ab 100644
--- a/cql3/variable_specifications.hh
+++ b/cql3/variable_specifications.hh
@@ -52,8 +52,8 @@ namespace cql3 {
 class variable_specifications final {
 private:
     std::vector<shared_ptr<column_identifier>> _variable_names;
-    std::vector<::shared_ptr<column_specification>> _specs;
-    std::vector<::shared_ptr<column_specification>> _target_columns;
+    std::vector<lw_shared_ptr<column_specification>> _specs;
+    std::vector<lw_shared_ptr<column_specification>> _target_columns;
 
 public:
 
@@ -68,13 +68,13 @@ public:
 
     size_t size() const;
 
-    std::vector<::shared_ptr<column_specification>> get_specifications() const &;
+    std::vector<lw_shared_ptr<column_specification>> get_specifications() const &;
 
-    std::vector<::shared_ptr<column_specification>> get_specifications() &&;
+    std::vector<lw_shared_ptr<column_specification>> get_specifications() &&;
 
     std::vector<uint16_t> get_partition_key_bind_indexes(const schema& schema) const;
 
-    void add(int32_t bind_index, ::shared_ptr<column_specification> spec);
+    void add(int32_t bind_index, lw_shared_ptr<column_specification> spec);
 
     void set_bound_variables(const std::vector<shared_ptr<column_identifier>>& bound_names);
 };
diff --git a/schema.cc b/schema.cc
index 1fa6bf4237..1dfeac9af1 100644
--- a/schema.cc
+++ b/schema.cc
@@ -144,10 +144,10 @@ bool schema::has_custom_partitioner() const {
     return _raw._partitioner.get().name() != default_partitioner_name;
 }
 
-::shared_ptr<cql3::column_specification>
+lw_shared_ptr<cql3::column_specification>
 schema::make_column_specification(const column_definition& def) {
     auto id = ::make_shared<cql3::column_identifier>(def.name(), column_name_type(def));
-    return ::make_shared<cql3::column_specification>(_raw._ks_name, _raw._cf_name, std::move(id), def.type);
+    return make_lw_shared<cql3::column_specification>(_raw._ks_name, _raw._cf_name, std::move(id), def.type);
 }
 
 v3_columns::v3_columns(std::vector<column_definition> cols, bool is_dense, bool is_compound)
@@ -194,7 +194,7 @@ v3_columns v3_columns::from_v2_schema(const schema& s) {
     for (column_definition& def : cols) {
         data_type name_type = def.is_static() ? static_column_name_type : utf8_type;
         auto id = ::make_shared<cql3::column_identifier>(def.name(), name_type);
-        def.column_specification = ::make_shared<cql3::column_specification>(s.ks_name(), s.cf_name(), std::move(id), def.type);
+        def.column_specification = make_lw_shared<cql3::column_specification>(s.ks_name(), s.cf_name(), std::move(id), def.type);
     }
 
     return v3_columns(std::move(cols), s.is_dense(), s.is_compound());
diff --git a/schema.hh b/schema.hh
index 596faba4ed..0df4a45583 100644
--- a/schema.hh
+++ b/schema.hh
@@ -315,7 +315,7 @@ public:
     ordinal_column_id ordinal_id;
 
     column_kind kind;
-    ::shared_ptr<cql3::column_specification> column_specification;
+    lw_shared_ptr<cql3::column_specification> column_specification;
 
     // NOTICE(sarna): This copy constructor is hand-written instead of default,
     // because it involves deep copying of the computation object.
@@ -685,7 +685,7 @@ public:
         data_type type;
     };
 private:
-    ::shared_ptr<cql3::column_specification> make_column_specification(const column_definition& def);
+    lw_shared_ptr<cql3::column_specification> make_column_specification(const column_definition& def);
     void rebuild();
     schema(const raw_schema&, std::optional<raw_view_info>);
 public:
diff --git a/transport/server.cc b/transport/server.cc
index a5155eec68..09718d57e3 100644
--- a/transport/server.cc
+++ b/transport/server.cc
@@ -1746,7 +1746,7 @@ void cql_server::response::write(const cql3::metadata& m, bool no_metadata) {
     }
 
     for (uint32_t i = 0; i < m.column_count(); ++i, ++names_i) {
-        ::shared_ptr<cql3::column_specification> name = *names_i;
+        lw_shared_ptr<cql3::column_specification> name = *names_i;
         if (!global_tables_spec) {
             write_string(name->ks_name);
             write_string(name->cf_name);
diff --git a/types.cc b/types.cc
index 79db5d7b3a..4711f0263a 100644
--- a/types.cc
+++ b/types.cc
@@ -502,19 +502,19 @@ empty_type_impl::empty_type_impl()
 logging::logger collection_type_impl::_logger("collection_type_impl");
 const size_t collection_type_impl::max_elements;
 
-shared_ptr<cql3::column_specification> collection_type_impl::make_collection_receiver(
+lw_shared_ptr<cql3::column_specification> collection_type_impl::make_collection_receiver(
         const cql3::column_specification& collection, bool is_key) const {
     struct visitor {
         const cql3::column_specification& collection;
         bool is_key;
-        shared_ptr<cql3::column_specification> operator()(const abstract_type&) { abort(); }
-        shared_ptr<cql3::column_specification> operator()(const list_type_impl&) {
+        lw_shared_ptr<cql3::column_specification> operator()(const abstract_type&) { abort(); }
+        lw_shared_ptr<cql3::column_specification> operator()(const list_type_impl&) {
             return cql3::lists::value_spec_of(collection);
         }
-        shared_ptr<cql3::column_specification> operator()(const map_type_impl&) {
+        lw_shared_ptr<cql3::column_specification> operator()(const map_type_impl&) {
             return is_key ? cql3::maps::key_spec_of(collection) : cql3::maps::value_spec_of(collection);
         }
-        shared_ptr<cql3::column_specification> operator()(const set_type_impl&) {
+        lw_shared_ptr<cql3::column_specification> operator()(const set_type_impl&) {
             return cql3::sets::value_spec_of(collection);
         }
     };
diff --git a/types.hh b/types.hh
index f34161cd93..c9ceece87d 100644
--- a/types.hh
+++ b/types.hh
@@ -64,7 +64,6 @@ class multiprecision_int;
 namespace cql3 {
 
 class cql3_type;
-class column_specification;
 
 }
 
diff --git a/types/collection.hh b/types/collection.hh
index efa7a7a7de..323e07f17f 100644
--- a/types/collection.hh
+++ b/types/collection.hh
@@ -30,6 +30,12 @@
 #include "utils/chunked_vector.hh"
 #include "schema_fwd.hh"
 
+namespace cql3 {
+
+class column_specification;
+
+}
+
 class collection_type_impl : public abstract_type {
     static logging::logger _logger;
 public:
@@ -43,7 +49,7 @@ public:
     bool is_multi_cell() const { return _is_multi_cell; }
     virtual data_type name_comparator() const = 0;
     virtual data_type value_comparator() const = 0;
-    shared_ptr<cql3::column_specification> make_collection_receiver(const cql3::column_specification& collection, bool is_key) const;
+    lw_shared_ptr<cql3::column_specification> make_collection_receiver(const cql3::column_specification& collection, bool is_key) const;
     virtual bool is_compatible_with_frozen(const collection_type_impl& previous) const = 0;
     virtual bool is_value_compatible_with_frozen(const collection_type_impl& previous) const = 0;
     template <typename BytesViewIterator>