wip

Author: Alex-PLACET

include/sparrow_ipc/arrow_interface/arrow_array_schema_common_release.hpp

@@ -28,6 +28,8 @@ namespace sparrow_ipc
         }
         SPARROW_ASSERT_TRUE(t.private_data != nullptr);
         const auto private_data = static_cast<const private_data_type*>(t.private_data);
+        delete private_data;
+        t.private_data = nullptr;
 
         if (t.dictionary)
         {

include/sparrow_ipc/config/sparrow_ipc_version.hpp

@@ -6,7 +6,7 @@ namespace sparrow_ipc
     constexpr int SPARROW_IPC_VERSION_MINOR = 1;
     constexpr int SPARROW_IPC_VERSION_PATCH = 0;
 
-    constexpr int SPARROW_IPC_BINARY_CURRENT = 9;
+    constexpr int SPARROW_IPC_BINARY_CURRENT = 1;
     constexpr int SPARROW_IPC_BINARY_REVISION = 0;
     constexpr int SPARROW_IPC_BINARY_AGE = 0;
 }

include/sparrow_ipc/deserialize.hpp

@@ -5,13 +5,31 @@
 
 #include <sparrow/record_batch.hpp>
 
-#include "config/config.hpp"
 #include "Message_generated.h"
+#include "sparrow_ipc/config/config.hpp"
 #include "sparrow_ipc/encapsulated_message.hpp"
-#include "SparseTensor_generated.h"
 
 namespace sparrow_ipc
 {
+    /**
+     * @brief Deserializes a schema message from Arrow IPC format data.
+     *
+     * This function parses an Arrow IPC schema message from a byte buffer, extracting
+     * the field name and custom metadata from the first (and expected only) field in the schema.
+     *
+     * @param data A span containing the raw byte data to deserialize from
+     * @param current_offset Reference to the current position in the data buffer, which will be
+     *                      updated to point past the processed schema message
+     * @param name Optional output parameter that will contain the field name if present
+     * @param metadata Optional output parameter that will contain the custom metadata
+     *                key-value pairs if present
+     *
+     * @throws std::runtime_error If the message is not a Schema message type
+     * @throws std::runtime_error If the schema does not contain exactly one field
+     *
+     * @note This function expects the data to start with a 4-byte length prefix followed
+     *       by the FlatBuffer schema message data
+     */
     SPARROW_IPC_API void deserialize_schema_message(
         std::span<const uint8_t> data,
         size_t& current_offset,
@@ -21,6 +39,25 @@ namespace sparrow_ipc
     [[nodiscard]] SPARROW_IPC_API const org::apache::arrow::flatbuf::RecordBatch*
     deserialize_record_batch_message(std::span<const uint8_t> data, size_t& current_offset);
 
+    /**
+     * @brief Deserializes an Arrow IPC stream from binary data into a vector of record batches.
+     *
+     * This function processes an Arrow IPC stream format, extracting schema information
+     * and record batch data. It handles encapsulated messages sequentially, first expecting
+     * a Schema message followed by one or more RecordBatch messages.
+     *
+     * @param data A span of bytes containing the serialized Arrow IPC stream data
+     *
+     * @return std::vector<sparrow::record_batch> A vector containing all deserialized record batches
+     *
+     * @throws std::runtime_error If:
+     *         - A RecordBatch message is encountered before a Schema message
+     *         - A RecordBatch message header is missing or invalid
+     *         - Unsupported message types are encountered (Tensor, DictionaryBatch, SparseTensor)
+     *         - An unknown message header type is encountered
+     *
+     * @note The function processes messages until an end-of-stream marker is detected
+     */
     [[nodiscard]] SPARROW_IPC_API std::vector<sparrow::record_batch>
     deserialize_stream(std::span<const uint8_t> data);
 }

include/sparrow_ipc/deserialize_primitive_array.hpp

@@ -40,6 +40,10 @@ namespace sparrow_ipc
             buffer_index++
         );
         const auto primitive_buffer_metadata = record_batch.buffers()->Get(buffer_index++);
