union_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 <algorithm>
#include <array>
#include <charconv>
#include <cstdint>
#include <functional>
#include <iterator>
#include <limits>
#include <numeric>
#include <optional>
#include <ranges>
#include <span>
#include <string>
#include <string_view>
#include <type_traits>
#include <utility>
#include <variant>
#include <vector>
 
#include "sparrow/array_api.hpp"
#include "sparrow/arrow_interface/arrow_flag_utils.hpp"
#include "sparrow/config/config.hpp"
#include "sparrow/layout/array_access.hpp"
#include "sparrow/layout/array_factory.hpp"
#include "sparrow/layout/array_helper.hpp"
#include "sparrow/layout/array_wrapper.hpp"
#include "sparrow/layout/layout_utils.hpp"
#include "sparrow/layout/nested_value_types.hpp"
#include "sparrow/u8_buffer.hpp"
#include "sparrow/utils/contracts.hpp"
#include "sparrow/utils/crtp_base.hpp"
#include "sparrow/utils/functor_index_iterator.hpp"
#include "sparrow/utils/memory.hpp"
#include "sparrow/utils/metadata.hpp"
#include "sparrow/utils/mp_utils.hpp"
#include "sparrow/utils/nullable.hpp"
 
namespace sparrow
{
    class dense_union_array;
    class sparse_union_array;
 
    namespace detail
    {
        template <>
        struct get_data_type_from_array<sparrow::dense_union_array>
        {
            [[nodiscard]] static constexpr sparrow::data_type get()
            {
                return sparrow::data_type::DENSE_UNION;
            }
        };
 
        template <>
        struct get_data_type_from_array<sparrow::sparse_union_array>
        {
            [[nodiscard]] static constexpr sparrow::data_type get()
            {
                return sparrow::data_type::SPARSE_UNION;
            }
        };
 
        inline constexpr std::size_t union_no_child = std::numeric_limits<std::size_t>::max();

        struct union_rebuild_value
        {
            std::size_t child = union_no_child;  
            std::size_t row = 0;                 
            bool force_null = false;             
            bool inserted = false;               
        };

        struct union_rebuild_payload
        {
            std::vector<std::vector<union_rebuild_value>> child_values;  
            u8_buffer<std::uint8_t> type_ids{};                          
            u8_buffer<std::uint32_t> offsets{};                          
        };
    }

    template <class T>
    constexpr bool is_dense_union_array_v = std::same_as<T, dense_union_array>;

    template <class T>
    constexpr bool is_sparse_union_array_v = std::same_as<T, sparse_union_array>;

    template <class DERIVED>
    class union_array_crtp_base : public crtp_base<DERIVED>
    {
    public:
 
        using self_type = union_array_crtp_base<DERIVED>;
        using derived_type = DERIVED;
        using inner_value_type = array_traits::inner_value_type;
        using value_type = array_traits::const_reference;
        using const_reference = array_traits::const_reference;
        using functor_type = detail::layout_bracket_functor<derived_type, value_type>;
        using const_functor_type = detail::layout_bracket_functor<const derived_type, value_type>;
        using iterator = functor_index_iterator<functor_type>;
        using const_iterator = functor_index_iterator<const_functor_type>;
        using const_reverse_iterator = std::reverse_iterator<const_iterator>;
        using size_type = std::size_t;
 
        using type_id_buffer_type = u8_buffer<std::uint8_t>;

        [[nodiscard]] constexpr std::optional<std::string_view> name() const;

        [[nodiscard]] SPARROW_CONSTEXPR_CLANG std::optional<key_value_view> metadata() const;

        [[nodiscard]] SPARROW_CONSTEXPR_CLANG value_type at(size_type i) const;

        [[nodiscard]] SPARROW_CONSTEXPR_CLANG value_type operator[](size_type i) const;

        [[nodiscard]] SPARROW_CONSTEXPR_CLANG value_type operator[](size_type i);

        [[nodiscard]] SPARROW_CONSTEXPR_CLANG value_type front() const;

        [[nodiscard]] SPARROW_CONSTEXPR_CLANG value_type back() const;

        [[nodiscard]] constexpr bool empty() const;

        [[nodiscard]] constexpr size_type size() const;

        [[nodiscard]] constexpr iterator begin();

        [[nodiscard]] constexpr iterator end();

        [[nodiscard]] constexpr const_iterator begin() const;

        [[nodiscard]] constexpr const_iterator end() const;

        [[nodiscard]] constexpr const_iterator cbegin() const;

        [[nodiscard]] constexpr const_iterator cend() const;

        [[nodiscard]] constexpr const_reverse_iterator rbegin() const;

        [[nodiscard]] constexpr const_reverse_iterator rend() const;

        [[nodiscard]] constexpr const_reverse_iterator crbegin() const;

        [[nodiscard]] constexpr const_reverse_iterator crend() const;

        iterator insert(const_iterator pos, const_reference value, size_type count = 1);

        iterator
        insert(const_iterator pos, const array_traits::value_type& value, size_type count = 1);

        template <std::input_iterator InputIt>
        iterator insert(const_iterator pos, InputIt first, InputIt last, size_type count = 1);

        iterator erase(const_iterator pos);

        iterator erase(const_iterator first, const_iterator last);

        void push_back(const_reference value);

        void push_back(const array_traits::value_type& value);

        void resize(size_type new_length);

        void resize(size_type new_length, const_reference value);

        void resize(size_type new_length, const array_traits::value_type& value);

        constexpr void zero_null_values(const inner_value_type& value)
        {
            sparrow::zero_null_values(*this, value);
        }
 
    protected:
 
        static constexpr size_t TYPE_ID_MAP_SIZE = 256;
 
        using type_id_map = std::array<std::uint8_t, TYPE_ID_MAP_SIZE>;

        static std::vector<std::uint8_t> child_type_ids_from_format(std::string_view format_string);

        static constexpr type_id_map make_type_id_map(std::span<const std::uint8_t> child_type_ids);

        template <std::ranges::input_range R>
        static constexpr type_id_map
        type_id_map_from_child_to_type_id(const std::optional<R>& child_index_to_type_id);

        template <std::ranges::input_range R>
            requires(std::convertible_to<std::ranges::range_value_t<R>, std::uint8_t>)
        static constexpr std::string
        make_format_string(bool dense, std::size_t n, const std::optional<R>& child_index_to_type_id);
 
        using children_type = std::vector<cloning_ptr<array_wrapper>>;

        constexpr children_type make_children(arrow_proxy& proxy);

        explicit union_array_crtp_base(arrow_proxy proxy);

        constexpr union_array_crtp_base(const self_type& rhs);

        constexpr self_type& operator=(const self_type& rhs);
 
        constexpr union_array_crtp_base(self_type&& rhs) = default;
        constexpr self_type& operator=(self_type&& rhs) = default;
 
        template <class VALUE>
        void resize_impl(size_type new_length, VALUE&& value)
        {
            const auto current_size = size();
            if (new_length < current_size)
            {
                erase_values(new_length, current_size - new_length);
            }
            else if (new_length > current_size)
            {
                insert(cend(), std::forward<VALUE>(value), new_length - current_size);
            }
        }
 
