Support arrow file format (#62)

Author: Alex-PLACET

CMakeLists.txt

@@ -122,6 +122,7 @@ set(SPARROW_IPC_HEADERS
     ${SPARROW_IPC_INCLUDE_DIR}/sparrow_ipc/serialize_utils.hpp
     ${SPARROW_IPC_INCLUDE_DIR}/sparrow_ipc/serialize.hpp
     ${SPARROW_IPC_INCLUDE_DIR}/sparrow_ipc/serializer.hpp
+    ${SPARROW_IPC_INCLUDE_DIR}/sparrow_ipc/stream_file_serializer.hpp
     ${SPARROW_IPC_INCLUDE_DIR}/sparrow_ipc/utils.hpp
 )
 
@@ -143,6 +144,7 @@ set(SPARROW_IPC_SRC
     ${SPARROW_IPC_SOURCE_DIR}/serialize_utils.cpp
     ${SPARROW_IPC_SOURCE_DIR}/serialize.cpp
     ${SPARROW_IPC_SOURCE_DIR}/serializer.cpp
+    ${SPARROW_IPC_SOURCE_DIR}/stream_file_serializer.cpp
     ${SPARROW_IPC_SOURCE_DIR}/utils.cpp
 )
 

include/sparrow_ipc/deserialize.hpp

@@ -2,15 +2,46 @@
 
 #include <optional>
 #include <string>
+#include <vector>
 
 #include <sparrow/record_batch.hpp>
 
 #include "Message_generated.h"
+#include "Schema_generated.h"
 #include "sparrow_ipc/config/config.hpp"
 #include "sparrow_ipc/encapsulated_message.hpp"
+#include "sparrow_ipc/metadata.hpp"
 
 namespace sparrow_ipc
 {
+    /**
+     * @brief Deserializes arrays from an Apache Arrow RecordBatch using the provided schema.
+     *
+     * This function processes each field in the schema and deserializes the corresponding
+     * data from the RecordBatch into sparrow::array objects. It handles various Arrow data
+     * types including primitive types (bool, integers, floating point), binary data, and
+     * string data with their respective size variants.
+     *
+     * @param record_batch The Apache Arrow FlatBuffer RecordBatch containing the serialized data
+     * @param schema The Apache Arrow FlatBuffer Schema defining the structure and types of the data
+     * @param encapsulated_message The message containing the binary data buffers
+     * @param field_metadata Metadata for each field
+     *
+     * @return std::vector<sparrow::array> A vector of deserialized arrays, one for each field in the schema
+     *
+     * @throws std::runtime_error If an unsupported data type, integer bit width, or floating point precision
+     * is encountered
+     *
+     * The function maintains a buffer index that is incremented as it processes each field
+     * to correctly map data buffers to their corresponding arrays.
+     */
+    [[nodiscard]] SPARROW_IPC_API std::vector<sparrow::array> get_arrays_from_record_batch(
+        const org::apache::arrow::flatbuf::RecordBatch& record_batch,
+        const org::apache::arrow::flatbuf::Schema& schema,
+        const encapsulated_message& encapsulated_message,
+        const std::vector<std::optional<std::vector<sparrow::metadata_pair>>>& field_metadata
+    );
+
     /**
      * @brief Deserializes an Arrow IPC stream from binary data into a vector of record batches.
      *

include/sparrow_ipc/flatbuffer_utils.hpp

@@ -280,7 +280,7 @@ namespace sparrow_ipc
     get_compressed_buffers(const sparrow::record_batch& record_batch, const CompressionType compression_type);
 
     /**
-     * @brief Calculates the total size of the body section for an Arrow array.
+     * @brief Calculates the total aligned size in bytes of all buffers in an Arrow array structure.
      *
      * This function recursively computes the total size needed for all buffers
      * in an Arrow array structure, including buffers from child arrays. Each

include/sparrow_ipc/magic_values.hpp

@@ -4,6 +4,7 @@
 #include <array>
 #include <cstdint>
 #include <istream>
+#include <ranges>
 
 namespace sparrow_ipc
 {
@@ -19,6 +20,19 @@ namespace sparrow_ipc
      */
     inline constexpr std::array<std::uint8_t, 8> end_of_stream = {0xFF, 0xFF, 0xFF, 0xFF, 0x00, 0x00, 0x00, 0x00};
 
+    /**
+     * Magic bytes for Arrow file format defined in the Arrow IPC specification:
+     * https://arrow.apache.org/docs/format/Columnar.html#ipc-file-format
+     * The magic string is "ARROW1" (6 bytes) followed by 2 padding bytes to reach 8-byte alignment
+     */
+    inline constexpr std::array<std::uint8_t, 6> arrow_file_magic = {'A', 'R', 'R', 'O', 'W', '1'};
+    inline constexpr std::size_t arrow_file_magic_size = arrow_file_magic.size();
+    
+    /**
+     * Magic bytes with padding for file header (8 bytes total for alignment)
+     */
+    inline constexpr std::array<std::uint8_t, 8> arrow_file_header_magic = {'A', 'R', 'R', 'O', 'W', '1', 0x00, 0x00};
+
     template <std::ranges::input_range R>
     [[nodiscard]] bool is_continuation(const R& buf)
     {
@@ -30,4 +44,15 @@ namespace sparrow_ipc
     {
         return std::ranges::equal(buf, end_of_stream);
     }
+
+    template <std::ranges::input_range R>
+    [[nodiscard]] bool is_arrow_file_magic(const R& buf)
+    {
+        if (std::ranges::size(buf) < arrow_file_magic_size)
+        {
+            return false;
+        }
+        auto buf_begin = std::ranges::begin(buf);
+        return std::equal(buf_begin, buf_begin + arrow_file_magic_size, arrow_file_magic.begin());
+    }
 }