Factorize more

Author: Hind-M

include/sparrow_ipc/compression.hpp

@@ -18,6 +18,6 @@ namespace sparrow_ipc
 
 //     CompressionType to_compression_type(org::apache::arrow::flatbuf::CompressionType compression_type);
 
-    std::vector<std::uint8_t> compress(org::apache::arrow::flatbuf::CompressionType compression_type, std::span<const std::uint8_t> data);
-    std::vector<std::uint8_t> decompress(org::apache::arrow::flatbuf::CompressionType compression_type, std::span<const std::uint8_t> data);
+    std::vector<std::uint8_t> compress(const org::apache::arrow::flatbuf::CompressionType compression_type, std::span<const std::uint8_t> data);
+    std::vector<std::uint8_t> decompress(const org::apache::arrow::flatbuf::CompressionType compression_type, std::span<const std::uint8_t> data);
 }

include/sparrow_ipc/serialize_utils.hpp

@@ -117,7 +117,23 @@ namespace sparrow_ipc
         return total_size;
     }
 
-    void fill_body_and_get_buffers_compressed(const sparrow::arrow_proxy& arrow_proxy, std::vector<uint8_t>& body, std::vector<org::apache::arrow::flatbuf::Buffer>& flatbuf_buffers, int64_t& offset, org::apache::arrow::flatbuf::CompressionType compression_type);
+    /**
+     * @brief Generates the compressed message body and buffer metadata for a record batch.
+     *
+     * This function traverses the record batch, compresses each buffer using the specified
+     * compression algorithm, and constructs the message body. For each compressed buffer,
+     * it is prefixed by its 8-byte uncompressed size. Padding is added after each
+     * compressed buffer to ensure 8-byte alignment.
+     *
+     * @param record_batch The record batch to serialize.
+     * @param compression_type The compression algorithm to use (e.g., LZ4_FRAME, ZSTD).
+     * @return A std::pair containing:
+     *         - first: A vector of bytes representing the complete compressed message body.
+     *         - second: A vector of FlatBuffer Buffer objects describing the offset and
+     *                   size of each buffer within the compressed body.
+     */
+    [[nodiscard]] SPARROW_IPC_API std::pair<std::vector<uint8_t>, std::vector<org::apache::arrow::flatbuf::Buffer>>
+    generate_compressed_body_and_buffers(const sparrow::record_batch& record_batch, const org::apache::arrow::flatbuf::CompressionType compression_type);
 
     /**
      * @brief Fills the body vector with serialized data from an arrow proxy and its children.

src/compression.cpp

@@ -20,7 +20,7 @@ namespace sparrow_ipc
 //         }
 //     }
 
-    std::vector<std::uint8_t> compress(org::apache::arrow::flatbuf::CompressionType compression_type, std::span<const std::uint8_t> data)
+    std::vector<std::uint8_t> compress(const org::apache::arrow::flatbuf::CompressionType compression_type, std::span<const std::uint8_t> data)
     {
         if (data.empty())
         {
@@ -30,7 +30,7 @@ namespace sparrow_ipc
         {
             case org::apache::arrow::flatbuf::CompressionType::LZ4_FRAME:
             {
-                std::int64_t uncompressed_size = data.size();
+                const std::int64_t uncompressed_size = data.size();
                 const size_t max_compressed_size = LZ4F_compressFrameBound(uncompressed_size, nullptr);
                 std::vector<std::uint8_t> compressed_data(max_compressed_size);
                 const size_t compressed_size = LZ4F_compressFrame(compressed_data.data(), max_compressed_size, data.data(), uncompressed_size, nullptr);
@@ -46,7 +46,7 @@ namespace sparrow_ipc
         }
     }
 
-    std::vector<std::uint8_t> decompress(org::apache::arrow::flatbuf::CompressionType compression_type, std::span<const std::uint8_t> data)
+    std::vector<std::uint8_t> decompress(const org::apache::arrow::flatbuf::CompressionType compression_type, std::span<const std::uint8_t> data)
     {
         if (data.empty())
         {

src/serialize_utils.cpp

@@ -1,3 +1,4 @@
+#include <tuple>
 #include <lz4frame.h>
 
 #include "sparrow_ipc/flatbuffer_utils.hpp"