        iterator iterator_at(size_type index)
        {
            return iterator(functor_type{&this->derived_cast()}, index);
        }

        void append_rebuild_entries(
            std::vector<detail::union_rebuild_value>& entries,
            size_type from,
            size_type to
        ) const
        {
            for (size_type index = from; index < to; ++index)
            {
                entries.push_back(detail::union_rebuild_value{
                    .child = m_type_id_map[p_type_ids[index]],
                    .row = this->derived_cast().element_offset(index)
                });
            }
        }

        iterator insert_materialized(
            const_iterator pos,
            std::vector<array_traits::value_type> inserted,
            size_type count
        );
 
        iterator erase_values(size_type first, size_type count);

        struct union_child_schema
        {
            data_type type;
            std::string format;
 
            [[nodiscard]] bool operator==(const union_child_schema& other) const = default;
        };

        std::vector<union_child_schema> make_child_schemas(
            std::vector<std::uint8_t>& child_type_ids,
            std::array<bool, TYPE_ID_MAP_SIZE>& used_type_ids
        ) const;

        static std::vector<std::size_t> resolve_inserted_children(
            std::span<const array_traits::value_type> inserted,
            std::vector<union_child_schema>& child_schemas,
            std::vector<std::uint8_t>& child_type_ids,
            std::array<bool, TYPE_ID_MAP_SIZE>& used_type_ids,
            std::vector<array>& new_child_templates
        );

        std::vector<detail::union_rebuild_value> make_union_fillers(
            size_type old_child_count,
            const std::vector<union_child_schema>& child_schemas,
            std::vector<std::size_t>& inserted_children,
            std::vector<array_traits::value_type>& inserted
        ) const;

        detail::union_rebuild_payload make_rebuild_payload(
            std::span<const detail::union_rebuild_value> values,
            std::span<const std::size_t> inserted_children,
            std::span<const std::uint8_t> child_type_ids,
            std::span<const detail::union_rebuild_value> fillers
        ) const;

        std::vector<array> build_rebuilt_children(
            const detail::union_rebuild_payload& payload,
            std::span<const array_traits::value_type> inserted,
            std::span<const std::size_t> inserted_children,
            const children_type& old_children,
            std::vector<array>& new_child_templates
        ) const;

        arrow_proxy create_rebuild_proxy(
            std::vector<array> children,
            detail::union_rebuild_payload&& payload,
            std::span<const std::uint8_t> child_type_ids
        ) const;

        iterator adopt_rebuilt_array(
            arrow_proxy replacement,
            std::vector<std::uint8_t> child_type_ids,
            size_type return_index
        );

        iterator rebuild_values(
            std::vector<detail::union_rebuild_value> values,
            std::vector<array_traits::value_type> inserted,
            size_type return_index
        );

        [[nodiscard]] constexpr arrow_proxy& get_arrow_proxy();

        [[nodiscard]] constexpr const arrow_proxy& get_arrow_proxy() const;
 
        arrow_proxy m_proxy;                                       
        const std::uint8_t* p_type_ids;                            
        children_type m_children;                                  
        std::vector<std::uint8_t> m_child_type_ids;                
        std::array<std::uint8_t, TYPE_ID_MAP_SIZE> m_type_id_map;  
 
        friend class detail::array_access;
 
#if defined(__cpp_lib_format)
        friend struct std::formatter<DERIVED>;
#endif
    };

    template <class D>
    constexpr bool operator==(const union_array_crtp_base<D>& lhs, const union_array_crtp_base<D>& rhs);

    class dense_union_array : public union_array_crtp_base<dense_union_array>
    {
    public:
 
        using base_type = union_array_crtp_base<dense_union_array>;
        using offset_buffer_type = u8_buffer<std::uint32_t>;
        using type_id_buffer_type = typename base_type::type_id_buffer_type;

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

        SPARROW_API explicit dense_union_array(arrow_proxy proxy);

        SPARROW_API dense_union_array(const dense_union_array& rhs);

        SPARROW_API dense_union_array& operator=(const dense_union_array& rhs);
 
        dense_union_array(dense_union_array&& rhs) = default;
        dense_union_array& operator=(dense_union_array&& rhs) = default;
 
    private:
 
        using type_id_map = typename base_type::type_id_map;

        template <
            std::ranges::input_range TYPE_MAPPING = std::vector<std::uint8_t>,
            input_metadata_container METADATA_RANGE = std::vector<metadata_pair>>
            requires(std::convertible_to<std::ranges::range_value_t<TYPE_MAPPING>, std::uint8_t>)
        [[nodiscard]] static auto create_proxy(
            std::vector<array>&& children,
            type_id_buffer_type&& element_type,
            offset_buffer_type&& offsets,
            std::optional<TYPE_MAPPING>&& type_mapping = std::nullopt,
            std::optional<std::string_view> name = std::nullopt,
            std::optional<METADATA_RANGE> metadata = std::nullopt
        ) -> arrow_proxy;

        template <
            std::ranges::input_range TYPE_ID_BUFFER_RANGE,
            std::ranges::input_range OFFSET_BUFFER_RANGE,
            std::ranges::input_range TYPE_MAPPING = std::vector<std::uint8_t>,
            input_metadata_container METADATA_RANGE = std::vector<metadata_pair>>
            requires(std::convertible_to<std::ranges::range_value_t<TYPE_MAPPING>, std::uint8_t>)
        [[nodiscard]] static arrow_proxy create_proxy(
            std::vector<array>&& children,
            TYPE_ID_BUFFER_RANGE&& element_type,
            OFFSET_BUFFER_RANGE&& offsets,
            std::optional<TYPE_MAPPING>&& type_mapping = std::nullopt,
            std::optional<std::string_view> name = std::nullopt,
            std::optional<METADATA_RANGE> metadata = std::nullopt
        )
        {
            SPARROW_ASSERT_TRUE(element_type.size() == offsets.size());
            type_id_buffer_type element_type_buffer{std::move(element_type)};
            offset_buffer_type offsets_buffer{std::move(offsets)};
            return dense_union_array::create_proxy(
                std::move(children),
                std::move(element_type_buffer),
                std::move(offsets_buffer),
                std::move(type_mapping),
                std::move(name),
                std::move(metadata)
            );
        }

        template <input_metadata_container METADATA_RANGE = std::vector<metadata_pair>>
        [[nodiscard]] static arrow_proxy create_proxy_impl(
            std::vector<array>&& children,
            type_id_buffer_type&& element_type,
            offset_buffer_type&& offsets,
            std::string&& format,
            std::optional<std::string_view> name = std::nullopt,
            std::optional<METADATA_RANGE> metadata = std::nullopt
        );

