wip

Author: Alex-PLACET

include/sparrow_ipc/serialize.hpp

@@ -10,6 +10,21 @@
 #include "sparrow_ipc/serialize_utils.hpp"
 #include "sparrow_ipc/utils.hpp"
 
+namespace sparrow_ipc
+{
+    /**
+     * @brief Information about a serialized record batch block.
+     *
+     * Contains the metadata length and body length of a serialized record batch,
+     * used for populating the Arrow IPC file format footer.
+     */
+    struct serialized_record_batch_info
+    {
+        int32_t metadata_length; ///< Length of the metadata (FlatBuffer message + padding)
+        int64_t body_length;     ///< Length of the record batch body (data buffers)
+    };
+}
+
 namespace sparrow_ipc
 {
     /**

include/sparrow_ipc/serializer.hpp

@@ -5,6 +5,7 @@
 
 #include "sparrow_ipc/any_output_stream.hpp"
 #include "sparrow_ipc/compression.hpp"
+#include "sparrow_ipc/serialize.hpp"
 #include "sparrow_ipc/serialize_utils.hpp"
 
 namespace sparrow_ipc

include/sparrow_ipc/stream_file_serializer.hpp

@@ -16,15 +16,29 @@
 
 namespace sparrow_ipc
 {
+    /**
+     * @brief Represents a block entry in the Arrow IPC file footer.
+     *
+     * Each block describes the location and size of a record batch in the file.
+     */
+    struct record_batch_block
+    {
+        int64_t offset;          ///< Offset from the start of the file to the record batch message
+        int32_t metadata_length; ///< Length of the metadata (FlatBuffer message)
+        int64_t body_length;     ///< Length of the record batch body (data buffers)
+    };
+
     /**
      * @brief Writes the Arrow IPC file footer.
      *
      * @param record_batch A record batch containing the schema for the footer
+     * @param record_batch_blocks Vector of block information for each record batch
      * @param stream The output stream to write the footer to
      * @return The size of the footer in bytes
      */
     SPARROW_IPC_API size_t write_footer(
         const sparrow::record_batch& record_batch,
+        const std::vector<record_batch_block>& record_batch_blocks,
         any_output_stream& stream
     );
 
@@ -184,7 +198,14 @@ namespace sparrow_ipc
                 {
                     throw std::invalid_argument("Record batch schema does not match file serializer schema");
                 }
-                serialize_record_batch(rb, m_stream, m_compression, compressed_buffers_cache);
+                
+                // Offset is from the start of the file to the record batch message
+                const int64_t offset = static_cast<int64_t>(m_stream.size());
+                
+                // Serialize and get block info
+                const auto info = serialize_record_batch(rb, m_stream, m_compression, compressed_buffers_cache);
+                
+                m_record_batch_blocks.push_back({offset, info.metadata_length, info.body_length});
             }
         }
 
@@ -280,6 +301,7 @@ namespace sparrow_ipc
         any_output_stream m_stream;
         bool m_ended{false};
         std::optional<CompressionType> m_compression;
