variable_size_binary_view_array.hpp

Go to the documentation of this file.

// Copyright 2024 Man Group Operations Limited
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
 
#pragma once
 
#include <cstddef>
#include <cstring>
#include <ranges>
#include <unordered_map>
 
#include "sparrow/arrow_interface/arrow_array.hpp"
#include "sparrow/arrow_interface/arrow_array_schema_proxy.hpp"
#include "sparrow/arrow_interface/arrow_schema.hpp"
#include "sparrow/buffer/dynamic_bitset.hpp"
#include "sparrow/layout/array_access.hpp"
#include "sparrow/layout/array_bitmap_base.hpp"
#include "sparrow/layout/layout_utils.hpp"
#include "sparrow/types/data_traits.hpp"
#include "sparrow/u8_buffer.hpp"
#include "sparrow/utils/extension.hpp"
#include "sparrow/utils/functor_index_iterator.hpp"
#include "sparrow/utils/iterator.hpp"
#include "sparrow/utils/mp_utils.hpp"
#include "sparrow/utils/nullable.hpp"
#include "sparrow/utils/ranges.hpp"
#include "sparrow/utils/repeat_container.hpp"
#include "sparrow/utils/sequence_view.hpp"
 
namespace sparrow
{
    template <std::ranges::sized_range T, class CR, typename Ext = empty_extension>
    class variable_size_binary_view_array_impl;

    using string_view_array = variable_size_binary_view_array_impl<
        arrow_traits<std::string>::value_type,
        arrow_traits<std::string>::const_reference>;

    using binary_view_array = variable_size_binary_view_array_impl<
        arrow_traits<std::vector<byte_t>>::value_type,
        arrow_traits<std::vector<byte_t>>::const_reference>;
 
    namespace detail
    {
        template <>
        struct get_data_type_from_array<sparrow::string_view_array>
        {
            [[nodiscard]] static constexpr sparrow::data_type get()
            {
                return sparrow::data_type::STRING_VIEW;
            }
        };
 
        template <>
        struct get_data_type_from_array<sparrow::binary_view_array>
        {
            [[nodiscard]] static constexpr sparrow::data_type get()
            {
                return sparrow::data_type::BINARY_VIEW;
            }
        };
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    struct array_inner_types<variable_size_binary_view_array_impl<T, CR, Ext>> : array_inner_types_base
    {
        using array_type = variable_size_binary_view_array_impl<T, CR, Ext>;
        using inner_value_type = T;
        using inner_reference = CR;
        using inner_const_reference = inner_reference;
 
        using value_iterator = functor_index_iterator<detail::layout_value_functor<array_type, inner_reference>>;
        using const_value_iterator = functor_index_iterator<
            detail::layout_value_functor<const array_type, inner_const_reference>>;
        using iterator_tag = std::random_access_iterator_tag;
    };
 
    template <class T>
    struct is_variable_size_binary_view_array_impl : std::false_type
    {
    };
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    struct is_variable_size_binary_view_array_impl<variable_size_binary_view_array_impl<T, CR, Ext>>
        : std::true_type
    {
    };

    template <class T>
    constexpr bool is_variable_size_binary_view_array = is_variable_size_binary_view_array_impl<T>::value;

