From 0b418fa7cf164e7fcbeab2d4ab1a3412eb8aeef0 Mon Sep 17 00:00:00 2001
From: Avi Kivity <avi@scylladb.com>
Date: Thu, 12 Jan 2023 18:17:31 +0200
Subject: [PATCH] cql3, transport, tests: remove "unset" from value type system

The CQL binary protocol introduced "unset" values in version 4
of the protocol. Unset values can be bound to variables, which
cause certain CQL fragments to be skipped. For example, the
fragment `SET a = :var` will not change the value of `a` if `:var`
is bound to an unset value.

Unsets, however, are very limited in where they can appear. They
can only appear at the top-level of an expression, and any computation
done with them is invalid. For example, `SET list_column = [3, :var]`
is invalid if `:var` is bound to unset.

This causes the code to be littered with checks for unset, and there
are plenty of tests dedicated to catching unsets. However, a simpler
way is possible - prevent the infiltration of unsets at the point of
entry (when evaluating a bind variable expression), and introduce
guards to check for the few cases where unsets are allowed.

This is what this long patch does. It performs the following:

(general)

1. unset is removed from the possible values of cql3::raw_value and
   cql3::raw_value_view.

(external->cql3)

2. query_options is fortified with a vector of booleans,
   unset_bind_variable_vector, where each boolean corresponds to a bind
   variable index and is true when it is unset.
3. To avoid churn, two compatiblity structs are introduced:
   cql3::raw_value{,_view}_vector_with_unset, which can be constructed
   from a std::vector<raw_value{,_view/}>, which is what most callers
   have. They can also be constructed with explicit unset vectors, for
   the few cases they are needed.

(cql3->variables)

4. query_options::get_value_at() now throws if the requested bind variable
   is unset. This replaces all the throwing checks in expression evaluation
   and statement execution, which are removed.
5. A new query_options::is_unset() is added for the users that can tolerate
   unset; though it is not used directly.
6. A new cql3::unset_operation_guard class guards against unsets. It accepts
   an expression, and can be queried whether an unset is present. Two
   conditions are checked: the expression must be a singleton bind
   variable, and at runtime it must be bound to an unset value.
7. The modification_statement operations are split into two, via two
   new subclasses of cql3::operation. cql3::operation_no_unset_support
   ignores unsets completely. cql3::operation_skip_if_unset checks if
   an operand is unset (luckily all operations have at most one operand that
   tolerates unset) and applies unset_operation_guard to it.
8. The various sites that accept expressions or operations are modified
   to check for should_skip_operation(). This are the loops around
   operations in update_statement and delete_statement, and the checks
   for unset in attributes (LIMIT and PER PARTITION LIMIT)

(tests)

9. Many unset tests are removed. It's now impossible to enter an
   unset value into the expression evaluation machinery (there's
   just no unset value), so it's impossible to test for it.
10. Other unset tests now have to be invoked via bind variables,
   since there's no way to create an unset cql3::expr::constant.
11. Many tests have their exception message match strings relaxed.
   Since unsets are now checked very early, we don't know the context
   where they happen. It would be possible to reintroduce it (by adding
   a format string parameter to cql3::unset_operation_guard), but it
   seems not to be worth the effort. Usage of unsets is rare, and it is
   explicit (at least with the Python driver, an unset cannot be
   introduced by ommission).

I tried as an alternative to wrap cql3::raw_value{,_view} (that doesn't
recognize unsets) with cql3::maybe_unset_value (that does), but that
caused huge amounts of churn, so I abandoned that in favor of the
current approach.

Closes #12517
---
 cql3/attributes.cc                            |  18 +-
 cql3/attributes.hh                            |   3 +
 cql3/column_condition.cc                      |  10 -
 cql3/constants.hh                             |  20 +-
 cql3/expr/expression.cc                       | 115 ++------
 cql3/expr/expression.hh                       |   1 -
 cql3/expr/unset.hh                            |  30 +++
 cql3/lists.cc                                 |  25 +-
 cql3/lists.hh                                 |  24 +-
 cql3/maps.cc                                  |  16 +-
 cql3/maps.hh                                  |  16 +-
 cql3/operation.cc                             |   4 +-
 cql3/operation.hh                             |  28 +-
 cql3/query_options.cc                         |  18 +-
 cql3/query_options.hh                         |  71 ++++-
 cql3/restrictions/statement_restrictions.cc   |   2 +-
 cql3/sets.cc                                  |   8 +-
 cql3/sets.hh                                  |  16 +-
 cql3/statements/delete_statement.cc           |   3 +
 cql3/statements/select_statement.cc           |   8 +-
 cql3/statements/select_statement.hh           |   9 +-
 cql3/statements/update_statement.cc           |   3 +
 cql3/user_types.cc                            |   7 -
 cql3/user_types.hh                            |  12 +-
 cql3/values.cc                                |  11 +-
 cql3/values.hh                                |  51 +---
 service/raft/raft_sys_table_storage.cc        |   2 +-
 test/boost/cql_query_test.cc                  |  62 +----
 test/boost/expr_test.cc                       | 254 ++++--------------
 test/boost/transport_test.cc                  |  12 +-
 .../validation/entities/collections_test.py   |   6 +-
 .../validation/entities/tuple_type_test.py    |   2 +-
 .../validation/entities/user_types_test.py    |   4 +-
 .../validation/operations/insert_test.py      |   2 +-
 .../select_single_column_relation_test.py     |   4 +-
 test/cql-pytest/test_filtering.py             |   2 +-
 test/lib/cql_test_env.cc                      |   2 +-
 test/lib/cql_test_env.hh                      |   2 +-
 test/lib/expr_test_utils.cc                   |  22 +-
 tracing/trace_keyspace_helper.cc              |   2 +-
 tracing/trace_state.cc                        |  24 +-
 tracing/trace_state.hh                        |   5 +-
 transport/request.hh                          |  40 ++-
 transport/server.cc                           |   7 +-
 44 files changed, 372 insertions(+), 611 deletions(-)
 create mode 100644 cql3/expr/unset.hh

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));