diff --git a/cpp/arcticdb/async/async_store.cpp b/cpp/arcticdb/async/async_store.cpp
index ce49757f4fe..0131be65eb5 100644
--- a/cpp/arcticdb/async/async_store.cpp
+++ b/cpp/arcticdb/async/async_store.cpp
@@ -20,8 +20,7 @@ std::pair<entity::VariantKey, std::optional<Segment>> lookup_match_in_dedup_map(
         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 std::make_pair(key_seg.atom_key(), std::make_optional(std::move(*key_seg.release_segment())));
     } else {
         ARCTICDB_DEBUG(log::version(),
                        "Found existing key with same contents: using existing object {}",
diff --git a/cpp/arcticdb/async/tasks.cpp b/cpp/arcticdb/async/tasks.cpp
index b209f8528ed..cdb1d5d0aad 100644
--- a/cpp/arcticdb/async/tasks.cpp
+++ b/cpp/arcticdb/async/tasks.cpp
@@ -37,17 +37,17 @@ 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(),
+                       seg->size(),
                        key);
-        auto &hdr = seg.header();
-        const auto& desc = seg.descriptor();
+        auto &hdr = seg->header();
+        const auto& desc = seg->descriptor();
         auto descriptor = async::get_filtered_descriptor(desc, columns_to_decode_);
         ranges_and_key_.col_range_.second = ranges_and_key_.col_range_.first + (descriptor.field_count() - descriptor.index().field_count());
         ARCTICDB_TRACE(log::codec(), "Creating segment");
         SegmentInMemory segment_in_memory(std::move(descriptor));
-        decode_into_memory_segment(seg, hdr, segment_in_memory, desc);
+        decode_into_memory_segment(*seg, hdr, segment_in_memory, desc);
         return pipelines::SegmentAndSlice(std::move(ranges_and_key_), std::move(segment_in_memory));
     }
 
diff --git a/cpp/arcticdb/async/tasks.hpp b/cpp/arcticdb/async/tasks.hpp
index 3e564e95045..af3ec5d438d 100644
--- a/cpp/arcticdb/async/tasks.hpp
+++ b/cpp/arcticdb/async/tasks.hpp
@@ -278,7 +278,12 @@ struct CopyCompressedTask : BaseTask {
     VariantKey copy() {
         return std::visit([that = this](const auto &source_key) {
             auto key_seg = that->lib_->read(source_key);
-            auto target_key_seg = stream::make_target_key<ClockType>(that->key_type_, that->stream_id_, that->version_id_, source_key, std::move(key_seg.segment()));
+            auto target_key_seg = stream::make_target_key<ClockType>(
+                that->key_type_,
+                that->stream_id_,
+                that->version_id_,
+                source_key,
+                std::move(*key_seg.release_segment()));
             auto return_key = target_key_seg.variant_key();
             that->lib_->write(Composite<storage::KeySegmentPair>{std::move(target_key_seg) });
             return return_key;
@@ -406,7 +411,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())};
     }
 };
 
