8066268338 Key segment pair refactor - do not move the segment out of the KeySegmentPair as it is a shared resource (#2166)

This allows us to re-enable some tests in `arcticdb-enterprise`
https://github.com/man-group/arcticdb-enterprise/pull/new/aseaton/keysegmentpair-refactor-enterprise
.

The changes are:

- Remove the mutable `Segment& segment()` in the `KeySegmentPair` - this
was often used to move out of the `Segment` owned by the shared pointer
in `KeySegmentPair`
- Pass `KeySegmentPair` by lvalue rather than rvalue reference where
possible. The `Task` layer still needs rvalue references as that is what
Folly expects (rightly, the Folly executors need to have ownership).
- Refactor `lookup_match_in_dedup_match` to return a variant
- Unrelated tidy up to `DecodeMetadataTask`
- `clone()` segments when adding them to the in in-memory and mock
storages

In particular this lets `CopyCompressedInterstore` safely copy the same
`KeySegmentPair` to several targets, whereas before the `Segment` could
be moved from during the copy to each target.

Author: poodlewars

cpp/arcticdb/async/async_store.cpp

@@ -7,26 +7,24 @@
 
 #include <arcticdb/async/async_store.hpp>
 #include <arcticdb/entity/variant_key.hpp>
-#include <arcticdb/codec/segment.hpp>
 #include <arcticdb/storage/key_segment_pair.hpp>
 #include <arcticdb/version/de_dup_map.hpp>
 
 namespace arcticdb::async {
-std::pair<entity::VariantKey, std::optional<Segment>> lookup_match_in_dedup_map(
+DeDupLookupResult lookup_match_in_dedup_map(
     const std::shared_ptr<DeDupMap> &de_dup_map,
-    storage::KeySegmentPair&& key_seg) {
+    storage::KeySegmentPair& key_seg) {
     std::optional<AtomKey> de_dup_key;
     if (!de_dup_map || !(de_dup_key = de_dup_map->get_key_if_present(key_seg.atom_key()))) {
         ARCTICDB_DEBUG(log::version(),
                        "No existing key with same contents: writing new object {}",
                        key_seg.atom_key());
-        return std::make_pair(std::move(key_seg.atom_key()), std::make_optional(std::move(key_seg.segment())));
-
+        return key_seg;
     } else {
         ARCTICDB_DEBUG(log::version(),
                        "Found existing key with same contents: using existing object {}",
                        *de_dup_key);
-        return std::make_pair(*de_dup_key, std::nullopt);
+        return *de_dup_key;
     }
 }
 }

cpp/arcticdb/async/async_store.hpp

@@ -24,9 +24,13 @@ namespace arcticdb::toolbox::apy{
 
 namespace arcticdb::async {
 
-std::pair<VariantKey, std::optional<Segment>> lookup_match_in_dedup_map(
+using ExistingObject = entity::VariantKey;
+using NewObject = storage::KeySegmentPair;
+using DeDupLookupResult = std::variant<ExistingObject, NewObject>;
+
+DeDupLookupResult lookup_match_in_dedup_map(
     const std::shared_ptr<DeDupMap> &de_dup_map,
-    storage::KeySegmentPair&& key_seg);
+    storage::KeySegmentPair& key_seg);
 
 template <typename Callable>
 auto read_and_continue(const VariantKey& key, std::shared_ptr<storage::Library> library, const storage::ReadKeyOpts& opts, Callable&& c) {
@@ -173,7 +177,7 @@ folly::Future<folly::Unit> write_compressed(storage::KeySegmentPair ks) override
 }
 
 void write_compressed_sync(storage::KeySegmentPair ks) override {
-    library_->write(std::move(ks));
+    library_->write(ks);
 }
 
 folly::Future<entity::VariantKey> update(const entity::VariantKey &key,
@@ -375,7 +379,6 @@ std::vector<folly::Future<bool>> batch_key_exists(
 folly::Future<SliceAndKey> async_write(
             folly::Future<std::tuple<PartialKey, SegmentInMemory, pipelines::FrameSlice>> &&input_fut,
             const std::shared_ptr<DeDupMap> &de_dup_map) override {
-        using KeyOptSegment = std::pair<VariantKey, std::optional<Segment>>;
         return std::move(input_fut).thenValue([this] (auto&& input) {
             auto [key, seg, slice] = std::forward<decltype(input)>(input);
             auto key_seg = EncodeAtomTask{
@@ -386,17 +389,20 @@ folly::Future<SliceAndKey> async_write(
                 encoding_version_}();
             return std::pair<storage::KeySegmentPair, FrameSlice>(std::move(key_seg), std::move(slice));
         })
-        .thenValue([de_dup_map](auto &&ks) -> std::pair<KeyOptSegment, pipelines::FrameSlice> {
-            auto [key_seg, slice] = std::forward<decltype(ks)>(ks);
-            return std::make_pair(lookup_match_in_dedup_map(de_dup_map, std::move(key_seg)), std::move(slice));
+        .thenValue([de_dup_map](auto &&ks) -> std::pair<DeDupLookupResult, pipelines::FrameSlice> {
+            auto& [key_seg, slice] = ks;
+            return std::make_pair<>(lookup_match_in_dedup_map(de_dup_map, key_seg), std::move(slice));
         })
         .via(&async::io_executor())
-        .thenValue([lib=library_](auto &&item) {
-            auto [key_opt_segment, slice] = std::forward<decltype(item)>(item);
-            if (key_opt_segment.second)
-                lib->write({VariantKey{key_opt_segment.first}, std::move(*key_opt_segment.second)});
-
-            return SliceAndKey{slice, to_atom(key_opt_segment.first)};
+        .thenValue([lib=library_](auto&& item) {
+            auto& [dedup_lookup, slice] = item;
+            return util::variant_match(dedup_lookup,
+               [&](NewObject& obj) {
+                   lib->write(obj);
+                   return SliceAndKey{slice, obj.atom_key()};
+               }, [&](ExistingObject& obj) {
+                    return SliceAndKey{slice, to_atom(std::move(obj))};
+               });
         });
     }
 

cpp/arcticdb/async/tasks.cpp

@@ -37,7 +37,7 @@ namespace arcticdb::async {
 
     pipelines::SegmentAndSlice DecodeSliceTask::decode_into_slice(storage::KeySegmentPair&& key_segment_pair) {
         auto key = key_segment_pair.atom_key();
-        auto& seg = key_segment_pair.segment();
+        auto& seg = *key_segment_pair.segment_ptr();
         ARCTICDB_DEBUG(log::storage(), "ReadAndDecodeAtomTask decoding segment of size {} with key {}",
                        seg.size(),
                        key);

cpp/arcticdb/async/tasks.hpp

@@ -183,7 +183,7 @@ struct WriteSegmentTask : BaseTask {
     VariantKey operator()(storage::KeySegmentPair &&key_seg) const {
         ARCTICDB_SAMPLE(WriteSegmentTask, 0)
         auto k = key_seg.variant_key();
-        lib_->write(std::move(key_seg));
+        lib_->write(key_seg);
         return k;
     }
 };
@@ -200,7 +200,7 @@ struct WriteIfNoneTask : BaseTask {
     VariantKey operator()(storage::KeySegmentPair &&key_seg) const {
         ARCTICDB_SAMPLE(WriteSegmentTask, 0)
         auto k = key_seg.variant_key();
-        lib_->write_if_none(std::move(key_seg));
+        lib_->write_if_none(key_seg);
         return k;
     }
 };
@@ -219,7 +219,7 @@ struct UpdateSegmentTask : BaseTask {
     VariantKey operator()(storage::KeySegmentPair &&key_seg) const {
         ARCTICDB_SAMPLE(UpdateSegmentTask, 0)
         auto k = key_seg.variant_key();
-        lib_->update(std::move(key_seg), opts_);
+        lib_->update(key_seg, opts_);
         return k;
     }
 };
@@ -305,9 +305,9 @@ struct CopyCompressedTask : BaseTask {
     VariantKey copy() {
         return std::visit([this](const auto &source_key) {
             auto key_seg = lib_->read_sync(source_key);
-            auto target_key_seg = stream::make_target_key<ClockType>(key_type_, stream_id_, version_id_, source_key, std::move(key_seg.segment()));
+            auto target_key_seg = stream::make_target_key<ClockType>(key_type_, stream_id_, version_id_, source_key, std::move(*key_seg.segment_ptr()));
             auto return_key = target_key_seg.variant_key();
-            lib_->write(std::move(target_key_seg));
+            lib_->write(target_key_seg);
             return return_key;
         }, source_key_);
     }
@@ -440,7 +440,7 @@ struct DecodeSegmentTask : BaseTask {
         ARCTICDB_DEBUG(log::storage(), "ReadAndDecodeAtomTask decoding segment with key {}",
                              variant_key_view(key_seg.variant_key()));
 
-        return {key_seg.variant_key(), decode_segment(std::move(key_seg.segment()))};
+        return {key_seg.variant_key(), decode_segment(*key_seg.segment_ptr())};
     }
 };
 
@@ -523,17 +523,10 @@ struct DecodeMetadataTask : BaseTask {
 
     DecodeMetadataTask() = default;
 
-    std::pair<std::optional<VariantKey>, std::optional<google::protobuf::Any>> operator()(storage::KeySegmentPair &&ks) const {
+    std::pair<std::optional<VariantKey>, std::optional<google::protobuf::Any>> operator()(storage::KeySegmentPair &&key_seg) const {
         ARCTICDB_SAMPLE(ReadMetadataTask, 0)
-        auto key_seg = std::move(ks);
         ARCTICDB_DEBUG(log::storage(), "ReadAndDecodeMetadataTask decoding segment with key {}", variant_key_view(key_seg.variant_key()));
-
-        auto meta = decode_metadata_from_segment(key_seg.segment());
-        std::pair<VariantKey, std::optional<google::protobuf::Any>> output;
-        output.first = key_seg.variant_key();
-        output.second = std::move(meta);
-
-        return output;
+        return std::make_pair<>(key_seg.variant_key(), decode_metadata_from_segment(key_seg.segment()));
     }
 };
 
@@ -542,16 +535,15 @@ struct DecodeTimeseriesDescriptorTask : BaseTask {
 
     DecodeTimeseriesDescriptorTask() = default;
 
-    std::pair<VariantKey, TimeseriesDescriptor> operator()(storage::KeySegmentPair &&ks) const {
+    std::pair<VariantKey, TimeseriesDescriptor> operator()(storage::KeySegmentPair &&key_seg) const {
         ARCTICDB_SAMPLE(DecodeTimeseriesDescriptorTask, 0)
-        auto key_seg = std::move(ks);
         ARCTICDB_DEBUG(log::storage(), "DecodeTimeseriesDescriptorTask decoding segment with key {}", variant_key_view(key_seg.variant_key()));
 
-        auto maybe_desc = decode_timeseries_descriptor(key_seg.segment());
+        auto maybe_desc = decode_timeseries_descriptor(*key_seg.segment_ptr());
 
         util::check(static_cast<bool>(maybe_desc), "Failed to decode timeseries descriptor");
         return std::make_pair(
-            std::move(key_seg.variant_key()),
+            key_seg.variant_key(),
             std::move(*maybe_desc));
 
     }
@@ -562,15 +554,14 @@ struct DecodeMetadataAndDescriptorTask : BaseTask {
 
     DecodeMetadataAndDescriptorTask() = default;
 
-    std::tuple<VariantKey, std::optional<google::protobuf::Any>, StreamDescriptor> operator()(storage::KeySegmentPair &&ks) const {
+    std::tuple<VariantKey, std::optional<google::protobuf::Any>, StreamDescriptor> operator()(storage::KeySegmentPair &&key_seg) const {
         ARCTICDB_SAMPLE(ReadMetadataAndDescriptorTask, 0)
         ARCTICDB_DEBUG_THROW(5)
-        auto key_seg = std::move(ks);
         ARCTICDB_DEBUG(log::storage(), "DecodeMetadataAndDescriptorTask decoding segment with key {}", variant_key_view(key_seg.variant_key()));
 
-        auto [any, descriptor] = decode_metadata_and_descriptor_fields(key_seg.segment());
+        auto [any, descriptor] = decode_metadata_and_descriptor_fields(*key_seg.segment_ptr());
         return std::make_tuple(
-            std::move(key_seg.variant_key()),
+            key_seg.variant_key(),
             std::move(any),
             std::move(descriptor)
             );
@@ -603,7 +594,7 @@ struct WriteCompressedTask : BaseTask {
     ARCTICDB_MOVE_ONLY_DEFAULT(WriteCompressedTask)
 
     folly::Future<folly::Unit> write() {
-        lib_->write(std::move(kv_));
+        lib_->write(kv_);
         return folly::makeFuture();
     }
 
@@ -628,7 +619,7 @@ struct WriteCompressedBatchTask : BaseTask {
 
     folly::Future<folly::Unit> write() {
         for(auto&& kv : kvs_)
-            lib_->write(std::move(kv));
+            lib_->write(kv);
 
         return folly::makeFuture();
     }

cpp/arcticdb/codec/codec.cpp

@@ -560,8 +560,7 @@ void decode_into_memory_segment(
         decode_v1(segment, hdr, res, desc);
 }
 
-SegmentInMemory decode_segment(Segment&& s) {
-    auto segment = std::move(s);
+SegmentInMemory decode_segment(Segment& segment) {
     auto &hdr = segment.header();
     ARCTICDB_TRACE(log::codec(), "Decoding descriptor: {}", segment.descriptor());
     auto descriptor = segment.descriptor();

cpp/arcticdb/codec/codec.hpp

@@ -57,7 +57,7 @@ EncodedFieldCollection decode_encoded_fields(
     const uint8_t* data,
     const uint8_t* begin ARCTICDB_UNUSED);
 
-SegmentInMemory decode_segment(Segment&& segment);
+SegmentInMemory decode_segment(Segment& segment);
 
 void decode_into_memory_segment(
     const Segment& segment,

cpp/arcticdb/codec/python_bindings.cpp

@@ -103,7 +103,7 @@ Segment encode_segment(SegmentInMemory segment_in_memory, const py::object &opts
 }
 
 SegmentInMemory decode_python_segment(Segment& segment) {
-    return decode_segment(std::move(segment));
+    return decode_segment(segment);
 }
 
 class BufferPairDataSink {

cpp/arcticdb/codec/test/test_codec.cpp

@@ -319,7 +319,7 @@ TYPED_TEST(SegmentStringEncodingTest, EncodeSingleString) {
     constexpr EncodingVersion encoding_version = TypeParam::value;
     Segment seg = encode_dispatch(s.clone(), opt, encoding_version);
 
-    SegmentInMemory res = decode_segment(std::move(seg));
+    SegmentInMemory res = decode_segment(seg);
     ASSERT_EQ(copy.string_at(0, 1), res.string_at(0, 1));
     ASSERT_EQ(std::string("happy"), res.string_at(0, 1));
 }
@@ -346,7 +346,7 @@ TYPED_TEST(SegmentStringEncodingTest, EncodeStringsBasic) {
     constexpr EncodingVersion encoding_version = TypeParam::value;
     Segment seg = encode_dispatch(SegmentInMemory{s}, opt, encoding_version);
 
-    SegmentInMemory res = decode_segment(std::move(seg));
+    SegmentInMemory res = decode_segment(seg);
     ASSERT_EQ(copy.string_at(0, 1), res.string_at(0, 1));
     ASSERT_EQ(std::string("happy"), res.string_at(0, 1));
     ASSERT_EQ(copy.string_at(1, 3), res.string_at(1, 3));

cpp/arcticdb/pipeline/read_frame.cpp

@@ -488,12 +488,11 @@ void check_data_left_for_subsequent_fields(
 void decode_into_frame_static(
     SegmentInMemory &frame,
     PipelineContextRow &context,
-    Segment &&s,
+    const Segment& seg,
     const DecodePathData& shared_data,
     std::any& handler_data,
     const ReadQuery& read_query,
     const ReadOptions& read_options) {
-    auto seg = std::move(s);
     ARCTICDB_SAMPLE_DEFAULT(DecodeIntoFrame)
     const uint8_t *data = seg.buffer().data();
     const uint8_t *begin = data;
@@ -626,14 +625,13 @@ void handle_type_promotion(
 void decode_into_frame_dynamic(
     SegmentInMemory& frame,
     PipelineContextRow& context,
-    Segment&& s,
+    const Segment& seg,
     const DecodePathData& shared_data,
     std::any& handler_data,
     const ReadQuery& read_query,
     const ReadOptions& read_options
 ) {
     ARCTICDB_SAMPLE_DEFAULT(DecodeIntoFrame)
-    auto seg = std::move(s);
     const uint8_t *data = seg.buffer().data();
     const uint8_t *begin = data;
     const uint8_t *end = begin + seg.buffer().bytes();
@@ -887,9 +885,9 @@ folly::Future<SegmentInMemory> fetch_data(
             [row=row, frame=frame, dynamic_schema=dynamic_schema, shared_data, &handler_data, read_query, read_options](auto &&ks) mutable {
                 auto key_seg = std::forward<storage::KeySegmentPair>(ks);
                 if(dynamic_schema) {
-                    decode_into_frame_dynamic(frame, row, std::move(key_seg.segment()), shared_data, handler_data, read_query, read_options);
+                    decode_into_frame_dynamic(frame, row, key_seg.segment(), shared_data, handler_data, read_query, read_options);
                 } else {
-                    decode_into_frame_static(frame, row, std::move(key_seg.segment()), shared_data, handler_data, read_query, read_options);
+                    decode_into_frame_static(frame, row, key_seg.segment(), shared_data, handler_data, read_query, read_options);
                 }
 
                 return key_seg.variant_key();

cpp/arcticdb/pipeline/read_frame.hpp

@@ -85,16 +85,16 @@ folly::Future<SegmentInMemory> fetch_data(
 void decode_into_frame_static(
     SegmentInMemory &frame,
     PipelineContextRow &context,
-    Segment &&seg,
+    const Segment& seg,
     const DecodePathData& shared_data,
     std::any& handler_data,
     const ReadQuery& read_query,
     const ReadOptions& read_options);
 
 void decode_into_frame_dynamic(
-    SegmentInMemory &frame,
+    const SegmentInMemory &frame,
     PipelineContextRow &context,
-    Segment &&seg,
+    const Segment& seg,
     const DecodePathData& shared_data,
     std::any& handler_data,
     const ReadQuery& read_query,

cpp/arcticdb/storage/azure/azure_client_impl.cpp

@@ -47,7 +47,7 @@ Azure::Storage::Blobs::BlobClientOptions RealAzureClient::get_client_options(con
 
 void RealAzureClient::write_blob(
         const std::string& blob_name,
-        Segment&& segment,
+        Segment& segment,
         const Azure::Storage::Blobs::UploadBlockBlobFromOptions& upload_option,
         unsigned int request_timeout) {
 

cpp/arcticdb/storage/azure/azure_client_impl.hpp

@@ -26,7 +26,7 @@ class RealAzureClient : public AzureClientWrapper {
 
     void write_blob(
             const std::string& blob_name,
-            Segment&& segment,
+            Segment& segment,
             const Azure::Storage::Blobs::UploadBlockBlobFromOptions& upload_option,
             unsigned int request_timeout) override;
 

cpp/arcticdb/storage/azure/azure_client_interface.hpp

@@ -50,7 +50,7 @@ class AzureClientWrapper {
     using Config = arcticdb::proto::azure_storage::Config;
     virtual void write_blob(
             const std::string& blob_name,
-            Segment&& segment,
+            Segment& segment,
             const Azure::Storage::Blobs::UploadBlockBlobFromOptions& upload_option,
             unsigned int request_timeout) = 0;