serialize.hpp

Go to the documentation of this file.

#pragma once
 
#include <ranges>
 
#include <sparrow/record_batch.hpp>
#include "sparrow_ipc/any_output_stream.hpp"
#include "sparrow_ipc/compression.hpp"
#include "sparrow_ipc/config/config.hpp"
#include "sparrow_ipc/magic_values.hpp"
#include "sparrow_ipc/serialize_utils.hpp"
#include "sparrow_ipc/utils.hpp"
 
namespace sparrow_ipc
{
    struct serialized_record_batch_info
    {
        int32_t metadata_length; 
        int64_t body_length;     
    };

    template <std::ranges::input_range R>
        requires std::same_as<std::ranges::range_value_t<R>, sparrow::record_batch>
    void serialize_record_batches_to_ipc_stream(const R& record_batches, any_output_stream& stream,
                                                std::optional<CompressionType> compression,
                                                std::optional<std::reference_wrapper<CompressionCache>> cache)
    {
        if (record_batches.empty())
        {
            return;
        }
 
        if (!utils::check_record_batches_consistency(record_batches))
        {
            throw std::invalid_argument(
                "All record batches must have the same schema to be serialized together."
            );
        }
        serialize_schema_message(record_batches[0], stream);
        for (const auto& rb : record_batches)
        {
            serialize_record_batch(rb, stream, compression, cache);
        }
        stream.write(end_of_stream);
    }

 
    SPARROW_IPC_API serialized_record_batch_info
    serialize_record_batch(const sparrow::record_batch& record_batch,
                           any_output_stream& stream,
                           std::optional<CompressionType> compression,
                           std::optional<std::reference_wrapper<CompressionCache>> cache);
    
    SPARROW_IPC_API void
    serialize_schema_message(const sparrow::record_batch& record_batch, any_output_stream& stream);
}