@@ -543,7 +548,7 @@ struct DecodeTimeseriesDescriptorTask : BaseTask {
         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(
@@ -566,7 +571,7 @@ struct DecodeTimeseriesDescriptorForIncompletesTask : BaseTask {
                 "DecodeTimeseriesDescriptorForIncompletesTask decoding segment with key {}",
                 variant_key_view(key_seg.variant_key()));
 
-        auto maybe_desc = decode_timeseries_descriptor_for_incompletes(key_seg.segment());
+        auto maybe_desc = decode_timeseries_descriptor_for_incompletes(*key_seg.segment_ptr());
 
         util::check(static_cast<bool>(maybe_desc), "Failed to decode timeseries descriptor");
         return std::make_pair(
@@ -585,7 +590,7 @@ struct DecodeMetadataAndDescriptorTask : BaseTask {
         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()),
             std::move(any),
diff --git a/cpp/arcticdb/codec/codec.cpp b/cpp/arcticdb/codec/codec.cpp
index 801ddbaf71b..6677f90418b 100644
--- a/cpp/arcticdb/codec/codec.cpp
+++ b/cpp/arcticdb/codec/codec.cpp
@@ -557,8 +557,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();
diff --git a/cpp/arcticdb/codec/codec.hpp b/cpp/arcticdb/codec/codec.hpp
index 446bacaed23..b2ca6f00b33 100644
--- a/cpp/arcticdb/codec/codec.hpp
+++ b/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,
diff --git a/cpp/arcticdb/codec/test/test_codec.cpp b/cpp/arcticdb/codec/test/test_codec.cpp
index dc8694b8545..ac84522f805 100644
--- a/cpp/arcticdb/codec/test/test_codec.cpp
+++ b/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));
diff --git a/cpp/arcticdb/pipeline/read_frame.cpp b/cpp/arcticdb/pipeline/read_frame.cpp
index 1358c6e3a34..0a0fbd21ba3 100644
--- a/cpp/arcticdb/pipeline/read_frame.cpp
+++ b/cpp/arcticdb/pipeline/read_frame.cpp
@@ -289,9 +289,8 @@ bool remaining_fields_empty(IteratorType it, const PipelineContextRow& context)
 void decode_into_frame_static(
     SegmentInMemory &frame,
     PipelineContextRow &context,
-    Segment &&s,
+    Segment& seg,
     const std::shared_ptr<BufferHolder>& buffers) {
-    auto seg = std::move(s);
     ARCTICDB_SAMPLE_DEFAULT(DecodeIntoFrame)
     const uint8_t *data = seg.buffer().data();
     const uint8_t *begin = data;
@@ -379,11 +378,10 @@ void decode_into_frame_static(
 void decode_into_frame_dynamic(
     SegmentInMemory& frame,
     PipelineContextRow& context,
-    Segment&& s,
+    Segment& seg,
     const std::shared_ptr<BufferHolder>& buffers
 ) {
     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();
@@ -1170,9 +1168,9 @@ folly::Future<std::vector<VariantKey>> fetch_data(
             [row=row, frame=frame, dynamic_schema=dynamic_schema, buffers](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()), buffers);
+                    decode_into_frame_dynamic(frame, row, *key_seg.segment_ptr(), buffers);
                 else
-                    decode_into_frame_static(frame, row, std::move(key_seg.segment()), buffers);
+                    decode_into_frame_static(frame, row, *key_seg.segment_ptr(), buffers);
                 return key_seg.variant_key();
             });
         }
diff --git a/cpp/arcticdb/pipeline/read_frame.hpp b/cpp/arcticdb/pipeline/read_frame.hpp
index d0c28348b86..2c4a4a906ab 100644
--- a/cpp/arcticdb/pipeline/read_frame.hpp
+++ b/cpp/arcticdb/pipeline/read_frame.hpp
@@ -81,14 +81,14 @@ folly::Future<std::vector<VariantKey>> fetch_data(
 void decode_into_frame_static(
     SegmentInMemory &frame,
     PipelineContextRow &context,
-    Segment &&seg,
+    Segment& seg,
     const std::shared_ptr<BufferHolder>& buffers
     );
 
 void decode_into_frame_dynamic(
         SegmentInMemory &frame,
         PipelineContextRow &context,
-        Segment &&seg,
+        Segment& seg,
         const std::shared_ptr<BufferHolder>& buffers
 );
 
diff --git a/cpp/arcticdb/storage/azure/azure_client_wrapper.hpp b/cpp/arcticdb/storage/azure/azure_client_wrapper.hpp
index bfae88ae648..3288effd849 100644
--- a/cpp/arcticdb/storage/azure/azure_client_wrapper.hpp
+++ b/cpp/arcticdb/storage/azure/azure_client_wrapper.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;
 
diff --git a/cpp/arcticdb/storage/azure/azure_mock_client.cpp b/cpp/arcticdb/storage/azure/azure_mock_client.cpp
index e3194eaa062..ca0880fd416 100644
--- a/cpp/arcticdb/storage/azure/azure_mock_client.cpp
+++ b/cpp/arcticdb/storage/azure/azure_mock_client.cpp
@@ -55,7 +55,7 @@ std::optional<Azure::Core::RequestFailedException> has_failure_trigger(const std
 
 void MockAzureClient::write_blob(
         const std::string& blob_name,
-        arcticdb::Segment&& segment,
+        arcticdb::Segment& segment,
         const Azure::Storage::Blobs::UploadBlockBlobFromOptions&,
         unsigned int) {
 
diff --git a/cpp/arcticdb/storage/azure/azure_mock_client.hpp b/cpp/arcticdb/storage/azure/azure_mock_client.hpp
index 147324f61e7..eaa539f9c27 100644
--- a/cpp/arcticdb/storage/azure/azure_mock_client.hpp
+++ b/cpp/arcticdb/storage/azure/azure_mock_client.hpp
@@ -24,7 +24,7 @@ class MockAzureClient : 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;
 
diff --git a/cpp/arcticdb/storage/azure/azure_real_client.cpp b/cpp/arcticdb/storage/azure/azure_real_client.cpp
index be354768ae4..7364985468c 100644
--- a/cpp/arcticdb/storage/azure/azure_real_client.cpp
+++ b/cpp/arcticdb/storage/azure/azure_real_client.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) {
 
diff --git a/cpp/arcticdb/storage/azure/azure_real_client.hpp b/cpp/arcticdb/storage/azure/azure_real_client.hpp
index 9fc8116de05..e63017a4e12 100644
--- a/cpp/arcticdb/storage/azure/azure_real_client.hpp
+++ b/cpp/arcticdb/storage/azure/azure_real_client.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;
 
diff --git a/cpp/arcticdb/storage/azure/azure_storage.cpp b/cpp/arcticdb/storage/azure/azure_storage.cpp
index 132a591385d..7af59526b1d 100644
--- a/cpp/arcticdb/storage/azure/azure_storage.cpp
+++ b/cpp/arcticdb/storage/azure/azure_storage.cpp
@@ -124,10 +124,9 @@ void do_write_impl(
                 for (auto& kv : group.values()) {
                     auto& k = kv.variant_key();
                     auto blob_name = object_path(b.bucketize(key_type_dir, k), k);
-                    auto& seg = kv.segment();
 
                     try {
-                        azure_client.write_blob(blob_name, std::move(seg), upload_option, request_timeout);
+                        azure_client.write_blob(blob_name, *kv.segment_ptr(), upload_option, request_timeout);
                     }
                     catch (const Azure::Core::RequestFailedException& e) {
                         raise_azure_exception(e, blob_name);
diff --git a/cpp/arcticdb/storage/file/mapped_file_storage.cpp b/cpp/arcticdb/storage/file/mapped_file_storage.cpp
index ba9c6e0db18..56dd4cbb331 100644
--- a/cpp/arcticdb/storage/file/mapped_file_storage.cpp
+++ b/cpp/arcticdb/storage/file/mapped_file_storage.cpp
@@ -56,7 +56,7 @@ void MappedFileStorage::init() {
 
 SegmentInMemory MappedFileStorage::read_segment(size_t offset, size_t bytes) const  {
     auto index_segment = Segment::from_bytes(file_.data() + offset, bytes);
-    return decode_segment(std::move(index_segment));
+    return decode_segment(index_segment);
 }
 
 void MappedFileStorage::do_load_header(size_t header_offset, size_t header_size) {
@@ -75,7 +75,7 @@ uint64_t MappedFileStorage::get_data_offset(const Segment& seg) {
     return previous_offset;
 }
 
-uint64_t MappedFileStorage::write_segment(Segment&& seg) {
+uint64_t MappedFileStorage::write_segment(Segment& seg) {
     auto segment = std::move(seg);
     auto offset = get_data_offset(segment);
     auto* data = file_.data() + offset;
@@ -89,7 +89,7 @@ void MappedFileStorage::do_write(Composite<KeySegmentPair>&& kvs) {
     ARCTICDB_SAMPLE(MappedFileStorageWriteValues, 0)
     auto key_values = std::move(kvs);
     key_values.broadcast([this] (auto key_seg) {
-        const auto offset = write_segment(std::move(key_seg.segment()));
+        const auto offset = write_segment(*key_seg.segment_ptr());
         const auto size = key_seg.segment().size();
         multi_segment_header_.add_key_and_offset(key_seg.atom_key(), offset, size);
     });
@@ -129,7 +129,7 @@ void MappedFileStorage::do_finalize(KeyData key_data)  {
     auto header_segment = encode_dispatch(multi_segment_header_.detach_segment(),
                                           config_.codec_opts(),
                                           EncodingVersion{static_cast<uint16_t>(config_.encoding_version())});
-    write_segment(std::move(header_segment));
+    write_segment(header_segment);
     auto pos = reinterpret_cast<KeyData*>(file_.data() + offset_);
     *pos = key_data;
     ARCTICDB_DEBUG(log::storage(), "Finalizing mapped file, writing key data {}", *pos);
diff --git a/cpp/arcticdb/storage/file/mapped_file_storage.hpp b/cpp/arcticdb/storage/file/mapped_file_storage.hpp
index 98a60a024a7..9a7e9414113 100644
--- a/cpp/arcticdb/storage/file/mapped_file_storage.hpp
+++ b/cpp/arcticdb/storage/file/mapped_file_storage.hpp
@@ -62,7 +62,7 @@ class MappedFileStorage final : public SingleFileStorage {
 
     void do_load_header(size_t header_offset, size_t header_size) override;
 
-    uint64_t write_segment(Segment&& seg);
+    uint64_t write_segment(Segment& seg);
 
     uint8_t* do_read_raw(size_t offset, size_t bytes) override;
 
diff --git a/cpp/arcticdb/storage/key_segment_pair.hpp b/cpp/arcticdb/storage/key_segment_pair.hpp
index 7baa92e6edc..aba4c5ce333 100644
--- a/cpp/arcticdb/storage/key_segment_pair.hpp
+++ b/cpp/arcticdb/storage/key_segment_pair.hpp
@@ -35,13 +35,15 @@ namespace arcticdb::storage {
 
         ARCTICDB_MOVE_COPY_DEFAULT(KeySegmentPair)
 
-        // TODO aseaton remove
-        Segment& segment() {
+        std::unique_ptr<Segment> release_segment() {
             util::check(segment_, "Attempting to access segment_ but it has not been set");
-            return *segment_;
+            auto tmp = std::make_unique<Segment>(std::move(*segment_));
+            segment_ = std::make_shared<Segment>();
+            return tmp;
         }
 
         [[nodiscard]] std::shared_ptr<Segment> segment_ptr() const {
+          util::check(segment_, "Attempting to access segment_ptr it is empty");
           return segment_;
         }
 
@@ -50,6 +52,10 @@ namespace arcticdb::storage {
           key_ = std::make_shared<VariantKey>(std::forward<T>(key));
         }
 
+        void set_segment(Segment&& segment) {
+          segment_ = std::make_shared<Segment>(std::move(segment));
+        }
+
         [[nodiscard]] const AtomKey &atom_key() const {
             util::check(std::holds_alternative<AtomKey>(variant_key()), "Expected atom key access");
             return std::get<AtomKey >(variant_key());
diff --git a/cpp/arcticdb/storage/library.hpp b/cpp/arcticdb/storage/library.hpp
index dd102e23586..093d4b2d2c6 100644
--- a/cpp/arcticdb/storage/library.hpp
+++ b/cpp/arcticdb/storage/library.hpp
@@ -125,7 +125,7 @@ class Library {
         KeySegmentPair res{VariantKey{key}};
         util::check(!std::holds_alternative<StringId>(variant_key_id(key)) || !std::get<StringId>(variant_key_id(key)).empty(), "Unexpected empty id");
         const ReadVisitor& visitor = [&res](const VariantKey&, Segment&& value) {
-            res.segment() = std::move(value);
+            res.set_segment(std::move(value));
         };
 
         read(Composite<VariantKey>(std::move(key)), visitor, opts);
diff --git a/cpp/arcticdb/storage/lmdb/lmdb_client_wrapper.hpp b/cpp/arcticdb/storage/lmdb/lmdb_client_wrapper.hpp
index e1860f3bf1e..15a52c3f643 100644
--- a/cpp/arcticdb/storage/lmdb/lmdb_client_wrapper.hpp
+++ b/cpp/arcticdb/storage/lmdb/lmdb_client_wrapper.hpp
@@ -37,7 +37,7 @@ class LmdbClientWrapper {
     virtual void write(
             const std::string& db_name,
             std::string& path,
-            Segment&& segment,
+            Segment& segment,
             ::lmdb::txn& txn,
             ::lmdb::dbi& dbi,
             int64_t overwrite_flag) = 0;
diff --git a/cpp/arcticdb/storage/lmdb/lmdb_mock_client.cpp b/cpp/arcticdb/storage/lmdb/lmdb_mock_client.cpp
index e403bc3758f..f125537c2ba 100644
--- a/cpp/arcticdb/storage/lmdb/lmdb_mock_client.cpp
+++ b/cpp/arcticdb/storage/lmdb/lmdb_mock_client.cpp
@@ -90,7 +90,7 @@ std::optional<Segment> MockLmdbClient::read(const std::string& db_name, std::str
     return std::make_optional<Segment>(lmdb_contents_.at(key).clone());
 }
 
-void MockLmdbClient::write(const std::string& db_name, std::string& path, arcticdb::Segment&& segment,
+void MockLmdbClient::write(const std::string& db_name, std::string& path, arcticdb::Segment& segment,
                            ::lmdb::txn&, ::lmdb::dbi&, int64_t) {
     LmdbKey key = {db_name, path};
             raise_if_has_failure_trigger(key, StorageOperation::WRITE);
@@ -98,7 +98,7 @@ void MockLmdbClient::write(const std::string& db_name, std::string& path, arctic
     if(has_key(key)) {
         raise_key_exists_error(lmdb_operation_string(StorageOperation::WRITE));
     } else {
-        lmdb_contents_.try_emplace(key, std::move(segment));
+        lmdb_contents_.try_emplace(key, segment.clone());
     }
 }
 
diff --git a/cpp/arcticdb/storage/lmdb/lmdb_mock_client.hpp b/cpp/arcticdb/storage/lmdb/lmdb_mock_client.hpp
index 6e81bf67303..c4f3e38174c 100644
--- a/cpp/arcticdb/storage/lmdb/lmdb_mock_client.hpp
+++ b/cpp/arcticdb/storage/lmdb/lmdb_mock_client.hpp
@@ -55,7 +55,7 @@ class MockLmdbClient : public LmdbClientWrapper {
     void write(
             const std::string& db_name,
             std::string& path,
-            Segment&& segment,
+            Segment& segment,
             ::lmdb::txn& txn,
             ::lmdb::dbi& dbi,
             int64_t overwrite_flag) override;
diff --git a/cpp/arcticdb/storage/lmdb/lmdb_real_client.cpp b/cpp/arcticdb/storage/lmdb/lmdb_real_client.cpp
index 547abcf1145..1e531836843 100644
--- a/cpp/arcticdb/storage/lmdb/lmdb_real_client.cpp
+++ b/cpp/arcticdb/storage/lmdb/lmdb_real_client.cpp
@@ -39,7 +39,7 @@ std::optional<Segment> RealLmdbClient::read(const std::string&, std::string& pat
     return segment;
 }
 
-void RealLmdbClient::write(const std::string&, std::string& path, arcticdb::Segment&& seg,
+void RealLmdbClient::write(const std::string&, std::string& path, arcticdb::Segment& seg,
                            ::lmdb::txn& txn, ::lmdb::dbi& dbi, int64_t overwrite_flag) {
     MDB_val mdb_key{path.size(), path.data()};
 
diff --git a/cpp/arcticdb/storage/lmdb/lmdb_real_client.hpp b/cpp/arcticdb/storage/lmdb/lmdb_real_client.hpp
index 279ed5f0066..2f7a6ff5ab4 100644
--- a/cpp/arcticdb/storage/lmdb/lmdb_real_client.hpp
+++ b/cpp/arcticdb/storage/lmdb/lmdb_real_client.hpp
@@ -34,7 +34,7 @@ class RealLmdbClient : public LmdbClientWrapper {
     void write(
             const std::string& db_name,
             std::string& path,
-            Segment&& segment,
+            Segment& segment,
             ::lmdb::txn& txn,
             ::lmdb::dbi& dbi,
             int64_t overwrite_flag) override;
diff --git a/cpp/arcticdb/storage/lmdb/lmdb_storage.cpp b/cpp/arcticdb/storage/lmdb/lmdb_storage.cpp
index 59929663ebf..11faabd6b79 100644
--- a/cpp/arcticdb/storage/lmdb/lmdb_storage.cpp
+++ b/cpp/arcticdb/storage/lmdb/lmdb_storage.cpp
@@ -75,10 +75,9 @@ void LmdbStorage::do_write_internal(Composite<KeySegmentPair>&& kvs, ::lmdb::txn
         for (auto &kv : group.values()) {
             ARCTICDB_DEBUG(log::storage(), "Lmdb storage writing segment with key {}", kv.key_view());
             auto k = to_serialized_key(kv.variant_key());
-            auto &seg = kv.segment();
             int64_t overwrite_flag = std::holds_alternative<RefKey>(kv.variant_key()) ? 0 : MDB_NOOVERWRITE;
             try {
-                lmdb_client_->write(db_name, k, std::move(seg), txn, dbi, overwrite_flag);
+                lmdb_client_->write(db_name, k, *kv.segment_ptr(), txn, dbi, overwrite_flag);
             } catch (const ::lmdb::key_exist_error& e) {
                 throw DuplicateKeyException(fmt::format("Key already exists: {}: {}", kv.variant_key(), e.what()));
             } catch (const ::lmdb::error& ex) {
diff --git a/cpp/arcticdb/storage/memory/memory_storage.cpp b/cpp/arcticdb/storage/memory/memory_storage.cpp
index f36ac83f957..330e6d864e6 100644
--- a/cpp/arcticdb/storage/memory/memory_storage.cpp
+++ b/cpp/arcticdb/storage/memory/memory_storage.cpp
@@ -18,11 +18,11 @@
 namespace arcticdb::storage::memory {
 
 void add_serialization_fields(KeySegmentPair& kv) {
-    auto& segment = kv.segment();
-    auto& hdr = segment.header();
-    (void)segment.calculate_size();
+    auto segment = kv.segment_ptr();
+    auto& hdr = segment->header();
+    (void)segment->calculate_size();
     if(hdr.encoding_version() == EncodingVersion::V2) {
-        const auto* src = segment.buffer().data();
+        const auto* src = segment->buffer().data();
         set_body_fields(hdr, src);
     }
 }
@@ -48,14 +48,14 @@ void add_serialization_fields(KeySegmentPair& kv) {
                                             key_vec.erase(it);
                                         }
                                         add_serialization_fields(kv);
-                                        key_vec.try_emplace(key, std::move(kv.segment()));
+                                        key_vec.try_emplace(key, std::move(*kv.release_segment()));
                                     },
                                     [&](const AtomKey &key) {
                                         if (key_vec.find(key) != key_vec.end()) {
                                             throw DuplicateKeyException(key);
                                         }
                                         add_serialization_fields(kv);
-                                        key_vec.try_emplace(key, std::move(kv.segment()));
+                                        key_vec.try_emplace(key, std::move(*kv.release_segment()));
                                     }
                 );
             }
diff --git a/cpp/arcticdb/storage/mongo/mongo_client.cpp b/cpp/arcticdb/storage/mongo/mongo_client.cpp
index 9d68414e5e2..496b852dead 100644
--- a/cpp/arcticdb/storage/mongo/mongo_client.cpp
+++ b/cpp/arcticdb/storage/mongo/mongo_client.cpp
@@ -134,12 +134,11 @@ auto build_document(storage::KeySegmentPair &kv) {
     using builder::stream::document;
 
     const auto &key = kv.variant_key();
-    auto &segment = kv.segment();
-    const auto total_size = segment.calculate_size();
+    const auto total_size = kv.segment_ptr()->calculate_size();
     /*thread_local*/ std::vector<uint8_t> buffer{};
     buffer.resize(total_size);
     bsoncxx::types::b_binary data = {};
-    kv.segment().write_to(buffer.data());
+    kv.segment_ptr()->write_to(buffer.data());
     data.size = uint32_t(total_size);
     data.bytes = buffer.data();
 
diff --git a/cpp/arcticdb/storage/mongo/mongo_mock_client.cpp b/cpp/arcticdb/storage/mongo/mongo_mock_client.cpp
index cedf82baf86..1c32691a7eb 100644
--- a/cpp/arcticdb/storage/mongo/mongo_mock_client.cpp
+++ b/cpp/arcticdb/storage/mongo/mongo_mock_client.cpp
@@ -107,7 +107,7 @@ bool MockMongoClient::write_segment(
         return false;
     }
 
-    mongo_contents.insert_or_assign(std::move(key), std::move(kv.segment()));
+    mongo_contents.insert_or_assign(std::move(key), kv.segment().clone());
     return true;
 }
 
@@ -129,7 +129,7 @@ UpdateResult MockMongoClient::update_segment(
         return {0}; // upsert is false, don't update and return 0 as modified_count
     }
 
-    mongo_contents.insert_or_assign(std::move(key), std::move(kv.segment()));
+    mongo_contents.insert_or_assign(std::move(key), kv.segment().clone());
     return {key_found ? 1 : 0};
 }
 
diff --git a/cpp/arcticdb/storage/mongo/mongo_storage.cpp b/cpp/arcticdb/storage/mongo/mongo_storage.cpp
index ff11cfe14a8..0c184ac6ba7 100644
--- a/cpp/arcticdb/storage/mongo/mongo_storage.cpp
+++ b/cpp/arcticdb/storage/mongo/mongo_storage.cpp
@@ -134,7 +134,7 @@ void MongoStorage::do_read(Composite<VariantKey>&& ks, const ReadVisitor& visito
                     keys_not_found.push_back(k);
                 }
                 else {
-                    visitor(k, std::move(kv.value().segment()));
+                    visitor(k,  std::move(*kv.value().release_segment()));
                 }
 
             } catch (const mongocxx::operation_exception& ex) {
diff --git a/cpp/arcticdb/storage/rocksdb/rocksdb_storage.cpp b/cpp/arcticdb/storage/rocksdb/rocksdb_storage.cpp
index 566c827c3e9..0c85cbd71bd 100644
--- a/cpp/arcticdb/storage/rocksdb/rocksdb_storage.cpp
+++ b/cpp/arcticdb/storage/rocksdb/rocksdb_storage.cpp
@@ -236,11 +236,10 @@ void RocksDBStorage::do_write_internal(Composite<KeySegmentPair>&& kvs) {
         for (auto &kv : group.values()) {
             auto k_str = to_serialized_key(kv.variant_key());
 
-            auto& seg = kv.segment();
-            auto total_sz = seg.calculate_size();
+            auto total_sz = kv.segment_ptr()->calculate_size();
             std::string seg_data;
             seg_data.resize(total_sz);
-            seg.write_to(reinterpret_cast<std::uint8_t *>(seg_data.data()));
+            kv.segment_ptr()->write_to(reinterpret_cast<std::uint8_t *>(seg_data.data()));
             auto allow_override = std::holds_alternative<RefKey>(kv.variant_key());
             if (!allow_override && do_key_exists(kv.variant_key())) {
                 throw DuplicateKeyException(kv.variant_key());
diff --git a/cpp/arcticdb/storage/s3/detail-inl.hpp b/cpp/arcticdb/storage/s3/detail-inl.hpp
index cc5913aed8f..472b25cf910 100644
--- a/cpp/arcticdb/storage/s3/detail-inl.hpp
+++ b/cpp/arcticdb/storage/s3/detail-inl.hpp
@@ -116,9 +116,8 @@ namespace s3 {
                         for (auto &kv: group.values()) {
                             auto &k = kv.variant_key();
                             auto s3_object_name = object_path(b.bucketize(key_type_dir, k), k);
-                            auto &seg = kv.segment();
 
-                            auto put_object_result = s3_client.put_object(s3_object_name, std::move(seg), bucket_name);
+                            auto put_object_result = s3_client.put_object(s3_object_name, *kv.segment_ptr(), bucket_name);
 
                             if (!put_object_result.is_success()) {
                                 auto& error = put_object_result.get_error();
diff --git a/cpp/arcticdb/storage/s3/s3_client_wrapper.hpp b/cpp/arcticdb/storage/s3/s3_client_wrapper.hpp
index 96c98e13ce6..71a77550ba3 100644
--- a/cpp/arcticdb/storage/s3/s3_client_wrapper.hpp
+++ b/cpp/arcticdb/storage/s3/s3_client_wrapper.hpp
@@ -56,7 +56,7 @@ class S3ClientWrapper {
 
     virtual S3Result<std::monostate> put_object(
             const std::string& s3_object_name,
-            Segment&& segment,
+            Segment& segment,
             const std::string& bucket_name) = 0;
 
     virtual S3Result<DeleteOutput> delete_objects(
diff --git a/cpp/arcticdb/storage/s3/s3_mock_client.cpp b/cpp/arcticdb/storage/s3/s3_mock_client.cpp
index 44d646cb42b..2f5f1e326f4 100644
--- a/cpp/arcticdb/storage/s3/s3_mock_client.cpp
+++ b/cpp/arcticdb/storage/s3/s3_mock_client.cpp
@@ -82,14 +82,14 @@ S3Result<Segment> MockS3Client::get_object(
 
 S3Result<std::monostate> MockS3Client::put_object(
         const std::string &s3_object_name,
-        Segment &&segment,
+        Segment &segment,
         const std::string &bucket_name) {
     auto maybe_error = has_failure_trigger(s3_object_name, StorageOperation::WRITE);
     if (maybe_error.has_value()) {
         return {maybe_error.value()};
     }
 
-    s3_contents.insert_or_assign({bucket_name, s3_object_name}, std::move(segment));
+    s3_contents.insert_or_assign({bucket_name, s3_object_name}, segment.clone());
 
     return {std::monostate()};
 }
diff --git a/cpp/arcticdb/storage/s3/s3_mock_client.hpp b/cpp/arcticdb/storage/s3/s3_mock_client.hpp
index 04c68122182..4ee33505beb 100644
--- a/cpp/arcticdb/storage/s3/s3_mock_client.hpp
+++ b/cpp/arcticdb/storage/s3/s3_mock_client.hpp
@@ -60,7 +60,7 @@ class MockS3Client : public S3ClientWrapper {
 
     S3Result<std::monostate> put_object(
             const std::string& s3_object_name,
-            Segment&& segment,
+            Segment& segment,
             const std::string& bucket_name) override;
 
     S3Result<DeleteOutput> delete_objects(
diff --git a/cpp/arcticdb/storage/s3/s3_real_client.cpp b/cpp/arcticdb/storage/s3/s3_real_client.cpp
index 7a070027d91..b21e8873657 100644
--- a/cpp/arcticdb/storage/s3/s3_real_client.cpp
+++ b/cpp/arcticdb/storage/s3/s3_real_client.cpp
@@ -132,7 +132,7 @@ S3Result<Segment> RealS3Client::get_object(
 
 S3Result<std::monostate> RealS3Client::put_object(
         const std::string &s3_object_name,
-        Segment &&segment,
+        Segment &segment,
         const std::string &bucket_name) {
 
     ARCTICDB_SUBSAMPLE(S3StorageWritePreamble, 0)
diff --git a/cpp/arcticdb/storage/s3/s3_real_client.hpp b/cpp/arcticdb/storage/s3/s3_real_client.hpp
index 57bfe87b65c..56f33b7bfd1 100644
--- a/cpp/arcticdb/storage/s3/s3_real_client.hpp
+++ b/cpp/arcticdb/storage/s3/s3_real_client.hpp
@@ -37,7 +37,7 @@ class RealS3Client : public S3ClientWrapper {
 
     S3Result<std::monostate> put_object(
             const std::string& s3_object_name,
-            Segment&& segment,
+            Segment& segment,
             const std::string& bucket_name) override;
 
     S3Result<DeleteOutput> delete_objects(
diff --git a/cpp/arcticdb/storage/storage.hpp b/cpp/arcticdb/storage/storage.hpp
index f58c0078281..c5f370472ce 100644
--- a/cpp/arcticdb/storage/storage.hpp
+++ b/cpp/arcticdb/storage/storage.hpp
@@ -129,7 +129,7 @@ class Storage {
         KeySegmentPair key_seg;
         const ReadVisitor& visitor = [&key_seg](const VariantKey & vk, Segment&& value) {
             key_seg.set_key(vk);
-            key_seg.segment() = std::move(value);
+            key_seg.set_segment(std::move(value));
         };
 
         read(std::forward<KeyType>(key), visitor, opts);
diff --git a/cpp/arcticdb/storage/test/test_embedded.cpp b/cpp/arcticdb/storage/test/test_embedded.cpp
index 05cd52f9c55..52375af130f 100644
--- a/cpp/arcticdb/storage/test/test_embedded.cpp
+++ b/cpp/arcticdb/storage/test/test_embedded.cpp
@@ -109,8 +109,8 @@ TEST_P(SimpleTestSuite, Example) {
     as::KeySegmentPair res;
     storage->read(k, [&](auto &&k, auto &&seg) {
         res.set_key(k);
-        res.segment() = std::move(seg);
-        res.segment().force_own_buffer(); // necessary since the non-owning buffer won't survive the visit
+        res.set_segment(std::move(seg));
+        res.segment_ptr()->force_own_buffer(); // necessary since the non-owning buffer won't survive the visit
     }, storage::ReadKeyOpts{});
 
     res = storage->read(k, as::ReadKeyOpts{});
@@ -133,8 +133,8 @@ TEST_P(SimpleTestSuite, Example) {
     as::KeySegmentPair update_res;
     storage->read(k, [&](auto &&k, auto &&seg) {
         update_res.set_key(k);
-        update_res.segment() = std::move(seg);
-        update_res.segment().force_own_buffer(); // necessary since the non-owning buffer won't survive the visit
+        update_res.set_segment(std::move(seg));
+        update_res.segment_ptr()->force_own_buffer(); // necessary since the non-owning buffer won't survive the visit
     }, as::ReadKeyOpts{});
 
     update_res = storage->read(k, as::ReadKeyOpts{});
@@ -189,11 +189,11 @@ TEST_P(SimpleTestSuite, Strings) {
     as::KeySegmentPair res;
     storage->read(save_k, [&](auto &&k, auto &&seg) {
         res.set_key(k);
-        res.segment() = std::move(seg);
-        res.segment().force_own_buffer(); // necessary since the non-owning buffer won't survive the visit
+        res.set_segment(std::move(seg));
+        res.segment_ptr()->force_own_buffer(); // necessary since the non-owning buffer won't survive the visit
     }, as::ReadKeyOpts{});
 
-    SegmentInMemory res_mem = decode_segment(std::move(res.segment()));
+    SegmentInMemory res_mem = decode_segment(*res.segment_ptr());
     ASSERT_EQ(s.string_at(0, 1), res_mem.string_at(0, 1));
     ASSERT_EQ(std::string("happy"), res_mem.string_at(0, 1));
     ASSERT_EQ(s.string_at(1, 3), res_mem.string_at(1, 3));
diff --git a/cpp/arcticdb/storage/test/test_mongo_storage.cpp b/cpp/arcticdb/storage/test/test_mongo_storage.cpp
index 4a9de615ba7..01660213d18 100644
--- a/cpp/arcticdb/storage/test/test_mongo_storage.cpp
+++ b/cpp/arcticdb/storage/test/test_mongo_storage.cpp
@@ -43,8 +43,8 @@ TEST(MongoStorage, ClientSession) {
 
     storage.read(k, [&](auto &&k, auto &&seg) {
         res.set_key(k);
-        res.segment() = std::move(seg);
-        res.segment().force_own_buffer(); // necessary since the non-owning buffer won't survive the visit
+        res.set_segment(std::move(seg));
+        res.segment_ptr()->force_own_buffer(); // necessary since the non-owning buffer won't survive the visit
     }, as::ReadKeyOpts{});
 
     res = storage.read(k, as::ReadKeyOpts{});
@@ -71,8 +71,8 @@ TEST(MongoStorage, ClientSession) {
 
     storage.read(k, [&](auto &&k, auto &&seg) {
         update_res.set_key(k);
-        update_res.segment() = std::move(seg);
-        update_res.segment().force_own_buffer(); // necessary since the non-owning buffer won't survive the visit
+        update_res.set_segment(std::move(seg));
+        update_res.segment_ptr()->force_own_buffer(); // necessary since the non-owning buffer won't survive the visit
     }, as::ReadKeyOpts{});
 
     update_res = storage.read(k, as::ReadKeyOpts{});
@@ -94,8 +94,8 @@ TEST(MongoStorage, ClientSession) {
 
     storage.read(numeric_k, [&](auto &&k, auto &&seg) {
         numeric_res.set_key(k);
-        numeric_res.segment() = std::move(seg);
-        numeric_res.segment().force_own_buffer(); // necessary since the non-owning buffer won't survive the visit
+        numeric_res.set_segment(std::move(seg));
+        numeric_res.segment_ptr()->force_own_buffer(); // necessary since the non-owning buffer won't survive the visit
     }, as::ReadKeyOpts{});
 
     numeric_res = storage.read(numeric_k, as::ReadKeyOpts{});
diff --git a/cpp/arcticdb/toolbox/library_tool.cpp b/cpp/arcticdb/toolbox/library_tool.cpp
index af1f9d37084..d5dda961ed9 100644
--- a/cpp/arcticdb/toolbox/library_tool.cpp
+++ b/cpp/arcticdb/toolbox/library_tool.cpp
@@ -30,7 +30,7 @@ LibraryTool::LibraryTool(std::shared_ptr<storage::Library> lib) {
 
 ReadResult LibraryTool::read(const VariantKey& key) {
     auto segment = read_to_segment(key);
-    auto segment_in_memory = decode_segment(std::move(segment));
+    auto segment_in_memory = decode_segment(segment);
     auto frame_and_descriptor = frame_and_descriptor_from_segment(std::move(segment_in_memory));
     return pipelines::make_read_result_from_frame(frame_and_descriptor, to_atom(key));
 }
@@ -38,8 +38,8 @@ ReadResult LibraryTool::read(const VariantKey& key) {
 Segment LibraryTool::read_to_segment(const VariantKey& key) {
     auto kv = store_->read_compressed_sync(key, storage::ReadKeyOpts{});
     util::check(kv.has_segment(), "Failed to read key: {}", key);
-    kv.segment().force_own_buffer();
-    return std::move(kv.segment());
+    kv.segment_ptr()->force_own_buffer();
+    return std::move(*kv.release_segment());
 }
 
 std::optional<google::protobuf::Any> LibraryTool::read_metadata(const VariantKey& key){