cql3, types: validate listlike collections (sets, lists) for storage

Lists allow NULL in some contexts (bind variables for LWT "IN ?"
conditions), but not in most others. Currently, the implementation
just disallows NULLs in list values, and the cases where it is allowed
are hacked around. To reduce the special cases, we'll allow lists
to have NULLs, and just restrict them for storage. This is similar
to how scalar values can be NULL, but not when they are part of a
partition key.

To prepare for the transition, identify the locations where lists
(and sets, which share the same storage) are stored as frozen
values and add a NULL check there. Non-frozen lists already have the
check. Since sets share the same format as lists, apply the same to
them.

No actual checks are done yet, since NULLs are impossible. This
is just a stub.

cql3/lists.cc

@@ -170,10 +170,14 @@ lists::do_append(const cql3::raw_value& list_value,
         }
         m.set_cell(prefix, column, appended.serialize(*ltype));
     } else {
+        auto ltype = static_cast<const list_type_impl*>(column.type.get());
         // for frozen lists, we're overwriting the whole cell value
         if (list_value.is_null()) {
             m.set_cell(prefix, column, params.make_dead_cell());
         } else {
+            list_value.view().with_value([&] (const FragmentedView auto& v) {
+                ltype->validate_for_storage(v);
+            });
             m.set_cell(prefix, column, params.make_cell(*column.type, list_value.view()));
         }
     }

cql3/sets.cc

@@ -62,6 +62,9 @@ sets::adder::do_add(mutation& m, const clustering_key_prefix& row_key, const upd
         m.set_cell(row_key, column, mut.serialize(set_type));
     } else if (!value.is_null()) {
         // for frozen sets, we're overwriting the whole cell
+        value.view().with_value([&] (const FragmentedView auto& v) {
+            set_type.validate_for_storage(v);
+        });
         m.set_cell(row_key, column, params.make_cell(*column.type, value.view()));
     } else {
         m.set_cell(row_key, column, params.make_dead_cell());

types.cc

@@ -580,6 +580,17 @@ bytes listlike_collection_type_impl::serialize_map(const map_type_impl& map_type
     return b;
 }
 
+void
+listlike_collection_type_impl::validate_for_storage(const FragmentedView auto& value) const {
+}
+
+template
+void listlike_collection_type_impl::validate_for_storage(const managed_bytes_view& value) const;
+
+template
+void listlike_collection_type_impl::validate_for_storage(const fragmented_temporary_buffer::view& value) const;
+
+
 static bool is_compatible_with_aux(const collection_type_impl& t, const abstract_type& previous) {
     if (t.get_kind() != previous.get_kind()) {
         return false;

types/collection.hh

@@ -88,6 +88,9 @@ public:
     // vector<pair<data_value, empty>> respectively. Serialize this representation
     // as a vector of values, not as a vector of pairs.
     bytes serialize_map(const map_type_impl& map_type, const data_value& value) const;
+
+    // Verify that there are no NULL elements. Throws if there are.
+    void validate_for_storage(const FragmentedView auto& value) const;
 };
 
 template <typename Iterator>
@@ -125,3 +128,11 @@ collection_type_impl::pack_fragmented(Iterator start, Iterator finish, int eleme
     }
     return out;
 }
+
+extern
+template
+void listlike_collection_type_impl::validate_for_storage(const managed_bytes_view& value) const;
+
+extern
+template
+void listlike_collection_type_impl::validate_for_storage(const fragmented_temporary_buffer::view& value) const;