    template <std::ranges::sized_range T, class CR, typename Ext>
    class variable_size_binary_view_array_impl final
        : public mutable_array_bitmap_base<variable_size_binary_view_array_impl<T, CR, Ext>>,
          public Ext
    {
    public:
 
        using self_type = variable_size_binary_view_array_impl<T, CR, Ext>;
        using base_type = mutable_array_bitmap_base<self_type>;
 
        using inner_types = array_inner_types<self_type>;
        using inner_value_type = typename inner_types::inner_value_type;
        using inner_reference = typename inner_types::inner_reference;
        using inner_const_reference = typename inner_types::inner_const_reference;
 
        using bitmap_type = typename base_type::bitmap_type;
        using bitmap_reference = typename base_type::bitmap_reference;
        using bitmap_const_reference = typename base_type::bitmap_const_reference;
        using bitmap_iterator = typename base_type::bitmap_iterator;
        using const_bitmap_iterator = typename base_type::const_bitmap_iterator;
        using bitmap_range = typename base_type::bitmap_range;
        using const_bitmap_range = typename base_type::const_bitmap_range;
 
        using value_type = nullable<inner_value_type>;
        using reference = nullable<inner_reference, bitmap_reference>;
        using const_reference = nullable<inner_const_reference, bitmap_const_reference>;
 
        using size_type = typename base_type::size_type;
        using difference_type = typename base_type::difference_type;
        using iterator_tag = typename base_type::iterator_tag;
 
        using value_iterator = typename base_type::value_iterator;
        using const_value_iterator = typename base_type::const_value_iterator;
 
        using iterator = typename base_type::iterator;
        using const_iterator = typename base_type::const_iterator;

        explicit variable_size_binary_view_array_impl(arrow_proxy);

        template <class... Args>
            requires(mpl::excludes_copy_and_move_ctor_v<variable_size_binary_view_array_impl<T, CR>, Args...>)
        explicit variable_size_binary_view_array_impl(Args&&... args)
            : variable_size_binary_view_array_impl(create_proxy(std::forward<Args>(args)...))
        {
        }
 
    private:

        struct buffers_collection
        {
            buffer<uint8_t> length_buffer;        
            buffer<uint8_t> long_string_storage;  
            u8_buffer<int64_t> buffer_sizes;      
        };

        [[nodiscard]] static constexpr std::string_view get_arrow_format()
        {
            return std::is_same_v<T, arrow_traits<std::string>::value_type> ? std::string_view("vu")
                                                                            : std::string_view("vz");
        }

        template <input_metadata_container METADATA_RANGE>
        [[nodiscard]] static ArrowSchema create_arrow_schema(
            std::optional<std::string_view> name,
            std::optional<METADATA_RANGE> metadata,
            std::optional<std::unordered_set<sparrow::ArrowFlag>> flags
        )
        {
            constexpr repeat_view<bool> children_ownership(true, 0);
            return make_arrow_schema(
                get_arrow_format(),
                std::move(name),
                std::move(metadata),
                flags,
                nullptr,  // children
                children_ownership,
                nullptr,  // dictionary
                true
            );
        }

        template <std::ranges::input_range R>
            requires std::convertible_to<std::ranges::range_value_t<R>, T>
        static buffers_collection create_buffers(R&& range);

        template <
            std::ranges::input_range R,
            validity_bitmap_input VB = validity_bitmap,
            input_metadata_container METADATA_RANGE = std::vector<metadata_pair>>
            requires std::convertible_to<std::ranges::range_value_t<R>, T>
        [[nodiscard]] static arrow_proxy create_proxy(
            R&& range,
            VB&& bitmap_input = validity_bitmap{validity_bitmap::default_allocator()},
            std::optional<std::string_view> name = std::nullopt,
            std::optional<METADATA_RANGE> metadata = std::nullopt
        );

        template <std::ranges::input_range NULLABLE_RANGE, input_metadata_container METADATA_RANGE = std::vector<metadata_pair>>
            requires std::convertible_to<std::ranges::range_value_t<NULLABLE_RANGE>, nullable<T>>
        [[nodiscard]] static arrow_proxy create_proxy(
            NULLABLE_RANGE&& nullable_range,
            std::optional<std::string_view> name = std::nullopt,
            std::optional<METADATA_RANGE> metadata = std::nullopt
        );

        template <std::ranges::input_range R, input_metadata_container METADATA_RANGE = std::vector<metadata_pair>>
            requires std::convertible_to<std::ranges::range_value_t<R>, T>
        [[nodiscard]] static arrow_proxy create_proxy(
            R&& range,
            bool = true,
            std::optional<std::string_view> name = std::nullopt,
            std::optional<METADATA_RANGE> metadata = std::nullopt
        );

        template <
            std::ranges::input_range VALUE_BUFFERS_RANGE,
            validity_bitmap_input VB,
            input_metadata_container METADATA_RANGE = std::vector<metadata_pair>>
            requires std::convertible_to<std::ranges::range_value_t<VALUE_BUFFERS_RANGE>, u8_buffer<uint8_t>>
        [[nodiscard]] static arrow_proxy create_proxy(
            size_t element_count,
            u8_buffer<uint8_t>&& buffer_view,
            VALUE_BUFFERS_RANGE&& value_buffers,
            VB&& validity_input,
            std::optional<std::string_view> name = std::nullopt,
            std::optional<METADATA_RANGE> metadata = std::nullopt
        );

        [[nodiscard]] constexpr inner_reference value(size_type i);

        [[nodiscard]] constexpr inner_const_reference value(size_type i) const;

        template <std::ranges::sized_range U>
            requires mpl::convertible_ranges<U, T>
        constexpr void assign(U&& rhs, size_type index);
 
        // Modifiers

        template <std::ranges::sized_range U>
            requires mpl::convertible_ranges<U, T>
        void resize_values(size_type new_length, U value);

        template <std::ranges::sized_range U>
            requires mpl::convertible_ranges<U, T>
        value_iterator insert_value(const_value_iterator pos, U value, size_type count);

        template <mpl::iterator_of_type<T> InputIt>
        value_iterator insert_values(const_value_iterator pos, InputIt first, InputIt last);

        value_iterator erase_values(const_value_iterator pos, size_type count);

        [[nodiscard]] constexpr value_iterator value_begin();

        [[nodiscard]] constexpr value_iterator value_end();

        [[nodiscard]] constexpr const_value_iterator value_cbegin() const;

        [[nodiscard]] constexpr const_value_iterator value_cend() const;
 
        static constexpr size_type LENGTH_BUFFER_INDEX = 1;            
        static constexpr std::size_t DATA_BUFFER_SIZE = 16;            
        static constexpr std::size_t SHORT_STRING_SIZE = 12;           
        static constexpr std::size_t PREFIX_SIZE = 4;                  
        static constexpr std::ptrdiff_t PREFIX_OFFSET = 4;             
        static constexpr std::ptrdiff_t SHORT_STRING_OFFSET = 4;       
        static constexpr std::ptrdiff_t BUFFER_INDEX_OFFSET = 8;       
        static constexpr std::ptrdiff_t BUFFER_OFFSET_OFFSET = 12;     
        static constexpr std::size_t FIRST_VAR_DATA_BUFFER_INDEX = 2;  
 
        friend base_type;
        friend base_type::base_type;
        friend base_type::base_type::base_type;
        friend class detail::layout_value_functor<self_type, inner_reference>;
        friend class detail::layout_value_functor<const self_type, inner_const_reference>;
    };
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    variable_size_binary_view_array_impl<T, CR, Ext>::variable_size_binary_view_array_impl(arrow_proxy proxy)
        : base_type(std::move(proxy))
    {
    }
 
    namespace
    {
        // Utility functions for safe unaligned access to int32 values
        inline std::int32_t read_int32_unaligned(const std::uint8_t* ptr)
        {
            std::int32_t value;
            std::memcpy(&value, ptr, sizeof(std::int32_t));
            return value;
        }
 
        inline void write_int32_unaligned(std::uint8_t* ptr, std::int32_t value)
        {
            std::memcpy(ptr, &value, sizeof(std::int32_t));
        }
 
        // Generic transformation function for type conversions
        template <typename To, typename From>
        inline constexpr To transform_to(const From& v)
        {
            return static_cast<To>(v);
        }
 