        template <
            std::ranges::input_range TYPE_ID_BUFFER_RANGE,
            std::ranges::input_range OFFSET_BUFFER_RANGE,
            input_metadata_container METADATA_RANGE = std::vector<metadata_pair>>
        [[nodiscard]] static arrow_proxy create_proxy_impl(
            std::vector<array>&& children,
            TYPE_ID_BUFFER_RANGE&& element_type,
            OFFSET_BUFFER_RANGE&& offsets,
            std::string&& format,
            std::optional<std::string_view> name = std::nullopt,
            std::optional<METADATA_RANGE> metadata = std::nullopt
        )
        {
            SPARROW_ASSERT_TRUE(std::ranges::distance(element_type) == std::ranges::distance(offsets));
            SPARROW_ASSERT_TRUE(std::ranges::distance(element_type) == children.size());
            type_id_buffer_type element_type_buffer{std::move(element_type)};
            offset_buffer_type offsets_buffer{std::move(offsets)};
            return dense_union_array::create_proxy_impl(
                std::move(children),
                std::move(element_type_buffer),
                std::move(offsets_buffer),
                std::move(format),
                std::move(name),
                std::move(metadata)
            );
        }

        SPARROW_API std::size_t element_offset(std::size_t i) const;
 
        SPARROW_API iterator rebuild_in_place(
            std::vector<detail::union_rebuild_value> values,
            std::vector<array_traits::value_type> inserted,
            size_type return_index
        );
 
        const std::int32_t* p_offsets;  
        friend class union_array_crtp_base<dense_union_array>;
    };

    class sparse_union_array : public union_array_crtp_base<sparse_union_array>
    {
    public:
 
        using base_type = union_array_crtp_base<sparse_union_array>;
        using type_id_buffer_type = typename base_type::type_id_buffer_type;

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

        SPARROW_API explicit sparse_union_array(arrow_proxy proxy);
 
        SPARROW_API sparse_union_array(const sparse_union_array&);
        SPARROW_API sparse_union_array& operator=(const sparse_union_array&);
 
    private:
 
        using type_id_map = typename base_type::type_id_map;

        template <
            std::ranges::input_range TYPE_MAPPING = std::vector<std::uint8_t>,
            input_metadata_container METADATA_RANGE = std::vector<metadata_pair>>
            requires(std::convertible_to<std::ranges::range_value_t<TYPE_MAPPING>, std::uint8_t>)
        static auto create_proxy(
            std::vector<array>&& children,
            type_id_buffer_type&& element_type,
            std::optional<TYPE_MAPPING>&& type_mapping = std::nullopt,
            std::optional<std::string_view> name = std::nullopt,
            std::optional<METADATA_RANGE> metadata = std::nullopt
        ) -> arrow_proxy;

        template <input_metadata_container METADATA_RANGE>
        static auto create_proxy_impl(
            std::vector<array>&& children,
            type_id_buffer_type&& element_type,
            std::string&& format,
            std::optional<std::string_view> name = std::nullopt,
            std::optional<METADATA_RANGE> metadata = std::nullopt
        ) -> arrow_proxy;

        [[nodiscard]] SPARROW_API std::size_t element_offset(std::size_t i) const;
 
        SPARROW_API iterator rebuild_in_place(
            std::vector<detail::union_rebuild_value> values,
            std::vector<array_traits::value_type> inserted,
            size_type return_index
        );
 
        friend class union_array_crtp_base<sparse_union_array>;
    };
 
    /****************************************
     * union_array_crtp_base implementation *
     ****************************************/
 
    template <class DERIVED>
    auto union_array_crtp_base<DERIVED>::child_type_ids_from_format(std::string_view format_string)
        -> std::vector<std::uint8_t>
    {
        SPARROW_ASSERT_TRUE(format_string.starts_with("+ud:") || format_string.starts_with("+us:"));
        format_string.remove_prefix(4);
 
        std::vector<std::uint8_t> result;
        while (!format_string.empty())
        {
            const auto separator = format_string.find(',');
            const auto token = format_string.substr(0, separator);
            std::uint32_t type_id = 0;
            const auto [end, error] = std::from_chars(token.data(), token.data() + token.size(), type_id);
            SPARROW_ASSERT_TRUE(error == std::errc{} && end == token.data() + token.size());
            SPARROW_ASSERT_TRUE(type_id <= std::numeric_limits<std::uint8_t>::max());
            result.push_back(static_cast<std::uint8_t>(type_id));
            if (separator == std::string_view::npos)
            {
                break;
            }
            format_string.remove_prefix(separator + 1);
        }
        return result;
    }
 
    template <class DERIVED>
    constexpr auto
    union_array_crtp_base<DERIVED>::make_type_id_map(std::span<const std::uint8_t> child_type_ids)
        -> type_id_map
    {
        type_id_map result{};
        std::array<bool, TYPE_ID_MAP_SIZE> seen{};
        for (std::size_t child_index = 0; child_index < child_type_ids.size(); ++child_index)
        {
            const auto type_id = child_type_ids[child_index];
            SPARROW_ASSERT_TRUE(child_index < TYPE_ID_MAP_SIZE);
            SPARROW_ASSERT_TRUE(!seen[type_id]);
            seen[type_id] = true;
            result[type_id] = static_cast<std::uint8_t>(child_index);
        }
        return result;
    }
 
    template <class DERIVED>
    template <std::ranges::input_range R>
    constexpr auto
    union_array_crtp_base<DERIVED>::type_id_map_from_child_to_type_id(const std::optional<R>& child_index_to_type_id)
        -> type_id_map
    {
        type_id_map ret{};
        if (!child_index_to_type_id.has_value())
        {
            constexpr std::array<std::uint8_t, TYPE_ID_MAP_SIZE> default_mapping = []
            {
                std::array<std::uint8_t, TYPE_ID_MAP_SIZE> arr{};
                std::iota(arr.begin(), arr.end(), 0);
                return arr;
            }();
            return default_mapping;
        }
        else
        {
            std::size_t child_index = 0;
            for (const auto type_id : *child_index_to_type_id)
            {
                SPARROW_ASSERT_TRUE(child_index < TYPE_ID_MAP_SIZE);
                const auto id = static_cast<std::size_t>(type_id);
                SPARROW_ASSERT_TRUE(id < TYPE_ID_MAP_SIZE);
                ret[id] = static_cast<std::uint8_t>(child_index++);
            }
        }
        return ret;
    }
 
    template <class DERIVED>
    template <std::ranges::input_range R>
        requires(std::convertible_to<std::ranges::range_value_t<R>, std::uint8_t>)
    constexpr std::string
    union_array_crtp_base<DERIVED>::make_format_string(bool dense, const std::size_t n, const std::optional<R>& range)
    {
        const auto range_size = range.has_value() ? std::ranges::size(*range) : 0;
        if (range_size != n && range_size != 0)
        {
            throw std::invalid_argument("Invalid type-id map");
        }
 
        std::string result = dense ? "+ud:" : "+us:";
        bool first = true;
        const auto append_type_id = [&result, &first](const auto type_id)
        {
            if (!first)
            {
                result.push_back(',');
            }
            result += std::to_string(type_id);
            first = false;
        };
 
        if (range_size == 0)
        {
            for (const auto type_id : std::views::iota(std::size_t{0}, n))
            {
                append_type_id(type_id);
            }
        }
        else
        {
            for (const auto type_id : *range)
            {
                append_type_id(type_id);
            }
        }
        return result;
    }
 
