diff --git a/cql3/attributes.cc b/cql3/attributes.cc
index b81bc3c3bb..84e80eca67 100644
--- a/cql3/attributes.cc
+++ b/cql3/attributes.cc
@@ -42,7 +42,7 @@ int64_t attributes::get_timestamp(int64_t now, const query_options& options) {
         return now;
     }
 
-    expr::constant tval = expr::evaluate(*_timestamp, options);
+    cql3::raw_value tval = expr::evaluate(*_timestamp, options);
     if (tval.is_null()) {
         throw exceptions::invalid_request_exception("Invalid null value of timestamp");
     }
@@ -60,7 +60,7 @@ int32_t attributes::get_time_to_live(const query_options& options) {
     if (!_time_to_live.has_value())
         return 0;
 
-    expr::constant tval = expr::evaluate(*_time_to_live, options);
+    cql3::raw_value tval = expr::evaluate(*_time_to_live, options);
     if (tval.is_null()) {
         throw exceptions::invalid_request_exception("Invalid null value of TTL");
     }
@@ -90,7 +90,7 @@ int32_t attributes::get_time_to_live(const query_options& options) {
 
 
 db::timeout_clock::duration attributes::get_timeout(const query_options& options) const {
-    expr::constant timeout = expr::evaluate(*_timeout, options);
+    cql3::raw_value timeout = expr::evaluate(*_timeout, options);
     if (timeout.is_null() || timeout.is_unset_value()) {
         throw exceptions::invalid_request_exception("Timeout value cannot be unset/null");
     }
diff --git a/cql3/column_condition.cc b/cql3/column_condition.cc
index df71e66c8d..ac4d7aefdc 100644
--- a/cql3/column_condition.cc
+++ b/cql3/column_condition.cc
@@ -137,7 +137,7 @@ bool column_condition::applies_to(const data_value* cell_value, const query_opti
         assert(cell_value->type()->is_collection());
         const collection_type_impl& cell_type = static_cast<const collection_type_impl&>(*cell_value->type());
 
-        expr::constant key_constant = expr::evaluate(*_collection_element, options);
+        cql3::raw_value key_constant = expr::evaluate(*_collection_element, options);
         cql3::raw_value_view key = key_constant.view();
         if (key.is_unset_value()) {
             throw exceptions::invalid_request_exception(
@@ -194,7 +194,7 @@ bool column_condition::applies_to(const data_value* cell_value, const query_opti
 
     if (is_compare(_op)) {
         // <, >, >=, <=, !=
-        expr::constant param = expr::evaluate(*_value, options);
+        cql3::raw_value param = expr::evaluate(*_value, options);
 
         if (param.is_unset_value()) {
             throw exceptions::invalid_request_exception("Invalid 'unset' value in condition");
@@ -241,11 +241,11 @@ bool column_condition::applies_to(const data_value* cell_value, const query_opti
     std::vector<bytes_opt> in_values;
 
     if (_value.has_value()) {
-        expr::constant lval = expr::evaluate(*_value, options);
+        cql3::raw_value lval = expr::evaluate(*_value, options);
         if (lval.is_null()) {
             throw exceptions::invalid_request_exception("Invalid null value for IN condition");
         }
-        for (const managed_bytes_opt& v : expr::get_elements(lval)) {
+        for (const managed_bytes_opt& v : expr::get_elements(lval, *type_of(*_value))) {
             if (v) {
                 in_values.push_back(to_bytes(*v));
             } else {
diff --git a/cql3/expr/expression.cc b/cql3/expr/expression.cc
index 9b1f3951e3..e7a1b8a0da 100644
--- a/cql3/expr/expression.cc
+++ b/cql3/expr/expression.cc
@@ -84,11 +84,11 @@ binary_operator::binary_operator(expression lhs, oper_t op, expression rhs, comp
 // Since column_identifier_raw is forward-declared in expression.hh, delay destructor instantiation here
 unresolved_identifier::~unresolved_identifier() = default;
 
-static constant evaluate(const bind_variable&, const evaluation_inputs&);
-static constant evaluate(const tuple_constructor&, const evaluation_inputs&);
-static constant evaluate(const collection_constructor&, const evaluation_inputs&);
-static constant evaluate(const usertype_constructor&, const evaluation_inputs&);
-static constant evaluate(const function_call&, const evaluation_inputs&);
+static cql3::raw_value evaluate(const bind_variable&, const evaluation_inputs&);
+static cql3::raw_value evaluate(const tuple_constructor&, const evaluation_inputs&);
+static cql3::raw_value evaluate(const collection_constructor&, const evaluation_inputs&);
+static cql3::raw_value evaluate(const usertype_constructor&, const evaluation_inputs&);
+static cql3::raw_value evaluate(const function_call&, const evaluation_inputs&);
 
 namespace {
 
@@ -178,12 +178,6 @@ get_value(const subscript& s, const evaluation_inputs& inputs) {
             format("Unsupported unset map key for column {}",
                 cdef->name_as_text()));
     }
-    if (key.type != key_type) {
-        // This can't happen, we always verify the index type earlier.
-        throw std::logic_error(
-            format("Tried to evaluate expression with wrong type for subscript of {}",
-                cdef->name_as_text()));
-    }
     if (col_type->is_map()) {
         const auto& data_map = value_cast<map_type_impl::native_type>(deserialized);
         const auto found = key.view().with_linearized([&] (bytes_view key_bv) {
@@ -213,7 +207,7 @@ bool equal(const expression& lhs, const managed_bytes_opt& rhs, const evaluation
     if (!rhs) {
         return false;
     }
-    const auto value = std::move(evaluate(lhs, inputs).value).to_managed_bytes_opt();
+    const auto value = evaluate(lhs, inputs).to_managed_bytes_opt();
     if (!value) {
         return false;
     }
@@ -222,16 +216,13 @@ bool equal(const expression& lhs, const managed_bytes_opt& rhs, const evaluation
 
 /// Convenience overload for expression.
 bool equal(const expression& lhs, const expression& rhs, const evaluation_inputs& inputs) {
-    return equal(lhs, evaluate(rhs, inputs).value.to_managed_bytes_opt(), inputs);
+    return equal(lhs, evaluate(rhs, inputs).to_managed_bytes_opt(), inputs);
 }
 
 /// True iff columns' values equal t.
 bool equal(const tuple_constructor& columns_tuple_lhs, const expression& t_rhs, const evaluation_inputs& inputs) {
-    const constant tup = evaluate(t_rhs, inputs);
-    if (!tup.type->is_tuple()) {
-        throw exceptions::invalid_request_exception("multi-column equality has right-hand side that isn't a tuple");
-    }
-    const auto& rhs = get_tuple_elements(tup);
+    const cql3::raw_value tup = evaluate(t_rhs, inputs);
+    const auto& rhs = get_tuple_elements(tup, *type_of(t_rhs));
     if (rhs.size() != columns_tuple_lhs.elements.size()) {
         throw exceptions::invalid_request_exception(
                 format("tuple equality size mismatch: {} elements on left-hand side, {} on right",
@@ -269,11 +260,11 @@ bool limits(const expression& col, oper_t op, const expression& rhs, const evalu
     if (!is_slice(op)) { // For EQ or NEQ, use equal().
         throw std::logic_error("limits() called on non-slice op");
     }
-    auto lhs = std::move(evaluate(col, inputs).value).to_managed_bytes_opt();
+    auto lhs = evaluate(col, inputs).to_managed_bytes_opt();
     if (!lhs) {
         return false;
     }
-    const auto b = evaluate(rhs, inputs).value.to_managed_bytes_opt();
+    const auto b = evaluate(rhs, inputs).to_managed_bytes_opt();
     return b ? limits(*lhs, op, *b, *type_of(col)) : false;
 }
 
@@ -283,12 +274,8 @@ bool limits(const tuple_constructor& columns_tuple, const oper_t op, const expre
     if (!is_slice(op)) { // For EQ or NEQ, use equal().
         throw std::logic_error("limits() called on non-slice op");
     }
-    const constant tup = evaluate(e, inputs);
-    if (!tup.type->is_tuple()) {
-        throw exceptions::invalid_request_exception(
-                "multi-column comparison has right-hand side that isn't a tuple");
-    }
-    const auto& rhs = get_tuple_elements(tup);
+    const cql3::raw_value tup = evaluate(e, inputs);
+    const auto& rhs = get_tuple_elements(tup, *type_of(e));
     if (rhs.size() != columns_tuple.elements.size()) {
         throw exceptions::invalid_request_exception(
                 format("tuple comparison size mismatch: {} elements on left-hand side, {} on right",
@@ -296,7 +283,7 @@ bool limits(const tuple_constructor& columns_tuple, const oper_t op, const expre
     }
     for (size_t i = 0; i < rhs.size(); ++i) {
         auto& cv = columns_tuple.elements[i];
-        auto lhs = std::move(evaluate(cv, inputs).value).to_managed_bytes_opt();
+        auto lhs = evaluate(cv, inputs).to_managed_bytes_opt();
         if (!lhs || !rhs[i]) {
             // CQL dictates that columns_tuple.elements[i] is a clustering column and non-null, but
             // let's not rely on grammar constraints that can be later relaxed.
@@ -456,13 +443,10 @@ bool like(const column_value& cv, const raw_value_view& pattern, const evaluatio
 
 /// True iff the column value is in the set defined by rhs.
 bool is_one_of(const expression& col, const expression& rhs, const evaluation_inputs& inputs) {
-    const constant in_list = evaluate(rhs, inputs);
+    const cql3::raw_value in_list = evaluate(rhs, inputs);
     statements::request_validations::check_false(
             in_list.is_null(), "Invalid null value for column {}", col);
 
-    if (!in_list.type->without_reversed().is_list()) {
-        throw std::logic_error("unexpected expression type in is_one_of(single column)");
-    }
     return boost::algorithm::any_of(get_list_elements(in_list), [&] (const managed_bytes_opt& b) {
         return equal(col, b, inputs);
     });
@@ -470,11 +454,8 @@ bool is_one_of(const expression& col, const expression& rhs, const evaluation_in
 
 /// True iff the tuple of column values is in the set defined by rhs.
 bool is_one_of(const tuple_constructor& tuple, const expression& rhs, const evaluation_inputs& inputs) {
-    constant in_list = evaluate(rhs, inputs);
-    if (!in_list.type->without_reversed().is_list()) {
-        throw std::logic_error("unexpected expression type in is_one_of(multi-column)");
-    }
-    return boost::algorithm::any_of(get_list_of_tuples_elements(in_list), [&] (const std::vector<managed_bytes_opt>& el) {
+    cql3::raw_value in_list = evaluate(rhs, inputs);
+    return boost::algorithm::any_of(get_list_of_tuples_elements(in_list, *type_of(rhs)), [&] (const std::vector<managed_bytes_opt>& el) {
         return boost::equal(tuple.elements, el, [&] (const expression& c, const managed_bytes_opt& b) {
             return equal(c, b, inputs);
         });
@@ -522,13 +503,13 @@ bool is_satisfied_by(const binary_operator& opr, const evaluation_inputs& inputs
                 } else if (is_slice(opr.op)) {
                     return limits(col, opr.op, opr.rhs, inputs);
                 } else if (opr.op == oper_t::CONTAINS) {
-                    constant val = evaluate(opr.rhs, inputs);
+                    cql3::raw_value val = evaluate(opr.rhs, inputs);
                     return contains(col, val.view(), inputs);
                 } else if (opr.op == oper_t::CONTAINS_KEY) {
-                    constant val = evaluate(opr.rhs, inputs);
+                    cql3::raw_value val = evaluate(opr.rhs, inputs);
                     return contains_key(col, val.view(), inputs);
                 } else if (opr.op == oper_t::LIKE) {
-                    constant val = evaluate(opr.rhs, inputs);
+                    cql3::raw_value val = evaluate(opr.rhs, inputs);
                     return like(col, val.view(), inputs);
                 } else if (opr.op == oper_t::IN) {
                     return is_one_of(col, opr.rhs, inputs);
@@ -696,14 +677,11 @@ const auto deref = boost::adaptors::transformed([] (const managed_bytes_opt& b)
 value_list get_IN_values(
         const expression& e, const query_options& options, const serialized_compare& comparator,
         sstring_view column_name) {
-    const constant in_list = evaluate(e, options);
+    const cql3::raw_value in_list = evaluate(e, options);
     if (in_list.is_unset_value()) {
         throw exceptions::invalid_request_exception(format("Invalid unset value for column {}", column_name));
     }
     statements::request_validations::check_false(in_list.is_null(), "Invalid null value for column {}", column_name);
-    if (!in_list.type->without_reversed().is_list()) {
-        throw std::logic_error(format("get_IN_values(single-column) on invalid expression {}", e));
-    }
     utils::chunked_vector<managed_bytes> list_elems = get_list_elements(in_list);
     return to_sorted_vector(std::move(list_elems) | non_null | deref, comparator);
 }
@@ -711,11 +689,8 @@ value_list get_IN_values(
 /// Returns possible values for k-th column from t, which must be RHS of IN.
 value_list get_IN_values(const expression& e, size_t k, const query_options& options,
                          const serialized_compare& comparator) {
-        const constant in_list = evaluate(e, options);
-    if (!in_list.type->without_reversed().is_list()) {
-        throw std::logic_error(format("get_IN_values(multi-column) on invalid expression {}", e));
-    }
-    const auto split_values = get_list_of_tuples_elements(in_list); // Need lvalue from which to make std::view.
+    const cql3::raw_value in_list = evaluate(e, options);
+    const auto split_values = get_list_of_tuples_elements(in_list, *type_of(e)); // Need lvalue from which to make std::view.
     const auto result_range = split_values
             | boost::adaptors::transformed([k] (const std::vector<managed_bytes_opt>& v) { return v[k]; }) | non_null | deref;
     return to_sorted_vector(std::move(result_range), comparator);
@@ -802,7 +777,7 @@ value_set possible_lhs_values(const column_definition* cdef, const expression& e
                                 return unbounded_value_set;
                             }
                             if (is_compare(oper.op)) {
-                                managed_bytes_opt val = evaluate(oper.rhs, options).value.to_managed_bytes_opt();
+                                managed_bytes_opt val = evaluate(oper.rhs, options).to_managed_bytes_opt();
                                 if (!val) {
                                     return empty_value_set; // All NULL comparisons fail; no column values match.
                                 }
@@ -828,7 +803,7 @@ value_set possible_lhs_values(const column_definition* cdef, const expression& e
                             const auto column_index_on_lhs = std::distance(tuple.elements.begin(), found);
                             if (is_compare(oper.op)) {
                                 // RHS must be a tuple due to upstream checks.
-                                managed_bytes_opt val = get_tuple_elements(evaluate(oper.rhs, options)).at(column_index_on_lhs);
+                                managed_bytes_opt val = get_tuple_elements(evaluate(oper.rhs, options), *type_of(oper.rhs)).at(column_index_on_lhs);
                                 if (!val) {
                                     return empty_value_set; // All NULL comparisons fail; no column values match.
                                 }
@@ -850,7 +825,7 @@ value_set possible_lhs_values(const column_definition* cdef, const expression& e
                             if (cdef) {
                                 return unbounded_value_set;
                             }
-                            const auto val = evaluate(oper.rhs, options).value.to_managed_bytes_opt();
+                            const auto val = evaluate(oper.rhs, options).to_managed_bytes_opt();
                             if (!val) {
                                 return empty_value_set; // All NULL comparisons fail; no token values match.
                             }
@@ -1108,7 +1083,7 @@ std::ostream& operator<<(std::ostream& os, const expression::printer& pr) {
                     } else if (opr.op == oper_t::IN && is<constant>(opr.rhs) && as<constant>(opr.rhs).type->without_reversed().is_list()) {
                         tuple_constructor rhs_tuple;
                         const list_type_impl* list_typ = dynamic_cast<const list_type_impl*>(&as<constant>(opr.rhs).type->without_reversed());
-                        for (const managed_bytes& elem : get_list_elements(as<constant>(opr.rhs))) {
+                        for (const managed_bytes& elem : get_list_elements(as<constant>(opr.rhs).value)) {
                             rhs_tuple.elements.push_back(constant(raw_value::make_value(elem), list_typ->get_elements_type()));
                         }
                         os << to_printer(opr.lhs) << ' ' << opr.op << ' ' << to_printer(rhs_tuple);
@@ -1647,48 +1622,47 @@ std::optional<bool> get_bool_value(const constant& constant_val) {
     return constant_val.view().deserialize<bool>(*boolean_type);
 }
 
-constant evaluate(const expression& e, const evaluation_inputs& inputs) {
+cql3::raw_value evaluate(const expression& e, const evaluation_inputs& inputs) {
     return expr::visit(overloaded_functor {
-        [](const binary_operator&) -> constant {
+        [](const binary_operator&) -> cql3::raw_value {
             on_internal_error(expr_logger, "Can't evaluate a binary_operator");
         },
-        [](const conjunction&) -> constant {
+        [](const conjunction&) -> cql3::raw_value {
             on_internal_error(expr_logger, "Can't evaluate a conjunction");
         },
-        [](const token&) -> constant {
+        [](const token&) -> cql3::raw_value {
             on_internal_error(expr_logger, "Can't evaluate token");
         },
-        [](const unresolved_identifier&) -> constant {
+        [](const unresolved_identifier&) -> cql3::raw_value {
             on_internal_error(expr_logger, "Can't evaluate unresolved_identifier");
         },
-        [](const column_mutation_attribute&) -> constant {
+        [](const column_mutation_attribute&) -> cql3::raw_value {
             on_internal_error(expr_logger, "Can't evaluate a column_mutation_attribute");
         },
-        [&](const cast& c) -> constant {
+        [&](const cast& c) -> cql3::raw_value {
             auto ret = evaluate(c.arg, inputs);
             auto type = std::get_if<data_type>(&c.type);
             if (!type) {
                 on_internal_error(expr_logger, "attempting to evaluate an unprepared cast");
             }
-            ret.type = *type;
             return ret;
         },
-        [](const field_selection&) -> constant {
+        [](const field_selection&) -> cql3::raw_value {
             on_internal_error(expr_logger, "Can't evaluate a field_selection");
         },
 
-        [&](const column_value& cv) -> constant {
-            return constant(raw_value::make_value(get_value(cv, inputs)), cv.col->type);
+        [&](const column_value& cv) -> cql3::raw_value {
+            return raw_value::make_value(get_value(cv, inputs));
         },
-        [&](const subscript& s) -> constant {
-            return constant(raw_value::make_value(get_value(s, inputs)), s.type);
+        [&](const subscript& s) -> cql3::raw_value {
+            return raw_value::make_value(get_value(s, inputs));
         },
-        [](const untyped_constant&) -> constant {
+        [](const untyped_constant&) -> cql3::raw_value {
             on_internal_error(expr_logger, "Can't evaluate a untyped_constant ");
         },
 
-        [](const null&) { return constant::make_null(); },
-        [](const constant& c) { return c; },
+        [](const null&) { return cql3::raw_value::make_null(); },
+        [](const constant& c) { return c.value; },
         [&](const bind_variable& bind_var) { return evaluate(bind_var, inputs); },
         [&](const tuple_constructor& tup) { return evaluate(tup, inputs); },
         [&](const collection_constructor& col) { return evaluate(col, inputs); },
@@ -1697,7 +1671,7 @@ constant evaluate(const expression& e, const evaluation_inputs& inputs) {
     }, e);
 }
 
-constant evaluate(const expression& e, const query_options& options) {
+cql3::raw_value evaluate(const expression& e, const query_options& options) {
     return evaluate(e, evaluation_inputs{.options = &options});
 }
 
@@ -1790,7 +1764,7 @@ static managed_bytes reserialize_value(View value_bytes,
         fmt::format("Reserializing type that shouldn't need reserialization: {}", type.name()));
 }
 
-static constant evaluate(const bind_variable& bind_var, const evaluation_inputs& inputs) {
+static cql3::raw_value evaluate(const bind_variable& bind_var, const evaluation_inputs& inputs) {
     if (bind_var.receiver.get() == nullptr) {
         on_internal_error(expr_logger,
             "evaluate(bind_variable) called with nullptr receiver, should be prepared first");
@@ -1799,11 +1773,11 @@ static constant evaluate(const bind_variable& bind_var, const evaluation_inputs&
     cql3::raw_value_view value = inputs.options->get_value_at(bind_var.bind_index);
 
     if (value.is_null()) {
-        return constant::make_null(bind_var.receiver->type);
+        return cql3::raw_value::make_null();
     }
 
     if (value.is_unset_value()) {
-        return constant::make_unset_value(bind_var.receiver->type);
+        return cql3::raw_value::make_unset_value();
     }
 
     const abstract_type& value_type = bind_var.receiver->type->without_reversed();
@@ -1819,13 +1793,13 @@ static constant evaluate(const bind_variable& bind_var, const evaluation_inputs&
             return reserialize_value(value_bytes, value_type, inputs.options->get_cql_serialization_format());
         });
 
-        return constant(raw_value::make_value(std::move(new_value)), bind_var.receiver->type);
+        return raw_value::make_value(std::move(new_value));
     }
 
-    return constant(raw_value::make_value(value), bind_var.receiver->type);
+    return raw_value::make_value(value);
 }
 
-static constant evaluate(const tuple_constructor& tuple, const evaluation_inputs& inputs) {
+static cql3::raw_value evaluate(const tuple_constructor& tuple, const evaluation_inputs& inputs) {
     if (tuple.type.get() == nullptr) {
         on_internal_error(expr_logger,
             "evaluate(tuple_constructor) called with nullptr type, should be prepared first");
@@ -1835,18 +1809,18 @@ static constant evaluate(const tuple_constructor& tuple, const evaluation_inputs
     tuple_elements.reserve(tuple.elements.size());
 
     for (size_t i = 0; i < tuple.elements.size(); i++) {
-        constant elem_val = evaluate(tuple.elements[i], inputs);
+        cql3::raw_value elem_val = evaluate(tuple.elements[i], inputs);
         if (elem_val.is_unset_value()) {
             throw exceptions::invalid_request_exception(format("Invalid unset value for tuple field number {:d}", i));
         }
 
-        tuple_elements.emplace_back(std::move(elem_val.value).to_managed_bytes_opt());
+        tuple_elements.emplace_back(std::move(elem_val).to_managed_bytes_opt());
     }
 
     cql3::raw_value raw_val =
         cql3::raw_value::make_value(tuple_type_impl::build_value_fragmented(std::move(tuple_elements)));
 
-    return constant(std::move(raw_val), tuple.type);
+    return raw_val;
 }
 
 // Range of managed_bytes
@@ -1873,14 +1847,14 @@ static managed_bytes serialize_listlike(const Range& elements, const char* colle
     );
 }
 
-static constant evaluate_list(const collection_constructor& collection,
+static cql3::raw_value evaluate_list(const collection_constructor& collection,
                               const evaluation_inputs& inputs,
                               bool skip_null = false) {
     std::vector<managed_bytes> evaluated_elements;
     evaluated_elements.reserve(collection.elements.size());
 
     for (const expression& element : collection.elements) {
-        constant evaluated_element = evaluate(element, inputs);
+        cql3::raw_value evaluated_element = evaluate(element, inputs);
 
         if (evaluated_element.is_unset_value()) {
             throw exceptions::invalid_request_exception("unset value is not supported inside collections");
@@ -1894,19 +1868,19 @@ static constant evaluate_list(const collection_constructor& collection,
             throw exceptions::invalid_request_exception("null is not supported inside collections");
         }
 
-        evaluated_elements.emplace_back(std::move(evaluated_element).value.to_managed_bytes());
+        evaluated_elements.emplace_back(std::move(evaluated_element).to_managed_bytes());
     }
 
     managed_bytes collection_bytes = serialize_listlike(evaluated_elements, "List");
-    return constant(raw_value::make_value(std::move(collection_bytes)), collection.type);
+    return raw_value::make_value(std::move(collection_bytes));
 }
 
-static constant evaluate_set(const collection_constructor& collection, const evaluation_inputs& inputs) {
+static cql3::raw_value evaluate_set(const collection_constructor& collection, const evaluation_inputs& inputs) {
     const set_type_impl& stype = dynamic_cast<const set_type_impl&>(collection.type->without_reversed());
     std::set<managed_bytes, serialized_compare> evaluated_elements(stype.get_elements_type()->as_less_comparator());
 
     for (const expression& element : collection.elements) {
-        constant evaluated_element = evaluate(element, inputs);
+        cql3::raw_value evaluated_element = evaluate(element, inputs);
 
         if (evaluated_element.is_null()) {
             throw exceptions::invalid_request_exception("null is not supported inside collections");
@@ -1926,21 +1900,21 @@ static constant evaluate_set(const collection_constructor& collection, const eva
                                                 evaluated_element.view().size_bytes()));
         }
 
-        evaluated_elements.emplace(std::move(evaluated_element).value.to_managed_bytes());
+        evaluated_elements.emplace(std::move(evaluated_element).to_managed_bytes());
     }
 
     managed_bytes collection_bytes = serialize_listlike(evaluated_elements, "Set");
-    return constant(raw_value::make_value(std::move(collection_bytes)), collection.type);
+    return raw_value::make_value(std::move(collection_bytes));
 }
 
-static constant evaluate_map(const collection_constructor& collection, const evaluation_inputs& inputs) {
+static cql3::raw_value evaluate_map(const collection_constructor& collection, const evaluation_inputs& inputs) {
     const map_type_impl& mtype = dynamic_cast<const map_type_impl&>(collection.type->without_reversed());
     std::map<managed_bytes, managed_bytes, serialized_compare> evaluated_elements(mtype.get_keys_type()->as_less_comparator());
 
     for (const expression& element : collection.elements) {
         if (auto tuple = expr::as_if<tuple_constructor>(&element)) {
-            constant key = evaluate(tuple->elements.at(0), inputs);
-            constant value = evaluate(tuple->elements.at(1), inputs);
+            cql3::raw_value key = evaluate(tuple->elements.at(0), inputs);
+            cql3::raw_value value = evaluate(tuple->elements.at(1), inputs);
 
             if (key.is_null() || value.is_null()) {
                 throw exceptions::invalid_request_exception("null is not supported inside collections");
@@ -1960,11 +1934,11 @@ static constant evaluate_map(const collection_constructor& collection, const eva
                                                    key.view().size_bytes()));
             }
 
-            evaluated_elements.emplace(std::move(key.value).to_managed_bytes(),
-                                       std::move(value.value).to_managed_bytes());
+            evaluated_elements.emplace(std::move(key).to_managed_bytes(),
+                                       std::move(value).to_managed_bytes());
         } else {
-            constant pair = evaluate(element, inputs);
-            std::vector<managed_bytes_opt> map_pair = get_tuple_elements(pair);
+            cql3::raw_value pair = evaluate(element, inputs);
+            std::vector<managed_bytes_opt> map_pair = get_tuple_elements(pair, *type_of(element));
 
             if (!map_pair.at(0).has_value() || !map_pair.at(1).has_value()) {
                 throw exceptions::invalid_request_exception("null is not supported inside collections");
@@ -1975,10 +1949,10 @@ static constant evaluate_map(const collection_constructor& collection, const eva
     }
 
     managed_bytes serialized_map = map_type_impl::serialize_to_managed_bytes(evaluated_elements);
-    return constant(raw_value::make_value(std::move(serialized_map)), collection.type);
+    return raw_value::make_value(std::move(serialized_map));
 }
 
-static constant evaluate(const collection_constructor& collection, const evaluation_inputs& inputs) {
+static cql3::raw_value evaluate(const collection_constructor& collection, const evaluation_inputs& inputs) {
     if (collection.type.get() == nullptr) {
         on_internal_error(expr_logger,
             "evaluate(collection_constructor) called with nullptr type, should be prepared first");
@@ -1997,7 +1971,7 @@ static constant evaluate(const collection_constructor& collection, const evaluat
     std::abort();
 }
 
-static constant evaluate(const usertype_constructor& user_val, const evaluation_inputs& inputs) {
+static cql3::raw_value evaluate(const usertype_constructor& user_val, const evaluation_inputs& inputs) {
     if (user_val.type.get() == nullptr) {
         on_internal_error(expr_logger,
             "evaluate(usertype_constructor) called with nullptr type, should be prepared first");
@@ -2022,20 +1996,20 @@ static constant evaluate(const usertype_constructor& user_val, const evaluation_
                 utype.field_name_as_string(i)));
         }
 
-        constant field_val = evaluate(cur_field->second, inputs);
+        cql3::raw_value field_val = evaluate(cur_field->second, inputs);
         if (field_val.is_unset_value()) {
             throw exceptions::invalid_request_exception(format(
                 "Invalid unset value for field '{}' of user defined type ", utype.field_name_as_string(i)));
         }
 
-        field_values.emplace_back(std::move(field_val.value).to_managed_bytes_opt());
+        field_values.emplace_back(std::move(field_val).to_managed_bytes_opt());
     }
 
     raw_value val_bytes = cql3::raw_value::make_value(tuple_type_impl::build_value_fragmented(field_values));
-    return constant(std::move(val_bytes), std::move(user_val.type));
+    return val_bytes;
 }
 
-static constant evaluate(const function_call& fun_call, const evaluation_inputs& inputs) {
+static cql3::raw_value evaluate(const function_call& fun_call, const evaluation_inputs& inputs) {
     const shared_ptr<functions::function>* fun = std::get_if<shared_ptr<functions::function>>(&fun_call.func);
     if (fun == nullptr) {
         throw std::runtime_error("Can't evaluate function call with name only, should be prepared earlier");
@@ -2051,12 +2025,12 @@ static constant evaluate(const function_call& fun_call, const evaluation_inputs&
     arguments.reserve(fun_call.args.size());
 
     for (const expression& arg : fun_call.args) {
-        constant arg_val = evaluate(arg, inputs);
+        cql3::raw_value arg_val = evaluate(arg, inputs);
         if (arg_val.is_null_or_unset()) {
             throw exceptions::invalid_request_exception(format("Invalid null or unset value for argument to {}", *scalar_fun));
         }
 
-        arguments.emplace_back(to_bytes_opt(std::move(arg_val.value)));
+        arguments.emplace_back(to_bytes_opt(std::move(arg_val)));
     }
 
     bool has_cache_id = fun_call.lwt_cache_id.get() != nullptr && fun_call.lwt_cache_id->has_value();
@@ -2064,7 +2038,7 @@ static constant evaluate(const function_call& fun_call, const evaluation_inputs&
         computed_function_values::mapped_type* cached_value =
             inputs.options->find_cached_pk_function_call(**fun_call.lwt_cache_id);
         if (cached_value != nullptr) {
-            return constant(raw_value::make_value(*cached_value), scalar_fun->return_type());
+            return raw_value::make_value(*cached_value);
         }
     }
 
@@ -2075,7 +2049,7 @@ static constant evaluate(const function_call& fun_call, const evaluation_inputs&
     }
 
     if (!result.has_value()) {
-        return constant::make_null(scalar_fun->return_type());
+        return cql3::raw_value::make_null();
     }
 
     try {
@@ -2087,18 +2061,11 @@ static constant evaluate(const function_call& fun_call, const evaluation_inputs&
                                        ));
     }
 
-    return constant(raw_value::make_value(std::move(*result)), scalar_fun->return_type());
+    return raw_value::make_value(std::move(*result));
 }
 
-static void ensure_can_get_value_elements(const constant& val,
-                                          abstract_type::kind expected_type_kind,
+static void ensure_can_get_value_elements(const cql3::raw_value& val,
                                           const char* caller_name) {
-    const abstract_type& val_type = val.type->without_reversed();
-
-    if (val_type.get_kind() != expected_type_kind) {
-        on_internal_error(expr_logger, fmt::format("{} called with wrong type: {}", caller_name, val_type.name()));
-    }
-
     if (val.is_null()) {
         on_internal_error(expr_logger, fmt::format("{} called with null value", caller_name));
     }
@@ -2108,44 +2075,44 @@ static void ensure_can_get_value_elements(const constant& val,
     }
 }
 
-utils::chunked_vector<managed_bytes> get_list_elements(const constant& val) {
-    ensure_can_get_value_elements(val, abstract_type::kind::list, "expr::get_list_elements");
+utils::chunked_vector<managed_bytes> get_list_elements(const cql3::raw_value& val) {
+    ensure_can_get_value_elements(val, "expr::get_list_elements");
 
     return val.view().with_value([](const FragmentedView auto& value_bytes) {
         return partially_deserialize_listlike(value_bytes, cql_serialization_format::internal());
     });
 }
 
-utils::chunked_vector<managed_bytes> get_set_elements(const constant& val) {
-    ensure_can_get_value_elements(val, abstract_type::kind::set, "expr::get_set_elements");
+utils::chunked_vector<managed_bytes> get_set_elements(const cql3::raw_value& val) {
+    ensure_can_get_value_elements(val, "expr::get_set_elements");
 
     return val.view().with_value([](const FragmentedView auto& value_bytes) {
         return partially_deserialize_listlike(value_bytes, cql_serialization_format::internal());
     });
 }
 
-std::vector<std::pair<managed_bytes, managed_bytes>> get_map_elements(const constant& val) {
-    ensure_can_get_value_elements(val, abstract_type::kind::map, "expr::get_map_elements");
+std::vector<std::pair<managed_bytes, managed_bytes>> get_map_elements(const cql3::raw_value& val) {
+    ensure_can_get_value_elements(val, "expr::get_map_elements");
 
     return val.view().with_value([](const FragmentedView auto& value_bytes) {
         return partially_deserialize_map(value_bytes, cql_serialization_format::internal());
     });
 }
 
-std::vector<managed_bytes_opt> get_tuple_elements(const constant& val) {
-    ensure_can_get_value_elements(val, abstract_type::kind::tuple, "expr::get_tuple_elements");
+std::vector<managed_bytes_opt> get_tuple_elements(const cql3::raw_value& val, const abstract_type& type) {
+    ensure_can_get_value_elements(val, "expr::get_tuple_elements");
 
     return val.view().with_value([&](const FragmentedView auto& value_bytes) {
-        const tuple_type_impl& ttype = static_cast<const tuple_type_impl&>(val.type->without_reversed());
+        const tuple_type_impl& ttype = static_cast<const tuple_type_impl&>(type.without_reversed());
         return ttype.split_fragmented(value_bytes);
     });
 }
 
-std::vector<managed_bytes_opt> get_user_type_elements(const constant& val) {
-    ensure_can_get_value_elements(val, abstract_type::kind::user, "expr::get_user_type_elements");
+std::vector<managed_bytes_opt> get_user_type_elements(const cql3::raw_value& val, const abstract_type& type) {
+    ensure_can_get_value_elements(val, "expr::get_user_type_elements");
 
     return val.view().with_value([&](const FragmentedView auto& value_bytes) {
-        const user_type_impl& utype = static_cast<const user_type_impl&>(val.type->without_reversed());
+        const user_type_impl& utype = static_cast<const user_type_impl&>(type.without_reversed());
         return utype.split_fragmented(value_bytes);
     });
 }
@@ -2155,8 +2122,8 @@ static std::vector<managed_bytes_opt> convert_listlike(utils::chunked_vector<man
                                           std::make_move_iterator(elements.end()));
 }
 
-std::vector<managed_bytes_opt> get_elements(const constant& val) {
-    const abstract_type& val_type = val.type->without_reversed();
+std::vector<managed_bytes_opt> get_elements(const cql3::raw_value& val, const abstract_type& type) {
+    const abstract_type& val_type = type.without_reversed();
 
     switch (val_type.get_kind()) {
         case abstract_type::kind::list:
@@ -2166,19 +2133,19 @@ std::vector<managed_bytes_opt> get_elements(const constant& val) {
             return convert_listlike(get_set_elements(val));
 
         case abstract_type::kind::tuple:
-            return get_tuple_elements(val);
+            return get_tuple_elements(val, type);
 
         case abstract_type::kind::user:
-            return get_user_type_elements(val);
+            return get_user_type_elements(val, type);
 
         default:
-            on_internal_error(expr_logger, fmt::format("expr::get_elements called on bad type: {}", val_type.name()));
+            on_internal_error(expr_logger, fmt::format("expr::get_elements called on bad type: {}", type.name()));
     }
 }
 
-utils::chunked_vector<std::vector<managed_bytes_opt>> get_list_of_tuples_elements(const constant& val) {
+utils::chunked_vector<std::vector<managed_bytes_opt>> get_list_of_tuples_elements(const cql3::raw_value& val, const abstract_type& type) {
     utils::chunked_vector<managed_bytes> elements = get_list_elements(val);
-    const list_type_impl& list_typ = dynamic_cast<const list_type_impl&>(val.type->without_reversed());
+    const list_type_impl& list_typ = dynamic_cast<const list_type_impl&>(type.without_reversed());
     const tuple_type_impl& tuple_typ = dynamic_cast<const tuple_type_impl&>(*list_typ.get_elements_type());
 
     utils::chunked_vector<std::vector<managed_bytes_opt>> tuples_list;
diff --git a/cql3/expr/expression.hh b/cql3/expr/expression.hh
index 34f118aa89..ff3149d003 100644
--- a/cql3/expr/expression.hh
+++ b/cql3/expr/expression.hh
@@ -673,22 +673,23 @@ data_type type_of(const expression& e);
 
 // Takes a prepared expression and calculates its value.
 // Evaluates bound values, calls functions and returns just the bytes and type.
-constant evaluate(const expression& e, const evaluation_inputs&);
+cql3::raw_value evaluate(const expression& e, const evaluation_inputs&);
 
-constant evaluate(const expression& e, const query_options&);
+cql3::raw_value evaluate(const expression& e, const query_options&);
 
-utils::chunked_vector<managed_bytes> get_list_elements(const constant&);
-utils::chunked_vector<managed_bytes> get_set_elements(const constant&);
-std::vector<managed_bytes_opt> get_tuple_elements(const constant&);
-std::vector<managed_bytes_opt> get_user_type_elements(const constant&);
-std::vector<std::pair<managed_bytes, managed_bytes>> get_map_elements(const constant&);
+utils::chunked_vector<managed_bytes> get_list_elements(const cql3::raw_value&);
+utils::chunked_vector<managed_bytes> get_set_elements(const cql3::raw_value&);
+std::vector<managed_bytes_opt> get_tuple_elements(const cql3::raw_value&, const abstract_type& type);
+std::vector<managed_bytes_opt> get_user_type_elements(const cql3::raw_value&, const abstract_type& type);
+std::vector<std::pair<managed_bytes, managed_bytes>> get_map_elements(const cql3::raw_value&);
 
 // Gets the elements of a constant which can be a list, set, tuple or user type
-std::vector<managed_bytes_opt> get_elements(const constant&);
+std::vector<managed_bytes_opt> get_elements(const cql3::raw_value&, const abstract_type& type);
 
 // Get elements of list<tuple<>> as vector<vector<managed_bytes_opt>
 // It is useful with IN restrictions like (a, b) IN [(1, 2), (3, 4)].
-utils::chunked_vector<std::vector<managed_bytes_opt>> get_list_of_tuples_elements(const constant&);
+// `type` parameter refers to the list<tuple<>> type.
+utils::chunked_vector<std::vector<managed_bytes_opt>> get_list_of_tuples_elements(const cql3::raw_value&, const abstract_type& type);
 
 // Retrieves information needed in prepare_context.
 // Collects the column specification for the bind variables in this expression.
diff --git a/cql3/expr/prepare_expr.cc b/cql3/expr/prepare_expr.cc
index 22a828356d..62f8892cc8 100644
--- a/cql3/expr/prepare_expr.cc
+++ b/cql3/expr/prepare_expr.cc
@@ -152,7 +152,7 @@ usertype_constructor_prepare_expression(const usertype_constructor& u, data_dict
     };
 
     if (all_terminal) {
-        return evaluate(value, query_options::DEFAULT);
+        return constant(evaluate(value, query_options::DEFAULT), value.type);
     } else {
         return value;
     }
@@ -274,7 +274,7 @@ map_prepare_expression(const collection_constructor& c, data_dictionary::databas
         .type = receiver->type
     };
     if (all_terminal) {
-        return evaluate(map_value, query_options::DEFAULT);
+        return constant(evaluate(map_value, query_options::DEFAULT), map_value.type);
     } else {
         return map_value;
     }
@@ -357,7 +357,7 @@ set_prepare_expression(const collection_constructor& c, data_dictionary::databas
                 .elements = {},
                 .type = receiver->type
             };
-            return expr::evaluate(map_value, query_options::DEFAULT);
+            return constant(expr::evaluate(map_value, query_options::DEFAULT), map_value.type);
         }
     }
 
@@ -383,7 +383,7 @@ set_prepare_expression(const collection_constructor& c, data_dictionary::databas
     };
     
     if (all_terminal) {
-        return evaluate(value, query_options::DEFAULT);
+        return constant(evaluate(value, query_options::DEFAULT), value.type);
     } else {
         return value;
     }
@@ -465,7 +465,7 @@ list_prepare_expression(const collection_constructor& c, data_dictionary::databa
         .type = receiver->type
     };
     if (all_terminal) {
-        return evaluate(value, query_options::DEFAULT);
+        return constant(evaluate(value, query_options::DEFAULT), value.type);
     } else {
         return value;
     }
@@ -549,7 +549,7 @@ tuple_constructor_prepare_nontuple(const tuple_constructor& tc, data_dictionary:
         .type = std::move(type),
     };
     if (all_terminal) {
-        return evaluate(value, query_options::DEFAULT);
+        return constant(evaluate(value, query_options::DEFAULT), value.type);
     } else {
         return value;
     }
@@ -830,7 +830,7 @@ prepare_function_call(const expr::function_call& fc, data_dictionary::database d
         .lwt_cache_id = fc.lwt_cache_id
     };
     if (all_terminal && scalar_fun->is_pure()) {
-        return expr::evaluate(fun_call, query_options::DEFAULT);
+        return constant(expr::evaluate(fun_call, query_options::DEFAULT), fun->return_type());
     } else {
         return fun_call;
     }
diff --git a/cql3/expr/to_restriction.cc b/cql3/expr/to_restriction.cc
index c0a58499b3..5352515f72 100644
--- a/cql3/expr/to_restriction.cc
+++ b/cql3/expr/to_restriction.cc
@@ -220,7 +220,7 @@ void validate_multi_column_relation(const std::vector<const column_definition*>&
             const list_type_impl* list_type = dynamic_cast<const list_type_impl*>(&rhs_list->type->without_reversed());
             const data_type& elements_type = list_type->get_elements_type();
 
-            utils::chunked_vector<managed_bytes> raw_elems = get_list_elements(*rhs_list);
+            utils::chunked_vector<managed_bytes> raw_elems = get_list_elements(rhs_list->value);
             std::vector<expression> list_elems;
             list_elems.reserve(raw_elems.size());
 
diff --git a/cql3/lists.cc b/cql3/lists.cc
index 0f6b030dd1..b554f8272a 100644
--- a/cql3/lists.cc
+++ b/cql3/lists.cc
@@ -36,7 +36,7 @@ lists::setter::execute(mutation& m, const clustering_key_prefix& prefix, const u
 }
 
 void
-lists::setter::execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params, const column_definition& column, const expr::constant& value) {
+lists::setter::execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params, const column_definition& column, const cql3::raw_value& value) {
     if (value.is_unset_value()) {
         return;
     }
@@ -132,10 +132,10 @@ lists::setter_by_uuid::execute(mutation& m, const clustering_key_prefix& prefix,
     mut.cells.reserve(1);
 
     if (value.is_null()) {
-        mut.cells.emplace_back(std::move(index.value).to_bytes(), params.make_dead_cell());
+        mut.cells.emplace_back(std::move(index).to_bytes(), params.make_dead_cell());
     } else {
         mut.cells.emplace_back(
-                    std::move(index.value).to_bytes(),
+                    std::move(index).to_bytes(),
                     params.make_cell(*ltype->value_comparator(), value.view(), atomic_cell::collection_member::yes));
     }
 
@@ -144,7 +144,7 @@ lists::setter_by_uuid::execute(mutation& m, const clustering_key_prefix& prefix,
 
 void
 lists::appender::execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) {
-    const expr::constant value = expr::evaluate(*_e, params._options);
+    const cql3::raw_value value = expr::evaluate(*_e, params._options);
     if (value.is_unset_value()) {
         return;
     }
@@ -153,7 +153,7 @@ lists::appender::execute(mutation& m, const clustering_key_prefix& prefix, const
 }
 
 void
-lists::do_append(const expr::constant& list_value,
+lists::do_append(const cql3::raw_value& list_value,
         mutation& m,
         const clustering_key_prefix& prefix,
         const column_definition& column,
@@ -198,11 +198,10 @@ lists::do_append(const expr::constant& list_value,
 void
 lists::prepender::execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) {
     assert(column.type->is_multi_cell()); // "Attempted to prepend to a frozen list";
-    expr::constant lvalue = expr::evaluate(*_e, params._options);
+    cql3::raw_value lvalue = expr::evaluate(*_e, params._options);
     if (lvalue.is_null_or_unset()) {
         return;
     }
-    assert(lvalue.type->is_list());
 
     // For prepend we need to be able to generate a unique but decreasing
     // timeuuid. We achieve that by by using a time in the past which
@@ -254,7 +253,7 @@ lists::discarder::execute(mutation& m, const clustering_key_prefix& prefix, cons
 
     auto&& existing_list = params.get_prefetched_list(m.key(), prefix, column);
     // We want to call bind before possibly returning to reject queries where the value provided is not a list.
-    expr::constant lvalue = expr::evaluate(*_e, params._options);
+    cql3::raw_value lvalue = expr::evaluate(*_e, params._options);
 
     if (!existing_list) {
         return;
@@ -270,8 +269,6 @@ lists::discarder::execute(mutation& m, const clustering_key_prefix& prefix, cons
         return;
     }
 
-    assert(lvalue.type->is_list());
-
     auto ltype = static_cast<const list_type_impl*>(column.type.get());
 
     // Note: below, we will call 'contains' on this toDiscard list for each element of existingList.
@@ -303,7 +300,7 @@ lists::discarder_by_index::requires_read() const {
 void
 lists::discarder_by_index::execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) {
     assert(column.type->is_multi_cell()); // "Attempted to delete an item by index from a frozen list";
-    expr::constant index = expr::evaluate(*_e, params._options);
+    cql3::raw_value index = expr::evaluate(*_e, params._options);
     if (index.is_null()) {
         throw exceptions::invalid_request_exception("Invalid null value for list index");
     }
diff --git a/cql3/lists.hh b/cql3/lists.hh
index 5831506521..549782d77a 100644
--- a/cql3/lists.hh
+++ b/cql3/lists.hh
@@ -33,7 +33,7 @@ public:
                 : operation(column, std::move(e)) {
         }
         virtual void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) override;
-        static void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params, const column_definition& column, const expr::constant& value);
+        static void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params, const column_definition& column, const cql3::raw_value& value);
     };
 
     class setter_by_index : public operation {
@@ -63,7 +63,7 @@ public:
         virtual void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) override;
     };
 
-    static void do_append(const expr::constant& list_value,
+    static void do_append(const cql3::raw_value& list_value,
             mutation& m,
             const clustering_key_prefix& prefix,
             const column_definition& column,
diff --git a/cql3/maps.cc b/cql3/maps.cc
index 7fe34458f3..889736e488 100644
--- a/cql3/maps.cc
+++ b/cql3/maps.cc
@@ -20,12 +20,12 @@
 namespace cql3 {
 void
 maps::setter::execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params) {
-    expr::constant value = expr::evaluate(*_e, params._options);
+    cql3::raw_value value = expr::evaluate(*_e, params._options);
     execute(m, row_key, params, column, value);
 }
 
 void
-maps::setter::execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params, const column_definition& column, const expr::constant& value) {
+maps::setter::execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params, const column_definition& column, const cql3::raw_value& value) {
     if (value.is_unset_value()) {
         return;
     }
@@ -64,7 +64,7 @@ maps::setter_by_key::execute(mutation& m, const clustering_key_prefix& prefix, c
                      params.make_cell(*ctype->get_values_type(), value.view(), atomic_cell::collection_member::yes)
                     : params.make_dead_cell();
     collection_mutation_description update;
-    update.cells.emplace_back(std::move(key.value).to_bytes(), std::move(avalue));
+    update.cells.emplace_back(std::move(key).to_bytes(), std::move(avalue));
 
     m.set_cell(prefix, column, update.serialize(*ctype));
 }
@@ -72,7 +72,7 @@ maps::setter_by_key::execute(mutation& m, const clustering_key_prefix& prefix, c
 void
 maps::putter::execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) {
     assert(column.type->is_multi_cell()); // "Attempted to add items to a frozen map";
-    expr::constant value = expr::evaluate(*_e, params._options);
+    cql3::raw_value value = expr::evaluate(*_e, params._options);
     if (!value.is_unset_value()) {
         do_put(m, prefix, params, value, column);
     }
@@ -80,7 +80,7 @@ maps::putter::execute(mutation& m, const clustering_key_prefix& prefix, const up
 
 void
 maps::do_put(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params,
-        const expr::constant& map_value, const column_definition& column) {
+        const cql3::raw_value& map_value, const column_definition& column) {
     if (column.type->is_multi_cell()) {
         if (map_value.is_null()) {
             return;
@@ -107,7 +107,7 @@ maps::do_put(mutation& m, const clustering_key_prefix& prefix, const update_para
 void
 maps::discarder_by_key::execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) {
     assert(column.type->is_multi_cell()); // "Attempted to delete a single key in a frozen map";
-    expr::constant key = expr::evaluate(*_e, params._options);
+    cql3::raw_value key = expr::evaluate(*_e, params._options);
     if (key.is_null()) {
         throw exceptions::invalid_request_exception("Invalid null map key");
     }
@@ -115,7 +115,7 @@ maps::discarder_by_key::execute(mutation& m, const clustering_key_prefix& prefix
         throw exceptions::invalid_request_exception("Invalid unset map key");
     }
     collection_mutation_description mut;
-    mut.cells.emplace_back(std::move(key.value).to_bytes(), params.make_dead_cell());
+    mut.cells.emplace_back(std::move(key).to_bytes(), params.make_dead_cell());
 
     m.set_cell(prefix, column, mut.serialize(*column.type));
 }
diff --git a/cql3/maps.hh b/cql3/maps.hh
index df8a05f804..c39c207f01 100644
--- a/cql3/maps.hh
+++ b/cql3/maps.hh
@@ -34,7 +34,7 @@ public:
         }
 
         virtual void execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params) override;
-        static void execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params, const column_definition& column, const expr::constant& value);
+        static void execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params, const column_definition& column, const cql3::raw_value& value);
     };
 
     class setter_by_key : public operation {
@@ -56,7 +56,7 @@ public:
     };
 
     static void do_put(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params,
-            const expr::constant& value, const column_definition& column);
+            const cql3::raw_value& value, const column_definition& column);
 
     class discarder_by_key : public operation {
     public:
diff --git a/cql3/operation.cc b/cql3/operation.cc
index 4d8dbed4f7..dc45044539 100644
--- a/cql3/operation.cc
+++ b/cql3/operation.cc
@@ -275,15 +275,11 @@ operation::set_counter_value_from_tuple_list::prepare(data_dictionary::database
             return true;
         }
         void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) override {
-            expr::constant list_value = expr::evaluate(*_e, params._options);
+            cql3::raw_value list_value = expr::evaluate(*_e, params._options);
             if (list_value.is_null()) {
                 throw std::invalid_argument("Invalid input data to counter set");
             }
 
-            if (!list_value.type->is_list()) {
-                throw std::invalid_argument("Invalid input data to counter set");
-            }
-
             utils::chunked_vector<managed_bytes> list_elements = expr::get_list_elements(list_value);
 
             counter_id last(utils::UUID(0, 0));
diff --git a/cql3/restrictions/multi_column_restriction.hh b/cql3/restrictions/multi_column_restriction.hh
index 2fcd232eee..8ec58797a8 100644
--- a/cql3/restrictions/multi_column_restriction.hh
+++ b/cql3/restrictions/multi_column_restriction.hh
@@ -198,8 +198,8 @@ public:
 #endif
 
     clustering_key_prefix composite_value(const query_options& options) const {
-        expr::constant t = expr::evaluate(_value, options);
-        auto values = expr::get_tuple_elements(t);
+        cql3::raw_value t = expr::evaluate(_value, options);
+        auto values = expr::get_tuple_elements(t, *type_of(_value));
         std::vector<managed_bytes> components;
         for (unsigned i = 0; i < values.size(); i++) {
             auto component = statements::request_validations::check_not_null(values[i],
diff --git a/cql3/restrictions/statement_restrictions.cc b/cql3/restrictions/statement_restrictions.cc
index 97998dc150..b00919983e 100644
--- a/cql3/restrictions/statement_restrictions.cc
+++ b/cql3/restrictions/statement_restrictions.cc
@@ -1004,7 +1004,7 @@ struct multi_column_range_accumulator {
 
     void operator()(const binary_operator& binop) {
         if (is_compare(binop.op)) {
-            auto opt_values = expr::get_tuple_elements(expr::evaluate(binop.rhs, options));
+            auto opt_values = expr::get_tuple_elements(expr::evaluate(binop.rhs, options), *type_of(binop.rhs));
             auto& lhs = expr::as<tuple_constructor>(binop.lhs);
             std::vector<managed_bytes> values(lhs.elements.size());
             for (size_t i = 0; i < lhs.elements.size(); ++i) {
@@ -1015,9 +1015,9 @@ struct multi_column_range_accumulator {
             }
             intersect_all(to_range(binop.op, clustering_key_prefix(std::move(values))));
         } else if (binop.op == oper_t::IN) {
-            const expr::constant tup = expr::evaluate(binop.rhs, options);
+            const cql3::raw_value tup = expr::evaluate(binop.rhs, options);
             statements::request_validations::check_false(tup.is_null(), "Invalid null value for IN restriction");
-            process_in_values(expr::get_list_of_tuples_elements(tup));
+            process_in_values(expr::get_list_of_tuples_elements(tup, *type_of(binop.rhs)));
         } else {
             on_internal_error(rlogger, format("multi_column_range_accumulator: unexpected atom {}", binop));
         }
@@ -1433,13 +1433,13 @@ query::clustering_range range_from_raw_bounds(
     opt_bound lb, ub;
     for (const auto& e : exprs) {
         if (auto b = find_clustering_order(e)) {
-            expr::constant tup_val = expr::evaluate(b->rhs, options);
+            cql3::raw_value tup_val = expr::evaluate(b->rhs, options);
             if (tup_val.is_null()) {
                 on_internal_error(rlogger, format("range_from_raw_bounds: unexpected atom {}", *b));
             }
 
             const auto r = to_range(
-                    b->op, clustering_key_prefix::from_optional_exploded(schema, expr::get_tuple_elements(tup_val)));
+                    b->op, clustering_key_prefix::from_optional_exploded(schema, expr::get_tuple_elements(tup_val, *type_of(b->rhs))));
             if (r.start()) {
                 lb = r.start();
             }
diff --git a/cql3/sets.cc b/cql3/sets.cc
index dd8da61973..57f604d2a1 100644
--- a/cql3/sets.cc
+++ b/cql3/sets.cc
@@ -15,12 +15,12 @@
 namespace cql3 {
 void
 sets::setter::execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params) {
-    expr::constant value = expr::evaluate(*_e, params._options);
+    cql3::raw_value value = expr::evaluate(*_e, params._options);
     execute(m, row_key, params, column, std::move(value));
 }
 
 void
-sets::setter::execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params, const column_definition& column, const expr::constant& value) {
+sets::setter::execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params, const column_definition& column, const cql3::raw_value& value) {
     if (value.is_unset_value()) {
         return;
     }
@@ -35,7 +35,7 @@ sets::setter::execute(mutation& m, const clustering_key_prefix& row_key, const u
 
 void
 sets::adder::execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params) {
-    const expr::constant value = expr::evaluate(*_e, params._options);
+    const cql3::raw_value value = expr::evaluate(*_e, params._options);
     if (value.is_unset_value()) {
         return;
     }
@@ -45,7 +45,7 @@ sets::adder::execute(mutation& m, const clustering_key_prefix& row_key, const up
 
 void
 sets::adder::do_add(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params,
-        const expr::constant& value, const column_definition& column) {
+        const cql3::raw_value& value, const column_definition& column) {
     auto& set_type = dynamic_cast<const set_type_impl&>(column.type->without_reversed());
     if (column.type->is_multi_cell()) {
         if (value.is_null()) {
@@ -78,13 +78,12 @@ void
 sets::discarder::execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params) {
     assert(column.type->is_multi_cell()); // "Attempted to remove items from a frozen set";
 
-    expr::constant svalue = expr::evaluate(*_e, params._options);
+    cql3::raw_value svalue = expr::evaluate(*_e, params._options);
     if (svalue.is_null_or_unset()) {
         return;
     }
 
     collection_mutation_description mut;
-    assert(svalue.type->is_set());
     utils::chunked_vector<managed_bytes> set_elements = expr::get_set_elements(svalue);
     mut.cells.reserve(set_elements.size());
     for (auto&& e : set_elements) {
@@ -96,12 +95,12 @@ sets::discarder::execute(mutation& m, const clustering_key_prefix& row_key, cons
 void sets::element_discarder::execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params)
 {
     assert(column.type->is_multi_cell() && "Attempted to remove items from a frozen set");
-    expr::constant elt = expr::evaluate(*_e, params._options);
+    cql3::raw_value elt = expr::evaluate(*_e, params._options);
     if (elt.is_null()) {
         throw exceptions::invalid_request_exception("Invalid null set element");
     }
     collection_mutation_description mut;
-    mut.cells.emplace_back(std::move(elt.value).to_bytes(), params.make_dead_cell());
+    mut.cells.emplace_back(std::move(elt).to_bytes(), params.make_dead_cell());
     m.set_cell(row_key, column, mut.serialize(*column.type));
 }
 
diff --git a/cql3/sets.hh b/cql3/sets.hh
index da23f80cc4..11e9d221ef 100644
--- a/cql3/sets.hh
+++ b/cql3/sets.hh
@@ -33,7 +33,7 @@ public:
                 : operation(column, std::move(e)) {
         }
         virtual void execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params) override;
-        static void execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params, const column_definition& column, const expr::constant& value);
+        static void execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params, const column_definition& column, const cql3::raw_value& value);
     };
 
     class adder : public operation {
@@ -43,7 +43,7 @@ public:
         }
         virtual void execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params) override;
         static void do_add(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params,
-                const expr::constant& value, const column_definition& column);
+                const cql3::raw_value& value, const column_definition& column);
     };
 
     // Note that this is reused for Map subtraction too (we subtract a set from a map)
diff --git a/cql3/statements/create_aggregate_statement.cc b/cql3/statements/create_aggregate_statement.cc
index 0ff81beb1a..0a3d849aa2 100644
--- a/cql3/statements/create_aggregate_statement.cc
+++ b/cql3/statements/create_aggregate_statement.cc
@@ -55,7 +55,7 @@ shared_ptr<functions::function> create_aggregate_statement::create(query_process
         auto dummy_ident = ::make_shared<column_identifier>("", true);
         auto column_spec = make_lw_shared<column_specification>("", "", dummy_ident, state_type);
         auto initcond_term = expr::evaluate(prepare_expression(_ival.value(), db, _name.keyspace, nullptr, {column_spec}), query_options::DEFAULT);
-        initcond = std::move(initcond_term.value).to_bytes();
+        initcond = std::move(initcond_term).to_bytes();
     }
 
     return ::make_shared<functions::user_aggregate>(_name, initcond, std::move(state_func), std::move(final_func));
diff --git a/cql3/statements/update_statement.cc b/cql3/statements/update_statement.cc
index e49fb4ea7a..3e85d5f845 100644
--- a/cql3/statements/update_statement.cc
+++ b/cql3/statements/update_statement.cc
@@ -149,7 +149,7 @@ void update_statement::add_update_for_key(mutation& m, const query::clustering_r
 }
 
 modification_statement::json_cache_opt insert_prepared_json_statement::maybe_prepare_json_cache(const query_options& options) const {
-    expr::constant c = expr::evaluate(_value, options);
+    cql3::raw_value c = expr::evaluate(_value, options);
     sstring json_string = utf8_type->to_string(to_bytes(c.view()));
     return json_helpers::parse(std::move(json_string), s->all_columns(), cql_serialization_format::internal());
 }
@@ -161,16 +161,16 @@ insert_prepared_json_statement::execute_set_value(mutation& m, const clustering_
     if (!value) {
         visit(*column.type, make_visitor(
         [&] (const list_type_impl&) {
-            lists::setter::execute(m, prefix, params, column, expr::constant::make_null());
+            lists::setter::execute(m, prefix, params, column, cql3::raw_value::make_null());
         },
         [&] (const set_type_impl&) {
-            sets::setter::execute(m, prefix, params, column, expr::constant::make_null());
+            sets::setter::execute(m, prefix, params, column, cql3::raw_value::make_null());
         },
         [&] (const map_type_impl&) {
-            maps::setter::execute(m, prefix, params, column, expr::constant::make_null());
+            maps::setter::execute(m, prefix, params, column, cql3::raw_value::make_null());
         },
         [&] (const user_type_impl&) {
-            user_types::setter::execute(m, prefix, params, column, expr::constant::make_null());
+            user_types::setter::execute(m, prefix, params, column, cql3::raw_value::make_null());
         },
         [&] (const abstract_type& type) {
             if (type.is_collection()) {
@@ -183,7 +183,7 @@ insert_prepared_json_statement::execute_set_value(mutation& m, const clustering_
     }
 
 
-    expr::constant val(raw_value::make_value(*value), column.type);
+    auto val = raw_value::make_value(*value);
     visit(*column.type, make_visitor(
     [&] (const list_type_impl& ltype) {
         lists::setter::execute(m, prefix, params, column, val);
diff --git a/cql3/user_types.cc b/cql3/user_types.cc
index d8dbc9a56f..57ca52f51a 100644
--- a/cql3/user_types.cc
+++ b/cql3/user_types.cc
@@ -20,11 +20,11 @@
 
 namespace cql3 {
 void user_types::setter::execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params) {
-    const expr::constant value = expr::evaluate(*_e, params._options);
+    const cql3::raw_value value = expr::evaluate(*_e, params._options);
     execute(m, row_key, params, column, value);
 }
 
-void user_types::setter::execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params, const column_definition& column, const expr::constant& ut_value) {
+void user_types::setter::execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params, const column_definition& column, const cql3::raw_value& ut_value) {
     if (ut_value.is_unset_value()) {
         return;
     }
@@ -49,7 +49,7 @@ void user_types::setter::execute(mutation& m, const clustering_key_prefix& row_k
         mut.tomb = params.make_tombstone_just_before();
 
         if (!ut_value.is_null()) {
-            const auto& elems = expr::get_user_type_elements(ut_value);
+            const auto& elems = expr::get_user_type_elements(ut_value, type);
             // There might be fewer elements given than fields in the type
             // (e.g. when the user uses a short tuple literal), but never more.
             assert(elems.size() <= type.size());
diff --git a/cql3/user_types.hh b/cql3/user_types.hh
index 6dfcf6724c..d328bf16bd 100644
--- a/cql3/user_types.hh
+++ b/cql3/user_types.hh
@@ -31,7 +31,7 @@ public:
         using operation::operation;
 
         virtual void execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params) override;
-        static void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params, const column_definition& column, const expr::constant& value);
+        static void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params, const column_definition& column, const cql3::raw_value& value);
     };
 
     class setter_by_field : public operation {
diff --git a/cql3/values.hh b/cql3/values.hh
index 65e0d1af90..0ddb5b3504 100644
--- a/cql3/values.hh
+++ b/cql3/values.hh
@@ -228,6 +228,9 @@ public:
     bool is_unset_value() const {
         return std::holds_alternative<unset_value>(_data);
     }
+    bool is_null_or_unset() const {
+        return !is_value();
+    }
     bool is_value() const {
         return _data.index() <= 1;
     }
@@ -265,6 +268,9 @@ public:
         }, std::move(_data));
     }
     raw_value_view to_view() const;
+    raw_value_view view() const {
+        return to_view();
+    }
     friend class raw_value_view;
 };
 
diff --git a/test/boost/expr_test.cc b/test/boost/expr_test.cc
index 33636fd1c4..b1e2eb206a 100644
--- a/test/boost/expr_test.cc
+++ b/test/boost/expr_test.cc
@@ -104,6 +104,10 @@ static sstring expr_print(const expression& e) {
     return format("{}", p);
 }
 
+static sstring value_print(const cql3::raw_value& v, const expression& e) {
+    return expr_print(constant(v, type_of(e)));
+}
+
 static constant make_int(int value) {
     return constant(raw_value::make_value(int32_type->decompose(value)), int32_type);
 }
@@ -194,11 +198,11 @@ BOOST_AUTO_TEST_CASE(expr_printer_list_test) {
     };
     BOOST_REQUIRE_EQUAL(expr_print(frozen_int_list), "[13, 45, 90]");
 
-    constant int_list_constant = evaluate(int_list, query_options::DEFAULT);
-    BOOST_REQUIRE_EQUAL(expr_print(int_list_constant), "[13, 45, 90]");
+    cql3::raw_value int_list_constant = evaluate(int_list, query_options::DEFAULT);
+    BOOST_REQUIRE_EQUAL(value_print(int_list_constant, int_list), "[13, 45, 90]");
 
-    constant frozen_int_list_constant = evaluate(frozen_int_list, query_options::DEFAULT);
-    BOOST_REQUIRE_EQUAL(expr_print(frozen_int_list_constant), "[13, 45, 90]");
+    cql3::raw_value frozen_int_list_constant = evaluate(frozen_int_list, query_options::DEFAULT);
+    BOOST_REQUIRE_EQUAL(value_print(frozen_int_list_constant, frozen_int_list), "[13, 45, 90]");
 }
 
 BOOST_AUTO_TEST_CASE(expr_printer_set_test) {
@@ -216,11 +220,11 @@ BOOST_AUTO_TEST_CASE(expr_printer_set_test) {
     };
     BOOST_REQUIRE_EQUAL(expr_print(frozen_int_set), "{13, 45, 90}");
 
-    constant int_set_constant = evaluate(int_set, query_options::DEFAULT);
-    BOOST_REQUIRE_EQUAL(expr_print(int_set_constant), "{13, 45, 90}");
+    cql3::raw_value int_set_constant = evaluate(int_set, query_options::DEFAULT);
+    BOOST_REQUIRE_EQUAL(value_print(int_set_constant, int_set), "{13, 45, 90}");
 
-    constant frozen_int_set_constant = evaluate(frozen_int_set, query_options::DEFAULT);
-    BOOST_REQUIRE_EQUAL(expr_print(frozen_int_set_constant), "{13, 45, 90}");
+    cql3::raw_value frozen_int_set_constant = evaluate(frozen_int_set, query_options::DEFAULT);
+    BOOST_REQUIRE_EQUAL(value_print(frozen_int_set_constant, frozen_int_set), "{13, 45, 90}");
 }
 
 BOOST_AUTO_TEST_CASE(expr_printer_map_test) {
@@ -245,11 +249,11 @@ BOOST_AUTO_TEST_CASE(expr_printer_map_test) {
     };
     BOOST_REQUIRE_EQUAL(expr_print(frozen_int_int_map), "{12:34, 56:78}");
 
-    constant int_int_map_const = evaluate(int_int_map, query_options::DEFAULT);
-    BOOST_REQUIRE_EQUAL(expr_print(int_int_map_const), "{12:34, 56:78}");
+    cql3::raw_value int_int_map_const = evaluate(int_int_map, query_options::DEFAULT);
+    BOOST_REQUIRE_EQUAL(value_print(int_int_map_const, int_int_map), "{12:34, 56:78}");
 
-    constant frozen_int_int_map_const = evaluate(frozen_int_int_map, query_options::DEFAULT);
-    BOOST_REQUIRE_EQUAL(expr_print(frozen_int_int_map_const), "{12:34, 56:78}");
+    cql3::raw_value frozen_int_int_map_const = evaluate(frozen_int_int_map, query_options::DEFAULT);
+    BOOST_REQUIRE_EQUAL(value_print(frozen_int_int_map_const, frozen_int_int_map), "{12:34, 56:78}");
 }
 
 BOOST_AUTO_TEST_CASE(expr_printer_tuple_test) {
@@ -259,8 +263,8 @@ BOOST_AUTO_TEST_CASE(expr_printer_tuple_test) {
     };
     BOOST_REQUIRE_EQUAL(expr_print(int_int_tuple), "(456, 789)");
 
-    constant int_int_tuple_const = evaluate(int_int_tuple, query_options::DEFAULT);
-    BOOST_REQUIRE_EQUAL(expr_print(int_int_tuple_const), "(456, 789)");
+    cql3::raw_value int_int_tuple_const = evaluate(int_int_tuple, query_options::DEFAULT);
+    BOOST_REQUIRE_EQUAL(value_print(int_int_tuple_const, int_int_tuple), "(456, 789)");
 }
 
 BOOST_AUTO_TEST_CASE(expr_printer_usertype_test) {
@@ -275,8 +279,8 @@ BOOST_AUTO_TEST_CASE(expr_printer_usertype_test) {
     };
     BOOST_REQUIRE_EQUAL(expr_print(user_typ), "{b:666, a:333}");
 
-    constant user_typ_const = evaluate(user_typ, query_options::DEFAULT);
-    BOOST_REQUIRE_EQUAL(expr_print(user_typ_const), "{a:333, b:666}");
+    cql3::raw_value user_typ_const = evaluate(user_typ, query_options::DEFAULT);
+    BOOST_REQUIRE_EQUAL(value_print(user_typ_const, user_typ), "{a:333, b:666}");
 }
 
 // When a list is printed as RHS of an IN binary_operator it should be printed as a tuple.
@@ -294,12 +298,12 @@ BOOST_AUTO_TEST_CASE(expr_printer_in_test) {
     };
     BOOST_REQUIRE_EQUAL(expr_print(a_in_int_list), "a IN (13, 45, 90)");
 
-    constant int_list_const = evaluate(int_list, query_options::DEFAULT);
+    cql3::raw_value int_list_const = evaluate(int_list, query_options::DEFAULT);
 
     binary_operator a_in_int_list_const {
         make_column("a"),
         oper_t::IN,
-        int_list_const
+        constant(int_list_const, type_of(int_list))
     };
     BOOST_REQUIRE_EQUAL(expr_print(a_in_int_list_const), "a IN (13, 45, 90)");
 }