diff --git a/cql3/attributes.cc b/cql3/attributes.cc
index 622f1be1c7..f2ea964a73 100644
--- a/cql3/attributes.cc
+++ b/cql3/attributes.cc
@@ -20,7 +20,9 @@ std::unique_ptr<attributes> attributes::none() {
 attributes::attributes(std::optional<cql3::expr::expression>&& timestamp,
                        std::optional<cql3::expr::expression>&& time_to_live,
                        std::optional<cql3::expr::expression>&& timeout)
-    : _timestamp{std::move(timestamp)}
+    : _timestamp_unset_guard(timestamp)
+    , _timestamp{std::move(timestamp)}
+    , _time_to_live_unset_guard(time_to_live)
     , _time_to_live{std::move(time_to_live)}
     , _timeout{std::move(timeout)}
 { }
@@ -38,7 +40,7 @@ bool attributes::is_timeout_set() const {
 }
 
 int64_t attributes::get_timestamp(int64_t now, const query_options& options) {
-    if (!_timestamp.has_value()) {
+    if (!_timestamp.has_value() || _timestamp_unset_guard.is_unset(options)) {
         return now;
     }
 
@@ -46,9 +48,6 @@ int64_t attributes::get_timestamp(int64_t now, const query_options& options) {
     if (tval.is_null()) {
         throw exceptions::invalid_request_exception("Invalid null value of timestamp");
     }
-    if (tval.is_unset_value()) {
-        return now;
-    }
     try {
         return tval.view().validate_and_deserialize<int64_t>(*long_type);
     } catch (marshal_exception& e) {
@@ -57,16 +56,13 @@ int64_t attributes::get_timestamp(int64_t now, const query_options& options) {
 }
 
 int32_t attributes::get_time_to_live(const query_options& options) {
-    if (!_time_to_live.has_value())
+    if (!_time_to_live.has_value() || _time_to_live_unset_guard.is_unset(options))
         return 0;
 
     cql3::raw_value tval = expr::evaluate(*_time_to_live, options);
     if (tval.is_null()) {
         throw exceptions::invalid_request_exception("Invalid null value of TTL");
     }
-    if (tval.is_unset_value()) {
-        return 0;
-    }
 
     int32_t ttl;
     try {
@@ -91,8 +87,8 @@ int32_t attributes::get_time_to_live(const query_options& options) {
 
 db::timeout_clock::duration attributes::get_timeout(const query_options& options) const {
     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");
+    if (timeout.is_null()) {
+        throw exceptions::invalid_request_exception("Timeout value cannot be null");
     }
     cql_duration duration = timeout.view().deserialize<cql_duration>(*duration_type);
     if (duration.months || duration.days) {
diff --git a/cql3/attributes.hh b/cql3/attributes.hh
index 7dd48b4982..dce6146afc 100644
--- a/cql3/attributes.hh
+++ b/cql3/attributes.hh
@@ -11,6 +11,7 @@
 #pragma once
 
 #include "cql3/expr/expression.hh"
+#include "cql3/expr/unset.hh"
 #include "db/timeout_clock.hh"
 
 namespace cql3 {
@@ -24,7 +25,9 @@ class prepare_context;
  */
 class attributes final {
 private:
+    expr::unset_bind_variable_guard _timestamp_unset_guard;
     std::optional<cql3::expr::expression> _timestamp;
+    expr::unset_bind_variable_guard _time_to_live_unset_guard;
     std::optional<cql3::expr::expression> _time_to_live;
     std::optional<cql3::expr::expression> _timeout;
 public:
diff --git a/cql3/column_condition.cc b/cql3/column_condition.cc
index 17cdc204d5..215b6e9ff6 100644
--- a/cql3/column_condition.cc
+++ b/cql3/column_condition.cc
@@ -139,10 +139,6 @@ bool column_condition::applies_to(const data_value* cell_value, const query_opti
 
         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(
-                    format("Invalid 'unset' value in {} element access", cell_type.cql3_type_name()));
-        }
         if (key.is_null()) {
             throw exceptions::invalid_request_exception(
                     format("Invalid null value for {} element access", cell_type.cql3_type_name()));
@@ -196,9 +192,6 @@ bool column_condition::applies_to(const data_value* cell_value, const query_opti
         // <, >, >=, <=, !=
         cql3::raw_value param = expr::evaluate(*_value, options);
 
-        if (param.is_unset_value()) {
-            throw exceptions::invalid_request_exception("Invalid 'unset' value in condition");
-        }
         if (param.is_null()) {
             if (_op == expr::oper_t::EQ) {
                 return cell_value == nullptr;
@@ -224,9 +217,6 @@ bool column_condition::applies_to(const data_value* cell_value, const query_opti
             return (*_matcher)(bytes_view(cell_value->serialize_nonnull()));
         } else {
             auto param = expr::evaluate(*_value, options);  // LIKE pattern
-            if (param.is_unset_value()) {
-                throw exceptions::invalid_request_exception("Invalid 'unset' value in LIKE pattern");
-            }
             if (param.is_null()) {
                 throw exceptions::invalid_request_exception("Invalid NULL value in LIKE pattern");
             }
diff --git a/cql3/constants.hh b/cql3/constants.hh
index e2caff1248..321f88222e 100644
--- a/cql3/constants.hh
+++ b/cql3/constants.hh
@@ -33,9 +33,9 @@ public:
     private static final Logger logger = LoggerFactory.getLogger(Constants.class);
 #endif
 public:
-    class setter : public operation {
+    class setter : public operation_skip_if_unset {
     public:
-        using operation::operation;
+        using operation_skip_if_unset::operation_skip_if_unset;
 
         virtual void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) override {
             auto value = expr::evaluate(*_e, params._options);
@@ -53,30 +53,26 @@ public:
         virtual void prepare_for_broadcast_tables(statements::broadcast_tables::prepared_update& query) const override;
     };
 
-    struct adder final : operation {
-        using operation::operation;
+    struct adder final : operation_skip_if_unset {
+        using operation_skip_if_unset::operation_skip_if_unset;
 
         virtual void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) override {
             auto value = expr::evaluate(*_e, params._options);
             if (value.is_null()) {
                 throw exceptions::invalid_request_exception("Invalid null value for counter increment");
-            } else if (value.is_unset_value()) {
-                return;
             }
             auto increment = value.view().deserialize<int64_t>(*long_type);
             m.set_cell(prefix, column, params.make_counter_update_cell(increment));
         }
     };
 
-    struct subtracter final : operation {
-        using operation::operation;
+    struct subtracter final : operation_skip_if_unset {
+        using operation_skip_if_unset::operation_skip_if_unset;
 
         virtual void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) override {
             auto value = expr::evaluate(*_e, params._options);
             if (value.is_null()) {
                 throw exceptions::invalid_request_exception("Invalid null value for counter increment");
-            } else if (value.is_unset_value()) {
-                return;
             }
             auto increment = value.view().deserialize<int64_t>(*long_type);
             if (increment == std::numeric_limits<int64_t>::min()) {
@@ -86,10 +82,10 @@ public:
         }
     };
 
-    class deleter : public operation {
+    class deleter : public operation_no_unset_support {
     public:
         deleter(const column_definition& column)
-            : operation(column, std::nullopt)
+            : operation_no_unset_support(column, std::nullopt)
         { }
 
         virtual void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) override;
diff --git a/cql3/expr/expression.cc b/cql3/expr/expression.cc
index f32f2635c7..a270b90dfd 100644
--- a/cql3/expr/expression.cc
+++ b/cql3/expr/expression.cc
@@ -180,18 +180,6 @@ get_value(const subscript& s, const evaluation_inputs& inputs) {
         // not an error.
         return std::nullopt;
     }
-    if (key.is_unset_value()) {
-        // An m[?] with ? bound to UNSET_VALUE is a invalid query.
-        // We could have detected it earlier while binding, but since
-        // we currently don't, we must protect the following code
-        // which can't work with an UNSET_VALUE. Note that the
-        // placement of this check here means that in an empty table,
-        // where we never need to evaluate the filter expression, this
-        // error will not be detected.
-        throw exceptions::invalid_request_exception(
-            format("Unsupported unset map key for column {}",
-                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) {
@@ -251,9 +239,6 @@ public:
 /// True iff lhs's value equals rhs.
 bool_or_null equal(const expression& lhs, const managed_bytes_opt& rhs_bytes, const evaluation_inputs& inputs) {
     raw_value lhs_value = evaluate(lhs, inputs);
-    if (lhs_value.is_unset_value()) {
-        throw exceptions::invalid_request_exception("unset value found on left-hand side of an equality operator");
-    }
     if (lhs_value.is_null() || !rhs_bytes.has_value()) {
         return bool_or_null::null();
     }
@@ -269,14 +254,6 @@ static std::optional<std::pair<managed_bytes, managed_bytes>> evaluate_binop_sid
     raw_value lhs_value = evaluate(lhs, inputs);
     raw_value rhs_value = evaluate(rhs, inputs);
 
-    if (lhs_value.is_unset_value()) {
-        throw exceptions::invalid_request_exception(
-            format("unset value found on left-hand side of a binary operator with operation {}", op));
-    }
-    if (rhs_value.is_unset_value()) {
-        throw exceptions::invalid_request_exception(
-            format("unset value found on right-hand side of a binary operator with operation {}", op));
-    }
     if (lhs_value.is_null() || rhs_value.is_null()) {
         return std::nullopt;
     }
@@ -492,14 +469,7 @@ bool_or_null is_one_of(const expression& lhs, const expression& rhs, const evalu
 
 bool is_not_null(const expression& lhs, const expression& rhs, const evaluation_inputs& inputs) {
     cql3::raw_value lhs_val = evaluate(lhs, inputs);
-    if (lhs_val.is_unset_value()) {
-        throw exceptions::invalid_request_exception("unset value found on left hand side of IS NOT operator");
-    }
-
     cql3::raw_value rhs_val = evaluate(rhs, inputs);
-    if (rhs_val.is_unset_value()) {
-        throw exceptions::invalid_request_exception("unset value found on right hand side of IS NOT operator");
-    }
     if (!rhs_val.is_null()) {
         throw exceptions::invalid_request_exception("IS NOT operator accepts only NULL as its right side");
     }
@@ -554,9 +524,6 @@ bool is_satisfied_by(const binary_operator& opr, const evaluation_inputs& inputs
     if (binop_eval_result.is_null()) {
         return false;
     }
-    if (binop_eval_result.is_unset_value()) {
-        on_internal_error(expr_logger, format("is_satisfied_by: binary operator evaluated to unset value: {}", opr));
-    }
     if (binop_eval_result.is_empty_value()) {
         on_internal_error(expr_logger, format("is_satisfied_by: binary operator evaluated to EMPTY_VALUE: {}", opr));
     }
@@ -644,9 +611,6 @@ value_list get_IN_values(
         const expression& e, const query_options& options, const serialized_compare& comparator,
         sstring_view column_name) {
     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));
-    }
     if (in_list.is_null()) {
         return value_list();
     }
@@ -1107,8 +1071,6 @@ std::ostream& operator<<(std::ostream& os, const expression::printer& pr) {
                 } else {
                     if (v.value.is_null()) {
                         os << "null";
-                    } else if (v.value.is_unset_value()) {
-                        os << "unset";
                     } else {
                         v.value.view().with_value([&](const FragmentedView auto& bytes_view) {
                             data_value deser_val = v.type->deserialize(bytes_view);
@@ -1618,10 +1580,6 @@ constant constant::make_null(data_type val_type) {
     return constant(cql3::raw_value::make_null(), std::move(val_type));
 }
 
-constant constant::make_unset_value(data_type val_type) {
-    return constant(cql3::raw_value::make_unset_value(), std::move(val_type));
-}
-
 constant constant::make_bool(bool bool_val) {
     return constant(raw_value::make_value(boolean_type->decompose(bool_val)), boolean_type);
 }
@@ -1630,22 +1588,14 @@ bool constant::is_null() const {
     return value.is_null();
 }
 
-bool constant::is_unset_value() const {
-    return value.is_unset_value();
-}
-
 bool constant::has_empty_value_bytes() const {
-    if (is_null_or_unset()) {
+    if (is_null()) {
         return false;
     }
 
     return value.view().size_bytes() == 0;
 }
 
-bool constant::is_null_or_unset() const {
-    return is_null() || is_unset_value();
-}
-
 cql3::raw_value_view constant::view() const {
     return value.view();
 }
@@ -1655,7 +1605,7 @@ std::optional<bool> get_bool_value(const constant& constant_val) {
         return std::nullopt;
     }
 
-    if (constant_val.is_null_or_unset()) {
+    if (constant_val.is_null()) {
         return std::nullopt;
     }
 
@@ -1714,7 +1664,7 @@ cql3::raw_value evaluate(const binary_operator& binop, const evaluation_inputs&
 // NULL is treated as an "unkown value" - maybe true maybe false.
 // `TRUE AND NULL` evaluates to NULL because it might be true but also might be false.
 // `FALSE AND NULL` evaluates to FALSE because no matter what value NULL acts as, the result will still be FALSE.
-// Unset and empty values are not allowed.
+// Empty values are not allowed.
 //
 // Usually in CQL the rule is that when NULL occurs in an operation the whole expression
 // becomes NULL, but here we decided to deviate from this behavior.
@@ -1735,9 +1685,6 @@ cql3::raw_value evaluate(const conjunction& conj, const evaluation_inputs& input
             has_null = true;
             continue;
         }
-        if (element_val.is_unset_value()) {
-            throw exceptions::invalid_request_exception("unset value found inside AND conjunction");
-        }
         if (element_val.is_empty_value()) {
             throw exceptions::invalid_request_exception("empty value found inside AND conjunction");
         }
@@ -1908,10 +1855,6 @@ static cql3::raw_value evaluate(const bind_variable& bind_var, const evaluation_
         return cql3::raw_value::make_null();
     }
 
-    if (value.is_unset_value()) {
-        return cql3::raw_value::make_unset_value();
-    }
-
     const abstract_type& value_type = bind_var.receiver->type->without_reversed();
     try {
         value.validate(value_type);
@@ -1942,10 +1885,6 @@ static cql3::raw_value evaluate(const tuple_constructor& tuple, const evaluation
 
     for (size_t i = 0; i < tuple.elements.size(); i++) {
         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).to_managed_bytes_opt());
     }
 
@@ -1987,10 +1926,6 @@ static cql3::raw_value evaluate_list(const collection_constructor& collection,
     for (const expression& element : collection.elements) {
         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");
-        }
-
         if (evaluated_element.is_null()) {
             if (skip_null) {
                 continue;
@@ -2017,10 +1952,6 @@ static cql3::raw_value evaluate_set(const collection_constructor& collection, co
             throw exceptions::invalid_request_exception("null is not supported inside collections");
         }
 
-        if (evaluated_element.is_unset_value()) {
-            throw exceptions::invalid_request_exception("unset value is not supported inside collections");
-        }
-
         if (evaluated_element.view().size_bytes() > std::numeric_limits<uint16_t>::max()) {
             // TODO: Behaviour copied from sets::delayed_value::bind(), but this seems incorrect
             // The original reasoning is:
@@ -2051,10 +1982,6 @@ static cql3::raw_value evaluate_map(const collection_constructor& collection, co
                 throw exceptions::invalid_request_exception("null is not supported inside collections");
             }
 
-            if (key.is_unset_value() || value.is_unset_value()) {
-                throw exceptions::invalid_request_exception("unset value is not supported inside collections");
-            }
-
             if (key.view().size_bytes() > std::numeric_limits<uint16_t>::max()) {
                 // TODO: Behaviour copied from maps::delayed_value::bind(), but this seems incorrect
                 // The original reasoning is:
@@ -2128,10 +2055,6 @@ static cql3::raw_value evaluate(const usertype_constructor& user_val, const eval
         }
 
         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).to_managed_bytes_opt());
     }
@@ -2157,8 +2080,8 @@ static cql3::raw_value evaluate(const function_call& fun_call, const evaluation_
 
     for (const expression& arg : fun_call.args) {
         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));
+        if (arg_val.is_null()) {
+            throw exceptions::invalid_request_exception(format("Invalid null value for argument to {}", *scalar_fun));
         }
 
         arguments.emplace_back(to_bytes_opt(std::move(arg_val)));
@@ -2200,10 +2123,6 @@ static void ensure_can_get_value_elements(const cql3::raw_value& val,
     if (val.is_null()) {
         on_internal_error(expr_logger, fmt::format("{} called with null value", caller_name));
     }
-
-    if (val.is_unset_value()) {
-        on_internal_error(expr_logger, fmt::format("{} called with unset value", caller_name));
-    }
 }
 
 utils::chunked_vector<managed_bytes> get_list_elements(const cql3::raw_value& val) {
@@ -2601,5 +2520,29 @@ bool has_only_eq_binops(const expression& e) {
 
     return non_eq_binop == nullptr;
 }
+
+unset_bind_variable_guard::unset_bind_variable_guard(const expr::expression& e) {
+    if (auto bv = expr::as_if<expr::bind_variable>(&e)) {
+        _var = *bv;
+    }
+}
+
+unset_bind_variable_guard::unset_bind_variable_guard(const std::optional<expr::expression>& e) {
+    if (!e) {
+        return;
+    }
+    if (auto bv = expr::as_if<expr::bind_variable>(&*e)) {
+        _var = *bv;
+    }
+}
+
+bool
+unset_bind_variable_guard::is_unset(const query_options& qo) const {
+    if (!_var) {
+        return false;
+    }
+    return qo.is_unset(_var->bind_index);
+}
+
 } // namespace expr
 } // namespace cql3
diff --git a/cql3/expr/expression.hh b/cql3/expr/expression.hh
index baf0303c99..f0d6ac34e4 100644
--- a/cql3/expr/expression.hh
+++ b/cql3/expr/expression.hh
@@ -372,7 +372,6 @@ struct constant {
 
     constant(cql3::raw_value value, data_type type);
     static constant make_null(data_type val_type = empty_type);
-    static constant make_unset_value(data_type val_type = empty_type);
     static constant make_bool(bool bool_val);
 
     bool is_null() const;
diff --git a/cql3/expr/unset.hh b/cql3/expr/unset.hh
new file mode 100644
index 0000000000..12bbeb0314
--- /dev/null
+++ b/cql3/expr/unset.hh
@@ -0,0 +1,30 @@
+// Copyright (C) 2023-present ScyllaDB
+// SPDX-License-Identifier: (AGPL-3.0-or-later and Apache-2.0)
+
+#pragma once
+
+#include <optional>
+#include "expression.hh"
+
+namespace cql3 {
+
+class query_options;
+
+}
+
+namespace cql3::expr {
+
+// Some expression users can behave differently if the expression is a bind variable
+// and if that bind variable is unset. unset_bind_variable_guard encapsulates the two
+// conditions.
+class unset_bind_variable_guard {
+    // Disengaged if the operand is not exactly a single bind variable.
+    std::optional<bind_variable> _var;
+public:
+    explicit unset_bind_variable_guard(const expr::expression& operand);
+    explicit unset_bind_variable_guard(std::nullopt_t) {}
+    explicit unset_bind_variable_guard(const std::optional<expr::expression>& operand);
+    bool is_unset(const query_options& qo) const;
+};
+
+}
diff --git a/cql3/lists.cc b/cql3/lists.cc
index b554f8272a..7d9d02e02a 100644
--- a/cql3/lists.cc
+++ b/cql3/lists.cc
@@ -37,9 +37,6 @@ 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 cql3::raw_value& value) {
-    if (value.is_unset_value()) {
-        return;
-    }
     if (column.type->is_multi_cell()) {
         // Delete all cells first, then append new ones
         collection_mutation_view_description mut;
@@ -70,13 +67,7 @@ lists::setter_by_index::execute(mutation& m, const clustering_key_prefix& prefix
     if (index.is_null()) {
         throw exceptions::invalid_request_exception("Invalid null value for list index");
     }
-    if (index.is_unset_value()) {
-        throw exceptions::invalid_request_exception("Invalid unset value for list index");
-    }
     auto value = expr::evaluate(*_e, params._options);
-    if (value.is_unset_value()) {
-        return;
-    }
 
     auto idx = index.view().deserialize<int32_t>(*int32_type);
     auto&& existing_list_opt = params.get_prefetched_list(m.key(), prefix, column);
@@ -122,10 +113,6 @@ lists::setter_by_uuid::execute(mutation& m, const clustering_key_prefix& prefix,
         throw exceptions::invalid_request_exception("Invalid null value for list index");
     }
 
-    if (index.is_unset_value()) {
-        throw exceptions::invalid_request_exception("Invalid unset value for list index");
-    }
-
     auto ltype = static_cast<const list_type_impl*>(column.type.get());
 
     collection_mutation_description mut;
@@ -145,9 +132,6 @@ 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 cql3::raw_value value = expr::evaluate(*_e, params._options);
-    if (value.is_unset_value()) {
-        return;
-    }
     assert(column.type->is_multi_cell()); // "Attempted to append to a frozen list";
     do_append(value, m, prefix, column, params);
 }
@@ -161,7 +145,7 @@ lists::do_append(const cql3::raw_value& list_value,
     if (column.type->is_multi_cell()) {
         // If we append null, do nothing. Note that for Setter, we've
         // already removed the previous value so we're good here too
-        if (list_value.is_null_or_unset()) {
+        if (list_value.is_null()) {
             return;
         }
 
@@ -199,7 +183,7 @@ 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";
     cql3::raw_value lvalue = expr::evaluate(*_e, params._options);
-    if (lvalue.is_null_or_unset()) {
+    if (lvalue.is_null()) {
         return;
     }
 
@@ -265,7 +249,7 @@ lists::discarder::execute(mutation& m, const clustering_key_prefix& prefix, cons
         return;
     }
 
-    if (lvalue.is_null_or_unset()) {
+    if (lvalue.is_null()) {
         return;
     }
 
@@ -304,9 +288,6 @@ lists::discarder_by_index::execute(mutation& m, const clustering_key_prefix& pre
     if (index.is_null()) {
         throw exceptions::invalid_request_exception("Invalid null value for list index");
     }
-    if (index.is_unset_value()) {
-        return;
-    }
 
     auto&& existing_list_opt = params.get_prefetched_list(m.key(), prefix, column);
     int32_t idx = index.view().deserialize<int32_t>(*int32_type);
diff --git a/cql3/lists.hh b/cql3/lists.hh
index 549782d77a..54c8a36c2b 100644
--- a/cql3/lists.hh
+++ b/cql3/lists.hh
@@ -27,21 +27,21 @@ public:
     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&);
 public:
-    class setter : public operation {
+    class setter : public operation_skip_if_unset {
     public:
         setter(const column_definition& column, expr::expression e)
-                : operation(column, std::move(e)) {
+                : operation_skip_if_unset(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 cql3::raw_value& value);
     };
 
-    class setter_by_index : public operation {
+    class setter_by_index : public operation_skip_if_unset {
     protected:
         expr::expression _idx;
     public:
         setter_by_index(const column_definition& column, expr::expression idx, expr::expression e)
-            : operation(column, std::move(e)), _idx(std::move(idx)) {
+            : operation_skip_if_unset(column, std::move(e)), _idx(std::move(idx)) {
         }
         virtual bool requires_read() const override;
         virtual void fill_prepare_context(prepare_context& ctx) override;
@@ -57,9 +57,9 @@ public:
         virtual void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) override;
     };
 
-    class appender : public operation {
+    class appender : public operation_skip_if_unset {
     public:
-        using operation::operation;
+        using operation_skip_if_unset::operation_skip_if_unset;
         virtual void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) override;
     };
 
@@ -69,25 +69,25 @@ public:
             const column_definition& column,
             const update_parameters& params);
 
-    class prepender : public operation {
+    class prepender : public operation_skip_if_unset {
     public:
-        using operation::operation;
+        using operation_skip_if_unset::operation_skip_if_unset;
         virtual void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) override;
     };
 
-    class discarder : public operation {
+    class discarder : public operation_skip_if_unset {
     public:
         discarder(const column_definition& column, expr::expression e)
-                : operation(column, std::move(e)) {
+                : operation_skip_if_unset(column, std::move(e)) {
         }
         virtual bool requires_read() const override;
         virtual void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) override;
     };
 
-    class discarder_by_index : public operation {
+    class discarder_by_index : public operation_skip_if_unset {
     public:
         discarder_by_index(const column_definition& column, expr::expression idx)
-                : operation(column, std::move(idx)) {
+                : operation_skip_if_unset(column, std::move(idx)) {
         }
         virtual bool requires_read() const override;
         virtual void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) override;
diff --git a/cql3/maps.cc b/cql3/maps.cc
index 889736e488..34f9967ee0 100644
--- a/cql3/maps.cc
+++ b/cql3/maps.cc
@@ -26,9 +26,6 @@ maps::setter::execute(mutation& m, const clustering_key_prefix& row_key, const u
 
 void
 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;
-    }
     if (column.type->is_multi_cell()) {
         // Delete all cells first, then put new ones
         collection_mutation_description mut;
@@ -50,12 +47,6 @@ maps::setter_by_key::execute(mutation& m, const clustering_key_prefix& prefix, c
     assert(column.type->is_multi_cell()); // "Attempted to set a value for a single key on a frozen map"m
     auto key = expr::evaluate(_k, params._options);
     auto value = expr::evaluate(*_e, params._options);
-    if (value.is_unset_value()) {
-        return;
-    }
-    if (key.is_unset_value()) {
-        throw invalid_request_exception("Invalid unset map key");
-    }
     if (key.is_null()) {
         throw invalid_request_exception("Invalid null map key");
     }
@@ -73,9 +64,7 @@ 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";
     cql3::raw_value value = expr::evaluate(*_e, params._options);
-    if (!value.is_unset_value()) {
-        do_put(m, prefix, params, value, column);
-    }
+    do_put(m, prefix, params, value, column);
 }
 
 void
@@ -111,9 +100,6 @@ maps::discarder_by_key::execute(mutation& m, const clustering_key_prefix& prefix
     if (key.is_null()) {
         throw exceptions::invalid_request_exception("Invalid null map key");
     }
-    if (key.is_unset_value()) {
-        throw exceptions::invalid_request_exception("Invalid unset map key");
-    }
     collection_mutation_description mut;
     mut.cells.emplace_back(std::move(key).to_bytes(), params.make_dead_cell());
 
diff --git a/cql3/maps.hh b/cql3/maps.hh
index c39c207f01..5f3825186a 100644
--- a/cql3/maps.hh
+++ b/cql3/maps.hh
@@ -27,30 +27,30 @@ public:
     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 setter : public operation {
+    class setter : public operation_skip_if_unset {
     public:
         setter(const column_definition& column, expr::expression e)
-                : operation(column, std::move(e)) {
+                : operation_skip_if_unset(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 cql3::raw_value& value);
     };
 
-    class setter_by_key : public operation {
+    class setter_by_key : public operation_skip_if_unset {
         expr::expression _k;
     public:
         setter_by_key(const column_definition& column, expr::expression k, expr::expression e)
-            : operation(column, std::move(e)), _k(std::move(k)) {
+            : operation_skip_if_unset(column, std::move(e)), _k(std::move(k)) {
         }
         virtual void fill_prepare_context(prepare_context& ctx) override;
         virtual void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) override;
     };
 
-    class putter : public operation {
+    class putter : public operation_skip_if_unset {
     public:
         putter(const column_definition& column, expr::expression e)
-            : operation(column, std::move(e)) {
+            : operation_skip_if_unset(column, std::move(e)) {
         }
         virtual void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) override;
     };
@@ -58,10 +58,10 @@ public:
     static void do_put(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params,
             const cql3::raw_value& value, const column_definition& column);
 
-    class discarder_by_key : public operation {
+    class discarder_by_key : public operation_no_unset_support {
     public:
         discarder_by_key(const column_definition& column, expr::expression k)
-                : operation(column, std::move(k)) {
+                : operation_no_unset_support(column, std::move(k)) {
         }
         virtual void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) override;
     };
diff --git a/cql3/operation.cc b/cql3/operation.cc
index f30f0929ee..bd8749c92b 100644
--- a/cql3/operation.cc
+++ b/cql3/operation.cc
@@ -268,9 +268,9 @@ operation::set_counter_value_from_tuple_list::prepare(data_dictionary::database
     auto v = prepare_expression(_value, db, keyspace, nullptr, spec);
 
     // Will not be used elsewhere, so make it local.
-    class counter_setter : public operation {
+    class counter_setter : public operation_no_unset_support {
     public:
-        using operation::operation;
+        using operation_no_unset_support::operation_no_unset_support;
 
         bool is_raw_counter_shard_write() const override {
             return true;
diff --git a/cql3/operation.hh b/cql3/operation.hh
index 5f9017cdfc..e159290bc4 100644
--- a/cql3/operation.hh
+++ b/cql3/operation.hh
@@ -17,6 +17,7 @@
 #include "update_parameters.hh"
 #include "cql3/column_identifier.hh"
 #include "cql3/expr/expression.hh"
+#include "cql3/expr/unset.hh"
 
 #include <optional>
 
@@ -54,10 +55,13 @@ protected:
     // may require none of more than one expression, but most need 1 so it simplify things a bit.
     std::optional<expr::expression> _e;
 
+    // A guard to check if the operation should be skipped due to unset operand.
+    expr::unset_bind_variable_guard _unset_guard;
 public:
-    operation(const column_definition& column_, std::optional<expr::expression> e)
+    operation(const column_definition& column_, std::optional<expr::expression> e, expr::unset_bind_variable_guard ubvg)
         : column{column_}
         , _e(std::move(e))
+        , _unset_guard(std::move(ubvg))
     { }
 
     virtual ~operation() {}
@@ -87,10 +91,14 @@ public:
     }
 
     /**
-     * Execute the operation.
+     * Execute the operation. Check should_skip_operation() first.
      */
     virtual void execute(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params) = 0;
-    
+
+    bool should_skip_operation(const query_options& qo) const {
+        return _unset_guard.is_unset(qo);
+    }
+
     virtual void prepare_for_broadcast_tables(statements::broadcast_tables::prepared_update&) const;
 
     /**
@@ -265,4 +273,18 @@ public:
     };
 };
 
+class operation_skip_if_unset : public operation {
+public:
+    operation_skip_if_unset(const column_definition& column, expr::expression e)
+            : operation(column, e, expr::unset_bind_variable_guard(e)) {
+    }
+};
+
+class operation_no_unset_support : public operation {
+public:
+    operation_no_unset_support(const column_definition& column, std::optional<expr::expression> e)
+            : operation(column, std::move(e), expr::unset_bind_variable_guard(std::nullopt)) {
+    }
+};
+
 }
diff --git a/cql3/query_options.cc b/cql3/query_options.cc
index a2736e07a2..47ea3725a8 100644
--- a/cql3/query_options.cc
+++ b/cql3/query_options.cc
@@ -30,6 +30,7 @@ query_options::query_options(const cql_config& cfg,
                            std::optional<std::vector<sstring_view>> names,
                            std::vector<cql3::raw_value> values,
                            std::vector<cql3::raw_value_view> value_views,
+                           cql3::unset_bind_variable_vector unset,
                            bool skip_metadata,
                            specific_options options
                            )
@@ -38,6 +39,7 @@ query_options::query_options(const cql_config& cfg,
    , _names(std::move(names))
    , _values(std::move(values))
    , _value_views(value_views)
+   , _unset(unset)
    , _skip_metadata(skip_metadata)
    , _options(std::move(options))
 {
@@ -46,15 +48,16 @@ query_options::query_options(const cql_config& cfg,
 query_options::query_options(const cql_config& cfg,
                              db::consistency_level consistency,
                              std::optional<std::vector<sstring_view>> names,
-                             std::vector<cql3::raw_value> values,
+                             cql3::raw_value_vector_with_unset values,
                              bool skip_metadata,
                              specific_options options
                              )
     : _cql_config(cfg)
     , _consistency(consistency)
     , _names(std::move(names))
-    , _values(std::move(values))
+    , _values(std::move(values.values))
     , _value_views()
+    , _unset(std::move(values.unset))
     , _skip_metadata(skip_metadata)
     , _options(std::move(options))
 {
@@ -64,7 +67,7 @@ query_options::query_options(const cql_config& cfg,
 query_options::query_options(const cql_config& cfg,
                              db::consistency_level consistency,
                              std::optional<std::vector<sstring_view>> names,
-                             std::vector<cql3::raw_value_view> value_views,
+                             cql3::raw_value_view_vector_with_unset value_views,
                              bool skip_metadata,
                              specific_options options
                              )
@@ -72,13 +75,14 @@ query_options::query_options(const cql_config& cfg,
     , _consistency(consistency)
     , _names(std::move(names))
     , _values()
-    , _value_views(std::move(value_views))
+    , _value_views(std::move(value_views.values))
+    , _unset(std::move(value_views.unset))
     , _skip_metadata(skip_metadata)
     , _options(std::move(options))
 {
 }
 
-query_options::query_options(db::consistency_level cl, std::vector<cql3::raw_value> values,
+query_options::query_options(db::consistency_level cl, cql3::raw_value_vector_with_unset values,
         specific_options options)
     : query_options(
           default_cql_config,
@@ -97,6 +101,7 @@ query_options::query_options(std::unique_ptr<query_options> qo, lw_shared_ptr<se
         std::move(qo->_names),
         std::move(qo->_values),
         std::move(qo->_value_views),
+        std::move(qo->_unset),
         qo->_skip_metadata,
         query_options::specific_options{qo->_options.page_size, paging_state, qo->_options.serial_consistency, qo->_options.timestamp}) {
 
@@ -108,12 +113,13 @@ query_options::query_options(std::unique_ptr<query_options> qo, lw_shared_ptr<se
         std::move(qo->_names),
         std::move(qo->_values),
         std::move(qo->_value_views),
+        std::move(qo->_unset),
         qo->_skip_metadata,
         query_options::specific_options{page_size, paging_state, qo->_options.serial_consistency, qo->_options.timestamp}) {
 
 }
 
-query_options::query_options(std::vector<cql3::raw_value> values)
+query_options::query_options(cql3::raw_value_vector_with_unset values)
     : query_options(
           db::consistency_level::ONE, std::move(values))
 {}
diff --git a/cql3/query_options.hh b/cql3/query_options.hh
index ddc5881ed1..c8e66e1a97 100644
--- a/cql3/query_options.hh
+++ b/cql3/query_options.hh
@@ -11,12 +11,14 @@
 #pragma once
 
 #include <concepts>
+#include <initializer_list>
 #include "timestamp.hh"
 #include "bytes.hh"
 #include "db/consistency_level_type.hh"
 #include "service/query_state.hh"
 #include "service/pager/paging_state.hh"
 #include "cql3/values.hh"
+#include "utils/small_vector.hh"
 
 namespace cql3 {
 
@@ -27,6 +29,38 @@ class column_specification;
 
 using computed_function_values = std::unordered_map<uint8_t, bytes_opt>;
 
+using unset_bind_variable_vector = utils::small_vector<bool, 16>;
+
+// Matches a raw_value_view with an unset vector to support CQL binary protocol
+// "unset" values.
+struct raw_value_view_vector_with_unset {
+    std::vector<raw_value_view> values;
+    unset_bind_variable_vector unset;
+
+    raw_value_view_vector_with_unset(std::vector<raw_value_view> values_, unset_bind_variable_vector unset_) : values(std::move(values_)), unset(std::move(unset_)) {}
+    // Constructor with no unset support, for tests and internal queries
+    raw_value_view_vector_with_unset(std::vector<raw_value_view> values_) : values(std::move(values_)) {
+        unset.resize(values.size());
+    }
+    raw_value_view_vector_with_unset() = default;
+};
+
+// Matches a raw_value with an unset vector to support CQL binary protocol
+// "unset" values.
+struct raw_value_vector_with_unset {
+    std::vector<raw_value> values;
+    unset_bind_variable_vector unset;
+
+    raw_value_vector_with_unset(std::vector<raw_value> values_, unset_bind_variable_vector unset_) : values(std::move(values_)), unset(std::move(unset_)) {}
+    // Constructor with no unset support, for tests and internal queries
+    raw_value_vector_with_unset(std::vector<raw_value> values_) : values(std::move(values_)) {
+        unset.resize(values.size());
+    }
+    // Mostly for testing.
+    raw_value_vector_with_unset(std::initializer_list<raw_value> values_) : raw_value_vector_with_unset(std::vector(values_)) {}
+    raw_value_vector_with_unset() = default;
+};
+
 /**
  * Options for a query.
  */
@@ -47,6 +81,7 @@ private:
     const std::optional<std::vector<sstring_view>> _names;
     std::vector<cql3::raw_value> _values;
     std::vector<cql3::raw_value_view> _value_views;
+    unset_bind_variable_vector _unset;
     const bool _skip_metadata;
     const specific_options _options;
     std::optional<std::vector<query_options>> _batch_options;
@@ -82,9 +117,9 @@ private:
     mutable computed_function_values _cached_pk_fn_calls;
 private:
     // Batch constructor.
-    template <typename Value>
-    requires std::same_as<Value, raw_value> || std::same_as<Value, raw_value_view>
-    explicit query_options(query_options&& o, std::vector<std::vector<Value>> values_ranges);
+    template <typename Values>
+    requires std::same_as<Values, raw_value_vector_with_unset> || std::same_as<Values, raw_value_view_vector_with_unset>
+    explicit query_options(query_options&& o, std::vector<Values> values_ranges);
 
 public:
     query_options(query_options&&) = default;
@@ -93,7 +128,7 @@ public:
     explicit query_options(const cql_config& cfg,
                            db::consistency_level consistency,
                            std::optional<std::vector<sstring_view>> names,
-                           std::vector<cql3::raw_value> values,
+                           raw_value_vector_with_unset values,
                            bool skip_metadata,
                            specific_options options
                            );
@@ -102,20 +137,21 @@ public:
                            std::optional<std::vector<sstring_view>> names,
                            std::vector<cql3::raw_value> values,
                            std::vector<cql3::raw_value_view> value_views,
+                           unset_bind_variable_vector unset,
                            bool skip_metadata,
                            specific_options options
                            );
     explicit query_options(const cql_config& cfg,
                            db::consistency_level consistency,
                            std::optional<std::vector<sstring_view>> names,
-                           std::vector<cql3::raw_value_view> value_views,
+                           raw_value_view_vector_with_unset value_views,
                            bool skip_metadata,
                            specific_options options
                            );
 
-    template <typename Value>
-    requires std::same_as<Value, raw_value> || std::same_as<Value, raw_value_view>
-    static query_options make_batch_options(query_options&& o, std::vector<std::vector<Value>> values_ranges) {
+    template <typename Values>
+    requires std::same_as<Values, raw_value_vector_with_unset> || std::same_as<Values, raw_value_view_vector_with_unset>
+    static query_options make_batch_options(query_options&& o, std::vector<Values> values_ranges) {
         return query_options(std::move(o), std::move(values_ranges));
     }
 
@@ -123,8 +159,8 @@ public:
     static thread_local query_options DEFAULT;
 
     // forInternalUse
-    explicit query_options(std::vector<cql3::raw_value> values);
-    explicit query_options(db::consistency_level, std::vector<cql3::raw_value> values, specific_options options = specific_options::DEFAULT);
+    explicit query_options(raw_value_vector_with_unset values);
+    explicit query_options(db::consistency_level, raw_value_vector_with_unset values, specific_options options = specific_options::DEFAULT);
     explicit query_options(std::unique_ptr<query_options>, lw_shared_ptr<service::pager::paging_state> paging_state);
     explicit query_options(std::unique_ptr<query_options>, lw_shared_ptr<service::pager::paging_state> paging_state, int32_t page_size);
 
@@ -133,7 +169,14 @@ public:
     }
 
     cql3::raw_value_view get_value_at(size_t idx) const {
-        return _value_views.at(idx);
+        if (_unset.at(idx)) {
+            throw exceptions::invalid_request_exception(fmt::format("Unexpected unset value for bind variable {}", idx));
+        }
+        return _value_views[idx];
+    }
+
+    bool is_unset(size_t idx) const {
+        return _unset.at(idx);
     }
 
     size_t get_values_count() const {
@@ -237,9 +280,9 @@ private:
     void fill_value_views();
 };
 
-template <typename Value>
-requires std::same_as<Value, raw_value> || std::same_as<Value, raw_value_view>
-query_options::query_options(query_options&& o, std::vector<std::vector<Value>> values_ranges)
+template <typename Values>
+requires std::same_as<Values, raw_value_vector_with_unset> || std::same_as<Values, raw_value_view_vector_with_unset>
+query_options::query_options(query_options&& o, std::vector<Values> values_ranges)
     : query_options(std::move(o))
 {
     std::vector<query_options> tmp;
diff --git a/cql3/restrictions/statement_restrictions.cc b/cql3/restrictions/statement_restrictions.cc
index 3fcec7f67c..b0fa0743c6 100644
--- a/cql3/restrictions/statement_restrictions.cc
+++ b/cql3/restrictions/statement_restrictions.cc
@@ -1788,7 +1788,7 @@ void statement_restrictions::prepare_indexed_global(const schema& idx_tbl_schema
             oper_t::EQ,
             // TODO: This should be a unique marker whose value we set at execution time.  There is currently no
             // handy mechanism for doing that in query_options.
-            expr::constant::make_unset_value(token_column->type));
+            expr::constant::make_null(token_column->type));
 }
 
 void statement_restrictions::prepare_indexed_local(const schema& idx_tbl_schema) {
diff --git a/cql3/sets.cc b/cql3/sets.cc
index 57f604d2a1..955a2a5954 100644
--- a/cql3/sets.cc
+++ b/cql3/sets.cc
@@ -21,9 +21,6 @@ sets::setter::execute(mutation& m, const clustering_key_prefix& row_key, const u
 
 void
 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;
-    }
     if (column.type->is_multi_cell()) {
         // Delete all cells first, then add new ones
         collection_mutation_description mut;
@@ -36,9 +33,6 @@ 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 cql3::raw_value value = expr::evaluate(*_e, params._options);
-    if (value.is_unset_value()) {
-        return;
-    }
     assert(column.type->is_multi_cell()); // "Attempted to add items to a frozen set";
     do_add(m, row_key, params, value, column);
 }
@@ -79,7 +73,7 @@ sets::discarder::execute(mutation& m, const clustering_key_prefix& row_key, cons
     assert(column.type->is_multi_cell()); // "Attempted to remove items from a frozen set";
 
     cql3::raw_value svalue = expr::evaluate(*_e, params._options);
-    if (svalue.is_null_or_unset()) {
+    if (svalue.is_null()) {
         return;
     }
 
diff --git a/cql3/sets.hh b/cql3/sets.hh
index 11e9d221ef..7a3b8f95e4 100644
--- a/cql3/sets.hh
+++ b/cql3/sets.hh
@@ -27,19 +27,19 @@ class sets {
 public:
     static lw_shared_ptr<column_specification> value_spec_of(const column_specification& column);
 
-    class setter : public operation {
+    class setter : public operation_skip_if_unset {
     public:
         setter(const column_definition& column, expr::expression e)
-                : operation(column, std::move(e)) {
+                : operation_skip_if_unset(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 cql3::raw_value& value);
     };
 
-    class adder : public operation {
+    class adder : public operation_skip_if_unset {
     public:
         adder(const column_definition& column, expr::expression e)
-            : operation(column, std::move(e)) {
+            : operation_skip_if_unset(column, std::move(e)) {
         }
         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,
@@ -47,18 +47,18 @@ public:
     };
 
     // Note that this is reused for Map subtraction too (we subtract a set from a map)
-    class discarder : public operation {
+    class discarder : public operation_skip_if_unset {
     public:
         discarder(const column_definition& column, expr::expression e)
-            : operation(column, std::move(e)) {
+            : operation_skip_if_unset(column, std::move(e)) {
         }
         virtual void execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params) override;
     };
 
-    class element_discarder : public operation {
+    class element_discarder : public operation_no_unset_support {
     public:
         element_discarder(const column_definition& column, expr::expression e)
-            : operation(column, std::move(e)) { }
+            : operation_no_unset_support(column, std::move(e)) { }
         virtual void execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params) override;
     };
 };
diff --git a/cql3/statements/delete_statement.cc b/cql3/statements/delete_statement.cc
index 034bbd9f3f..aebc63bd66 100644
--- a/cql3/statements/delete_statement.cc
+++ b/cql3/statements/delete_statement.cc
@@ -48,6 +48,9 @@ void delete_statement::add_update_for_key(mutation& m, const query::clustering_r
     }
 
     for (auto&& op : _column_operations) {
+        if (op->should_skip_operation(params._options)) {
+            continue;
+        }
         op->execute(m, range.start() ? std::move(range.start()->value()) : clustering_key_prefix::make_empty(), params);
     }
 }
diff --git a/cql3/statements/select_statement.cc b/cql3/statements/select_statement.cc
index 23a61cd71c..949530ca42 100644
--- a/cql3/statements/select_statement.cc
+++ b/cql3/statements/select_statement.cc
@@ -171,7 +171,9 @@ select_statement::select_statement(schema_ptr schema,
     , _restrictions_need_filtering(_restrictions->need_filtering())
     , _group_by_cell_indices(group_by_cell_indices)
     , _is_reversed(is_reversed)
+    , _limit_unset_guard(limit)
     , _limit(std::move(limit))
+    , _per_partition_limit_unset_guard(per_partition_limit)
     , _per_partition_limit(std::move(per_partition_limit))
     , _ordering_comparator(std::move(ordering_comparator))
     , _stats(stats)
@@ -268,8 +270,9 @@ select_statement::make_partition_slice(const query_options& options) const
 
 uint64_t select_statement::do_get_limit(const query_options& options,
                                         const std::optional<expr::expression>& limit,
+                                        const expr::unset_bind_variable_guard& limit_unset_guard,
                                         uint64_t default_limit) const {
-    if (!limit.has_value() || _selection->is_aggregate()) {
+    if (!limit.has_value() || limit_unset_guard.is_unset(options) || _selection->is_aggregate()) {
         return default_limit;
     }
 
@@ -277,9 +280,6 @@ uint64_t select_statement::do_get_limit(const query_options& options,
     if (val.is_null()) {
         throw exceptions::invalid_request_exception("Invalid null value of limit");
     }
-    if (val.is_unset_value()) {
-        return default_limit;
-    }
     try {
         auto l = val.view().validate_and_deserialize<int32_t>(*int32_type);
         if (l <= 0) {
diff --git a/cql3/statements/select_statement.hh b/cql3/statements/select_statement.hh
index 546d5bf75e..29e1dbe53e 100644
--- a/cql3/statements/select_statement.hh
+++ b/cql3/statements/select_statement.hh
@@ -11,6 +11,7 @@
 #pragma once
 
 #include "cql3/statements/raw/select_statement.hh"
+#include "cql3/expr/unset.hh"
 #include "cql3/cql_statement.hh"
 #include "cql3/stats.hh"
 #include <seastar/core/shared_ptr.hh>
@@ -60,7 +61,9 @@ protected:
     const bool _restrictions_need_filtering;
     ::shared_ptr<std::vector<size_t>> _group_by_cell_indices; ///< Indices in result row of cells holding GROUP BY values.
     bool _is_reversed;
+    expr::unset_bind_variable_guard _limit_unset_guard;
     std::optional<expr::expression> _limit;
+    expr::unset_bind_variable_guard _per_partition_limit_unset_guard;
     std::optional<expr::expression> _per_partition_limit;
 
     template<typename T>
@@ -141,12 +144,12 @@ public:
     db::timeout_clock::duration get_timeout(const service::client_state& state, const query_options& options) const;
 
 protected:
-    uint64_t do_get_limit(const query_options& options, const std::optional<expr::expression>& limit, uint64_t default_limit) const;
+    uint64_t do_get_limit(const query_options& options, const std::optional<expr::expression>& limit, const expr::unset_bind_variable_guard& unset_guard, uint64_t default_limit) const;
     uint64_t get_limit(const query_options& options) const {
-        return do_get_limit(options, _limit, query::max_rows);
+        return do_get_limit(options, _limit, _limit_unset_guard, query::max_rows);
     }
     uint64_t get_per_partition_limit(const query_options& options) const {
-        return do_get_limit(options, _per_partition_limit, query::partition_max_rows);
+        return do_get_limit(options, _per_partition_limit, _per_partition_limit_unset_guard, query::partition_max_rows);
     }
     bool needs_post_query_ordering() const;
     virtual void update_stats_rows_read(int64_t rows_read) const {
diff --git a/cql3/statements/update_statement.cc b/cql3/statements/update_statement.cc
index 073dfd6c5b..cc2d45a873 100644
--- a/cql3/statements/update_statement.cc
+++ b/cql3/statements/update_statement.cc
@@ -96,6 +96,9 @@ bool update_statement::allow_clustering_key_slices() const {
 
 void update_statement::execute_operations_for_key(mutation& m, const clustering_key_prefix& prefix, const update_parameters& params, const json_cache_opt& json_cache) const {
     for (auto&& update : _column_operations) {
+        if (update->should_skip_operation(params._options)) {
+            continue;
+        }
         update->execute(m, prefix, params);
     }
 }
diff --git a/cql3/user_types.cc b/cql3/user_types.cc
index 57ca52f51a..06d6c2c8d4 100644
--- a/cql3/user_types.cc
+++ b/cql3/user_types.cc
@@ -25,10 +25,6 @@ void user_types::setter::execute(mutation& m, const clustering_key_prefix& row_k
 }
 
 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;
-    }
-
     auto& type = static_cast<const user_type_impl&>(*column.type);
     if (type.is_multi_cell()) {
         // Non-frozen user defined type.
@@ -79,9 +75,6 @@ void user_types::setter_by_field::execute(mutation& m, const clustering_key_pref
     assert(column.type->is_user_type() && column.type->is_multi_cell());
 
     auto value = expr::evaluate(*_e, params._options);
-    if (value.is_unset_value()) {
-        return;
-    }
 
     auto& type = static_cast<const user_type_impl&>(*column.type);
 
diff --git a/cql3/user_types.hh b/cql3/user_types.hh
index d328bf16bd..d83afa8610 100644
--- a/cql3/user_types.hh
+++ b/cql3/user_types.hh
@@ -26,29 +26,29 @@ class user_types {
 public:
     static lw_shared_ptr<column_specification> field_spec_of(const column_specification& column, size_t field);
 
-    class setter : public operation {
+    class setter : public operation_skip_if_unset {
     public:
-        using operation::operation;
+        using operation_skip_if_unset::operation_skip_if_unset;
 
         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 cql3::raw_value& value);
     };
 
-    class setter_by_field : public operation {
+    class setter_by_field : public operation_skip_if_unset {
         size_t _field_idx;
     public:
         setter_by_field(const column_definition& column, size_t field_idx, expr::expression e)
-            : operation(column, std::move(e)), _field_idx(field_idx) {
+            : operation_skip_if_unset(column, std::move(e)), _field_idx(field_idx) {
         }
 
         virtual void execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params) override;
     };
 
-    class deleter_by_field : public operation {
+    class deleter_by_field : public operation_no_unset_support {
         size_t _field_idx;
     public:
         deleter_by_field(const column_definition& column, size_t field_idx)
-            : operation(column, std::nullopt), _field_idx(field_idx) {
+            : operation_no_unset_support(column, std::nullopt), _field_idx(field_idx) {
         }
 
         virtual void execute(mutation& m, const clustering_key_prefix& row_key, const update_parameters& params) override;
diff --git a/cql3/values.cc b/cql3/values.cc
index a944a22373..87674cca74 100644
--- a/cql3/values.cc
+++ b/cql3/values.cc
@@ -19,8 +19,6 @@ std::ostream& operator<<(std::ostream& os, const raw_value_view& value) {
         os << " }";
     }, [&] (null_value) {
         os << "{ null }";
-    }, [&] (unset_value) {
-        os << "{ unset }";
     });
     return os;
 }
@@ -30,7 +28,7 @@ raw_value_view raw_value::view() const {
     case 0:  return raw_value_view::make_value(managed_bytes_view(bytes_view(std::get<bytes>(_data))));
     case 1:  return raw_value_view::make_value(managed_bytes_view(std::get<managed_bytes>(_data)));
     case 2:  return raw_value_view::make_null();
-    default: return raw_value_view::make_unset_value();
+    default: throw std::runtime_error(fmt::format("raw_value_view::view bad index: {}", _data.index()));
     }
 }
 
@@ -39,7 +37,7 @@ raw_value raw_value::make_value(const raw_value_view& view) {
     case 0:  return raw_value::make_value(managed_bytes(std::get<fragmented_temporary_buffer::view>(view._data)));
     case 1:  return raw_value::make_value(managed_bytes(std::get<managed_bytes_view>(view._data)));
     case 2:  return raw_value::make_null();
-    default: return raw_value::make_unset_value();
+    default: throw std::runtime_error(fmt::format("raw_value_view::make_value bad index: {}", view._data.index()));
     }
 }
 
@@ -48,7 +46,6 @@ raw_value_view raw_value_view::make_temporary(raw_value&& value) {
     case 0:  return raw_value_view(managed_bytes(std::get<bytes>(value._data)));
     case 1:  return raw_value_view(std::move(std::get<managed_bytes>(value._data)));
     case 2:  return raw_value_view::make_null();
-    case 3: return raw_value_view::make_unset_value();
     default: throw std::runtime_error(fmt::format("raw_value_view::make_temporary bad index: {}", value._data.index()));
     }
 }
@@ -59,10 +56,6 @@ raw_value_view::raw_value_view(managed_bytes&& tmp) {
 }
 
 bool operator==(const raw_value& v1, const raw_value& v2) {
-    if (v1.is_unset_value() && v2.is_unset_value()) {
-        return true;
-    }
-
     if (v1.is_null() && v2.is_null()) {
         return true; // note: this is not CQL comparison which would return NULL here
     }
diff --git a/cql3/values.hh b/cql3/values.hh
index 337d618d84..1ff680c1f9 100644
--- a/cql3/values.hh
+++ b/cql3/values.hh
@@ -26,16 +26,12 @@ struct null_value {
     friend bool operator==(const null_value&, const null_value) { return true; }
 };
 
-struct unset_value {
-    friend bool operator==(const unset_value&, const unset_value) { return true; }
-};
-
 class raw_value;
 /// \brief View to a raw CQL protocol value.
 ///
 /// \see raw_value
 class raw_value_view {
-    std::variant<fragmented_temporary_buffer::view, managed_bytes_view, null_value, unset_value> _data;
+    std::variant<fragmented_temporary_buffer::view, managed_bytes_view, null_value> _data;
     // Temporary storage is only useful if a raw_value_view needs to be instantiated
     // with a value which lifetime is bounded only to the view itself.
     // This hack is introduced in order to avoid storing temporary storage
@@ -50,9 +46,6 @@ class raw_value_view {
     raw_value_view(null_value data)
         : _data{std::move(data)}
     {}
-    raw_value_view(unset_value data)
-        : _data{std::move(data)}
-    {}
     raw_value_view(fragmented_temporary_buffer::view data)
         : _data{data}
     {}
@@ -66,9 +59,6 @@ public:
     static raw_value_view make_null() {
         return raw_value_view{null_value{}};
     }
-    static raw_value_view make_unset_value() {
-        return raw_value_view{unset_value{}};
-    }
     static raw_value_view make_value(fragmented_temporary_buffer::view view) {
         return raw_value_view{view};
     }
@@ -82,13 +72,10 @@ public:
     bool is_null() const {
         return std::holds_alternative<null_value>(_data);
     }
-    bool is_unset_value() const {
-        return std::holds_alternative<unset_value>(_data);
-    }
-    // An empty value is not null or unset, but it has 0 bytes of data.
+    // An empty value is not null, but it has 0 bytes of data.
     // An empty int value can be created in CQL using blobasint(0x).
     bool is_empty_value() const {
-        if (is_null() || is_unset_value()) {
+        if (is_null()) {
             return false;
         }
         return size_bytes() == 0;
@@ -174,17 +161,14 @@ public:
 /// \brief Raw CQL protocol value.
 ///
 /// The `raw_value` type represents an uninterpreted value from the CQL wire
-/// protocol. A raw value can hold either a null value, an unset value, or a byte
+/// protocol. A raw value can hold either a null value, or a byte
 /// blob that represents the value.
 class raw_value {
-    std::variant<bytes, managed_bytes, null_value, unset_value> _data;
+    std::variant<bytes, managed_bytes, null_value> _data;
 
     raw_value(null_value&& data)
         : _data{std::move(data)}
     {}
-    raw_value(unset_value&& data)
-        : _data{std::move(data)}
-    {}
     raw_value(bytes&& data)
         : _data{std::move(data)}
     {}
@@ -201,9 +185,6 @@ public:
     static raw_value make_null() {
         return raw_value{null_value{}};
     }
-    static raw_value make_unset_value() {
-        return raw_value{unset_value{}};
-    }
     static raw_value make_value(const raw_value_view& view);
     static raw_value make_value(managed_bytes&& mb) {
         return raw_value{std::move(mb)};
@@ -235,16 +216,10 @@ public:
     bool is_null() const {
         return std::holds_alternative<null_value>(_data);
     }
-    bool is_unset_value() const {
-        return std::holds_alternative<unset_value>(_data);
-    }
-    bool is_null_or_unset() const {
-        return !is_value();
-    }
-    // An empty value is not null or unset, but it has 0 bytes of data.
+    // An empty value is not null, but it has 0 bytes of data.
     // An empty int value can be created in CQL using blobasint(0x).
     bool is_empty_value() const {
-        if (is_null_or_unset()) {
+        if (is_null()) {
             return false;
         }
         return view().size_bytes() == 0;
@@ -262,9 +237,6 @@ public:
             [](null_value&&) -> bytes {
                 throw std::runtime_error("to_bytes() called on raw value that is null");
             },
-            [](unset_value&&) -> bytes {
-                throw std::runtime_error("to_bytes() called on raw value that is unset_value");
-            }
         }, std::move(_data));
     }
     bytes_opt to_bytes_opt() && {
@@ -274,9 +246,6 @@ public:
             [](null_value&&) -> bytes_opt {
                 return std::nullopt;
             },
-            [](unset_value&&) -> bytes_opt {
-                return std::nullopt;
-            }
         }, std::move(_data));
     }
     managed_bytes to_managed_bytes() && {
@@ -286,9 +255,6 @@ public:
             [](null_value&&) -> managed_bytes {
                 throw std::runtime_error("to_managed_bytes() called on raw value that is null");
             },
-            [](unset_value&&) -> managed_bytes {
-                throw std::runtime_error("to_managed_bytes() called on raw value that is unset_value");
-            }
         }, std::move(_data));
     }
     managed_bytes_opt to_managed_bytes_opt() && {
@@ -298,9 +264,6 @@ public:
             [](null_value&&) -> managed_bytes_opt {
                 return std::nullopt;
             },
-            [](unset_value&&) -> managed_bytes_opt {
-                return std::nullopt;
-            }
         }, std::move(_data));
     }
     raw_value_view view() const;
diff --git a/service/raft/raft_sys_table_storage.cc b/service/raft/raft_sys_table_storage.cc
index 57495c513b..f921b507fd 100644
--- a/service/raft/raft_sys_table_storage.cc
+++ b/service/raft/raft_sys_table_storage.cc
@@ -204,7 +204,7 @@ future<> raft_sys_table_storage::do_store_log_entries(const std::vector<raft::lo
     // fragmented storage for log_entries data
     std::vector<fragmented_temporary_buffer> stmt_data_views;
     // statement value views -- required for `query_options` to consume `fragmented_temporary_buffer::view`
-    std::vector<std::vector<cql3::raw_value_view>> stmt_value_views;
+    std::vector<cql3::raw_value_view_vector_with_unset> stmt_value_views;
     const size_t entries_size = entries.size();
     batch_stmts.reserve(entries_size);
     stmt_values.reserve(entries_size);
diff --git a/test/boost/cql_query_test.cc b/test/boost/cql_query_test.cc
index 97660ed8db..e152b13440 100644
--- a/test/boost/cql_query_test.cc
+++ b/test/boost/cql_query_test.cc
@@ -544,10 +544,8 @@ SEASTAR_TEST_CASE(test_bound_var_in_collection_literal) {
                 exceptions::invalid_request_exception
             );
 
-            // Unset value is not allowed as a collections element
-            const auto unset_value = cql3::raw_value::make_unset_value();
             BOOST_REQUIRE_THROW(
-                e.execute_prepared(stmt, {unset_value}).get(),
+                e.execute_prepared(stmt, cql3::raw_value_vector_with_unset({null_value}, {true})).get(),
                 exceptions::invalid_request_exception
             );
 
@@ -5456,8 +5454,6 @@ cql3::raw_value make_collection_raw_value(size_t size_to_write, const std::vecto
     for (const cql3::raw_value& val : elements_to_write) {
         if (val.is_null()) {
                 write_int32(out, -1);
-        } else if (val.is_unset_value()) {
-                write_int32(out, -2);
         } else {
             val.view().with_value([&](const FragmentedView auto& val_view) {
                 write_collection_value(out, linearized(val_view));
@@ -5478,7 +5474,7 @@ SEASTAR_TEST_CASE(test_null_and_unset_in_collections) {
         };
 
         auto check_unset_msg = [](std::source_location loc = std::source_location::current()) {
-            return exception_predicate::message_equals("unset value is not supported inside collections", loc);
+            return exception_predicate::message_contains("unset", loc);
         };
 
         // Test null when specified inside a collection literal
@@ -5500,7 +5496,6 @@ SEASTAR_TEST_CASE(test_null_and_unset_in_collections) {
         auto insert_map_with_value_marker = e.prepare("INSERT INTO null_in_col (p, m) VALUES (0, {0:1, 2:?, 4:5})").get0();
 
         cql3::raw_value null_value = cql3::raw_value::make_null();
-        cql3::raw_value unset_value = cql3::raw_value::make_unset_value();
 
         BOOST_REQUIRE_EXCEPTION(e.execute_prepared(insert_list_with_marker, {null_value}).get(),
                                 exceptions::invalid_request_exception, check_null_msg());
@@ -5511,13 +5506,15 @@ SEASTAR_TEST_CASE(test_null_and_unset_in_collections) {
         BOOST_REQUIRE_EXCEPTION(e.execute_prepared(insert_map_with_value_marker, {null_value}).get(),
                                 exceptions::invalid_request_exception, check_null_msg());
 
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(insert_list_with_marker, {unset_value}).get(),
+        auto bind_variable_list_with_unset = cql3::raw_value_vector_with_unset({null_value}, {true});
+
+        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(insert_list_with_marker, bind_variable_list_with_unset).get(),
                                 exceptions::invalid_request_exception, check_unset_msg());
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(insert_set_with_marker, {unset_value}).get(),
+        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(insert_set_with_marker, bind_variable_list_with_unset).get(),
                                 exceptions::invalid_request_exception, check_unset_msg());
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(insert_map_with_key_marker, {unset_value}).get(),
+        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(insert_map_with_key_marker, bind_variable_list_with_unset).get(),
                                 exceptions::invalid_request_exception, check_unset_msg());
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(insert_map_with_value_marker, {unset_value}).get(),
+        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(insert_map_with_value_marker, bind_variable_list_with_unset).get(),
                                 exceptions::invalid_request_exception, check_unset_msg());
 
 
@@ -5551,27 +5548,6 @@ SEASTAR_TEST_CASE(test_null_and_unset_in_collections) {
                                 exceptions::invalid_request_exception, check_null_msg());
 
 
-        cql3::raw_value list_with_unset = make_collection_raw_value(3, {make_int(1), unset_value, make_int(2)});
-        cql3::raw_value set_with_unset = make_collection_raw_value(3, {make_int(1), unset_value, make_int(2)});
-
-        cql3::raw_value map_with_unset_key = make_collection_raw_value(3, {make_int(0), make_int(1),
-                                                                     unset_value, make_int(3),
-                                                                     make_int(4), make_int(5)});
-
-        cql3::raw_value map_with_unset_value = make_collection_raw_value(3, {make_int(0), make_int(1),
-                                                                       make_int(2), unset_value,
-                                                                       make_int(4), make_int(5)});
-
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(insert_list, {list_with_unset}).get(),
-                                exceptions::invalid_request_exception, check_unset_msg());
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(insert_set, {set_with_unset}).get(),
-                                exceptions::invalid_request_exception, check_unset_msg());
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(insert_map, {map_with_unset_key}).get(),
-                                exceptions::invalid_request_exception, check_unset_msg());
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(insert_map, {map_with_unset_value}).get(),
-                                exceptions::invalid_request_exception, check_unset_msg());
-
-
         // Update setting to bad collection value
         BOOST_REQUIRE_EXCEPTION(e.execute_cql("UPDATE null_in_col SET l = [1, null, 2] WHERE p = 0").get(),
                                 exceptions::invalid_request_exception, check_null_msg());
@@ -5617,14 +5593,13 @@ SEASTAR_TEST_CASE(test_null_and_unset_in_collections) {
         BOOST_REQUIRE_EXCEPTION(e.execute_prepared(add_map_with_value_marker, {null_value}).get(),
                                 exceptions::invalid_request_exception, check_null_msg());
 
-
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(add_list_with_marker, {unset_value}).get(),
+        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(add_list_with_marker, bind_variable_list_with_unset).get(),
                                 exceptions::invalid_request_exception, check_unset_msg());
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(add_set_with_marker, {unset_value}).get(),
+        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(add_set_with_marker, bind_variable_list_with_unset).get(),
                                 exceptions::invalid_request_exception, check_unset_msg());
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(add_map_with_key_marker, {unset_value}).get(),
+        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(add_map_with_key_marker, bind_variable_list_with_unset).get(),
                                 exceptions::invalid_request_exception, check_unset_msg());
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(add_map_with_value_marker, {unset_value}).get(),
+        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(add_map_with_value_marker, bind_variable_list_with_unset).get(),
                                 exceptions::invalid_request_exception, check_unset_msg());
 
         // Update adding a collection value with bad bind marker
@@ -5641,15 +5616,6 @@ SEASTAR_TEST_CASE(test_null_and_unset_in_collections) {
         BOOST_REQUIRE_EXCEPTION(e.execute_prepared(add_map, {map_with_null_value}).get(),
                                 exceptions::invalid_request_exception, check_null_msg());
 
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(add_list, {list_with_unset}).get(),
-                                exceptions::invalid_request_exception, check_unset_msg());
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(add_set, {set_with_unset}).get(),
-                                exceptions::invalid_request_exception, check_unset_msg());
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(add_map, {map_with_unset_key}).get(),
-                                exceptions::invalid_request_exception, check_unset_msg());
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(add_map, {map_with_unset_value}).get(),
-                                exceptions::invalid_request_exception, check_unset_msg());
-
         // List of IN values is also a list that can't contain nulls
         BOOST_REQUIRE_EXCEPTION(e.execute_cql("SELECT * FROM null_in_col WHERE p IN (1, null, 2)").get(),
                                 exceptions::invalid_request_exception, check_null_msg());
@@ -5658,15 +5624,13 @@ SEASTAR_TEST_CASE(test_null_and_unset_in_collections) {
 
         BOOST_REQUIRE_EXCEPTION(e.execute_prepared(where_in_list_with_marker, {null_value}).get(),
                                 exceptions::invalid_request_exception, check_null_msg());
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(where_in_list_with_marker, {unset_value}).get(),
+        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(where_in_list_with_marker, bind_variable_list_with_unset).get(),
                                 exceptions::invalid_request_exception, check_unset_msg());
 
         auto where_in_list_marker = e.prepare("SELECT * FROM null_in_col WHERE p IN ?").get0();
 
         BOOST_REQUIRE_EXCEPTION(e.execute_prepared(where_in_list_marker, {list_with_null}).get(),
                                 exceptions::invalid_request_exception, check_null_msg());
-        BOOST_REQUIRE_EXCEPTION(e.execute_prepared(where_in_list_marker, {list_with_unset}).get(),
-                                exceptions::invalid_request_exception, check_unset_msg());
     });
 }
 
diff --git a/test/boost/expr_test.cc b/test/boost/expr_test.cc
index ad314d7006..b3dea87c52 100644
--- a/test/boost/expr_test.cc
+++ b/test/boost/expr_test.cc
@@ -21,10 +21,10 @@ using namespace cql3;
 using namespace cql3::expr;
 using namespace cql3::expr::test_utils;
 
-bind_variable new_bind_variable(int bind_index) {
+bind_variable new_bind_variable(int bind_index, data_type type = int32_type) {
     return bind_variable {
         .bind_index = bind_index,
-        .receiver = nullptr
+        .receiver = make_lw_shared<column_specification>("ks", "tab", make_shared<column_identifier>("?", true), std::move(type)),
     };
 }
 
@@ -383,11 +383,6 @@ BOOST_AUTO_TEST_CASE(evaluate_constant_null) {
     BOOST_REQUIRE_EQUAL(evaluate(constant_null_with_type, evaluation_inputs{}), raw_value::make_null());
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_constant_unset) {
-    expression constant_unset = constant::make_unset_value();
-    BOOST_REQUIRE_EQUAL(evaluate(constant_unset, evaluation_inputs{}), raw_value::make_unset_value());
-}
-
 BOOST_AUTO_TEST_CASE(evaluate_constant_empty) {
     expression constant_empty_bool = constant(raw_value::make_value(bytes()), boolean_type);
     BOOST_REQUIRE(evaluate(constant_empty_bool, evaluation_inputs{}).is_empty_value());
@@ -532,6 +527,11 @@ BOOST_AUTO_TEST_CASE(evaluate_bind_variable_performs_validation) {
     BOOST_REQUIRE_THROW(evaluate(bind_var, inputs), exceptions::invalid_request_exception);
 }
 
+BOOST_AUTO_TEST_CASE(evaluate_bind_variable_vs_unset) {
+    auto qo = query_options(cql3::raw_value_vector_with_unset({raw_value::make_null()}, {true}));
+    BOOST_REQUIRE_THROW(evaluate(new_bind_variable(0), evaluation_inputs{.options = &qo}), exceptions::invalid_request_exception);
+}
+
 BOOST_AUTO_TEST_CASE(evaluate_list_collection_constructor_empty) {
     // TODO: Empty multi-cell collections are trated as NULL in the database,
     // should the conversion happen in evaluate?
@@ -556,12 +556,6 @@ BOOST_AUTO_TEST_CASE(evaluate_list_collection_constructor_with_null) {
     BOOST_REQUIRE_THROW(evaluate(list_with_null, evaluation_inputs{}), exceptions::invalid_request_exception);
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_list_collection_constructor_with_unset) {
-    expression list_with_unset = make_list_constructor(
-        {make_int_const(1), constant::make_unset_value(int32_type), make_int_const(3)}, int32_type);
-    BOOST_REQUIRE_THROW(evaluate(list_with_unset, evaluation_inputs{}), exceptions::invalid_request_exception);
-}
-
 BOOST_AUTO_TEST_CASE(evaluate_list_collection_constructor_with_empty_value) {
     expression list_with_empty =
         make_list_constructor({make_int_const(1), make_empty_const(int32_type), make_int_const(3)}, int32_type);
@@ -593,12 +587,6 @@ BOOST_AUTO_TEST_CASE(evaluate_set_collection_constructor_with_null) {
     BOOST_REQUIRE_THROW(evaluate(set_with_null, evaluation_inputs{}), exceptions::invalid_request_exception);
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_set_collection_constructor_with_unset) {
-    expression set_with_unset = make_set_constructor(
-        {make_int_const(1), constant::make_unset_value(int32_type), make_int_const(3)}, int32_type);
-    BOOST_REQUIRE_THROW(evaluate(set_with_unset, evaluation_inputs{}), exceptions::invalid_request_exception);
-}
-
 BOOST_AUTO_TEST_CASE(evaluate_set_collection_constructor_with_empty) {
     expression set_with_only_one_empty = make_set_constructor({make_empty_const(int32_type)}, int32_type);
     BOOST_REQUIRE_EQUAL(evaluate(set_with_only_one_empty, evaluation_inputs{}), make_set_raw({make_empty_raw()}));
@@ -662,28 +650,6 @@ BOOST_AUTO_TEST_CASE(evaluate_map_collection_constructor_with_null_value) {
     BOOST_REQUIRE_THROW(evaluate(map_with_null_value, evaluation_inputs{}), exceptions::invalid_request_exception);
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_map_collection_constructor_with_unset_key) {
-    expression map_with_unset_key = make_map_constructor(
-        {
-            {make_int_const(1), make_int_const(2)},
-            {constant::make_unset_value(int32_type), make_int_const(4)},
-            {make_int_const(5), make_int_const(6)},
-        },
-        int32_type, int32_type);
-    BOOST_REQUIRE_THROW(evaluate(map_with_unset_key, evaluation_inputs{}), exceptions::invalid_request_exception);
-}
-
-BOOST_AUTO_TEST_CASE(evaluate_map_collection_constructor_with_unset_value) {
-    expression map_with_unset_value = make_map_constructor(
-        {
-            {make_int_const(1), make_int_const(2)},
-            {make_int_const(3), constant::make_unset_value(int32_type)},
-            {make_int_const(5), make_int_const(6)},
-        },
-        int32_type, int32_type);
-    BOOST_REQUIRE_THROW(evaluate(map_with_unset_value, evaluation_inputs{}), exceptions::invalid_request_exception);
-}
-
 BOOST_AUTO_TEST_CASE(evaluate_map_collection_constructor_with_empty_key) {
     expression map_with_empty_key = make_map_constructor(
         {
@@ -736,12 +702,6 @@ BOOST_AUTO_TEST_CASE(evaluate_tuple_constructor_with_null) {
                         make_tuple_raw({make_int_raw(12), raw_value::make_null(), make_int_raw(34)}));
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_tuple_constructor_with_unset) {
-    expression tuple_with_unset =
-        make_tuple_constructor({make_int_const(12), constant::make_unset_value(int32_type)}, {int32_type, utf8_type});
-    BOOST_REQUIRE_THROW(evaluate(tuple_with_unset, evaluation_inputs{}), exceptions::invalid_request_exception);
-}
-
 BOOST_AUTO_TEST_CASE(evaluate_tuple_constructor_with_empty) {
     expression tuple_with_empty = make_tuple_constructor(
         {make_int_const(12), make_empty_const(int32_type), make_int_const(34)}, {int32_type, utf8_type, int32_type});
@@ -776,13 +736,6 @@ BOOST_AUTO_TEST_CASE(evaluate_usertype_constructor_with_null) {
                         make_tuple_raw({make_int_raw(123), raw_value::make_null(), make_bool_raw(true)}));
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_usertype_constructor_with_unset) {
-    expression usertype_with_unset = make_usertype_constructor({{"field1", make_int_const(123)},
-                                                                {"field2", constant::make_unset_value(utf8_type)},
-                                                                {"field3", make_bool_const(true)}});
-    BOOST_REQUIRE_THROW(evaluate(usertype_with_unset, evaluation_inputs{}), exceptions::invalid_request_exception);
-}
-
 BOOST_AUTO_TEST_CASE(evaluate_usertype_constructor_with_empty) {
     expression usertype_with_null = make_usertype_constructor(
         {{"field1", make_int_const(123)}, {"field2", make_empty_const(utf8_type)}, {"field3", make_bool_const(true)}});
@@ -818,8 +771,6 @@ BOOST_AUTO_TEST_CASE(evalaute_subscripted_empty_list) {
     BOOST_REQUIRE_EQUAL(evaluate_subscripted(list, make_int_const(std::numeric_limits<int32_t>::min())),
                         raw_value::make_null());
     BOOST_REQUIRE_EQUAL(evaluate_subscripted(list, constant::make_null(int32_type)), raw_value::make_null());
-    BOOST_REQUIRE_THROW(evaluate_subscripted(list, constant::make_unset_value(int32_type)),
-                        exceptions::invalid_request_exception);
     
     // TODO: Should empty value list indexes cause an error? Why not return NULL?
     BOOST_REQUIRE_THROW(evaluate_subscripted(list, make_empty_const(int32_type)), empty_value_exception);
@@ -873,12 +824,6 @@ BOOST_AUTO_TEST_CASE(evaluate_subscripted_list_null_index) {
     BOOST_REQUIRE_EQUAL(evaluate_subscripted(list, constant::make_null(int32_type)), raw_value::make_null());
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_subscripted_list_unset_index) {
-    constant list = make_subscript_test_list();
-    BOOST_REQUIRE_THROW(evaluate_subscripted(list, constant::make_unset_value(int32_type)),
-                        exceptions::invalid_request_exception);
-}
-
 BOOST_AUTO_TEST_CASE(evaluate_subscripted_list_empty_index) {
     constant list = make_subscript_test_list();
     // TODO: Should empty value list indexes cause an error? Why not return NULL?
@@ -890,9 +835,6 @@ BOOST_AUTO_TEST_CASE(evaluate_subscripted_list_null_list) {
     BOOST_REQUIRE_EQUAL(evaluate_subscripted(list, make_int_const(0)), raw_value::make_null());
     BOOST_REQUIRE_EQUAL(evaluate_subscripted(list, make_empty_const(int32_type)), raw_value::make_null());
     BOOST_REQUIRE_EQUAL(evaluate_subscripted(list, constant::make_null(int32_type)), raw_value::make_null());
-
-    // TODO: Shouldn't this throw an error?
-    BOOST_REQUIRE_EQUAL(evaluate_subscripted(list, constant::make_unset_value(int32_type)), raw_value::make_null());
 }
 
 BOOST_AUTO_TEST_CASE(evaluate_subscripted_empty_map) {
@@ -907,8 +849,6 @@ BOOST_AUTO_TEST_CASE(evaluate_subscripted_empty_map) {
     BOOST_REQUIRE_EQUAL(evaluate_subscripted(map, make_int_const(std::numeric_limits<int32_t>::max())),
                         raw_value::make_null());
     BOOST_REQUIRE_EQUAL(evaluate_subscripted(map, constant::make_null(int32_type)), raw_value::make_null());
-    BOOST_REQUIRE_THROW(evaluate_subscripted(map, constant::make_unset_value(int32_type)),
-                        exceptions::invalid_request_exception);
     BOOST_REQUIRE_EQUAL(evaluate_subscripted(map, make_empty_const(int32_type)), raw_value::make_null());
 }
 
@@ -954,12 +894,6 @@ BOOST_AUTO_TEST_CASE(evaluate_subscripted_map_null_index) {
     BOOST_REQUIRE_EQUAL(evaluate_subscripted(map, constant::make_null(int32_type)), raw_value::make_null());
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_subscripted_map_unset_index) {
-    constant map = make_subscript_test_map();
-    BOOST_REQUIRE_THROW(evaluate_subscripted(map, constant::make_unset_value(int32_type)),
-                        exceptions::invalid_request_exception);
-}
-
 BOOST_AUTO_TEST_CASE(evaluate_subscripted_map_empty) {
     // Empty list values seem to not be allowed.
     constant map = make_empty_const(map_type_impl::get_instance(int32_type, int32_type, true));
@@ -971,9 +905,6 @@ BOOST_AUTO_TEST_CASE(evaluate_subscripted_map_null_map) {
     BOOST_REQUIRE_EQUAL(evaluate_subscripted(map, make_int_const(0)), raw_value::make_null());
     BOOST_REQUIRE_EQUAL(evaluate_subscripted(map, make_empty_const(int32_type)), raw_value::make_null());
     BOOST_REQUIRE_EQUAL(evaluate_subscripted(map, constant::make_null(int32_type)), raw_value::make_null());
-
-    // TODO: Shouldn't this throw an error?
-    BOOST_REQUIRE_EQUAL(evaluate_subscripted(map, constant::make_unset_value(int32_type)), raw_value::make_null());
 }
 
 enum expected_invalid_or_valid { expected_valid, expected_invalid };
@@ -989,7 +920,7 @@ static void check_bind_variable_evaluate(constant check_value, expected_invalid_
     expression bind_var = bind_variable{.bind_index = 0, .receiver = make_receiver(check_value.type, "bind_var")};
 
     auto [inputs, inputs_data] = make_evaluation_inputs(
-        test_schema, {{"pk", make_int_raw(0)}, {"r", raw_value::make_null()}}, {check_value.value});
+        test_schema, {{"pk", make_int_raw(0)}, {"r", cql3::raw_value::make_null()}}, {check_value.value});
 
     switch (expected_validity) {
         case expected_valid:
@@ -1007,12 +938,6 @@ BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_no_null_in_list) {
     check_bind_variable_evaluate(list_with_null, expected_invalid);
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_no_unset_in_list) {
-    constant list_with_unset =
-        make_list_const({make_int_const(1), constant::make_unset_value(int32_type), make_int_const(2)}, int32_type);
-    check_bind_variable_evaluate(list_with_unset, expected_invalid);
-}
-
 BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_empty_in_list) {
     constant lis_with_empty =
         make_list_const({make_int_const(1), make_empty_const(int32_type), make_int_const(2)}, int32_type);
@@ -1025,12 +950,6 @@ BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_no_null_in_set) {
     check_bind_variable_evaluate(set_with_null, expected_invalid);
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_no_unset_in_set) {
-    constant set_with_unset =
-        make_set_const({make_int_const(1), constant::make_unset_value(int32_type), make_int_const(2)}, int32_type);
-    check_bind_variable_evaluate(set_with_unset, expected_invalid);
-}
-
 // TODO: This fails, but I feel like this is a bug.
 // BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_empty_in_set) {
 //     constant set_with_empty =
@@ -1046,14 +965,6 @@ BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_no_null_key_in_map) {
     check_bind_variable_evaluate(map_with_null_key, expected_invalid);
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_no_unset_key_in_map) {
-    constant map_with_unset_key = make_map_const({{make_int_const(1), make_int_const(2)},
-                                                  {constant::make_unset_value(int32_type), make_int_const(4)},
-                                                  {make_int_const(5), make_int_const(6)}},
-                                                 int32_type, int32_type);
-    check_bind_variable_evaluate(map_with_unset_key, expected_invalid);
-}
-
 BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_empty_key_in_map) {
     constant map_with_empty_key = make_map_const({{make_empty_const(int32_type), make_int_const(4)},
                                                   {make_int_const(1), make_int_const(2)},
@@ -1070,14 +981,6 @@ BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_no_null_value_in_map) {
     check_bind_variable_evaluate(map_with_null_value, expected_invalid);
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_no_unset_value_in_map) {
-    constant map_with_unset_value = make_map_const({{make_int_const(1), make_int_const(2)},
-                                                    {make_int_const(3), constant::make_unset_value(int32_type)},
-                                                    {make_int_const(5), make_int_const(6)}},
-                                                   int32_type, int32_type);
-    check_bind_variable_evaluate(map_with_unset_value, expected_invalid);
-}
-
 BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_empty_value_in_map) {
     constant map_with_empty_value = make_map_const({{make_int_const(1), make_int_const(2)},
                                                     {make_int_const(3), make_empty_const(int32_type)},
@@ -1113,11 +1016,6 @@ BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_null_in_lists_recursively)
     check_bind_variable_evaluate(list_with_null, expected_invalid);
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_unset_in_lists_recursively) {
-    constant list_with_unset = create_nested_list_with_value(constant::make_unset_value(int32_type));
-    check_bind_variable_evaluate(list_with_unset, expected_invalid);
-}
-
 // TODO: This fails, but I feel like this is a bug.
 // BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_empty_in_lists_recursively) {
 //     constant list_with_empty = create_nested_list_or_set_with_value(make_empty_const(int32_type));
@@ -1150,11 +1048,6 @@ BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_null_in_sets_recursively)
     check_bind_variable_evaluate(set_with_null, expected_invalid);
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_unset_in_sets_recursively) {
-    constant set_with_unset = create_nested_set_with_value(constant::make_unset_value(int32_type));
-    check_bind_variable_evaluate(set_with_unset, expected_invalid);
-}
-
 BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_empty_in_sets_recursively) {
     constant set_with_empty = create_nested_set_with_value(make_empty_const(int32_type));
     check_bind_variable_evaluate(set_with_empty, expected_valid);
@@ -1207,16 +1100,6 @@ BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_null_key_in_maps_recursive
     check_bind_variable_evaluate(map_with_null_key2, expected_invalid);
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_unset_key_in_maps_recursively) {
-    constant map_with_unset_key1 =
-        create_nested_map_with_key(constant::make_unset_value(int32_type), make_int_const(13));
-    check_bind_variable_evaluate(map_with_unset_key1, expected_invalid);
-
-    constant map_with_unset_key2 =
-        create_nested_map_with_key(make_int_const(1), constant::make_unset_value(int32_type));
-    check_bind_variable_evaluate(map_with_unset_key2, expected_invalid);
-}
-
 BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_empty_key_in_maps_recursively) {
     constant map_with_empty_key1 = create_nested_map_with_key(make_empty_const(int32_type), make_int_const(13));
     check_bind_variable_evaluate(map_with_empty_key1, expected_valid);
@@ -1272,16 +1155,6 @@ BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_null_value_in_maps_recursi
     check_bind_variable_evaluate(map_with_null_value2, expected_invalid);
 }
 
-BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_unset_value_in_maps_recursively) {
-    constant map_with_unset_value1 =
-        create_nested_map_with_value(constant::make_unset_value(int32_type), make_int_const(13));
-    check_bind_variable_evaluate(map_with_unset_value1, expected_invalid);
-
-    constant map_with_unset_value2 =
-        create_nested_map_with_value(make_int_const(1), constant::make_unset_value(int32_type));
-    check_bind_variable_evaluate(map_with_unset_value2, expected_invalid);
-}
-
 BOOST_AUTO_TEST_CASE(evaluate_bind_variable_validates_empty_value_in_maps_recursively) {
     constant map_with_empty_value1 = create_nested_map_with_value(make_empty_const(int32_type), make_int_const(13));
     check_bind_variable_evaluate(map_with_empty_value1, expected_valid);
@@ -2515,14 +2388,11 @@ BOOST_AUTO_TEST_CASE(prepare_usertype_constructor_with_bind_variable_and_missing
     BOOST_REQUIRE_EQUAL(prepared, expected);
 }
 
-// Test how evaluating a given binary operator behaves when null and unset are present.
+// Test how evaluating a given binary operator behaves when null is present.
 // A binary with null on either side should evaluate to null.
-// When UNSET_VALUE is present evaluating should throw an exception.
-static void test_evaluate_binop_null_unset(oper_t op, expression valid_lhs, expression valid_rhs) {
+static void test_evaluate_binop_null(oper_t op, expression valid_lhs, expression valid_rhs) {
     constant lhs_null_val = constant::make_null(type_of(valid_lhs));
     constant rhs_null_val = constant::make_null(type_of(valid_rhs));
-    constant lhs_unset_val = constant::make_unset_value(type_of(valid_lhs));
-    constant rhs_unset_val = constant::make_unset_value(type_of(valid_rhs));
 
     expression valid_binop = binary_operator(valid_lhs, op, valid_rhs);
     BOOST_REQUIRE(evaluate(valid_binop, evaluation_inputs{}).is_value());
@@ -2535,21 +2405,6 @@ static void test_evaluate_binop_null_unset(oper_t op, expression valid_lhs, expr
 
     expression binop_both_null = binary_operator(lhs_null_val, op, rhs_null_val);
     BOOST_REQUIRE_EQUAL(evaluate(binop_both_null, evaluation_inputs{}), raw_value::make_null());
-
-    expression binop_lhs_unset = binary_operator(lhs_unset_val, op, valid_rhs);
-    BOOST_REQUIRE_THROW(evaluate(binop_lhs_unset, evaluation_inputs{}), exceptions::invalid_request_exception);
-
-    expression binop_rhs_unset = binary_operator(valid_lhs, op, rhs_unset_val);
-    BOOST_REQUIRE_THROW(evaluate(binop_rhs_unset, evaluation_inputs{}), exceptions::invalid_request_exception);
-
-    expression binop_both_unset = binary_operator(lhs_unset_val, op, rhs_unset_val);
-    BOOST_REQUIRE_THROW(evaluate(binop_both_unset, evaluation_inputs{}), exceptions::invalid_request_exception);
-
-    expression binop_lhs_null_rhs_unset = binary_operator(lhs_null_val, op, rhs_unset_val);
-    BOOST_REQUIRE_THROW(evaluate(binop_lhs_null_rhs_unset, evaluation_inputs{}), exceptions::invalid_request_exception);
-
-    expression binop_lhs_unset_rhs_null = binary_operator(lhs_unset_val, op, rhs_null_val);
-    BOOST_REQUIRE_THROW(evaluate(binop_lhs_unset_rhs_null, evaluation_inputs{}), exceptions::invalid_request_exception);
 }
 
 BOOST_AUTO_TEST_CASE(evaluate_binary_operator_eq) {
@@ -2565,7 +2420,7 @@ BOOST_AUTO_TEST_CASE(evaluate_binary_operator_eq) {
     expression empty_neq = binary_operator(make_int_const(0), oper_t::EQ, make_empty_const(int32_type));
     BOOST_REQUIRE_EQUAL(evaluate(empty_neq, evaluation_inputs{}), make_bool_raw(false));
 
-    test_evaluate_binop_null_unset(oper_t::EQ, make_int_const(123), make_int_const(456));
+    test_evaluate_binop_null(oper_t::EQ, make_int_const(123), make_int_const(456));
 }
 
 BOOST_AUTO_TEST_CASE(evaluate_binary_operator_neq) {
@@ -2582,7 +2437,7 @@ BOOST_AUTO_TEST_CASE(evaluate_binary_operator_neq) {
     expression empty_neq_0 = binary_operator(make_empty_const(int32_type), oper_t::NEQ, make_int_const(0));
     BOOST_REQUIRE_EQUAL(evaluate(empty_neq_0, evaluation_inputs{}), make_bool_raw(true));
 
-    test_evaluate_binop_null_unset(oper_t::NEQ, make_int_const(123), make_int_const(456));
+    test_evaluate_binop_null(oper_t::NEQ, make_int_const(123), make_int_const(456));
 }
 
 BOOST_AUTO_TEST_CASE(evaluate_binary_operator_lt) {
@@ -2602,7 +2457,7 @@ BOOST_AUTO_TEST_CASE(evaluate_binary_operator_lt) {
         binary_operator(make_empty_const(int32_type), oper_t::LT, make_int_const(std::numeric_limits<int32_t>::min()));
     BOOST_REQUIRE_EQUAL(evaluate(empty_lt_int_min, evaluation_inputs{}), make_bool_raw(true));
 
-    test_evaluate_binop_null_unset(oper_t::LT, make_int_const(123), make_int_const(456));
+    test_evaluate_binop_null(oper_t::LT, make_int_const(123), make_int_const(456));
 }
 
 BOOST_AUTO_TEST_CASE(evaluate_binary_operator_lte) {
@@ -2623,7 +2478,7 @@ BOOST_AUTO_TEST_CASE(evaluate_binary_operator_lte) {
         binary_operator(make_empty_const(int32_type), oper_t::LT, make_int_const(std::numeric_limits<int32_t>::min()));
     BOOST_REQUIRE_EQUAL(evaluate(empty_lte_int_min, evaluation_inputs{}), make_bool_raw(true));
 
-    test_evaluate_binop_null_unset(oper_t::LTE, make_int_const(123), make_int_const(456));
+    test_evaluate_binop_null(oper_t::LTE, make_int_const(123), make_int_const(456));
 }
 
 BOOST_AUTO_TEST_CASE(evaluate_binary_operator_gt) {
@@ -2643,7 +2498,7 @@ BOOST_AUTO_TEST_CASE(evaluate_binary_operator_gt) {
         binary_operator(make_int_const(std::numeric_limits<int32_t>::min()), oper_t::GT, make_empty_const(int32_type));
     BOOST_REQUIRE_EQUAL(evaluate(int_min_gt_empty, evaluation_inputs{}), make_bool_raw(true));
 
-    test_evaluate_binop_null_unset(oper_t::GT, make_int_const(234), make_int_const(-3434));
+    test_evaluate_binop_null(oper_t::GT, make_int_const(234), make_int_const(-3434));
 }
 
 BOOST_AUTO_TEST_CASE(evaluate_binary_operator_gte) {
@@ -2664,7 +2519,7 @@ BOOST_AUTO_TEST_CASE(evaluate_binary_operator_gte) {
         binary_operator(make_int_const(std::numeric_limits<int32_t>::min()), oper_t::GTE, make_empty_const(int32_type));
     BOOST_REQUIRE_EQUAL(evaluate(int_min_gte_empty, evaluation_inputs{}), make_bool_raw(true));
 
-    test_evaluate_binop_null_unset(oper_t::GTE, make_int_const(234), make_int_const(-3434));
+    test_evaluate_binop_null(oper_t::GTE, make_int_const(234), make_int_const(-3434));
 }
 
 BOOST_AUTO_TEST_CASE(evaluate_binary_operator_in) {
@@ -2691,7 +2546,7 @@ BOOST_AUTO_TEST_CASE(evaluate_binary_operator_in) {
     expression nonexisting_int_in_list_with_empty = binary_operator(make_int_const(321), oper_t::IN, list_with_empty);
     BOOST_REQUIRE_EQUAL(evaluate(nonexisting_int_in_list_with_empty, evaluation_inputs{}), make_bool_raw(false));
 
-    test_evaluate_binop_null_unset(oper_t::IN, make_int_const(5), in_list);
+    test_evaluate_binop_null(oper_t::IN, make_int_const(5), in_list);
 }
 
 BOOST_AUTO_TEST_CASE(evaluate_binary_operator_list_contains) {
@@ -2720,7 +2575,7 @@ BOOST_AUTO_TEST_CASE(evaluate_binary_operator_list_contains) {
         binary_operator(list_with_empty, oper_t::CONTAINS, make_int_const(321));
     BOOST_REQUIRE_EQUAL(evaluate(list_with_empty_contains_nonexisting_int, evaluation_inputs{}), make_bool_raw(false));
 
-    test_evaluate_binop_null_unset(oper_t::CONTAINS, list_val, make_int_const(5));
+    test_evaluate_binop_null(oper_t::CONTAINS, list_val, make_int_const(5));
 }
 
 BOOST_AUTO_TEST_CASE(evaluate_binary_operator_set_contains) {
@@ -2749,7 +2604,7 @@ BOOST_AUTO_TEST_CASE(evaluate_binary_operator_set_contains) {
         binary_operator(set_with_empty, oper_t::CONTAINS, make_int_const(321));
     BOOST_REQUIRE_EQUAL(evaluate(set_with_empty_contains_nonexisting_int, evaluation_inputs{}), make_bool_raw(false));
 
-    test_evaluate_binop_null_unset(oper_t::CONTAINS, set_val, make_int_const(5));
+    test_evaluate_binop_null(oper_t::CONTAINS, set_val, make_int_const(5));
 }
 
 BOOST_AUTO_TEST_CASE(evaluate_binary_operator_map_contains) {
@@ -2779,7 +2634,7 @@ BOOST_AUTO_TEST_CASE(evaluate_binary_operator_map_contains) {
         binary_operator(map_with_empty, oper_t::CONTAINS, make_int_const(3));
     BOOST_REQUIRE_EQUAL(evaluate(map_with_empty_contains_nonexisting_int, evaluation_inputs{}), make_bool_raw(false));
 
-    test_evaluate_binop_null_unset(oper_t::CONTAINS, map_val, make_int_const(5));
+    test_evaluate_binop_null(oper_t::CONTAINS, map_val, make_int_const(5));
 }
 
 BOOST_AUTO_TEST_CASE(evaluate_binary_operator_map_contains_key) {
@@ -2810,7 +2665,7 @@ BOOST_AUTO_TEST_CASE(evaluate_binary_operator_map_contains_key) {
     BOOST_REQUIRE_EQUAL(evaluate(map_with_empty_contains_key_nonexisting_int, evaluation_inputs{}),
                         make_bool_raw(false));
 
-    test_evaluate_binop_null_unset(oper_t::CONTAINS_KEY, map_val, make_int_const(5));
+    test_evaluate_binop_null(oper_t::CONTAINS_KEY, map_val, make_int_const(5));
 }
 
 BOOST_AUTO_TEST_CASE(evaluate_binary_operator_is_not) {
@@ -2827,18 +2682,6 @@ BOOST_AUTO_TEST_CASE(evaluate_binary_operator_is_not) {
     expression empty_is_not_null =
         binary_operator(make_empty_const(int32_type), oper_t::IS_NOT, constant::make_null(int32_type));
     BOOST_REQUIRE_EQUAL(evaluate(empty_is_not_null, evaluation_inputs{}), make_bool_raw(true));
-
-    expression unset_is_not_null =
-        binary_operator(constant::make_unset_value(int32_type), oper_t::IS_NOT, constant::make_null(int32_type));
-    BOOST_REQUIRE_THROW(evaluate(unset_is_not_null, evaluation_inputs{}), exceptions::invalid_request_exception);
-
-    expression int_is_not_unset =
-        binary_operator(make_int_const(123), oper_t::IS_NOT, constant::make_unset_value(int32_type));
-    BOOST_REQUIRE_THROW(evaluate(int_is_not_unset, evaluation_inputs{}), exceptions::invalid_request_exception);
-
-    expression unset_is_not_unset =
-        binary_operator(constant::make_unset_value(int32_type), oper_t::IS_NOT, constant::make_unset_value(int32_type));
-    BOOST_REQUIRE_THROW(evaluate(unset_is_not_unset, evaluation_inputs{}), exceptions::invalid_request_exception);
 }
 
 BOOST_AUTO_TEST_CASE(evaluate_binary_operator_like) {
@@ -2862,7 +2705,7 @@ BOOST_AUTO_TEST_CASE(evaluate_binary_operator_like) {
         binary_operator(make_empty_const(utf8_type), oper_t::LIKE, make_empty_const(utf8_type));
     BOOST_REQUIRE_EQUAL(evaluate(empty_like_empty, evaluation_inputs{}), make_bool_raw(true));
 
-    test_evaluate_binop_null_unset(oper_t::LIKE, make_text_const("some_text"), make_text_const("some_%"));
+    test_evaluate_binop_null(oper_t::LIKE, make_text_const("some_text"), make_text_const("some_%"));
 }
 
 // An empty conjunction should evaluate to true
@@ -2990,22 +2833,6 @@ BOOST_AUTO_TEST_CASE(evaluate_conjunction_with_null) {
     BOOST_REQUIRE_EQUAL(evaluate(conj_with_null, evaluation_inputs{}), raw_value::make_null());
 }
 
-// Evaluating a conjunction that contains a single unset value should throw an error
-BOOST_AUTO_TEST_CASE(evaluate_conjunction_one_unset) {
-    expression conj_one_unset = conjunction{.children = {constant::make_unset_value(boolean_type)}};
-
-    BOOST_REQUIRE_THROW(evaluate(conj_one_unset, evaluation_inputs{}), exceptions::invalid_request_exception);
-}
-
-// Evaluating 'true AND true AND true AND UNSET_VALUE AND ...' should throw an erorr
-BOOST_AUTO_TEST_CASE(evaluate_conjunction_with_unset) {
-    expression conj_with_unset = conjunction{
-        .children = {make_bool_const(true), make_bool_const(true), make_bool_const(true),
-                     constant::make_unset_value(boolean_type), make_bool_const(false), make_bool_const(true)}};
-
-    BOOST_REQUIRE_THROW(evaluate(conj_with_unset, evaluation_inputs{}), exceptions::invalid_request_exception);
-}
-
 // Evaluating a conjunction that contains a single empty value should throw an error
 BOOST_AUTO_TEST_CASE(evaluate_conjunction_one_empty) {
     expression conj_one_empty = conjunction{.children = {make_empty_const(boolean_type)}};
@@ -3022,32 +2849,42 @@ BOOST_AUTO_TEST_CASE(evaluate_conjunction_with_empty) {
     BOOST_REQUIRE_THROW(evaluate(conj_with_empty, evaluation_inputs{}), exceptions::invalid_request_exception);
 }
 
+static cql3::query_options query_options_with_unset_bind_variable() {
+    return cql3::query_options(cql3::raw_value_vector_with_unset({cql3::raw_value::make_null()}, {true}));
+}
+
 // Short circuiting on false ignores all further values, even though they could make the expression invalid
 BOOST_AUTO_TEST_CASE(evaluate_conjunction_short_circuit_on_false_does_not_detect_invalid_values) {
-    // An expression which would throw an error when evaluated
-    expression invalid_to_evaluate = conjunction{.children = {constant::make_unset_value(boolean_type)}};
+    auto qo = query_options_with_unset_bind_variable();
+    auto inputs = evaluation_inputs{.options = &qo};
 
-    BOOST_REQUIRE_THROW(evaluate(invalid_to_evaluate, evaluation_inputs{}), exceptions::invalid_request_exception);
+    // An expression which would throw an error when evaluated
+    expression invalid_to_evaluate = conjunction{.children = {new_bind_variable(0)}};
+
+    BOOST_REQUIRE_THROW(evaluate(invalid_to_evaluate, inputs), exceptions::invalid_request_exception);
 
     expression conj_with_false_then_invalid =
         conjunction{.children = {make_bool_const(true), make_bool_const(false), make_empty_const(boolean_type),
-                                 constant::make_unset_value(boolean_type), invalid_to_evaluate, make_bool_const(true)}};
+                                 new_bind_variable(0), invalid_to_evaluate, make_bool_const(true)}};
 
-    BOOST_REQUIRE_EQUAL(evaluate(conj_with_false_then_invalid, evaluation_inputs{}), make_bool_raw(false));
+    BOOST_REQUIRE_EQUAL(evaluate(conj_with_false_then_invalid, inputs), make_bool_raw(false));
 }
 
 // Null doesn't short-circuit
 BOOST_AUTO_TEST_CASE(evaluate_conjunction_doesnt_short_circuit_on_null) {
-    // An expression which would throw an error when evaluated
-    expression invalid_to_evaluate = conjunction{.children = {constant::make_unset_value(boolean_type)}};
+    auto qo = query_options_with_unset_bind_variable();
+    auto inputs = evaluation_inputs{.options = &qo};
 
-    BOOST_REQUIRE_THROW(evaluate(invalid_to_evaluate, evaluation_inputs{}), exceptions::invalid_request_exception);
+    // An expression which would throw an error when evaluated
+    expression invalid_to_evaluate = conjunction{.children = {new_bind_variable(0)}};
+
+    BOOST_REQUIRE_THROW(evaluate(invalid_to_evaluate, inputs), exceptions::invalid_request_exception);
 
     expression conj_with_null_then_invalid = conjunction{
         .children = {make_bool_const(true), constant::make_null(boolean_type), make_empty_const(boolean_type),
-                     constant::make_unset_value(boolean_type), invalid_to_evaluate, make_bool_const(true)}};
+                     new_bind_variable(0), invalid_to_evaluate, make_bool_const(true)}};
 
-    BOOST_REQUIRE_THROW(evaluate(conj_with_null_then_invalid, evaluation_inputs{}),
+    BOOST_REQUIRE_THROW(evaluate(conj_with_null_then_invalid, inputs),
                         exceptions::invalid_request_exception);
 }
 
@@ -3100,9 +2937,12 @@ BOOST_AUTO_TEST_CASE(evaluate_conjunction_of_conjunctions_to_null) {
 
 // Evaluating '() AND (true AND true) AND (true AND UNSET_VALUE) AND (false)' throws an error
 BOOST_AUTO_TEST_CASE(evaluate_conjunction_of_conjunctions_with_invalid) {
+    auto qo = query_options_with_unset_bind_variable();
+    auto inputs = evaluation_inputs{.options = &qo};
+
     expression conj1 = conjunction{.children = {}};
 
-    expression conj2 = conjunction{.children = {make_bool_const(true), constant::make_unset_value(boolean_type)}};
+    expression conj2 = conjunction{.children = {make_bool_const(true), new_bind_variable(0)}};
 
     expression conj3 = conjunction{.children = {make_bool_const(true), make_bool_const(true)}};
 
@@ -3110,7 +2950,7 @@ BOOST_AUTO_TEST_CASE(evaluate_conjunction_of_conjunctions_with_invalid) {
 
     expression conj_of_conjs = conjunction{.children = {conj1, conj2, conj3, conj4}};
 
-    BOOST_REQUIRE_THROW(evaluate(conj_of_conjs, evaluation_inputs{}), exceptions::invalid_request_exception);
+    BOOST_REQUIRE_THROW(evaluate(conj_of_conjs, inputs), exceptions::invalid_request_exception);
 }
 
 // It should be possible to prepare an empty conjunction
diff --git a/test/boost/transport_test.cc b/test/boost/transport_test.cc
index a11a3a3841..2f47356345 100644
--- a/test/boost/transport_test.cc
+++ b/test/boost/transport_test.cc
@@ -100,13 +100,17 @@ SEASTAR_THREAD_TEST_CASE(test_response_request_reader) {
     BOOST_CHECK_EQUAL(unsigned(req.read_byte()), unsigned(opcode));
     BOOST_CHECK_EQUAL(req.read_int() + 9, total_length);
 
-    BOOST_CHECK(req.read_value_view(version).is_null());
-    BOOST_CHECK(req.read_value_view(version).is_unset_value());
-    BOOST_CHECK_EQUAL(to_bytes(req.read_value_view(version)), value);
+    auto v1 = req.read_value_view(version);
+    BOOST_CHECK(!v1.unset && v1.value.is_null());
+    auto v2 = req.read_value_view(version);
+    BOOST_CHECK(v2.unset);
+    BOOST_CHECK_EQUAL(to_bytes(req.read_value_view(version).value), value);
 
     std::vector<sstring_view> names;
     std::vector<cql3::raw_value_view> values;
-    req.read_name_and_value_list(version, names, values);
+    cql3::unset_bind_variable_vector unset;
+    req.read_name_and_value_list(version, names, values, unset);
+    BOOST_CHECK(std::none_of(unset.begin(), unset.end(), std::identity()));
     BOOST_CHECK_EQUAL(names, names_and_values | boost::adaptors::transformed([] (auto& name_and_value) {
         return sstring_view(name_and_value.first);
     }));
diff --git a/test/cql-pytest/cassandra_tests/validation/entities/collections_test.py b/test/cql-pytest/cassandra_tests/validation/entities/collections_test.py
index 9a1593d483..2ee1011745 100644
--- a/test/cql-pytest/cassandra_tests/validation/entities/collections_test.py
+++ b/test/cql-pytest/cassandra_tests/validation/entities/collections_test.py
@@ -273,10 +273,10 @@ def testMapWithUnsetValues(cql, test_keyspace):
         # test unset variables in a map update operation, should not delete the contents
         execute(cql, table, "UPDATE %s SET m['k'] = ? WHERE k = 10", UNSET_VALUE)
         assert_rows(execute(cql, table, "SELECT m FROM %s WHERE k = 10"), [m])
-        assert_invalid_message(cql, table, "Invalid unset map key", "UPDATE %s SET m[?] = 'foo' WHERE k = 10", UNSET_VALUE)
+        assert_invalid_message(cql, table, "unset", "UPDATE %s SET m[?] = 'foo' WHERE k = 10", UNSET_VALUE)
 
         # test unset value for map key
-        assert_invalid_message(cql, table, "Invalid unset map key", "DELETE m[?] FROM %s WHERE k = 10", UNSET_VALUE)
+        assert_invalid_message(cql, table, "unset", "DELETE m[?] FROM %s WHERE k = 10", UNSET_VALUE)
 
 def testListWithUnsetValues(cql, test_keyspace):
     with create_table(cql, test_keyspace, "(k int PRIMARY KEY, l list<text>)") as table:
@@ -294,7 +294,7 @@ def testListWithUnsetValues(cql, test_keyspace):
         assert_rows(execute(cql, table, "SELECT l FROM %s WHERE k = 10"), [l])
 
         # set in index
-        assert_invalid_message(cql, table, "Invalid unset value for list index", "UPDATE %s SET l[?] = 'foo' WHERE k = 10", UNSET_VALUE)
+        assert_invalid_message(cql, table, "unset value", "UPDATE %s SET l[?] = 'foo' WHERE k = 10", UNSET_VALUE)
 
         # remove element by index
         execute(cql, table, "DELETE l[?] FROM %s WHERE k = 10", UNSET_VALUE)
diff --git a/test/cql-pytest/cassandra_tests/validation/entities/tuple_type_test.py b/test/cql-pytest/cassandra_tests/validation/entities/tuple_type_test.py
index 90dded60a7..58122fbd93 100644
--- a/test/cql-pytest/cassandra_tests/validation/entities/tuple_type_test.py
+++ b/test/cql-pytest/cassandra_tests/validation/entities/tuple_type_test.py
@@ -92,7 +92,7 @@ def testInvalidQueries(cql, test_keyspace):
 def testTupleWithUnsetValues(cql, test_keyspace):
     with create_table(cql, test_keyspace, "(k int PRIMARY KEY, t tuple<int, text, double>)") as table:
         # invalid positional field substitution
-        assert_invalid_message(cql, table, "Invalid unset value for tuple field number 1",
+        assert_invalid_message(cql, table, "unset",
                              "INSERT INTO %s (k, t) VALUES(0, (3, ?, 2.1))", UNSET_VALUE)
 
         #FIXME: The Python driver doesn't agree to send such a command to the server,
diff --git a/test/cql-pytest/cassandra_tests/validation/entities/user_types_test.py b/test/cql-pytest/cassandra_tests/validation/entities/user_types_test.py
index 3cec73c819..66eeba7b9b 100644
--- a/test/cql-pytest/cassandra_tests/validation/entities/user_types_test.py
+++ b/test/cql-pytest/cassandra_tests/validation/entities/user_types_test.py
@@ -159,10 +159,10 @@ def testUDTWithUnsetValues(cql, test_keyspace):
     with create_type(cql, test_keyspace, "(x int, y int)") as myType:
         with create_type(cql, test_keyspace, f"(a frozen<{myType}>)") as myOtherType:
             with create_table(cql, test_keyspace, f"(k int PRIMARY KEY, v frozen<{myType}>, z frozen<{myOtherType}>)") as table:
-                assert_invalid_message(cql, table, "Invalid unset value for field 'y' of user defined type ",
+                assert_invalid_message(cql, table, "unset",
                     "INSERT INTO %s (k, v) VALUES (10, {x:?, y:?})", 1, UNSET_VALUE)
                 # Reproduces issue #9671:
-                assert_invalid_message(cql, table, "Invalid unset value for field 'y' of user defined type ",
+                assert_invalid_message(cql, table, "unset",
                     "INSERT INTO %s (k, v, z) VALUES (10, {x:?, y:?}, {a:{x: ?, y: ?}})", 1, 1, 1, UNSET_VALUE)
 
 def testAlteringUserTypeNestedWithinMap(cql, test_keyspace):
diff --git a/test/cql-pytest/cassandra_tests/validation/operations/insert_test.py b/test/cql-pytest/cassandra_tests/validation/operations/insert_test.py
index 775696f067..8fe59f7c52 100644
--- a/test/cql-pytest/cassandra_tests/validation/operations/insert_test.py
+++ b/test/cql-pytest/cassandra_tests/validation/operations/insert_test.py
@@ -44,7 +44,7 @@ def testInsertWithUnset(cql, test_keyspace):
         # has "Invalid unset value for argument in call to function blobasint"
         # Scylla has "Invalid null or unset value for argument to
         # system.blobasint : (blob) -> int"
-        assertInvalidMessageRE(cql, table, "Invalid.*unset value for argument.*blobasint", "SELECT * FROM %s WHERE k = blobAsInt(?)", unset())
+        assertInvalidMessageRE(cql, table, "unset", "SELECT * FROM %s WHERE k = blobAsInt(?)", unset())
 
 # Both Scylla and Cassandra define MAX_TTL or max_ttl with the same formula,
 # 20 years in seconds. In both systems, it is not configurable.
diff --git a/test/cql-pytest/cassandra_tests/validation/operations/select_single_column_relation_test.py b/test/cql-pytest/cassandra_tests/validation/operations/select_single_column_relation_test.py
index 8096b7f099..77f0c037ef 100644
--- a/test/cql-pytest/cassandra_tests/validation/operations/select_single_column_relation_test.py
+++ b/test/cql-pytest/cassandra_tests/validation/operations/select_single_column_relation_test.py
@@ -474,9 +474,9 @@ def testFunctionCallWithUnset(cql, test_keyspace):
     with create_table(cql, test_keyspace, "(k int PRIMARY KEY, s text, i int)") as table:
         # The error messages in Scylla and Cassandra here are slightly
         # different.
-        assert_invalid_message(cql, table, "unset value for argument",
+        assert_invalid_message(cql, table, "unset",
                              "SELECT * FROM %s WHERE token(k) >= token(?)", UNSET_VALUE)
-        assert_invalid_message(cql, table, "unset value for argument",
+        assert_invalid_message(cql, table, "unset",
                              "SELECT * FROM %s WHERE k = blobAsInt(?)", UNSET_VALUE)
 
 def testLimitWithUnset(cql, test_keyspace):
diff --git a/test/cql-pytest/test_filtering.py b/test/cql-pytest/test_filtering.py
index fc7d35731e..5d792eb022 100644
--- a/test/cql-pytest/test_filtering.py
+++ b/test/cql-pytest/test_filtering.py
@@ -220,7 +220,7 @@ def test_filtering_with_subscript(cql, test_keyspace, cassandra_bug):
         # the scan brings up several rows, it may exercise different code
         # paths.
         assert list(cql.execute(f"select p from {table} where m1[null] = 2 ALLOW FILTERING")) == []
-        with pytest.raises(InvalidRequest, match='Unsupported unset map key for column m1'):
+        with pytest.raises(InvalidRequest, match='unset'):
             cql.execute(stmt, [UNSET_VALUE])
 
         # check subscripted list filtering
diff --git a/test/lib/cql_test_env.cc b/test/lib/cql_test_env.cc
index 0c5b17be22..738f26b243 100644
--- a/test/lib/cql_test_env.cc
+++ b/test/lib/cql_test_env.cc
@@ -241,7 +241,7 @@ public:
 
     virtual future<::shared_ptr<cql_transport::messages::result_message>> execute_prepared(
         cql3::prepared_cache_key_type id,
-        std::vector<cql3::raw_value> values,
+        cql3::raw_value_vector_with_unset values,
         db::consistency_level cl = db::consistency_level::ONE) override {
 
         const auto& so = cql3::query_options::specific_options::DEFAULT;
diff --git a/test/lib/cql_test_env.hh b/test/lib/cql_test_env.hh
index 2f288147cc..89967824e9 100644
--- a/test/lib/cql_test_env.hh
+++ b/test/lib/cql_test_env.hh
@@ -113,7 +113,7 @@ public:
 
     virtual future<::shared_ptr<cql_transport::messages::result_message>> execute_prepared(
         cql3::prepared_cache_key_type id,
-        std::vector<cql3::raw_value> values,
+        cql3::raw_value_vector_with_unset values,
         db::consistency_level cl = db::consistency_level::ONE) = 0;
 
     virtual future<::shared_ptr<cql_transport::messages::result_message>> execute_prepared_with_qo(
diff --git a/test/lib/expr_test_utils.cc b/test/lib/expr_test_utils.cc
index e3ee03dec5..d7cfae49be 100644
--- a/test/lib/expr_test_utils.cc
+++ b/test/lib/expr_test_utils.cc
@@ -81,7 +81,7 @@ constant make_text_const(const sstring_view& text) {
 }
 
 // This function implements custom serialization of collection values.
-// Some tests require the collection to contain unset_value or an empty value,
+// Some tests require the collection to contain an empty value,
 // which is impossible to express using the existing code.
 cql3::raw_value make_collection_raw(size_t size_to_write, const std::vector<cql3::raw_value>& elements_to_write) {
     size_t serialized_len = 0;
@@ -100,8 +100,6 @@ cql3::raw_value make_collection_raw(size_t size_to_write, const std::vector<cql3
     for (const cql3::raw_value& val : elements_to_write) {
         if (val.is_null()) {
             write_int32(out, -1);
-        } else if (val.is_unset_value()) {
-            write_int32(out, -2);
         } else {
             val.view().with_value(
                 [&](const FragmentedView auto& val_view) { write_collection_value(out, linearized(val_view)); });
@@ -129,7 +127,7 @@ raw_value make_map_raw(const std::vector<std::pair<raw_value, raw_value>>& value
 }
 
 // This function implements custom serialization of tuples.
-// Some tests require the tuple to contain unset_value or an empty value,
+// Some tests require the tuple to contain an empty value,
 // which is impossible to express using the existing code.
 raw_value make_tuple_raw(const std::vector<raw_value>& values) {
     size_t serialized_len = 0;
@@ -144,8 +142,6 @@ raw_value make_tuple_raw(const std::vector<raw_value>& values) {
     for (const raw_value& val : values) {
         if (val.is_null()) {
             write<int32_t>(out, -1);
-        } else if (val.is_unset_value()) {
-            write<int32_t>(out, -2);
         } else {
             val.view().with_value([&](const FragmentedView auto& bytes_view) {
                 write<int32_t>(out, bytes_view.size_bytes());
@@ -344,10 +340,6 @@ std::pair<evaluation_inputs, std::unique_ptr<evaluation_inputs_data>> make_evalu
             throw_error("Passed NULL as value for {}. This is not allowed for partition key columns.",
                         pk_col.name_as_text());
         }
-        if (col_value.is_unset_value()) {
-            throw_error("Passed UNSET_VALUE as value for {}. This is not allowed for partition key columns.",
-                        pk_col.name_as_text());
-        }
         partition_key.push_back(raw_value(col_value).to_bytes());
     }
 
@@ -359,10 +351,6 @@ std::pair<evaluation_inputs, std::unique_ptr<evaluation_inputs_data>> make_evalu
             throw_error("Passed NULL as value for {}. This is not allowed for clustering key columns.",
                         ck_col.name_as_text());
         }
-        if (col_value.is_unset_value()) {
-            throw_error("Passed UNSET_VALUE as value for {}. This is not allowed for clustering key columns.",
-                        ck_col.name_as_text());
-        }
         clustering_key.push_back(raw_value(col_value).to_bytes());
     }
 
@@ -381,18 +369,12 @@ std::pair<evaluation_inputs, std::unique_ptr<evaluation_inputs_data>> make_evalu
 
     for (const column_definition& col : table_schema->regular_columns()) {
         const raw_value& col_value = get_col_val(col);
-        if (col_value.is_unset_value()) {
-            throw_error("Passed UNSET_VALUE as value for {}. This is not allowed.", col.name_as_text());
-        }
         int32_t index = selection->index_of(col);
         static_and_regular_columns[index] = raw_value(col_value).to_managed_bytes_opt();
     }
 
     for (const column_definition& col : table_schema->static_columns()) {
         const raw_value& col_value = get_col_val(col);
-        if (col_value.is_unset_value()) {
-            throw_error("Passed UNSET_VALUE as value for {}. This is not allowed.", col.name_as_text());
-        }
         int32_t index = selection->index_of(col);
         static_and_regular_columns[index] = raw_value(col_value).to_managed_bytes_opt();
     }
diff --git a/tracing/trace_keyspace_helper.cc b/tracing/trace_keyspace_helper.cc
index ad3bfd6898..ecad64af39 100644
--- a/tracing/trace_keyspace_helper.cc
+++ b/tracing/trace_keyspace_helper.cc
@@ -392,7 +392,7 @@ future<> trace_keyspace_helper::apply_events_mutation(cql3::query_processor& qp,
         tlogger.trace("{}: storing {} events records: parent_id {} span_id {}", records->session_id, events_records.size(), records->parent_id, records->my_span_id);
 
         std::vector<cql3::statements::batch_statement::single_statement> modifications(events_records.size(), cql3::statements::batch_statement::single_statement(_events.insert_stmt(), false));
-        std::vector<std::vector<cql3::raw_value>> values;
+        std::vector<cql3::raw_value_vector_with_unset> values;
 
         values.reserve(events_records.size());
         std::for_each(events_records.begin(), events_records.end(), [&values, all_records = records, this] (event_record& one_event_record) { values.emplace_back(make_event_mutation_data(*all_records, one_event_record)); });
diff --git a/tracing/trace_state.cc b/tracing/trace_state.cc
index e530424cb0..5562193ccb 100644
--- a/tracing/trace_state.cc
+++ b/tracing/trace_state.cc
@@ -27,7 +27,7 @@ struct trace_state::params_values {
     struct prepared_statement_info {
         prepared_checked_weak_ptr statement;
         std::optional<std::vector<sstring_view>> query_option_names;
-        std::vector<cql3::raw_value_view> query_option_values;
+        cql3::raw_value_view_vector_with_unset query_option_values;
         explicit prepared_statement_info(prepared_checked_weak_ptr statement) : statement(std::move(statement)) {}
     };
 
@@ -194,7 +194,7 @@ void trace_state::build_parameters_map() {
 
 void trace_state::build_parameters_map_for_one_prepared(const prepared_checked_weak_ptr& prepared_ptr,
         std::optional<std::vector<sstring_view>>& names_opt,
-        std::vector<cql3::raw_value_view>& values, const sstring& param_name_prefix) {
+        cql3::raw_value_view_vector_with_unset& values, const sstring& param_name_prefix) {
     auto& params_map = _records->session_rec.parameters;
     size_t i = 0;
 
@@ -202,17 +202,17 @@ void trace_state::build_parameters_map_for_one_prepared(const prepared_checked_w
     // Such an eviction is a very unlikely event, however if it happens, since we are unable to recover their types, trace raw representations of the values.
 
     if (names_opt) {
-        if (names_opt->size() != values.size()) {
-            throw std::logic_error(format("Number of \"names\" ({}) doesn't match the number of positional variables ({})", names_opt->size(), values.size()).c_str());
+        if (names_opt->size() != values.values.size()) {
+            throw std::logic_error(format("Number of \"names\" ({}) doesn't match the number of positional variables ({})", names_opt->size(), values.values.size()).c_str());
         }
 
         auto& names = names_opt.value();
-        for (; i < values.size(); ++i) {
-            params_map.emplace(format("{}[{:d}]({})", param_name_prefix, i, names[i]), raw_value_to_sstring(values[i], prepared_ptr ? prepared_ptr->bound_names[i]->type : nullptr));
+        for (; i < values.values.size(); ++i) {
+            params_map.emplace(format("{}[{:d}]({})", param_name_prefix, i, names[i]), raw_value_to_sstring(values.values[i], values.unset[i], prepared_ptr ? prepared_ptr->bound_names[i]->type : nullptr));
         }
     } else {
-        for (; i < values.size(); ++i) {
-            params_map.emplace(format("{}[{:d}]", param_name_prefix, i), raw_value_to_sstring(values[i], prepared_ptr ? prepared_ptr->bound_names[i]->type : nullptr));
+        for (; i < values.values.size(); ++i) {
+            params_map.emplace(format("{}[{:d}]", param_name_prefix, i), raw_value_to_sstring(values.values[i], values.unset[i], prepared_ptr ? prepared_ptr->bound_names[i]->type : nullptr));
         }
     }
 }
@@ -281,13 +281,15 @@ void trace_state::stop_foreground_and_write() noexcept {
     }
 }
 
-sstring trace_state::raw_value_to_sstring(const cql3::raw_value_view& v, const data_type& t) {
+sstring trace_state::raw_value_to_sstring(const cql3::raw_value_view& v, bool is_unset, const data_type& t) {
     static constexpr int max_val_bytes = 64;
 
+    if (is_unset) {
+        return "unset value";
+    }
+
     if (v.is_null()) {
         return "null";
-    } else if (v.is_unset_value()) {
-        return "unset value";
     } else {
       return v.with_linearized([&] (bytes_view val) {
         sstring str_rep;
diff --git a/tracing/trace_state.hh b/tracing/trace_state.hh
index df7cd61f5d..1121c8adf9 100644
--- a/tracing/trace_state.hh
+++ b/tracing/trace_state.hh
@@ -25,6 +25,7 @@
 namespace cql3{
 class query_options;
 struct raw_value_view;
+struct raw_value_view_vector_with_unset;
 
 namespace statements {
 class prepared_statement;
@@ -319,7 +320,7 @@ private:
      * @param t type object corresponding to the given raw value.
      * @return the string with the representation of the given raw value.
      */
-    sstring raw_value_to_sstring(const cql3::raw_value_view& v, const data_type& t);
+    sstring raw_value_to_sstring(const cql3::raw_value_view& v, bool is_unset, const data_type& t);
 
     /**
      * Stores a page size of a query being traced.
@@ -420,7 +421,7 @@ private:
      */
     void build_parameters_map_for_one_prepared(const prepared_checked_weak_ptr& prepared_ptr,
             std::optional<std::vector<sstring_view>>& names_opt,
-            std::vector<cql3::raw_value_view>& values, const sstring& param_name_prefix);
+            cql3::raw_value_view_vector_with_unset& values, const sstring& param_name_prefix);
 
     /**
      * The actual trace message storing method.
diff --git a/transport/request.hh b/transport/request.hh
index 0c0d16d845..21fa91b5ef 100644
--- a/transport/request.hh
+++ b/transport/request.hh
@@ -15,6 +15,16 @@
 
 namespace cql_transport {
 
+struct unset_tag {};
+
+struct value_view_and_unset {
+    cql3::raw_value_view value;
+    bool unset = false;
+
+    value_view_and_unset(cql3::raw_value_view value) : value(std::move(value)) {}
+    value_view_and_unset(unset_tag) : value(cql3::raw_value_view::make_null()), unset(true) {}
+};
+
 class request_reader {
     fragmented_temporary_buffer::istream _in;
     bytes_ostream* _linearization_buffer;
@@ -123,7 +133,7 @@ public:
         return b;
     }
 
-    cql3::raw_value_view read_value_view(uint8_t version) {
+    value_view_and_unset read_value_view(uint8_t version) {
         auto len = read_int();
         if (len < 0) {
             if (version < 4) {
@@ -132,7 +142,7 @@ public:
             if (len == -1) {
                 return cql3::raw_value_view::make_null();
             } else if (len == -2) {
-                return cql3::raw_value_view::make_unset_value();
+                return value_view_and_unset(unset_tag());
             } else {
                 throw exceptions::protocol_exception(format("invalid value length: {:d}", len));
             }
@@ -140,13 +150,17 @@ public:
         return cql3::raw_value_view::make_value(_in.read_view(len, exception_thrower()));
     }
 
-    void read_name_and_value_list(uint8_t version, std::vector<sstring_view>& names, std::vector<cql3::raw_value_view>& values) {
+    void read_name_and_value_list(uint8_t version, std::vector<sstring_view>& names, std::vector<cql3::raw_value_view>& values,
+            cql3::unset_bind_variable_vector& unset) {
         uint16_t size = read_short();
         names.reserve(size);
+        unset.reserve(size);
         values.reserve(size);
         for (uint16_t i = 0; i < size; i++) {
             names.emplace_back(read_string_view());
-            values.emplace_back(read_value_view(version));
+            auto&& [value, is_unset] = read_value_view(version);
+            values.emplace_back(std::move(value));
+            unset.emplace_back(is_unset);
         }
     }
 
@@ -158,11 +172,14 @@ public:
         }
     }
 
-    void read_value_view_list(uint8_t version, std::vector<cql3::raw_value_view>& values) {
+    void read_value_view_list(uint8_t version, std::vector<cql3::raw_value_view>& values, cql3::unset_bind_variable_vector& unset) {
         uint16_t size = read_short();
         values.reserve(size);
+        unset.reserve(size);
         for (uint16_t i = 0; i < size; i++) {
-            values.emplace_back(read_value_view(version));
+            auto&& [value, is_unset] = read_value_view(version);
+            values.emplace_back(std::move(value));
+            unset.emplace_back(is_unset);
         }
     }
 
@@ -208,13 +225,14 @@ public:
         auto consistency = read_consistency();
         auto flags = enum_set<options_flag_enum>::from_mask(read_byte());
         std::vector<cql3::raw_value_view> values;
+        cql3::unset_bind_variable_vector unset;
         std::vector<sstring_view> names;
 
         if (flags.contains<options_flag::VALUES>()) {
             if (flags.contains<options_flag::NAMES_FOR_VALUES>()) {
-                read_name_and_value_list(version, names, values);
+                read_name_and_value_list(version, names, values, unset);
             } else {
-                read_value_view_list(version, values);
+                read_value_view_list(version, values, unset);
             }
         }
 
@@ -248,10 +266,12 @@ public:
             if (!names.empty()) {
                 onames = std::move(names);
             }
-            options = std::make_unique<cql3::query_options>(cql_config, consistency, std::move(onames), std::move(values), skip_metadata,
+            options = std::make_unique<cql3::query_options>(cql_config, consistency, std::move(onames),
+                cql3::raw_value_view_vector_with_unset(std::move(values), std::move(unset)), skip_metadata,
                 cql3::query_options::specific_options{page_size, std::move(paging_state), serial_consistency, ts});
         } else {
-            options = std::make_unique<cql3::query_options>(cql_config, consistency, std::nullopt, std::move(values), skip_metadata,
+            options = std::make_unique<cql3::query_options>(cql_config, consistency, std::nullopt,
+                cql3::raw_value_view_vector_with_unset(std::move(values), std::move(unset)), skip_metadata,
                 cql3::query_options::specific_options::DEFAULT);
         }
 
diff --git a/transport/server.cc b/transport/server.cc
index 0b49000387..2268368f14 100644
--- a/transport/server.cc
+++ b/transport/server.cc
@@ -1040,7 +1040,7 @@ process_batch_internal(service::client_state& client_state, distributed<cql3::qu
     const unsigned n = in.read_short();
 
     std::vector<cql3::statements::batch_statement::single_statement> modifications;
-    std::vector<std::vector<cql3::raw_value_view>> values;
+    std::vector<cql3::raw_value_view_vector_with_unset> values;
     std::unordered_map<cql3::prepared_cache_key_type, cql3::authorized_prepared_statements_cache::value_type> pending_authorization_entries;
 
     modifications.reserve(n);
@@ -1106,14 +1106,15 @@ process_batch_internal(service::client_state& client_state, distributed<cql3::qu
         modifications.emplace_back(std::move(modif_statement_ptr), needs_authorization);
 
         std::vector<cql3::raw_value_view> tmp;
-        in.read_value_view_list(version, tmp);
+        cql3::unset_bind_variable_vector unset;
+        in.read_value_view_list(version, tmp, unset);
 
         auto stmt = ps->statement;
         if (stmt->get_bound_terms() != tmp.size()) {
             throw exceptions::invalid_request_exception(format("There were {:d} markers(?) in CQL but {:d} bound variables",
                             stmt->get_bound_terms(), tmp.size()));
         }
-        values.emplace_back(std::move(tmp));
+        values.emplace_back(cql3::raw_value_view_vector_with_unset(std::move(tmp), std::move(unset)));
     }
 
     auto q_state = std::make_unique<cql_query_state>(client_state, trace_state, std::move(permit));