    template <class DERIVED>
    constexpr std::optional<std::string_view> union_array_crtp_base<DERIVED>::name() const
    {
        return m_proxy.name();
    }
 
    template <class DERIVED>
    SPARROW_CONSTEXPR_CLANG std::optional<key_value_view> union_array_crtp_base<DERIVED>::metadata() const
    {
        return m_proxy.metadata();
    }
 
    template <class DERIVED>
    constexpr arrow_proxy& union_array_crtp_base<DERIVED>::get_arrow_proxy()
    {
        return m_proxy;
    }
 
    template <class DERIVED>
    constexpr const arrow_proxy& union_array_crtp_base<DERIVED>::get_arrow_proxy() const
    {
        return m_proxy;
    }
 
    template <class DERIVED>
    union_array_crtp_base<DERIVED>::union_array_crtp_base(arrow_proxy proxy)
        : m_proxy(std::move(proxy))
        , p_type_ids(reinterpret_cast<std::uint8_t*>(m_proxy.buffers()[0 /*index of type-ids*/].data()))
        , m_children(make_children(m_proxy))
        , m_child_type_ids(child_type_ids_from_format(m_proxy.format()))
        , m_type_id_map(make_type_id_map(m_child_type_ids))
    {
    }
 
    template <class DERIVED>
    constexpr union_array_crtp_base<DERIVED>::union_array_crtp_base(const self_type& rhs)
        : self_type(rhs.m_proxy)
    {
    }
 
    template <class DERIVED>
    constexpr auto union_array_crtp_base<DERIVED>::operator=(const self_type& rhs) -> self_type&
    {
        if (this != &rhs)
        {
            m_proxy = rhs.m_proxy;
            p_type_ids = reinterpret_cast<std::uint8_t*>(m_proxy.buffers()[0 /*index of type-ids*/].data());
            m_children = make_children(m_proxy);
            m_child_type_ids = rhs.m_child_type_ids;
            m_type_id_map = rhs.m_type_id_map;
        }
        return *this;
    }
 
    template <class DERIVED>
    SPARROW_CONSTEXPR_CLANG auto union_array_crtp_base<DERIVED>::operator[](std::size_t i) const -> value_type
    {
        const auto type_id = static_cast<std::size_t>(p_type_ids[i + m_proxy.offset()]);
        const auto child_index = m_type_id_map[type_id];
        const auto offset = this->derived_cast().element_offset(i);
        return array_element(*m_children[child_index], static_cast<std::size_t>(offset));
    }
 
    template <class DERIVED>
    SPARROW_CONSTEXPR_CLANG auto union_array_crtp_base<DERIVED>::operator[](std::size_t i) -> value_type
    {
        return static_cast<const derived_type&>(*this)[i];
    }
 
    template <class DERIVED>
    constexpr std::size_t union_array_crtp_base<DERIVED>::size() const
    {
        return m_proxy.length();
    }
 
    template <class DERIVED>
    constexpr bool union_array_crtp_base<DERIVED>::empty() const
    {
        return size() == 0;
    }
 
    template <class DERIVED>
    constexpr auto union_array_crtp_base<DERIVED>::begin() -> iterator
    {
        return iterator(functor_type{&(this->derived_cast())}, 0);
    }
 
    template <class DERIVED>
    constexpr auto union_array_crtp_base<DERIVED>::end() -> iterator
    {
        return iterator(functor_type{&(this->derived_cast())}, this->size());
    }
 
    template <class DERIVED>
    constexpr auto union_array_crtp_base<DERIVED>::begin() const -> const_iterator
    {
        return cbegin();
    }
 
    template <class DERIVED>
    constexpr auto union_array_crtp_base<DERIVED>::end() const -> const_iterator
    {
        return cend();
    }
 
    template <class DERIVED>
    constexpr auto union_array_crtp_base<DERIVED>::cbegin() const -> const_iterator
    {
        return const_iterator(const_functor_type{&(this->derived_cast())}, 0);
    }
 
    template <class DERIVED>
    constexpr auto union_array_crtp_base<DERIVED>::cend() const -> const_iterator
    {
        return const_iterator(const_functor_type{&(this->derived_cast())}, this->size());
    }
 
    template <class DERIVED>
    constexpr auto union_array_crtp_base<DERIVED>::rbegin() const -> const_reverse_iterator
    {
        return const_reverse_iterator{cend()};
    }
 
    template <class DERIVED>
    constexpr auto union_array_crtp_base<DERIVED>::rend() const -> const_reverse_iterator
    {
        return const_reverse_iterator{cbegin()};
    }
 
    template <class DERIVED>
    constexpr auto union_array_crtp_base<DERIVED>::crbegin() const -> const_reverse_iterator
    {
        return rbegin();
    }
 
    template <class DERIVED>
    constexpr auto union_array_crtp_base<DERIVED>::crend() const -> const_reverse_iterator
    {
        return rend();
    }
 
    template <class DERIVED>
    SPARROW_CONSTEXPR_CLANG auto union_array_crtp_base<DERIVED>::front() const -> value_type
    {
        return (*this)[0];
    }
 
    template <class DERIVED>
    SPARROW_CONSTEXPR_CLANG auto union_array_crtp_base<DERIVED>::back() const -> value_type
    {
        return (*this)[this->size() - 1];
    }
 
    template <class DERIVED>
    constexpr auto union_array_crtp_base<DERIVED>::make_children(arrow_proxy& proxy) -> children_type
    {
        children_type children(proxy.children().size(), nullptr);
        for (std::size_t i = 0; i < children.size(); ++i)
        {
            children[i] = array_factory(proxy.children()[i].view());
        }
        return children;
    }
 
    template <class D>
    constexpr bool operator==(const union_array_crtp_base<D>& lhs, const union_array_crtp_base<D>& rhs)
    {
        return std::ranges::equal(lhs, rhs);
    }
 
    template <class DERIVED>
    template <std::input_iterator InputIt>
    auto union_array_crtp_base<DERIVED>::insert(const_iterator pos, InputIt first, InputIt last, size_type count)
        -> iterator
    {
        // A no-op insertion must not require offset zero (see insert_materialized).
        if (count == 0)
        {
            return iterator_at(static_cast<size_type>(pos - cbegin()));
        }
        return insert_materialized(pos, std::vector<array_traits::value_type>(first, last), count);
    }
 
    template <class DERIVED>
    void union_array_crtp_base<DERIVED>::push_back(const_reference value)
    {
        insert(cend(), value);
    }
 
    template <class DERIVED>
    void union_array_crtp_base<DERIVED>::push_back(const array_traits::value_type& value)
    {
        insert(cend(), value);
    }
 
    template <class DERIVED>
    void union_array_crtp_base<DERIVED>::resize(size_type new_length)
    {
        resize_impl(new_length, array_traits::value_type{});
    }
 