        // Helper function to update buffer offsets for long strings after a specific offset
        template <typename SizeType>
        inline void update_buffer_offsets_after(
            std::uint8_t* view_data,
            SizeType array_size,
            std::size_t data_buffer_size,
            std::size_t short_string_size,
            std::ptrdiff_t buffer_offset_offset,
            std::size_t threshold_offset,
            std::ptrdiff_t offset_adjustment,
            SizeType skip_index = static_cast<SizeType>(-1)
        )
        {
            for (SizeType i = 0; i < array_size; ++i)
            {
                if (i == skip_index)
                {
                    continue;
                }
 
                auto* view_ptr = view_data + (i * data_buffer_size);
                std::int32_t length;
                std::memcpy(&length, view_ptr, sizeof(std::int32_t));
 
                if (static_cast<std::size_t>(length) > short_string_size)
                {
                    std::int32_t current_offset;
                    std::memcpy(&current_offset, view_ptr + buffer_offset_offset, sizeof(std::int32_t));
 
                    if (static_cast<std::size_t>(current_offset) > threshold_offset)
                    {
                        current_offset += static_cast<std::int32_t>(offset_adjustment);
                        std::memcpy(view_ptr + buffer_offset_offset, &current_offset, sizeof(std::int32_t));
                    }
                }
            }
        }
 
        // Helper function to update buffer sizes metadata
        template <typename Buffer>
        inline void update_buffer_sizes_metadata(Buffer& buffer_sizes_buffer, std::int64_t new_size)
        {
            auto buffer_sizes_ptr = buffer_sizes_buffer.template data<std::int64_t>();
            *buffer_sizes_ptr = new_size;
        }
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    template <std::ranges::input_range R>
        requires std::convertible_to<std::ranges::range_value_t<R>, T>
    auto variable_size_binary_view_array_impl<T, CR, Ext>::create_buffers(R&& range) -> buffers_collection
    {
#ifdef __GNUC__
#    pragma GCC diagnostic push
#    pragma GCC diagnostic ignored "-Wcast-align"
#endif
 
        const auto size = range_size(range);
        buffer<uint8_t> length_buffer(size * DATA_BUFFER_SIZE, typename buffer<uint8_t>::default_allocator());
 
        std::size_t long_string_storage_size = 0;
        std::size_t i = 0;
        for (auto&& val : range)
        {
            auto val_casted = val
                              | std::ranges::views::transform(transform_to<std::uint8_t, typename T::value_type>);
 
            const auto length = val.size();
            auto length_ptr = length_buffer.data() + (i * DATA_BUFFER_SIZE);
 
            // write length
            write_int32_unaligned(length_ptr, static_cast<std::int32_t>(length));
 
            if (length <= SHORT_STRING_SIZE)
            {
                // write data itself
                sparrow::ranges::copy(val_casted, length_ptr + SHORT_STRING_OFFSET);
                std::fill(
                    length_ptr + SHORT_STRING_OFFSET + length,
                    length_ptr + DATA_BUFFER_SIZE,
                    std::uint8_t(0)
                );
            }
            else
            {
                // write the prefix of the data
                auto prefix_sub_range = val_casted | std::ranges::views::take(PREFIX_SIZE);
                sparrow::ranges::copy(prefix_sub_range, length_ptr + PREFIX_OFFSET);
 
                // write the buffer index
                write_int32_unaligned(length_ptr + BUFFER_INDEX_OFFSET, 0);
 
                // write the buffer offset
                write_int32_unaligned(
                    length_ptr + BUFFER_OFFSET_OFFSET,
                    static_cast<std::int32_t>(long_string_storage_size)
                );
 
                // count the size of the long string storage
                long_string_storage_size += length;
            }
            ++i;
        }
 
        // write the long string storage
        buffer<uint8_t> long_string_storage(long_string_storage_size, buffer<uint8_t>::default_allocator());
        std::size_t long_string_storage_offset = 0;
        for (auto&& val : range)
        {
            const auto length = val.size();
            if (length > SHORT_STRING_SIZE)
            {
                auto val_casted = val
                                  | std::ranges::views::transform(
                                      transform_to<std::uint8_t, typename T::value_type>
                                  );
                sparrow::ranges::copy(val_casted, long_string_storage.data() + long_string_storage_offset);
                long_string_storage_offset += length;
            }
        }
 
        // For binary or utf-8 view arrays, an extra buffer is appended which stores
        // the lengths of each variadic data buffer as int64_t.
        // This buffer is necessary since these buffer lengths are not trivially
        // extractable from other data in an array of binary or utf-8 view type.
        u8_buffer<int64_t> buffer_sizes(
            static_cast<std::size_t>(1),
            static_cast<int64_t>(long_string_storage_size)
        );
 
        return {std::move(length_buffer), std::move(long_string_storage), std::move(buffer_sizes)};
 
#ifdef __GNUC__
#    pragma GCC diagnostic pop
#endif
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    template <std::ranges::input_range R, validity_bitmap_input VB, input_metadata_container METADATA_RANGE>
        requires std::convertible_to<std::ranges::range_value_t<R>, T>
    arrow_proxy variable_size_binary_view_array_impl<T, CR, Ext>::create_proxy(
        R&& range,
        VB&& validity_input,
        std::optional<std::string_view> name,
        std::optional<METADATA_RANGE> metadata
    )
    {
        const auto size = range_size(range);
        validity_bitmap vbitmap = ensure_validity_bitmap(size, std::forward<VB>(validity_input));
        const auto null_count = vbitmap.null_count();
 
        static const std::optional<std::unordered_set<sparrow::ArrowFlag>> flags{{ArrowFlag::NULLABLE}};
 
        // create arrow schema
        ArrowSchema schema = create_arrow_schema(std::move(name), std::move(metadata), flags);
 
        // create buffers
        auto buffers_parts = create_buffers(std::forward<R>(range));
 
        std::vector<buffer<uint8_t>> buffers{
            std::move(vbitmap).extract_storage(),
            std::move(buffers_parts.length_buffer),
            std::move(buffers_parts.long_string_storage),
            std::move(buffers_parts.buffer_sizes).extract_storage()
        };
 
        constexpr repeat_view<bool> children_ownership(true, 0);
 
        // create arrow array
        ArrowArray arr = make_arrow_array(
            static_cast<std::int64_t>(size),  // length
            static_cast<int64_t>(null_count),
            0,  // offset
            std::move(buffers),
            nullptr,  // children
            children_ownership,
            nullptr,  // dictionary
            true
        );
 
        arrow_proxy proxy{std::move(arr), std::move(schema)};
        Ext::init(proxy);
        return proxy;
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    template <std::ranges::input_range NULLABLE_RANGE, input_metadata_container METADATA_RANGE>
        requires std::convertible_to<std::ranges::range_value_t<NULLABLE_RANGE>, nullable<T>>
    [[nodiscard]] arrow_proxy variable_size_binary_view_array_impl<T, CR, Ext>::create_proxy(
        NULLABLE_RANGE&& nullable_range,
        std::optional<std::string_view> name,
        std::optional<METADATA_RANGE> metadata
    )
    {
        auto values = nullable_range
                      | std::views::transform(
                          [](const auto& v)
                          {
                              return static_cast<T>(v.value());
                          }
                      );
 
        auto is_non_null = nullable_range
                           | std::views::transform(
                               [](const auto& v)
                               {
                                   return v.has_value();
                               }
                           );
 
        return create_proxy(
            std::forward<decltype(values)>(values),
            std::forward<decltype(is_non_null)>(is_non_null),
            name,
            metadata
        );
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    template <std::ranges::input_range R, input_metadata_container METADATA_RANGE>
        requires std::convertible_to<std::ranges::range_value_t<R>, T>
    [[nodiscard]] arrow_proxy variable_size_binary_view_array_impl<T, CR, Ext>::create_proxy(
        R&& range,
        bool nullable,
        std::optional<std::string_view> name,
        std::optional<METADATA_RANGE> metadata
    )
    {
        if (nullable)
        {
            return create_proxy(
                std::forward<R>(range),
                validity_bitmap{validity_bitmap::default_allocator()},
                std::move(name),
                std::move(metadata)
            );
        }
 
        // create arrow schema
        ArrowSchema schema = create_arrow_schema(std::move(name), std::move(metadata), std::nullopt);
 
        // create buffers
        auto buffers_parts = create_buffers(std::forward<R>(range));
 
        std::vector<buffer<uint8_t>> buffers{
            buffer<uint8_t>{nullptr, 0, buffer<uint8_t>::default_allocator()},  // validity bitmap
            std::move(buffers_parts.length_buffer),
            std::move(buffers_parts.long_string_storage),
            std::move(buffers_parts.buffer_sizes).extract_storage()
        };
        const auto size = range_size(range);
 
        constexpr repeat_view<bool> children_ownership(true, 0);
 
        // create arrow array
        ArrowArray arr = make_arrow_array(
            static_cast<std::int64_t>(size),  // length
            static_cast<int64_t>(0),
            0,  // offset
            std::move(buffers),
            nullptr,  // children
            children_ownership,
            nullptr,  // dictionary
            true
        );
 
        arrow_proxy proxy{std::move(arr), std::move(schema)};
        Ext::init(proxy);
        return proxy;
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    template <std::ranges::input_range VALUE_BUFFERS_RANGE, validity_bitmap_input VB, input_metadata_container METADATA_RANGE>
        requires std::convertible_to<std::ranges::range_value_t<VALUE_BUFFERS_RANGE>, u8_buffer<uint8_t>>
    arrow_proxy variable_size_binary_view_array_impl<T, CR, Ext>::create_proxy(
        size_t element_count,
        u8_buffer<uint8_t>&& buffer_view,
        VALUE_BUFFERS_RANGE&& value_buffers,
        VB&& validity_input,
        std::optional<std::string_view> name,
        std::optional<METADATA_RANGE> metadata
    )
    {
        const auto size = buffer_view.size() / DATA_BUFFER_SIZE;
        SPARROW_ASSERT_TRUE(size == element_count);
 
        static const std::optional<std::unordered_set<sparrow::ArrowFlag>> flags{{ArrowFlag::NULLABLE}};
 
        ArrowSchema schema = create_arrow_schema(std::move(name), std::move(metadata), flags);
 
        auto bitmap = ensure_validity_bitmap(size, std::forward<VB>(validity_input));
        std::vector<buffer<uint8_t>> buffers{
            std::move(bitmap).extract_storage(),
            std::move(buffer_view).extract_storage()
        };
        for (auto&& buf : value_buffers)
        {
            buffers.emplace_back(std::forward<decltype(buf)>(buf), typename buffer<uint8_t>::default_allocator());
        }
 
        // Create buffer sizes for the variadic buffers
        u8_buffer<int64_t> buffer_sizes(value_buffers.size());
        for (std::size_t i = 0; i < value_buffers.size(); ++i)
        {
            buffer_sizes[i] = static_cast<int64_t>(value_buffers[i].size());
        }
        buffers.push_back(std::move(buffer_sizes).extract_storage());
 
        constexpr repeat_view<bool> children_ownership(true, 0);
 
        ArrowArray arr = make_arrow_array(
            static_cast<std::int64_t>(size),                 // length
            static_cast<std::int64_t>(bitmap.null_count()),  // null_count
            0,                                               // offset
            std::move(buffers),
            nullptr,  // children
            children_ownership,
            nullptr,  // dictionary
            true
        );
 
        arrow_proxy proxy{std::move(arr), std::move(schema)};
        Ext::init(proxy);
        return proxy;
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    constexpr auto variable_size_binary_view_array_impl<T, CR, Ext>::value(size_type i) -> inner_reference
    {
        return static_cast<const self_type*>(this)->value(i);
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    constexpr auto variable_size_binary_view_array_impl<T, CR, Ext>::value(size_type i) const
        -> inner_const_reference
    {
#ifdef __GNUC__
#    pragma GCC diagnostic push
#    pragma GCC diagnostic ignored "-Wcast-align"
#endif
 
        SPARROW_ASSERT_TRUE(i < this->size());
        using char_or_byte = typename inner_const_reference::value_type;
 
        auto data_ptr = this->get_arrow_proxy().buffers()[LENGTH_BUFFER_INDEX].template data<uint8_t>()
                        + (i * DATA_BUFFER_SIZE);
        const auto length = static_cast<std::size_t>(read_int32_unaligned(data_ptr));
 
        if (length <= SHORT_STRING_SIZE)
        {
            const auto ptr = reinterpret_cast<const char_or_byte*>(data_ptr);
            const auto ret = inner_const_reference(ptr + SHORT_STRING_OFFSET, length);
            return ret;
        }
        else
        {
            const auto buffer_index = static_cast<std::size_t>(
                read_int32_unaligned(data_ptr + BUFFER_INDEX_OFFSET)
            );
            const auto buffer_offset = static_cast<std::size_t>(
                read_int32_unaligned(data_ptr + BUFFER_OFFSET_OFFSET)
            );
            const auto buffer = this->get_arrow_proxy()
                                    .buffers()[buffer_index + FIRST_VAR_DATA_BUFFER_INDEX]
                                    .template data<const char_or_byte>();
            return inner_const_reference(buffer + buffer_offset, length);
        }
 
#ifdef __GNUC__
#    pragma GCC diagnostic pop
#endif
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    constexpr auto variable_size_binary_view_array_impl<T, CR, Ext>::value_begin() -> value_iterator
    {
        return value_iterator(detail::layout_value_functor<self_type, inner_reference>(), 0);
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    constexpr auto variable_size_binary_view_array_impl<T, CR, Ext>::value_end() -> value_iterator
    {
        return value_iterator(detail::layout_value_functor<self_type, inner_reference>(this), this->size());
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    constexpr auto variable_size_binary_view_array_impl<T, CR, Ext>::value_cbegin() const
        -> const_value_iterator
    {
        return const_value_iterator(detail::layout_value_functor<const self_type, inner_const_reference>(this), 0);
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    constexpr auto variable_size_binary_view_array_impl<T, CR, Ext>::value_cend() const -> const_value_iterator
    {
        return const_value_iterator(
            detail::layout_value_functor<const self_type, inner_const_reference>(this),
            this->size()
        );
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    template <std::ranges::sized_range U>
        requires mpl::convertible_ranges<U, T>
    constexpr void variable_size_binary_view_array_impl<T, CR, Ext>::assign(U&& rhs, size_type index)
    {
        SPARROW_ASSERT_TRUE(index < this->size());
        const auto new_length = static_cast<std::size_t>(std::ranges::size(rhs));
 
        auto& length_buffer = this->get_arrow_proxy().get_array_private_data()->buffers()[LENGTH_BUFFER_INDEX];
        auto view_ptr = length_buffer.data() + (index * DATA_BUFFER_SIZE);
        const auto current_length = static_cast<std::size_t>(read_int32_unaligned(view_ptr));
 
        // Update the length in the view structure
        write_int32_unaligned(view_ptr, static_cast<std::int32_t>(new_length));
 
        if (new_length <= SHORT_STRING_SIZE)
        {
            auto data_ptr = view_ptr + SHORT_STRING_OFFSET;
            std::ranges::copy(rhs, reinterpret_cast<typename T::value_type*>(data_ptr));
 
            // Clear any remaining bytes in the inline storage
            if (new_length < SHORT_STRING_SIZE)
            {
                std::fill_n(
                    reinterpret_cast<typename T::value_type*>(data_ptr) + new_length,
                    SHORT_STRING_SIZE - new_length,
                    typename T::value_type{}
                );
            }
        }
        else
        {
            // Handle assignment of long strings (> 12 bytes)
            // This requires managing the variadic buffers and potentially reorganizing the layout
 
            auto& buffers = this->get_arrow_proxy().get_array_private_data()->buffers();
            auto& var_data_buffer = buffers[FIRST_VAR_DATA_BUFFER_INDEX];
            auto& buffer_sizes_buffer = buffers[buffers.size() - 1];  // Last buffer contains sizes
 
            const bool was_long_string = current_length > SHORT_STRING_SIZE;
            std::size_t current_buffer_offset = 0;
 
            if (was_long_string)
            {
                current_buffer_offset = static_cast<std::size_t>(
                    read_int32_unaligned(view_ptr + BUFFER_OFFSET_OFFSET)
                );
            }
 
            auto transformed_data = rhs
                                    | std::ranges::views::transform(
                                        transform_to<typename T::value_type, typename T::value_type>
                                    );
 
            // Check for memory reuse optimization: if the new value is identical to existing data
            bool can_reuse_memory = false;
            if (was_long_string && new_length == current_length)
            {
                const auto* existing_data = var_data_buffer.data() + current_buffer_offset;
                can_reuse_memory = std::ranges::equal(
                    transformed_data,
                    std::span<const typename T::value_type>(
                        reinterpret_cast<const typename T::value_type*>(existing_data),
                        new_length
                    )
                );
            }
 
            if (can_reuse_memory)
            {
                // Data is identical - just update the view structure prefix and we're done
                auto prefix_range = rhs | std::ranges::views::take(PREFIX_SIZE);
                auto prefix_transformed = prefix_range
                                          | std::ranges::views::transform(
                                              transform_to<std::uint8_t, typename T::value_type>
                                          );
                std::ranges::copy(prefix_transformed, view_ptr + PREFIX_OFFSET);
                return;  // Early exit - no buffer management needed
            }
 
            // Calculate space requirements and buffer management strategy
            const auto length_diff = static_cast<std::ptrdiff_t>(new_length)
                                     - static_cast<std::ptrdiff_t>(current_length);
            const bool can_fit_in_place = was_long_string && length_diff <= 0;
 
            std::size_t final_offset = 0;
 
            if (can_fit_in_place)
            {
                // We can reuse the existing space (new data is same size or smaller)
                final_offset = current_buffer_offset;
 
                // If the new data is smaller, we need to compact the buffer
                if (length_diff < 0)
                {
                    const auto bytes_to_compact = static_cast<std::size_t>(-length_diff);
                    const auto move_start = current_buffer_offset + current_length;
                    const auto move_end = var_data_buffer.size();
                    const auto bytes_to_move = move_end - move_start;
 
                    if (bytes_to_move > 0)
                    {
                        std::move(
                            var_data_buffer.data() + move_start,
                            var_data_buffer.data() + move_end,
                            var_data_buffer.data() + move_start - bytes_to_compact
                        );
                    }
 
                    var_data_buffer.resize(var_data_buffer.size() - bytes_to_compact);
 
                    // Update buffer offsets for all elements that come after this one
                    update_buffer_offsets_after(
                        length_buffer.data(),
                        this->size(),
                        DATA_BUFFER_SIZE,
                        SHORT_STRING_SIZE,
                        BUFFER_OFFSET_OFFSET,
                        current_buffer_offset + current_length,
                        -static_cast<std::ptrdiff_t>(bytes_to_compact),
                        index
                    );
 
                    // Update buffer sizes metadata
                    update_buffer_sizes_metadata(
                        buffer_sizes_buffer,
                        static_cast<std::int64_t>(var_data_buffer.size())
                    );
                }
            }
            else
            {
                // Need to expand buffer or assign to new location
                const auto expansion_needed = was_long_string ? length_diff
                                                              : static_cast<std::ptrdiff_t>(new_length);
                const auto new_var_buffer_size = var_data_buffer.size() + expansion_needed;
 
                if (was_long_string && length_diff > 0)
                {
                    // Expand in-place: move data after current element to make space
                    final_offset = current_buffer_offset;
                    const auto expansion_bytes = static_cast<std::size_t>(length_diff);
                    const auto move_start = current_buffer_offset + current_length;
                    const auto bytes_to_move = var_data_buffer.size() - move_start;
 
                    // Resize buffer first
                    var_data_buffer.resize(new_var_buffer_size);
 
                    if (bytes_to_move > 0)
                    {
                        // Move data to make space for expansion
                        std::move_backward(
                            var_data_buffer.data() + move_start,
                            var_data_buffer.data() + move_start + bytes_to_move,
                            var_data_buffer.data() + move_start + bytes_to_move + expansion_bytes
                        );
                    }
 
                    // Update buffer offsets for all elements that come after this one
                    update_buffer_offsets_after(
                        length_buffer.data(),
                        this->size(),
                        DATA_BUFFER_SIZE,
                        SHORT_STRING_SIZE,
                        BUFFER_OFFSET_OFFSET,
                        move_start - 1,  // threshold is just before move_start
                        static_cast<std::ptrdiff_t>(expansion_bytes),
                        index
                    );
                }
                else
                {
                    // Append to end of buffer (new long string)
                    final_offset = var_data_buffer.size();
                    var_data_buffer.resize(new_var_buffer_size);
                }
 
                // Update buffer sizes metadata
                update_buffer_sizes_metadata(buffer_sizes_buffer, static_cast<std::int64_t>(new_var_buffer_size));
            }
 
            std::ranges::copy(transformed_data, var_data_buffer.data() + final_offset);
 
            // Update view structure for long string format
            // Write prefix (first 4 bytes)
            auto prefix_range = rhs | std::ranges::views::take(PREFIX_SIZE);
            auto prefix_transformed = prefix_range
                                      | std::ranges::views::transform(
                                          transform_to<std::uint8_t, typename T::value_type>
                                      );
            std::ranges::copy(prefix_transformed, view_ptr + PREFIX_OFFSET);
 
            write_int32_unaligned(
                view_ptr + BUFFER_INDEX_OFFSET,
                static_cast<std::int32_t>(FIRST_VAR_DATA_BUFFER_INDEX)
            );
 
            write_int32_unaligned(view_ptr + BUFFER_OFFSET_OFFSET, static_cast<std::int32_t>(final_offset));
        }
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    template <std::ranges::sized_range U>
        requires mpl::convertible_ranges<U, T>
    void variable_size_binary_view_array_impl<T, CR, Ext>::resize_values(size_type new_length, U value)
    {
        const size_t current_size = this->size();
 
        if (new_length == current_size)
        {
            return;
        }
 
        if (new_length < current_size)
        {
            erase_values(sparrow::next(value_cbegin(), new_length), current_size - new_length);
        }
        else
        {
            insert_value(value_cend(), value, new_length - current_size);
        }
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    template <std::ranges::sized_range U>
        requires mpl::convertible_ranges<U, T>
    auto
    variable_size_binary_view_array_impl<T, CR, Ext>::insert_value(const_value_iterator pos, U value, size_type count)
        -> value_iterator
    {
        const auto repeat_view = sparrow::repeat_view<U>(value, count);
        return insert_values(pos, std::ranges::begin(repeat_view), std::ranges::end(repeat_view));
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    template <mpl::iterator_of_type<T> InputIt>
    auto variable_size_binary_view_array_impl<T, CR, Ext>::insert_values(
        const_value_iterator pos,
        InputIt first,
        InputIt last
    ) -> value_iterator
    {
        SPARROW_ASSERT_TRUE(first <= last);
        const size_type count = static_cast<size_type>(std::distance(first, last));
        if (count == 0)
        {
            const auto insert_index = std::distance(value_cbegin(), pos);
            return value_begin() + insert_index;
        }
 
        const auto insert_index = static_cast<size_t>(std::distance(value_cbegin(), pos));
        const auto current_size = this->size();
        const auto new_size = current_size + count;
 
        // Calculate total additional variadic storage needed
        std::size_t additional_var_storage = 0;
        std::vector<std::size_t> value_lengths;
        value_lengths.reserve(count);
 
        for (auto it = first; it != last; ++it)
        {
            const auto length = static_cast<std::size_t>(std::ranges::size(*it));
            value_lengths.push_back(length);
            if (length > SHORT_STRING_SIZE)
            {
                additional_var_storage += length;
            }
        }
 
        auto& proxy = this->get_arrow_proxy();
        auto* private_data = proxy.get_array_private_data();
        auto& buffers = private_data->buffers();
 
        const auto new_view_buffer_size = new_size * DATA_BUFFER_SIZE;
        buffers[LENGTH_BUFFER_INDEX].resize(new_view_buffer_size);
 
        if (additional_var_storage > 0)
        {
            const auto current_var_size = buffers[FIRST_VAR_DATA_BUFFER_INDEX].size();
            buffers[FIRST_VAR_DATA_BUFFER_INDEX].resize(current_var_size + additional_var_storage);
        }
 
        auto& buffer_sizes = buffers[buffers.size() - 1];
        update_buffer_sizes_metadata(
            buffer_sizes,
            static_cast<std::int64_t>(buffers[FIRST_VAR_DATA_BUFFER_INDEX].size())
        );
 
        // Shift existing view structures after insertion point
        auto* view_data = buffers[LENGTH_BUFFER_INDEX].data();
        if (insert_index < current_size)
        {
            const auto bytes_to_move = (current_size - insert_index) * DATA_BUFFER_SIZE;
            const auto src_offset = insert_index * DATA_BUFFER_SIZE;
            const auto dst_offset = (insert_index + count) * DATA_BUFFER_SIZE;
 
            std::memmove(view_data + dst_offset, view_data + src_offset, bytes_to_move);
 
            // Update buffer offsets for moved long strings
            if (additional_var_storage > 0)
            {
                for (size_type i = insert_index + count; i < new_size; ++i)
                {
                    auto* view_ptr = view_data + (i * DATA_BUFFER_SIZE);
                    std::int32_t length;
                    std::memcpy(&length, view_ptr, sizeof(std::int32_t));
 
                    if (static_cast<std::size_t>(length) > SHORT_STRING_SIZE)
                    {
                        std::int32_t current_offset;
                        std::memcpy(&current_offset, view_ptr + BUFFER_OFFSET_OFFSET, sizeof(std::int32_t));
                        current_offset += static_cast<std::int32_t>(additional_var_storage);
                        std::memcpy(view_ptr + BUFFER_OFFSET_OFFSET, &current_offset, sizeof(std::int32_t));
                    }
                }
            }
        }
 
        // Insert new view structures
        std::size_t var_offset = buffers[FIRST_VAR_DATA_BUFFER_INDEX].size() - additional_var_storage;
        size_type value_idx = 0;
 
        for (auto it = first; it != last; ++it, ++value_idx)
        {
            const auto view_index = insert_index + value_idx;
            auto* view_ptr = view_data + (view_index * DATA_BUFFER_SIZE);
            const auto value_length = value_lengths[value_idx];
 
            const auto& current_value = *it;
 
            // Write length
            const std::int32_t value_length_int32 = static_cast<std::int32_t>(value_length);
            std::memcpy(view_ptr, &value_length_int32, sizeof(std::int32_t));
 
            if (value_length <= SHORT_STRING_SIZE)
            {
                // Store inline - convert and copy elements manually
                std::ranges::transform(
                    current_value,
                    view_ptr + SHORT_STRING_OFFSET,
                    transform_to<std::uint8_t, typename T::value_type>
                );
 
                std::fill(
                    view_ptr + SHORT_STRING_OFFSET + value_length,
                    view_ptr + DATA_BUFFER_SIZE,
                    std::uint8_t(0)
                );
            }
            else
            {
                // Store prefix - copy first PREFIX_SIZE elements manually
                std::ranges::transform(
                    current_value | std::views::take(PREFIX_SIZE),
                    view_ptr + PREFIX_OFFSET,
                    transform_to<std::uint8_t, typename T::value_type>
                );
 
                // Set buffer index
                const std::int32_t buffer_index_zero = 0;
                std::memcpy(view_ptr + BUFFER_INDEX_OFFSET, &buffer_index_zero, sizeof(std::int32_t));
 
                // Set buffer offset
                const std::int32_t var_offset_int32 = static_cast<std::int32_t>(var_offset);
                std::memcpy(view_ptr + BUFFER_OFFSET_OFFSET, &var_offset_int32, sizeof(std::int32_t));
 
                // Copy data to variadic buffer - convert and copy manually
                std::ranges::transform(
                    current_value,
                    buffers[FIRST_VAR_DATA_BUFFER_INDEX].data() + var_offset,
                    transform_to<std::uint8_t, typename T::value_type>
                );
 
                var_offset += value_length;
            }
        }
 
        // Update buffers
        proxy.update_buffers();
 
        return value_begin() + static_cast<difference_type>(insert_index);
    }
 
    template <std::ranges::sized_range T, class CR, typename Ext>
    auto
    variable_size_binary_view_array_impl<T, CR, Ext>::erase_values(const_value_iterator pos, size_type count)
        -> value_iterator
    {
        const size_t erase_index = static_cast<size_t>(std::distance(value_cbegin(), pos));
        const size_t current_size = this->size();
 
        // Validate bounds and handle zero count
        if (erase_index + count > current_size)
        {
            count = current_size - erase_index;
        }
 
        if (count == 0)
        {
            return value_begin() + static_cast<difference_type>(erase_index);
        }
 
        const auto new_size = current_size - count;
 
        // Calculate how much variadic storage will be freed
        std::size_t freed_var_storage = 0;
        auto& proxy = this->get_arrow_proxy();
        auto* private_data = proxy.get_array_private_data();
        auto& buffers = private_data->buffers();
        auto* view_data = buffers[LENGTH_BUFFER_INDEX].data();
 
        // Calculate freed storage from elements being erased
        for (size_type i = erase_index; i < erase_index + count; ++i)
        {
            auto* view_ptr = view_data + (i * DATA_BUFFER_SIZE);
            std::int32_t length;
            std::memcpy(&length, view_ptr, sizeof(std::int32_t));
            if (static_cast<std::size_t>(length) > SHORT_STRING_SIZE)
            {
                freed_var_storage += static_cast<std::size_t>(length);
            }
        }
 
        // Handle empty array case
        if (new_size == 0)
        {
            // Resize all buffers to empty
            if (buffers[0].size() > 0)
            {
                buffers[0].clear();
            }
            buffers[LENGTH_BUFFER_INDEX].clear();
            buffers[FIRST_VAR_DATA_BUFFER_INDEX].clear();
 
            auto& buffer_sizes = buffers[buffers.size() - 1];
            update_buffer_sizes_metadata(buffer_sizes, 0);
 
            proxy.update_buffers();
            return value_begin();
        }
 
        // Compact variadic buffer if needed
        if (freed_var_storage > 0)
        {
            auto& var_buffer = buffers[FIRST_VAR_DATA_BUFFER_INDEX];
            std::size_t write_offset = 0;
 
            // Create mapping of old offsets to new offsets
            std::unordered_map<std::size_t, std::size_t> offset_mapping;
            offset_mapping.reserve(current_size - count);
 
            for (size_type i = 0; i < current_size; ++i)
            {
                if (i >= erase_index && i < erase_index + count)
                {
                    // Skip erased elements
                    continue;
                }
 
                auto* view_ptr = view_data + (i * DATA_BUFFER_SIZE);
                std::int32_t length;
                std::memcpy(&length, view_ptr, sizeof(std::int32_t));
                if (static_cast<std::size_t>(length) > SHORT_STRING_SIZE)
                {
                    std::int32_t old_offset_int32;
                    std::memcpy(&old_offset_int32, view_ptr + BUFFER_OFFSET_OFFSET, sizeof(std::int32_t));
                    const auto old_offset = static_cast<std::size_t>(old_offset_int32);
 
                    // Record mapping for updating view structures later
                    offset_mapping[old_offset] = write_offset;
 
                    // Move data if needed
                    if (write_offset != old_offset)
                    {
                        std::memmove(
                            var_buffer.data() + write_offset,
                            var_buffer.data() + old_offset,
                            static_cast<std::size_t>(length)
                        );
                    }
 
                    write_offset += static_cast<std::size_t>(length);
                }
            }
 
            // Resize variadic buffer
            var_buffer.resize(var_buffer.size() - freed_var_storage);
 
            // Update buffer sizes metadata
            auto& buffer_sizes = buffers[buffers.size() - 1];
            update_buffer_sizes_metadata(buffer_sizes, static_cast<std::int64_t>(var_buffer.size()));
 
            // Update view structure offsets
            for (size_type i = 0; i < current_size; ++i)
            {
                if (i >= erase_index && i < erase_index + count)
                {
                    continue;  // Skip erased elements
                }
 
                auto* view_ptr = view_data + (i * DATA_BUFFER_SIZE);
                std::int32_t length;
                std::memcpy(&length, view_ptr, sizeof(std::int32_t));
                if (static_cast<std::size_t>(length) > SHORT_STRING_SIZE)
                {
                    std::int32_t old_offset_int32;
                    std::memcpy(&old_offset_int32, view_ptr + BUFFER_OFFSET_OFFSET, sizeof(std::int32_t));
                    const auto old_offset = static_cast<std::size_t>(old_offset_int32);
                    auto it = offset_mapping.find(old_offset);
                    if (it != offset_mapping.end())
                    {
                        const std::int32_t new_offset = static_cast<std::int32_t>(it->second);
                        std::memcpy(view_ptr + BUFFER_OFFSET_OFFSET, &new_offset, sizeof(std::int32_t));
                    }
                }
            }
        }
 
        // Compact view buffer - move elements after erase range
        if (erase_index + count < current_size)
        {
            const auto src_offset = (erase_index + count) * DATA_BUFFER_SIZE;
            const auto dst_offset = erase_index * DATA_BUFFER_SIZE;
            const auto bytes_to_move = (current_size - erase_index - count) * DATA_BUFFER_SIZE;
 
            std::memmove(view_data + dst_offset, view_data + src_offset, bytes_to_move);
        }
 
        // Resize view buffer
        buffers[LENGTH_BUFFER_INDEX].resize(new_size * DATA_BUFFER_SIZE);
 
        // Update buffers
        proxy.update_buffers();
 
        // Return iterator to element after last erased, or end if we erased to the end
        return erase_index < new_size ? sparrow::next(value_begin(), erase_index) : value_end();
    }
 
}