From 898cea4cdeab7e69244a898f20d2516eed8c310f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Micha=C5=82=20Chojnowski?= <michal.chojnowski@scylladb.com>
Date: Sun, 22 Nov 2020 02:08:19 +0100
Subject: [PATCH] types: deserialize tuple types from FragmentedView

A part of the transition of deserialize from bytes_view to FragmentedView.
---
 types.cc       | 19 +++++++++++++------
 types/tuple.hh |  9 +++++++++
 2 files changed, 22 insertions(+), 6 deletions(-)

diff --git a/types.cc b/types.cc
index e0514a7575..771e3fe790 100644
--- a/types.cc
+++ b/types.cc
@@ -1833,19 +1833,22 @@ static void serialize(const abstract_type& t, const void* value, bytes::iterator
     return ::serialize(t, value, out, cql_serialization_format::internal());
 }
 
-static data_value deserialize_aux(const tuple_type_impl& t, bytes_view v) {
+template <FragmentedView View>
+data_value deserialize_aux(const tuple_type_impl& t, View v) {
     tuple_type_impl::native_type ret;
     ret.reserve(t.all_types().size());
     auto ti = t.all_types().begin();
-    auto vi = tuple_deserializing_iterator::start(v);
-    while (ti != t.all_types().end() && vi != tuple_deserializing_iterator::finish(v)) {
+    while (ti != t.all_types().end() && v.size_bytes()) {
         data_value obj = data_value::make_null(*ti);
-        if (*vi) {
-            obj = (*ti)->deserialize(**vi);
+        std::optional<View> e = read_tuple_element(v);
+        if (e) {
+            // FIXME: don't linearize
+            with_linearized(*e, [&] (bytes_view bv) {
+                obj = (*ti)->deserialize(bv);
+            });
         }
         ret.push_back(std::move(obj));
         ++ti;
-        ++vi;
     }
     while (ti != t.all_types().end()) {
         ret.push_back(data_value::make_null(*ti++));
@@ -1853,6 +1856,10 @@ static data_value deserialize_aux(const tuple_type_impl& t, bytes_view v) {
     return data_value::make(t.shared_from_this(), std::make_unique<tuple_type_impl::native_type>(std::move(ret)));
 }
 
+data_value deserialize_aux(const tuple_type_impl& t, bytes_view v) {
+    return deserialize_aux(t, single_fragmented_view(v));
+}
+
 template<FragmentedView View>
 utils::multiprecision_int deserialize_value(const varint_type_impl&, View v) {
     bool negative = v.current_fragment().front() < 0;
diff --git a/types/tuple.hh b/types/tuple.hh
index d9ae5aadc1..cee0ab5e0c 100644
--- a/types/tuple.hh
+++ b/types/tuple.hh
@@ -93,6 +93,15 @@ private:
     }
 };
 
+template <FragmentedView View>
+std::optional<View> read_tuple_element(View& v) {
+    auto s = read_simple<int32_t>(v);
+    if (s < 0) {
+        return std::nullopt;
+    }
+    return read_simple_bytes(v, s);
+}
+
 class tuple_type_impl : public concrete_type<std::vector<data_value>> {
     using intern = type_interning_helper<tuple_type_impl, std::vector<data_type>>;
 protected: