Merge branch 'master' into memray_head_tail_tests

Author: grusev

cpp/arcticdb/codec/lz4.hpp

@@ -50,7 +50,7 @@ struct Lz4BlockEncoder {
         util::check_arg(compressed_bytes > 0 || (compressed_bytes == 0 && block_utils.bytes_ == 0),
             "expected compressed bytes >= 0, actual {}",
             compressed_bytes);
-        ARCTICDB_TRACE(log::storage(), "Block of size {} compressed to {} bytes", block_utils.bytes_, compressed_bytes);
+        ARCTICDB_TRACE(log::storage(), "Block of size {} compressed to {} bytes: {}", block_utils.bytes_, compressed_bytes, dump_bytes(out, compressed_bytes, 10U));
         hasher(in, block_utils.count_);
         pos += ssize_t(compressed_bytes);
         copy_codec(*out_codec.mutable_lz4(), opts);

cpp/arcticdb/codec/segment.cpp

@@ -30,7 +30,7 @@ arcticdb::proto::encoding::SegmentHeader generate_v1_header(const SegmentHeader&
     segment_header.set_compacted(header.compacted());
     segment_header.set_encoding_version(static_cast<uint16_t>(header.encoding_version()));
 
-    ARCTICDB_TRACE(log::codec(), "Encoded segment header {}", segment_header.DebugString());
+    ARCTICDB_TRACE(log::codec(), "Encoded segment header bytes {}: {}", segment_header.ByteSizeLong(), segment_header.DebugString());
     return segment_header;
 }
 
@@ -153,7 +153,8 @@ DeserializedSegmentData decode_header_and_fields(const uint8_t*& src, bool copy_
         auto segment_header = deserialize_segment_header_from_proto(proto_wrapper->proto());
         util::check(segment_header.encoding_version() == EncodingVersion::V1, "Expected v1 header to contain legacy encoding version");
         auto fields = std::make_shared<FieldCollection>(field_collection_from_proto(proto_wrapper->proto().stream_descriptor().fields()));
-        src += FIXED_HEADER_SIZE + fixed_hdr->header_bytes;
+        const auto total_header_size = FIXED_HEADER_SIZE + fixed_hdr->header_bytes;
+        src += total_header_size;
         auto stream_id = stream_id_from_proto(proto_wrapper->proto().stream_descriptor());
         return {std::move(segment_header), std::move(fields), std::move(data), std::move(proto_wrapper), stream_id};
     } else {
@@ -239,12 +240,12 @@ Segment Segment::from_buffer(const std::shared_ptr<Buffer>& buffer) {
     return{std::move(seg_hdr), buffer, std::move(desc_data), std::move(fields), stream_id, readable_size};
 }
 
-size_t Segment::write_proto_header(uint8_t* dst) {
+size_t Segment::write_proto_header(uint8_t* dst, size_t header_size) {
     const auto& header = generate_header_proto();
     const auto hdr_size = proto_size();
     FixedHeader hdr = {MAGIC_NUMBER, HEADER_VERSION_V1, std::uint32_t(hdr_size)};
     write_fixed_header(dst, hdr);
-
+    util::check(header_size == proto_size(), "Header size mismatch: {} != {}", header_size, proto_size());
     google::protobuf::io::ArrayOutputStream aos(dst + FIXED_HEADER_SIZE, static_cast<int>(hdr_size));
     header.SerializeToZeroCopyStream(&aos);
     return hdr_size;
@@ -269,12 +270,18 @@ std::pair<uint8_t*, size_t> Segment::serialize_header_v2(size_t expected_bytes)
     return std::make_pair(buffer->preamble(), calculate_size());
 }
 
-std::pair<uint8_t*, size_t> Segment::serialize_v1_header_in_place(size_t total_hdr_size) {
+std::pair<uint8_t*, size_t> Segment::serialize_v1_header_in_place(size_t hdr_size) {
+    const auto total_hdr_size = hdr_size + FIXED_HEADER_SIZE;
     const auto &buffer = buffer_.get_owning_buffer();
     auto base_ptr = buffer->preamble() + (buffer->preamble_bytes() - total_hdr_size);
+    util::check(buffer->data() != nullptr, "Unexpected null base pointer in v1 header serialization");
     util::check(base_ptr + total_hdr_size == buffer->data(), "Expected base ptr to align with data ptr, {} != {}",fmt::ptr(base_ptr + total_hdr_size),fmt::ptr(buffer->data()));
-    write_proto_header(base_ptr);
-    ARCTICDB_TRACE(log::storage(), "Header fits in internal buffer {:x} with {} bytes space: {}", intptr_t (base_ptr), buffer->preamble_bytes() - total_hdr_size,dump_bytes(buffer->data(), buffer->bytes(), 100u));
+    auto red_zone = *buffer->data();
+    auto header_bytes_written = write_proto_header(base_ptr, hdr_size);
+    ARCTICDB_TRACE(log::storage(), "Header fits in internal buffer {:x} with {} bytes space: {}", intptr_t (base_ptr), buffer->preamble_bytes() - total_hdr_size,dump_bytes(buffer->data(), buffer->bytes(), 10u));
+    auto check_red_zone = *buffer->data();
+    util::check(red_zone == check_red_zone, "Data overwrite occurred {} != {}", check_red_zone, red_zone);
+    util::check(header_bytes_written == hdr_size, "Wrote unexpected number of header bytes {} != {}", header_bytes_written, total_hdr_size);
     return std::make_pair(base_ptr, calculate_size());
 }
 
@@ -284,46 +291,38 @@ std::tuple<uint8_t*, size_t, std::unique_ptr<Buffer>> Segment::serialize_v1_head
     auto offset = FIXED_HEADER_SIZE + hdr_size;
 
     auto total_size = offset + bytes_to_copy;
-
-    // Verify we have enough space for everything
     tmp->ensure(total_size);
-    util::check(tmp->available() >= total_size,
-        "Buffer available space {} is less than required size {}", 
-        tmp->available(), 
-        total_size);
 
-    // This is both a sanity check and a way to populate the segment with the correct size
+    util::check(tmp->available() >= total_size, "Buffer available space {} is less than required size {}",tmp->available(), total_size);
+
     auto calculated_size = calculate_size();
     util::check(total_size == calculated_size, "Expected total size {} to be equal to calculated size {}", total_size, calculated_size);
 
     auto* dst = tmp->preamble();
     util::check(dst != nullptr, "Expected dst to be non-null");
+    auto header_bytes_written = write_proto_header(dst, hdr_size);
 
-    auto written_hdr_size = write_proto_header(dst);
-    util::check(written_hdr_size == hdr_size, "Expected written header size {} to be equal to expected header size {}", written_hdr_size, hdr_size);
+    // This is a bit redundant since the size is also checked in write_proto_header, but the consequences of getting
+    // it wrong are pretty bad (corrupt data) so will leave it in for future-proofing
+    util::check(header_bytes_written == hdr_size, "Expected written header size {} to be equal to expected header size {}", header_bytes_written, hdr_size);
 
-    auto *final_dst = dst + offset;
-
-    auto *src = buffer().data();
-    if (src != nullptr) {   
-        std::memcpy(final_dst,
-                    src,
-                    bytes_to_copy);
+    if(buffer().data() != nullptr) {
+        std::memcpy(dst + offset, buffer().data(), buffer().bytes());
     } else {
         util::check(bytes_to_copy == 0, "Expected bytes_to_copy to be 0 when src is nullptr");
         ARCTICDB_DEBUG(log::codec(), "src is nullptr, skipping memcpy");
     }
 
-    return std::make_tuple(dst, total_size, std::move(tmp));
+    return std::make_tuple(tmp->preamble(), calculate_size(), std::move(tmp));
 }
 
 std::tuple<uint8_t*, size_t, std::unique_ptr<Buffer>> Segment::serialize_header_v1() {
-    auto proto_header = generate_v1_header(header_, desc_);
-    const auto hdr_size = proto_header.ByteSizeLong();
+    auto proto_header = generate_header_proto();
+    const auto hdr_size = proto_size();
     auto total_hdr_size = hdr_size + FIXED_HEADER_SIZE;
 
     if (buffer_.is_owning_buffer() && buffer_.preamble_bytes() >= total_hdr_size) {
-        auto [dst, size] = serialize_v1_header_in_place(total_hdr_size);
+        auto [dst, size] = serialize_v1_header_in_place(hdr_size);
         return std::make_tuple(dst, size, std::unique_ptr<Buffer>());
     } else {
         return serialize_v1_header_to_buffer(hdr_size);
@@ -352,8 +351,10 @@ std::tuple<uint8_t*, size_t, std::unique_ptr<Buffer>> Segment::serialize_header(
 }
 
 const arcticdb::proto::encoding::SegmentHeader& Segment::generate_header_proto() {
-    if(!proto_)
+    if(!proto_) {
         proto_ = std::make_unique<arcticdb::proto::encoding::SegmentHeader>(generate_v1_header(header_, desc_));
+        proto_size_ = proto_->ByteSizeLong();
+    }
 
     return *proto_;
 }
@@ -364,7 +365,7 @@ void Segment::write_to(std::uint8_t* dst) {
 
     size_t header_size;
     if(header_.encoding_version() == EncodingVersion::V1)
-        header_size = write_proto_header(dst);
+        header_size = write_proto_header(dst, proto_size());
     else
         header_size = write_binary_header(dst);
 

cpp/arcticdb/codec/segment.hpp

@@ -143,7 +143,7 @@ class Segment {
 
     std::tuple<uint8_t*, size_t, std::unique_ptr<Buffer>> serialize_header();
 
-    size_t write_proto_header(uint8_t* dst);
+    size_t write_proto_header(uint8_t* dst, size_t total_header_size);
 
     [[nodiscard]] std::size_t size() const {
         util::check(size_.has_value(), "Segment size has not been set");
@@ -160,8 +160,7 @@ class Segment {
 
     [[nodiscard]] size_t proto_size() {
         util::check(static_cast<bool>(proto_), "Proto has not been generated");
-
-        return proto_->ByteSizeLong();
+        return proto_size_;
     }
 
     [[nodiscard]] std::size_t segment_header_bytes_size() {
@@ -263,6 +262,7 @@ class Segment {
     StreamDescriptor desc_;
     std::any keepalive_;
     std::unique_ptr<arcticdb::proto::encoding::SegmentHeader> proto_;
+    size_t proto_size_ = 0UL;
     std::optional<size_t> size_;
 };
 

cpp/arcticdb/column_store/block.hpp

@@ -62,7 +62,7 @@ struct MemBlock {
             util::check(is_external(), "Cannot free inline allocated block");
             if(external_data_ != nullptr) {
                 log::version().warn("Unexpected release of detachable block memory");
-                free(reinterpret_cast<void *>(external_data_));
+                delete[] external_data_;
             }
         }
     }

cpp/arcticdb/log/log.hpp

@@ -10,7 +10,6 @@
 #include <memory>
 
 #include <spdlog/spdlog.h>
-
 #ifdef DEBUG_BUILD
 #define ARCTICDB_DEBUG(logger, ...) logger.debug(__VA_ARGS__)
 #define ARCTICDB_TRACE(logger, ...) logger.trace(__VA_ARGS__)

cpp/arcticdb/pipeline/read_frame.cpp

@@ -142,6 +142,7 @@ size_t get_index_field_count(const SegmentInMemory& frame) {
 
 const uint8_t* skip_heading_fields(const SegmentHeader & hdr, const uint8_t*& data) {
     const auto has_magic_numbers = hdr.encoding_version() == EncodingVersion::V2;
+    const auto start [[maybe_unused]] = data;
     if(has_magic_numbers)
         util::check_magic<MetadataMagic>(data);
 
@@ -172,6 +173,7 @@ const uint8_t* skip_heading_fields(const SegmentHeader & hdr, const uint8_t*& da
         ARCTICDB_DEBUG(log::version(), "Skipping {} bytes of index descriptor", index_fields_size);
             data += index_fields_size;
     }
+    ARCTICDB_DEBUG(log::version(), "Skip header fields skipped {} bytes", data - start);
     return data;
 }
 
@@ -486,14 +488,16 @@ void check_data_left_for_subsequent_fields(
 }
 
 void decode_into_frame_static(
-    SegmentInMemory &frame,
-    PipelineContextRow &context,
-    const Segment& seg,
-    const DecodePathData& shared_data,
-    std::any& handler_data,
-    const ReadQuery& read_query,
-    const ReadOptions& read_options) {
+        SegmentInMemory &frame,
+        PipelineContextRow &context,
+        const storage::KeySegmentPair& key_seg,
+        const DecodePathData& shared_data,
+        std::any& handler_data,
+        const ReadQuery& read_query,
+        const ReadOptions& read_options) {
     ARCTICDB_SAMPLE_DEFAULT(DecodeIntoFrame)
+    ARCTICDB_DEBUG(log::version(), "Statically decoding segment with key {}", key_seg.atom_key());
+    const auto& seg = key_seg.segment();
     const uint8_t *data = seg.buffer().data();
     const uint8_t *begin = data;
     const uint8_t *end = begin + seg.buffer().bytes();
@@ -623,15 +627,16 @@ void handle_type_promotion(
 }
 
 void decode_into_frame_dynamic(
-    SegmentInMemory& frame,
-    PipelineContextRow& context,
-    const Segment& seg,
-    const DecodePathData& shared_data,
-    std::any& handler_data,
-    const ReadQuery& read_query,
-    const ReadOptions& read_options
-) {
+        SegmentInMemory& frame,
+        PipelineContextRow& context,
+        const storage::KeySegmentPair& key_seg,
+        const DecodePathData& shared_data,
+        std::any& handler_data,
+        const ReadQuery& read_query,
+        const ReadOptions& read_options) {
     ARCTICDB_SAMPLE_DEFAULT(DecodeIntoFrame)
+    ARCTICDB_DEBUG(log::version(), "Dynamically decoding segment with key {}", key_seg.atom_key());
+    const auto& seg = key_seg.segment();
     const uint8_t *data = seg.buffer().data();
     const uint8_t *begin = data;
     const uint8_t *end = begin + seg.buffer().bytes();
@@ -885,9 +890,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, key_seg.segment(), shared_data, handler_data, read_query, read_options);
+                    decode_into_frame_dynamic(frame, row, key_seg, shared_data, handler_data, read_query, read_options);
                 } else {
-                    decode_into_frame_static(frame, row, key_seg.segment(), shared_data, handler_data, read_query, read_options);
+                    decode_into_frame_static(frame, row, key_seg, shared_data, handler_data, read_query, read_options);
                 }
 
                 return key_seg.variant_key();

cpp/arcticdb/pipeline/read_frame.hpp

@@ -85,7 +85,7 @@ folly::Future<SegmentInMemory> fetch_data(
 void decode_into_frame_static(
     SegmentInMemory &frame,
     PipelineContextRow &context,
-    const Segment& seg,
+    const storage::KeySegmentPair& key_seg,
     const DecodePathData& shared_data,
     std::any& handler_data,
     const ReadQuery& read_query,
@@ -94,7 +94,7 @@ void decode_into_frame_static(
 void decode_into_frame_dynamic(
     const SegmentInMemory &frame,
     PipelineContextRow &context,
-    const Segment& seg,
+    const storage::KeySegmentPair& key_seg,
     const DecodePathData& shared_data,
     std::any& handler_data,
     const ReadQuery& read_query,

cpp/arcticdb/storage/s3/s3_client_impl.cpp

@@ -188,8 +188,8 @@ S3Result<std::monostate> S3ClientImpl::put_object(
         request.SetIfNoneMatch("*");
     }
     ARCTICDB_RUNTIME_DEBUG(log::storage(), "Set s3 key {}", request.GetKey().c_str());
-
     auto [dst, write_size, buffer] = segment.serialize_header();
+
     auto body = std::make_shared<boost::interprocess::bufferstream>(reinterpret_cast<char *>(dst), write_size);
     util::check(body->good(), "Overflow of bufferstream with size {}", write_size);
     request.SetBody(body);

cpp/arcticdb/storage/s3/s3_client_wrapper.cpp

@@ -138,7 +138,6 @@ folly::Future<S3Result<std::monostate>> S3ClientTestWrapper::delete_object(
 
 // Using a fixed page size since it's only being used for simple tests.
 // If we ever need to configure it we should move it to the s3 proto config instead.
-constexpr auto page_size = 10;
 S3Result<ListObjectsOutput> S3ClientTestWrapper::list_objects(
         const std::string& name_prefix,
         const std::string& bucket_name,

cpp/arcticdb/util/buffer.hpp

@@ -88,6 +88,7 @@ struct Buffer : public BaseBuffer<Buffer, true> {
     }
 
     void set_preamble(size_t pos) {
+        util::check(preamble_bytes_ == 0, "Cannot reset buffer preabmle");
         util::check(pos <= capacity_, "Can't set preamble past the end of the buffer");
         preamble_bytes_ = pos;
         ptr_ += pos;
@@ -177,7 +178,7 @@ struct Buffer : public BaseBuffer<Buffer, true> {
             resize(total_size);
         } else {
             ARCTICDB_TRACE(log::version(), "Buffer {} has sufficient bytes for {}, ptr {} data {}, capacity {}",
-                                uintptr_t(this), bytes, uintptr_t(ptr_), uintptr_t(data_), capacity_, body_bytes_);
+                                uintptr_t(this), bytes, uintptr_t(ptr_), uintptr_t(data_), capacity_);
         }
 
         body_bytes_ = bytes;
@@ -239,7 +240,7 @@ struct Buffer : public BaseBuffer<Buffer, true> {
             body_bytes_ = bytes;
             capacity_ = body_bytes_ + preamble_bytes_;
             ARCTICDB_TRACE(log::version(), "Buffer {} did realloc for {}, ptr {} data {}, capacity {}",
-                                uintptr_t(this), bytes,  uintptr_t(ptr_), uintptr_t(data_), capacity_, body_bytes_);
+                                uintptr_t(this), bytes,  uintptr_t(ptr_), uintptr_t(data_), capacity_);
         } else {
             throw std::bad_alloc();
         }

cpp/arcticdb/util/dump_bytes.hpp

@@ -12,23 +12,25 @@
 
 // based on this: https://codereview.stackexchange.com/questions/165120/printing-hex-dumps-for-diagnostics
 namespace arcticdb {
-inline std::ostream &hex_dump(std::ostream &os, const void *buffer,
-                       std::size_t buf_size, bool show_printable_chars = true) {
-    if (buffer == nullptr) {
+inline std::ostream &hex_dump(
+        std::ostream &os, const void *buffer,
+        std::size_t buf_size,
+        bool show_printable_chars = true) {
+    if (buffer == nullptr)
         return os;
-    }
+
     auto old_format = os.flags();
     auto old_fill_char = os.fill();
     constexpr std::size_t max_line{8};
+
     // create a place to store text version of string
     char render_string[max_line + 1];
     char *rsptr{render_string};
+
     // convenience cast
     const unsigned char *buf{reinterpret_cast<const unsigned char *>(buffer)};
-
     for (std::size_t line_count = max_line; buf_size; --buf_size, ++buf) {
-        os << std::setw(2) << std::setfill('0') << std::hex
-           << static_cast<unsigned>(*buf) << ' ';
+        os << std::setw(2) << std::setfill('0') << std::hex << static_cast<unsigned>(*buf) << ' ';
         *rsptr++ = std::isprint(*buf) ? *buf : '.';
         if (--line_count == 0) {
             *rsptr++ = '\0';  // terminate string
@@ -40,6 +42,7 @@ inline std::ostream &hex_dump(std::ostream &os, const void *buffer,
             line_count = std::min(max_line, buf_size);
         }
     }
+
     // emit newline if we haven't already
     if (rsptr != render_string) {
         if (show_printable_chars) {
@@ -56,7 +59,7 @@ inline std::ostream &hex_dump(std::ostream &os, const void *buffer,
     return os;
 }
 
-inline std::string dump_bytes(const void* data, size_t size, size_t max_size = 20u) {
+[[nodiscard]] inline std::string dump_bytes(const void* data, size_t size, size_t max_size = 20u) {
     const auto sz = std::min(max_size, size);
     std::ostringstream strm;
     hex_dump(strm, data, sz);

python/tests/stress/arcticdb/version_store/test_stress_write_and_reread.py

@@ -0,0 +1,21 @@
+import numpy as np
+import pandas as pd
+from pandas.testing import assert_series_equal
+
+
+def get_random_series() -> pd.Series:
+    dates = pd.date_range("2018-08-17", "2019-01-10", name="date")
+    securities = pd.Index(np.arange(100), name="security_id")
+    index = pd.MultiIndex.from_product([dates, securities])
+    np.random.seed(42)
+    random_series = pd.Series(np.random.randn(len(index)), index=index, name="stuff")
+    return random_series
+
+
+def test_batch_roundtrip(s3_version_store_v1):
+    df = get_random_series()
+    symbols = [f"symbol_{i}" for i in range(500)]
+    data_vector = [df for _ in symbols]
+    s3_version_store_v1.batch_write(symbols, data_vector)
+    s3_version_store_v1.batch_read(symbols)
+