+        if (body.size() < (primitive_buffer_metadata->offset() + primitive_buffer_metadata->length()))
+        {
+            throw std::runtime_error("Primitive buffer exceeds body size");
+        }
         auto primitives_ptr = const_cast<uint8_t*>(body.data() + primitive_buffer_metadata->offset());
         std::vector<std::uint8_t*> buffers = {bitmap_ptr, primitives_ptr};
         ArrowArray array = make_non_owning_arrow_array(

include/sparrow_ipc/deserialize_utils.hpp

@@ -1,36 +1,33 @@
 #pragma once
 
 #include <span>
+#include <utility>
 
 #include <sparrow/buffer/dynamic_bitset/dynamic_bitset_view.hpp>
 #include <sparrow/u8_buffer.hpp>
-#include <utility>
 
 #include "Message_generated.h"
 #include "Schema_generated.h"
 
 namespace sparrow_ipc::utils
 {
-    template <typename T>
-    [[nodiscard]] sparrow::u8_buffer<T> message_buffer_to_u8buffer(
-        const org::apache::arrow::flatbuf::RecordBatch* record_batch,
-        std::span<const uint8_t> body,
-        size_t index
-    )
-    {
-        const auto buffer_metadata = record_batch->buffers()->Get(index);
-        auto ptr = const_cast<uint8_t*>(body.data() + buffer_metadata->offset());
-        auto casted_ptr = reinterpret_cast<T*>(ptr);
-        const std::size_t count = static_cast<std::size_t>(buffer_metadata->length() / sizeof(T));
-        return sparrow::u8_buffer<T>{casted_ptr, count};
-    }
-
-    [[nodiscard]] const sparrow::dynamic_bitset_view<const std::uint8_t> message_buffer_to_validity_bitmap(
-        const org::apache::arrow::flatbuf::RecordBatch* record_batch,
-        std::span<const uint8_t> body,
-        size_t index
-    );
-
+    /**
+     * @brief Extracts bitmap pointer and null count from a RecordBatch buffer.
+     *
+     * This function retrieves a bitmap buffer from the specified index in the RecordBatch's
+     * buffer list and calculates the number of null values represented by the bitmap.
+     *
+     * @param record_batch The Arrow RecordBatch containing buffer metadata
+     * @param body The raw buffer data as a byte span
+     * @param index The index of the bitmap buffer in the RecordBatch's buffer list
+     *
+     * @return A pair containing:
+     *         - First: Pointer to the bitmap data (nullptr if buffer is empty)
+     *         - Second: Count of null values in the bitmap (0 if buffer is empty)
+     *
+     * @note If the bitmap buffer has zero length, returns {nullptr, 0}
+     * @note The returned pointer is a non-const cast of the original const data
+     */
     [[nodiscard]] std::pair<std::uint8_t*, int64_t> get_bitmap_pointer_and_null_count(
         const org::apache::arrow::flatbuf::RecordBatch& record_batch,
         std::span<const uint8_t> body,

include/sparrow_ipc/deserialize_variable_size_binary_array.hpp

@@ -36,9 +36,18 @@ namespace sparrow_ipc
             body,
             buffer_index++
         );
+
         const auto offset_metadata = record_batch.buffers()->Get(buffer_index++);
+        if ((offset_metadata->offset() + offset_metadata->length()) > body.size())
+        {
+            throw std::runtime_error("Offset buffer exceeds body size");
+        }
         auto offset_ptr = const_cast<uint8_t*>(body.data() + offset_metadata->offset());
         const auto buffer_metadata = record_batch.buffers()->Get(buffer_index++);
+        if ((buffer_metadata->offset() + buffer_metadata->length()) > body.size())
+        {
+            throw std::runtime_error("Data buffer exceeds body size");
+        }
         auto buffer_ptr = const_cast<uint8_t*>(body.data() + buffer_metadata->offset());
         std::vector<std::uint8_t*> buffers = {bitmap_ptr, offset_ptr, buffer_ptr};
         ArrowArray array = make_non_owning_arrow_array(

include/sparrow_ipc/magic_values.hpp

@@ -6,15 +6,25 @@
 
 namespace sparrow_ipc
 {
+
+    /**
+     * Continuation value defined in the Arrow IPC specification:
+     * https://arrow.apache.org/docs/format/Columnar.html#encapsulated-message-format
+     */
     constexpr std::array<uint8_t, 4> continuation = {0xFF, 0xFF, 0xFF, 0xFF};
 
+    /**
+     * End-of-stream marker defined in the Arrow IPC specification:
+     * https://arrow.apache.org/docs/format/Columnar.html#ipc-streaming-format
+     */
     constexpr std::array<uint8_t, 8> end_of_stream = {0xFF, 0xFF, 0xFF, 0xFF, 0x00, 0x00, 0x00, 0x00};
 
     template <std::ranges::input_range R>
     [[nodiscard]] bool is_continuation(const R& buf)
     {
         return std::ranges::equal(buf, continuation);
     }
+
     template <std::ranges::input_range R>
     [[nodiscard]] bool is_end_of_stream(const R& buf)
     {

include/sparrow_ipc/metadata.hpp

@@ -10,6 +10,20 @@
 
 namespace sparrow_ipc
 {
+    /**
+     * @brief Converts FlatBuffers metadata to Sparrow metadata format.
+     *
+     * This function takes a FlatBuffers vector containing key-value pairs from Apache Arrow
+     * format and converts them into a vector of Sparrow metadata pairs. Each key-value pair
+     * from the FlatBuffers structure is extracted and stored as a sparrow::metadata_pair.
+     *
+     * @param metadata A FlatBuffers vector containing KeyValue pairs from Apache Arrow format
+     * @return std::vector<sparrow::metadata_pair> A vector of Sparrow metadata pairs containing
+     *         the converted key-value data
+     *
+     * @note The function reserves space in the output vector to match the input size for
+     *       optimal memory allocation performance.
+     */
     std::vector<sparrow::metadata_pair> to_sparrow_metadata(
         const ::flatbuffers::Vector<::flatbuffers::Offset<org::apache::arrow::flatbuf::KeyValue>>& metadata
     );

src/arrow_interface/arrow_array.cpp

@@ -16,12 +16,6 @@ namespace sparrow_ipc
         SPARROW_ASSERT_TRUE(array->release == std::addressof(release_non_owning_arrow_array))
 
         release_common_non_owning_arrow(*array);
-        if (array->private_data != nullptr)
-        {
-            const auto private_data = static_cast<non_owning_arrow_array_private_data*>(array->private_data);
-            delete private_data;
-            array->private_data = nullptr;
-        }
         array->buffers = nullptr;  // The buffers were deleted with the private data
     }
 

src/arrow_interface/arrow_schema.cpp

@@ -9,12 +9,6 @@ namespace sparrow_ipc
         SPARROW_ASSERT_FALSE(schema == nullptr);
         SPARROW_ASSERT_TRUE(schema->release == std::addressof(release_non_owning_arrow_schema));
         release_common_non_owning_arrow(*schema);
-        if (schema->private_data != nullptr)
-        {
-            const auto private_data = static_cast<non_owning_arrow_schema_private_data*>(schema->private_data);
-            delete private_data;
-            schema->private_data = nullptr;
-        }
         *schema = {};
     }
 }