    template <class DERIVED>
    void union_array_crtp_base<DERIVED>::resize(size_type new_length, const_reference value)
    {
        resize_impl(new_length, value);
    }
 
    template <class DERIVED>
    void union_array_crtp_base<DERIVED>::resize(size_type new_length, const array_traits::value_type& value)
    {
        resize_impl(new_length, value);
    }
 
    template <class DERIVED>
    auto union_array_crtp_base<DERIVED>::erase(const_iterator pos) -> iterator
    {
        const auto index = static_cast<size_type>(pos - cbegin());
        SPARROW_ASSERT_TRUE(index < size());
        return erase_values(index, 1);
    }
 
    template <class DERIVED>
    auto union_array_crtp_base<DERIVED>::erase(const_iterator first, const_iterator last) -> iterator
    {
        const auto first_index = static_cast<size_type>(first - cbegin());
        const auto last_index = static_cast<size_type>(last - cbegin());
        SPARROW_ASSERT_TRUE(first_index <= last_index);
        SPARROW_ASSERT_TRUE(last_index <= size());
        return erase_values(first_index, last_index - first_index);
    }
 
    template <class DERIVED>
    auto union_array_crtp_base<DERIVED>::make_child_schemas(
        std::vector<std::uint8_t>& child_type_ids,
        std::array<bool, TYPE_ID_MAP_SIZE>& used_type_ids
    ) const -> std::vector<union_child_schema>
    {
        const auto child_count = m_children.size();
        child_type_ids = m_child_type_ids;
        SPARROW_ASSERT_TRUE(child_type_ids.size() == child_count);
 
        std::vector<union_child_schema> child_schemas;
        child_schemas.reserve(child_count);
        for (std::size_t child_index = 0; child_index < child_count; ++child_index)
        {
            used_type_ids[child_type_ids[child_index]] = true;
            child_schemas.emplace_back(
                m_children[child_index]->data_type(),
                std::string(m_children[child_index]->get_arrow_proxy().format())
            );
        }
        return child_schemas;
    }
 
    template <class DERIVED>
    auto union_array_crtp_base<DERIVED>::erase_values(size_type first, size_type count) -> iterator
    {
        using rebuild_value = detail::union_rebuild_value;
        SPARROW_ASSERT_TRUE(m_proxy.offset() == 0);
        const auto current_size = size();
        SPARROW_ASSERT_TRUE(first <= current_size);
        SPARROW_ASSERT_TRUE(count <= current_size - first);
        if (count == 0)
        {
            return iterator_at(first);
        }
 
        std::vector<rebuild_value> values;
        values.reserve(current_size - count);
        this->append_rebuild_entries(values, 0, first);
        this->append_rebuild_entries(values, first + count, current_size);
        return this->derived_cast().rebuild_in_place(std::move(values), {}, first);
    }
 
    namespace detail::union_rebuild
    {
        using dynamic_value = array_traits::value_type;
        using rebuild_value = detail::union_rebuild_value;
        using union_rebuild_payload = detail::union_rebuild_payload;
 
        // Defined in union_array.cpp: array.hpp includes this header, so the null child cannot be built here.
        [[nodiscard]] SPARROW_API array make_null_child(std::size_t count, const arrow_proxy& child_proxy);

        inline std::size_t next_free_type_id(std::span<const bool> used_type_ids)
        {
            return static_cast<std::size_t>(std::ranges::find(used_type_ids, false) - used_type_ids.begin());
        }

        inline std::size_t child_of(const rebuild_value& value, std::span<const std::size_t> inserted_children)
        {
            if (value.inserted)
            {
                SPARROW_ASSERT_TRUE(value.row < inserted_children.size());
                return inserted_children[value.row];
            }
            return value.child;
        }
 
        inline array make_empty_child(const array_wrapper& child)
        {
            const array view{child.get_arrow_proxy().view()};
            return array_empty_like(view);
        }

        template <class T>
        dynamic_value make_null_dynamic_value(const T& value)
        {
            using stored_type = std::remove_cvref_t<T>;
            return dynamic_value(nullable<stored_type>(stored_type(value), false));
        }

        inline dynamic_value make_null_like(const dynamic_value& value)
        {
            return std::visit(
                [](const auto& typed_value) -> dynamic_value
                {
                    return make_null_dynamic_value(typed_value.get());
                },
#if SPARROW_GCC_11_2_WORKAROUND
                static_cast<const dynamic_value::base_type&>(value)
#else
                value
#endif
            );
        }

        inline union_rebuild_payload make_dense_rebuild_payload(
            std::span<const rebuild_value> values,
            std::span<const std::size_t> inserted_children,
            std::span<const std::uint8_t> child_type_ids
        )
        {
            union_rebuild_payload payload;
            payload.child_values.resize(child_type_ids.size());
            payload.type_ids.reserve(values.size());
            payload.offsets.reserve(values.size());
 
            std::vector<std::size_t> child_value_counts(child_type_ids.size(), 0);
            for (const auto& value : values)
            {
                ++child_value_counts[child_of(value, inserted_children)];
            }
            for (std::size_t child_index = 0; child_index < child_type_ids.size(); ++child_index)
            {
                payload.child_values[child_index].reserve(child_value_counts[child_index]);
            }
 
            for (const auto& value : values)
            {
                const auto child_index = child_of(value, inserted_children);
                payload.type_ids.push_back(child_type_ids[child_index]);
                SPARROW_ASSERT_TRUE(
                    payload.child_values[child_index].size()
                    <= static_cast<std::size_t>(std::numeric_limits<std::int32_t>::max())
                );
                payload.offsets.push_back(static_cast<std::uint32_t>(payload.child_values[child_index].size()));
                payload.child_values[child_index].push_back(value);
            }
            return payload;
        }

        inline union_rebuild_payload make_sparse_rebuild_payload(
            std::span<const rebuild_value> values,
            std::span<const std::size_t> inserted_children,
            std::span<const std::uint8_t> child_type_ids,
            std::span<const rebuild_value> fillers
        )
        {
            union_rebuild_payload payload;
            payload.child_values.resize(child_type_ids.size());
            payload.type_ids.reserve(values.size());
            for (const auto& value : values)
            {
                payload.type_ids.push_back(child_type_ids[child_of(value, inserted_children)]);
            }
 
            for (std::size_t child_index = 0; child_index < child_type_ids.size(); ++child_index)
            {
                SPARROW_ASSERT_TRUE(child_index < fillers.size());
                auto& child_values = payload.child_values[child_index];
                child_values.reserve(values.size());
                for (const auto& value : values)
                {
                    child_values.push_back(
                        child_of(value, inserted_children) == child_index ? value : fillers[child_index]
                    );
                }
            }
            return payload;
        }

        inline array make_filler_row(const array& old_child_view, std::size_t row)
        {
            array filler = old_child_view.slice(row, row + 1);
            auto& proxy = detail::array_access::get_arrow_proxy(filler);
            if (auto& bitmap = proxy.bitmap(); bitmap.has_value())
            {
                *bitmap->begin() = false;
                proxy.set_null_count(1);
            }
            return filler;
        }