+        std::vector<record_batch_block> m_record_batch_blocks;
     };
 
     /**

integration_tests/test_integration_tools.cpp

@@ -173,7 +173,7 @@ TEST_SUITE("Integration Tools Tests")
 
     TEST_CASE("json_to_stream and validate - Round-trip with validation")
     {
-        const std::filesystem::path json_file = tests_resources_files_path / "generated_binary.json";
+        const std::filesystem::path json_file = tests_resources_files_path / "generated_primitive.json";
 
         if (!std::filesystem::exists(json_file))
         {
@@ -190,11 +190,17 @@ TEST_SUITE("Integration Tools Tests")
             json_file,
             std::span<const uint8_t>(arrow_file_data)
         );
+        // write to a temp file
+        std::filesystem::path temp_file = std::filesystem::temp_directory_path() / "temp.arrow_file";
+        std::ofstream out(temp_file, std::ios::binary);
+        out.write(reinterpret_cast<const char*>(arrow_file_data.data()), arrow_file_data.size());
+        out.close();
+
         CHECK(matches);
 
         // Also verify by deserializing
-        auto batches = sparrow_ipc::deserialize_file(std::span<const uint8_t>(arrow_file_data));
-        CHECK_GT(batches.size(), 0);
+        const auto batches = sparrow_ipc::deserialize_file(std::span<const uint8_t>(arrow_file_data));
+        CHECK_EQ(batches.size(), 2);
     }
 
     TEST_CASE("compare_record_batch - Identical batches")

src/serialize.cpp

@@ -2,6 +2,7 @@
 
 #include "sparrow_ipc/serialize.hpp"
 #include "sparrow_ipc/flatbuffer_utils.hpp"
+#include "sparrow_ipc/utils.hpp"
 
 namespace sparrow_ipc
 {
@@ -23,11 +24,33 @@ namespace sparrow_ipc
         common_serialize(get_schema_message_builder(record_batch), stream);
     }
 
-    void serialize_record_batch(const sparrow::record_batch& record_batch, any_output_stream& stream,
+    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)
     {
-        common_serialize(get_record_batch_message_builder(record_batch, compression, cache), stream);
+        // Build and serialize metadata
+        flatbuffers::FlatBufferBuilder builder = get_record_batch_message_builder(record_batch, compression, cache);
+        const flatbuffers::uoffset_t flatbuffer_size = builder.GetSize();
+        
+        // Calculate metadata length: flatbuffer size + padding to 8-byte alignment
+        // Note: The metadata_length in the Arrow file footer includes the size prefix (4 bytes)
+        // but not the continuation bytes
+        const int32_t metadata_length = static_cast<int32_t>(
+            sizeof(uint32_t) + utils::align_to_8(static_cast<size_t>(flatbuffer_size))
+        );
+        
+        // Write metadata
+        common_serialize(builder, stream);
+        
+        // Track position before body to calculate body length
+        const size_t body_start = stream.size();
+        
+        // Write body
         generate_body(record_batch, stream, compression, cache);
+        
+        // Calculate body length
+        const int64_t body_length = static_cast<int64_t>(stream.size() - body_start);
+        
+        return {metadata_length, body_length};
     }
 }

src/stream_file_serializer.cpp

@@ -47,8 +47,8 @@ namespace sparrow_ipc
         // Write end-of-stream marker
         m_stream.write(end_of_stream);
 
-        // Write footer using the first record batch for schema
-        const size_t footer_size = write_footer(m_first_record_batch.value(), m_stream);
+        // Write footer using the first record batch for schema and the tracked blocks
+        const size_t footer_size = write_footer(m_first_record_batch.value(), m_record_batch_blocks, m_stream);
 
         // Write footer size (int32, little-endian)
         const int32_t footer_size_i32 = static_cast<int32_t>(footer_size);
@@ -64,6 +64,7 @@ namespace sparrow_ipc
 
     size_t write_footer(
         const sparrow::record_batch& record_batch,
+        const std::vector<record_batch_block>& record_batch_blocks,
         any_output_stream& stream
     )
     {
@@ -83,9 +84,14 @@ namespace sparrow_ipc
             std::vector<org::apache::arrow::flatbuf::Block>{}
         );
 
-        // Create record batches vector
-        std::vector<org::apache::arrow::flatbuf::Block> record_batch_blocks;
-        auto record_batches_fb = footer_builder.CreateVectorOfStructs(record_batch_blocks);
+        // Create record batches vector from tracked blocks
+        std::vector<org::apache::arrow::flatbuf::Block> fb_blocks;
+        fb_blocks.reserve(record_batch_blocks.size());
+        for (const auto& block : record_batch_blocks)
+        {
+            fb_blocks.emplace_back(block.offset, block.metadata_length, block.body_length);
+        }
+        auto record_batches_fb = footer_builder.CreateVectorOfStructs(fb_blocks);
 
         // Create footer
         auto footer = org::apache::arrow::flatbuf::CreateFooter(

tests/test_serialize_utils.cpp

@@ -7,6 +7,7 @@
 #include "sparrow_ipc/flatbuffer_utils.hpp"
 #include "sparrow_ipc/magic_values.hpp"
 #include "sparrow_ipc/memory_output_stream.hpp"
+#include "sparrow_ipc/serialize.hpp"
 #include "sparrow_ipc/serialize_utils.hpp"
 #include "sparrow_ipc/utils.hpp"
 #include "sparrow_ipc_tests_helpers.hpp"