utils/managed_bytes, serializer: add conversion between buffer_view<bytes_ostream> and managed_bytes_view

The codebase evolved to have several different ways to hold a fragmented
buffer: fragmented_temporary_buffer (for data received from the network;
not relevant for this discussion); bytes_ostream (for fragmented data that
is built incrementally; also used for a serialized result_set), and
managed_bytes (used for lsa and serialized individual values in
expression evaluation).

One problem with this state of affairs is that using data in one
fragmented form with functions that accept another fragmented form
requires either a copy, or templating everything. The former is
unpalatable for fast-path code, and the latter is undesirable for
compile time and run-time code footprint. So we'd like to make
the various forms compatible.

In 53e0dc7530 ("bytes_ostream: base on managed_bytes") we changed
bytes_ostream to have the same underlying data structure as
managed_bytes, so all that remains is to add the right API. This
is somewhat difficult as the data is hidden in multiple layers:
ser::buffer_view<> is used to abstract a slice of bytes_ostream,
and this is further abstracted by using iterators into bytes_ostream
rather than directly using the internals. Likewise, it's impossible
to construct a managed_bytes_view from the internals.

Hack through all of these by adding extract_implementation() methods,
and a build_managed_bytes_view_from_internals() helper. These are all
used by new APIs buffer_view_to_managed_bytes_view() that extract
the internals and put them back together again.

Ideally we wouldn't need any of this, but unifying the type system
in this area is quite an undertaking, so we need some shortcuts.

bytes_ostream.hh

@@ -53,6 +53,10 @@ public:
         using difference_type = std::ptrdiff_t;
         using pointer = bytes_view*;
         using reference = bytes_view&;
+
+        struct implementation {
+            blob_storage* current_chunk;
+        };
     private:
         chunk* _current = nullptr;
     public:
@@ -76,6 +80,11 @@ public:
             return tmp;
         }
         bool operator==(const fragment_iterator&) const = default;
+        implementation extract_implementation() const {
+            return implementation {
+                .current_chunk = _current,
+            };
+        }
     };
     using const_iterator = fragment_iterator;
 

serializer.cc

@@ -12,4 +12,27 @@ namespace ser {
 
 logging::logger serlog("serializer");
 
-} // namespace ser
+} // namespace ser
+
+namespace utils {
+
+managed_bytes_view
+buffer_view_to_managed_bytes_view(ser::buffer_view<bytes_ostream::fragment_iterator> bv) {
+    auto impl = bv.extract_implementation();
+    return build_managed_bytes_view_from_internals(
+            impl.current,
+            impl.next.extract_implementation().current_chunk,
+            impl.size
+    );
+}
+
+managed_bytes_view_opt
+buffer_view_to_managed_bytes_view(std::optional<ser::buffer_view<bytes_ostream::fragment_iterator>> bvo) {
+    if (!bvo) {
+        return std::nullopt;
+    }
+    return buffer_view_to_managed_bytes_view(*bvo);
+}
+
+
+} // namespace utils

serializer.hh

@@ -45,6 +45,12 @@ class buffer_view {
 public:
     using fragment_type = bytes_view;
 
+    struct implementation {
+        bytes_view current;
+        FragmentIterator next;
+        size_t size;
+    };
+
     class iterator {
         bytes_view _current;
         size_t _left = 0;
@@ -177,6 +183,14 @@ public:
         }
         return fn(bv);
     }
+
+    implementation extract_implementation() const {
+        return implementation {
+            .current = _first,
+            .next = _next,
+            .size = _total_size,
+        };
+    }
 };
 static_assert(FragmentedView<buffer_view<bytes_ostream::fragment_iterator>>);
 

utils/buffer_view-to-managed_bytes_view.hh

@@ -0,0 +1,18 @@
+// Copyright 2023-present ScyllaDB
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+#pragma once
+
+#include "managed_bytes.hh"
+#include "serializer.hh"
+#include "bytes_ostream.hh"
+
+namespace utils {
+
+managed_bytes_view
+buffer_view_to_managed_bytes_view(ser::buffer_view<bytes_ostream::fragment_iterator> bv);
+
+managed_bytes_view_opt
+buffer_view_to_managed_bytes_view(std::optional<ser::buffer_view<bytes_ostream::fragment_iterator>> bvo);
+
+}

utils/managed_bytes.hh

@@ -374,6 +374,12 @@ private:
     fragment_type _current_fragment = {};
     blob_storage* _next_fragments = nullptr;
     size_t _size = 0;
+private:
+    managed_bytes_basic_view(fragment_type current_fragment, blob_storage* next_fragments, size_t size)
+        : _current_fragment(current_fragment)
+        , _next_fragments(next_fragments)
+        , _size(size) {
+    }
 public:
     managed_bytes_basic_view() = default;
     managed_bytes_basic_view(const managed_bytes_basic_view&) = default;
@@ -465,6 +471,8 @@ public:
         });
         return func(bv);
     }
+
+    friend managed_bytes_basic_view<mutable_view::no> build_managed_bytes_view_from_internals(bytes_view current_fragment, blob_storage* next_fragment, size_t size);
 };
 static_assert(FragmentedView<managed_bytes_view>);
 static_assert(FragmentedMutableView<managed_bytes_mutable_view>);
@@ -497,6 +505,12 @@ inline managed_bytes::managed_bytes(View v) : managed_bytes(initialized_later(),
     write_fragmented(self, v);
 }
 
+inline
+managed_bytes_view
+build_managed_bytes_view_from_internals(bytes_view current_fragment, blob_storage* next_fragment, size_t size) {
+    return managed_bytes_view(current_fragment, next_fragment, size);
+}
+
 template<>
 struct appending_hash<managed_bytes_view> {
     template<Hasher Hasher>