        inline void
        append_old_child_values(array& destination, const array& old_child_view, std::size_t first_row, std::size_t count)
        {
            const array rows = old_child_view.slice_view(first_row, first_row + count);
            destination.insert(destination.cend(), rows.cbegin(), rows.cend());
        }

        template <typename CHILDREN>
        void append_rebuild_values(
            array& destination,
            std::span<const rebuild_value> values,
            std::span<const dynamic_value> inserted,
            std::span<const std::size_t> inserted_children,
            const CHILDREN& old_children
        )
        {
            // View over the old child the rows are read from, and the filler row reused by every
            // filler run of that child (interleaved filler rows would otherwise be re-sliced once
            // per row).
            std::optional<array> old_child_view;
            std::optional<array> filler_row;
 
            for (std::size_t index = 0; index < values.size();)
            {
                const auto child_index = child_of(values[index], inserted_children);
                const bool is_inserted = values[index].inserted;
                const bool force_null = values[index].force_null;
                const auto row = values[index].row;
 
                // A run repeats the same row (filler, inserted value) or reads consecutive rows.
                std::size_t count = 1;
                while (index + count < values.size())
                {
                    const auto& next = values[index + count];
                    if (child_of(next, inserted_children) != child_index || next.inserted != is_inserted
                        || next.force_null != force_null
                        || next.row != (force_null || is_inserted ? row : row + count))
                    {
                        break;
                    }
                    ++count;
                }
 
                if (is_inserted)
                {
                    const array inserted_row = array_make_from_element(inserted[row]);
                    destination.insert(destination.cend(), inserted_row.cbegin(), inserted_row.cend(), count);
                    index += count;
                    continue;
                }
 
                if (!old_child_view.has_value())
                {
                    old_child_view = array{old_children[child_index]->get_arrow_proxy().view()};
                }
 
                if (force_null)
                {
                    if (!filler_row.has_value())
                    {
                        filler_row = make_filler_row(*old_child_view, row);
                    }
                    destination.insert(destination.cend(), filler_row->cbegin(), filler_row->cend(), count);
                }
                else
                {
                    append_old_child_values(destination, *old_child_view, row, count);
                }
                index += count;
            }
        }
    }
 
    template <class DERIVED>
    auto union_array_crtp_base<DERIVED>::resolve_inserted_children(
        std::span<const array_traits::value_type> inserted,
        std::vector<union_child_schema>& child_schemas,
        std::vector<std::uint8_t>& child_type_ids,
        std::array<bool, TYPE_ID_MAP_SIZE>& used_type_ids,
        std::vector<array>& new_child_templates
    ) -> std::vector<std::size_t>
    {
        std::vector<std::size_t> inserted_children;
        inserted_children.reserve(inserted.size());
        for (const auto& value : inserted)
        {
            auto child = array_make_from_element(value);
            const union_child_schema schema{
                child.data_type(),
                std::string(detail::array_access::get_arrow_proxy(child).format())
            };
            const auto existing = std::ranges::find(child_schemas, schema);
            if (existing != child_schemas.end())
            {
                inserted_children.push_back(static_cast<std::size_t>(existing - child_schemas.begin()));
                continue;
            }
 
            SPARROW_ASSERT_TRUE(child_schemas.size() < TYPE_ID_MAP_SIZE);
            const auto type_id = detail::union_rebuild::next_free_type_id(used_type_ids);
            SPARROW_ASSERT_TRUE(type_id < TYPE_ID_MAP_SIZE);
            used_type_ids[type_id] = true;
 
            inserted_children.push_back(child_schemas.size());
            child_schemas.push_back(schema);
            child_type_ids.push_back(static_cast<std::uint8_t>(type_id));
            new_child_templates.push_back(std::move(child));
        }
        return inserted_children;
    }
 
    template <class DERIVED>
    auto union_array_crtp_base<DERIVED>::make_union_fillers(
        size_type old_child_count,
        const std::vector<union_child_schema>& child_schemas,
        std::vector<std::size_t>& inserted_children,
        std::vector<array_traits::value_type>& inserted
    ) const -> std::vector<detail::union_rebuild_value>
    {
        std::vector<detail::union_rebuild_value> fillers(child_schemas.size());
        for (std::size_t child_index = 0; child_index < fillers.size(); ++child_index)
        {
            if (child_index < old_child_count && array_size(*m_children[child_index]) > 0)
            {
                fillers[child_index] = detail::union_rebuild_value{
                    .child = child_index,
                    .row = 0,
                    .force_null = true
                };
                continue;
            }
 
            // A child created for an inserted value has no rows yet: derive its null from the
            // value that created it, an empty old child from its own default element.
            array_traits::value_type default_value;
            if (child_index < old_child_count)
            {
                default_value = array_default_value(*m_children[child_index]);
            }
            else
            {
                const auto creator = std::ranges::find(inserted_children, child_index);
                SPARROW_ASSERT_TRUE(creator != inserted_children.end());
                default_value = inserted[static_cast<std::size_t>(creator - inserted_children.begin())];
            }
            inserted.push_back(detail::union_rebuild::make_null_like(default_value));
            // Keep the two parallel lists in sync: entries refer to the filler by inserted value
            // index, and their child index is read back from inserted_children.
            inserted_children.push_back(child_index);
            fillers[child_index] = detail::union_rebuild_value{.row = inserted.size() - 1, .inserted = true};
        }
        return fillers;
    }
 
    template <class DERIVED>
    auto union_array_crtp_base<DERIVED>::make_rebuild_payload(
        std::span<const detail::union_rebuild_value> values,
        std::span<const std::size_t> inserted_children,
        std::span<const std::uint8_t> child_type_ids,
        std::span<const detail::union_rebuild_value> fillers
    ) const -> detail::union_rebuild_payload
    {
        if constexpr (is_dense_union_array_v<DERIVED>)
        {
            return detail::union_rebuild::make_dense_rebuild_payload(values, inserted_children, child_type_ids);
        }
        else
        {
            return detail::union_rebuild::make_sparse_rebuild_payload(
                values,
                inserted_children,
                child_type_ids,
                fillers
            );
        }
    }
 
    template <class DERIVED>
    auto union_array_crtp_base<DERIVED>::build_rebuilt_children(
        const detail::union_rebuild_payload& payload,
        std::span<const array_traits::value_type> inserted,
        std::span<const std::size_t> inserted_children,
        const children_type& old_children,
        std::vector<array>& new_child_templates
    ) const -> std::vector<array>
    {
        const auto old_child_count = old_children.size();
        std::vector<array> new_children;
        new_children.reserve(payload.child_values.size());
        for (std::size_t child_index = 0; child_index < payload.child_values.size(); ++child_index)
        {
            array child = child_index < old_child_count
                              ? detail::union_rebuild::make_empty_child(*old_children[child_index])
                              : std::move(new_child_templates[child_index - old_child_count]);
            auto& values_for_child = payload.child_values[child_index];
            if (child.data_type() == data_type::NA)
            {
                const auto& child_proxy = detail::array_access::get_arrow_proxy(child);
                child = detail::union_rebuild::make_null_child(values_for_child.size(), child_proxy);
            }
            else
            {
                child.erase(child.cbegin(), child.cend());
                detail::union_rebuild::append_rebuild_values(
                    child,
                    values_for_child,
                    inserted,
                    inserted_children,
                    old_children
                );
            }
            new_children.push_back(std::move(child));
        }
        return new_children;
    }
 
    template <class DERIVED>
    auto union_array_crtp_base<DERIVED>::create_rebuild_proxy(
        std::vector<array> children,
        detail::union_rebuild_payload&& payload,
        std::span<const std::uint8_t> child_type_ids
    ) const -> arrow_proxy
    {
        auto metadata = m_proxy.metadata();
        std::optional<std::vector<std::uint8_t>> type_id_mapping{
            std::vector<std::uint8_t>{child_type_ids.begin(), child_type_ids.end()}
        };
        if constexpr (is_dense_union_array_v<DERIVED>)
        {
            return DERIVED::create_proxy(
                std::move(children),
                std::move(payload.type_ids),
                std::move(payload.offsets),
                std::move(type_id_mapping),
                m_proxy.name(),
                std::move(metadata)
            );
        }
        else
        {
            return DERIVED::create_proxy(
                std::move(children),
                std::move(payload.type_ids),
                std::move(type_id_mapping),
                m_proxy.name(),
                std::move(metadata)
            );
        }
    }
 
    template <class DERIVED>
    auto union_array_crtp_base<DERIVED>::adopt_rebuilt_array(
        arrow_proxy replacement,
        std::vector<std::uint8_t> child_type_ids,
        size_type return_index
    ) -> iterator
    {
        m_proxy = std::move(replacement);
        p_type_ids = reinterpret_cast<std::uint8_t*>(m_proxy.buffers()[0].data());
        m_children = make_children(m_proxy);
        m_child_type_ids = std::move(child_type_ids);
        m_type_id_map = make_type_id_map(m_child_type_ids);
        if constexpr (is_dense_union_array_v<DERIVED>)
        {
            this->derived_cast().p_offsets = reinterpret_cast<std::int32_t*>(m_proxy.buffers()[1].data());
        }
        return iterator(functor_type{&this->derived_cast()}, return_index);
    }
 
    template <class DERIVED>
    auto union_array_crtp_base<DERIVED>::rebuild_values(
        std::vector<detail::union_rebuild_value> values,
        std::vector<array_traits::value_type> inserted,
        size_type return_index
    ) -> iterator
    {
        const auto old_child_count = m_children.size();
 
        std::vector<std::uint8_t> child_type_ids;
        std::array<bool, TYPE_ID_MAP_SIZE> used_type_ids{};
        std::vector<union_child_schema> child_schemas = make_child_schemas(child_type_ids, used_type_ids);
 
        // Inserted values are resolved once each, however often the plan refers to them.
        std::vector<array> new_child_templates;
        std::vector<std::size_t> inserted_children = resolve_inserted_children(
            inserted,
            child_schemas,
            child_type_ids,
            used_type_ids,
            new_child_templates
        );
 
        std::vector<detail::union_rebuild_value> fillers;
        if constexpr (!is_dense_union_array_v<DERIVED>)
        {
            fillers = make_union_fillers(old_child_count, child_schemas, inserted_children, inserted);
        }
 
        detail::union_rebuild_payload payload = make_rebuild_payload(
            std::span<const detail::union_rebuild_value>{values},
            std::span<const std::size_t>{inserted_children},
            child_type_ids,
            std::span<const detail::union_rebuild_value>{fillers}
        );
 
        std::vector<array> new_children = build_rebuilt_children(
            payload,
            inserted,
            inserted_children,
            m_children,
            new_child_templates
        );
 
        auto replacement = create_rebuild_proxy(std::move(new_children), std::move(payload), child_type_ids);
        return adopt_rebuilt_array(std::move(replacement), std::move(child_type_ids), return_index);
    }
 
    template <class DERIVED>
    auto union_array_crtp_base<DERIVED>::insert(const_iterator pos, const_reference value, size_type count)
        -> iterator
    {
        // A no-op insertion must not require offset zero (see insert_materialized).
        if (count == 0)
        {
            return iterator_at(static_cast<size_type>(pos - cbegin()));
        }
        return insert(pos, array_traits::value_type(value), count);
    }
 
    template <class DERIVED>
    auto union_array_crtp_base<DERIVED>::insert(
        const_iterator pos,
        const array_traits::value_type& value,
        size_type count
    ) -> iterator
    {
        // A no-op insertion must not require offset zero (see insert_materialized).
        if (count == 0)
        {
            return iterator_at(static_cast<size_type>(pos - cbegin()));
        }
        return insert_materialized(pos, std::vector<array_traits::value_type>{value}, count);
    }
 
    template <class DERIVED>
    auto union_array_crtp_base<DERIVED>::insert_materialized(
        const_iterator pos,
        std::vector<array_traits::value_type> inserted,
        size_type count
    ) -> iterator
    {
        const auto pos_index = static_cast<size_type>(pos - cbegin());
        if (inserted.empty())
        {
            return iterator_at(pos_index);
        }
 
        SPARROW_ASSERT_TRUE(count > 0);
        SPARROW_ASSERT_TRUE(m_proxy.offset() == 0);
        SPARROW_ASSERT_TRUE(pos_index <= size());
 
        const auto old_size = size();
        const auto value_count = static_cast<size_type>(inserted.size());
        std::vector<detail::union_rebuild_value> entries;
        entries.reserve(old_size + value_count * count);
 
        append_rebuild_entries(entries, 0, pos_index);
        for (size_type repetition = 0; repetition < count; ++repetition)
        {
            for (size_type value_index = 0; value_index < value_count; ++value_index)
            {
                entries.push_back(detail::union_rebuild_value{
                    .row = value_index,
                    .inserted = true
                });
            }
        }
        append_rebuild_entries(entries, pos_index, old_size);
        return this->derived_cast().rebuild_in_place(std::move(entries), std::move(inserted), pos_index);
    }
 
    /************************************
     * Union array shared implementation *
     ************************************/
 
    namespace detail
    {
        template <input_metadata_container METADATA_RANGE = std::vector<metadata_pair>>
        arrow_proxy create_union_proxy_impl(
            std::vector<array>&& children,
            std::vector<buffer<std::uint8_t>>&& buffers,
            std::size_t size,
            std::string&& format,
            std::optional<std::string_view> name,
            std::optional<METADATA_RANGE> metadata
        )
        {
            const auto n_children = children.size();
            ArrowSchema** child_schemas = n_children == 0 ? nullptr : new ArrowSchema*[n_children];
            ArrowArray** child_arrays = n_children == 0 ? nullptr : new ArrowArray*[n_children];
 
            for (std::size_t i = 0; i < n_children; ++i)
            {
                auto& child = children[i];
                auto [flat_arr, flat_schema] = extract_arrow_structures(std::move(child));
                child_arrays[i] = new ArrowArray(std::move(flat_arr));
                child_schemas[i] = new ArrowSchema(std::move(flat_schema));
            }
 
            const bool is_nullable = n_children == 0
                                     || std::all_of(
                                         child_schemas,
                                         child_schemas + n_children,
                                         [](const ArrowSchema* schema)
                                         {
                                             return to_set_of_ArrowFlags(schema->flags).contains(
                                                 ArrowFlag::NULLABLE
                                             );
                                         }
                                     );
 
            const std::optional<std::unordered_set<sparrow::ArrowFlag>>
                flags = is_nullable
                            ? std::make_optional(std::unordered_set<sparrow::ArrowFlag>{ArrowFlag::NULLABLE})
                            : std::nullopt;
 
            ArrowSchema schema = make_arrow_schema(
                std::move(format),
                std::move(name),                      // name
                std::move(metadata),                  // metadata
                flags,                                // flags,
                child_schemas,                        // children
                repeat_view<bool>(true, n_children),  // children_ownership
                nullptr,                              // dictionary,
                true                                  // dictionary ownership
            );
 
            ArrowArray arr = make_arrow_array(
                static_cast<std::int64_t>(size),  // length
                0,                                // null_count: always 0 as the nullability is in children
                0,                                // offset
                std::move(buffers),
                child_arrays,                         // children
                repeat_view<bool>(true, n_children),  // children_ownership
                nullptr,                              // dictionary
                true
            );
 
            return arrow_proxy{std::move(arr), std::move(schema)};
        }
    }
 
    /************************************
     * dense_union_array implementation *
     ************************************/
 
    template <std::ranges::input_range TYPE_MAPPING, input_metadata_container METADATA_RANGE>
        requires(std::convertible_to<std::ranges::range_value_t<TYPE_MAPPING>, std::uint8_t>)
    auto dense_union_array::create_proxy(
        std::vector<array>&& children,
        type_id_buffer_type&& element_type,
        offset_buffer_type&& offsets,
        std::optional<TYPE_MAPPING>&& child_index_to_type_id,
        std::optional<std::string_view> name,
        std::optional<METADATA_RANGE> metadata
    ) -> arrow_proxy
    {
        SPARROW_ASSERT_TRUE(element_type.size() == offsets.size());
        const auto n_children = children.size();
 
        std::string format = make_format_string(true /*dense union*/, n_children, child_index_to_type_id);
 
        return create_proxy_impl(
            std::move(children),
            std::move(element_type),
            std::move(offsets),
            std::move(format),
            std::move(name),
            std::move(metadata)
        );
    }
 
    template <input_metadata_container METADATA_RANGE>
    auto dense_union_array::create_proxy_impl(
        std::vector<array>&& children,
        type_id_buffer_type&& element_type,
        offset_buffer_type&& offsets,
        std::string&& format,
        std::optional<std::string_view> name,
        std::optional<METADATA_RANGE> metadata
    ) -> arrow_proxy
    {
        SPARROW_ASSERT_TRUE(element_type.size() == offsets.size());
        const auto size = element_type.size();
 
        std::vector<buffer<std::uint8_t>> arr_buffs;
        arr_buffs.reserve(2);
        arr_buffs.emplace_back(std::move(element_type).extract_storage());
        arr_buffs.emplace_back(std::move(offsets).extract_storage());
 
        return detail::create_union_proxy_impl(
            std::move(children),
            std::move(arr_buffs),
            size,
            std::move(format),
            std::move(name),
            std::move(metadata)
        );
    }
 
    /*************************************
     * sparse_union_array implementation *
     *************************************/
 
    template <std::ranges::input_range TYPE_MAPPING, input_metadata_container METADATA_RANGE>
        requires(std::convertible_to<std::ranges::range_value_t<TYPE_MAPPING>, std::uint8_t>)
    auto sparse_union_array::create_proxy(
        std::vector<array>&& children,
        type_id_buffer_type&& element_type,
        std::optional<TYPE_MAPPING>&& child_index_to_type_id,
        std::optional<std::string_view> name,
        std::optional<METADATA_RANGE> metadata
    ) -> arrow_proxy
    {
        const auto n_children = children.size();
        if (child_index_to_type_id.has_value())
        {
            SPARROW_ASSERT_TRUE((*child_index_to_type_id).size() == n_children);
        }
 
        std::string format = make_format_string(false /*is dense union*/, n_children, child_index_to_type_id);
 
        return create_proxy_impl(
            std::move(children),
            std::move(element_type),
            std::move(format),
            std::move(name),
            std::move(metadata)
        );
    }
 
    template <input_metadata_container METADATA_RANGE>
    auto sparse_union_array::create_proxy_impl(
        std::vector<array>&& children,
        type_id_buffer_type&& element_type,
        std::string&& format,
        std::optional<std::string_view> name,
        std::optional<METADATA_RANGE> metadata
    ) -> arrow_proxy
    {
        for (const auto& child : children)
        {
            SPARROW_ASSERT_TRUE(child.size() == element_type.size());
        }
        const auto size = element_type.size();
 
        std::vector<buffer<std::uint8_t>> arr_buffs;
        arr_buffs.reserve(1);
        arr_buffs.emplace_back(std::move(element_type).extract_storage());
 
        return detail::create_union_proxy_impl(
            std::move(children),
            std::move(arr_buffs),
            size,
            std::move(format),
            std::move(name),
            std::move(metadata)
        );
    }
}
 
#if defined(__cpp_lib_format)

template <typename U>
    requires std::derived_from<U, sparrow::union_array_crtp_base<U>>
struct std::formatter<U>
{
    constexpr auto parse(std::format_parse_context& ctx)
    {
        return ctx.begin();  // Simple implementation
    }

    auto format(const U& ar, std::format_context& ctx) const
    {
        if constexpr (std::is_same_v<U, sparrow::dense_union_array>)
        {
            std::format_to(ctx.out(), "DenseUnion");
        }
        else if constexpr (std::is_same_v<U, sparrow::sparse_union_array>)
        {
            std::format_to(ctx.out(), "SparseUnion");
        }
        else
        {
            static_assert(sparrow::mpl::dependent_false<U>::value, "Unknown union array type");
            sparrow::mpl::unreachable();
        }
        const auto& proxy = ar.get_arrow_proxy();
        std::format_to(ctx.out(), " [name={} | size={}] <", proxy.name().value_or("nullptr"), proxy.length());
 
        std::for_each(
            ar.cbegin(),
            std::prev(ar.cend()),
            [&ctx](const auto& value)
            {
                std::format_to(ctx.out(), "{}, ", value);
            }
        );
 
        return std::format_to(ctx.out(), "{}>", ar.back());
    }
};
 
namespace sparrow
{
    template <typename U>
        requires std::derived_from<U, union_array_crtp_base<U>>
    std::ostream& operator<<(std::ostream& os, const U& value)
    {
        os << std::format("{}", value);
        return os;
    }
}
 
#endif