Use std::span

Alex-PLACET · Alex-PLACET · commit e618b0b8951a · 2025-09-05T11:51:10.000+02:00
diff --git a/include/sparrow_ipc/deserialize.hpp b/include/sparrow_ipc/deserialize.hpp
@@ -13,13 +13,14 @@
 namespace sparrow_ipc
 {
     SPARROW_IPC_API void deserialize_schema_message(
-        const uint8_t* buf_ptr,
+        std::span<const uint8_t> data,
         size_t& current_offset,
         std::optional<std::string>& name,
         std::optional<std::vector<sparrow::metadata_pair>>& metadata
     );
     [[nodiscard]] SPARROW_IPC_API const org::apache::arrow::flatbuf::RecordBatch*
-    deserialize_record_batch_message(const uint8_t* buf_ptr, size_t& current_offset);
+    deserialize_record_batch_message(std::span<const uint8_t> data, size_t& current_offset);
 
-    [[nodiscard]] SPARROW_IPC_API std::vector<sparrow::record_batch> deserialize_stream(const uint8_t* buf_ptr);
+    [[nodiscard]] SPARROW_IPC_API std::vector<sparrow::record_batch>
+    deserialize_stream(std::span<const uint8_t> data);
 }
diff --git a/include/sparrow_ipc/encapsulated_message.hpp b/include/sparrow_ipc/encapsulated_message.hpp
@@ -9,7 +9,7 @@ namespace sparrow_ipc
     {
     public:
 
-        EncapsulatedMessage(const uint8_t* buf_ptr);
+        EncapsulatedMessage(std::span<const uint8_t> data);
 
         [[nodiscard]] const org::apache::arrow::flatbuf::Message* flat_buffer_message() const;
 
@@ -36,8 +36,8 @@ namespace sparrow_ipc
 
     private:
 
-        const uint8_t* m_buf_ptr;
+        std::span<const uint8_t> m_data;
     };
 
-    [[nodiscard]] EncapsulatedMessage create_encapsulated_message(const uint8_t* buf_ptr);
+    [[nodiscard]] EncapsulatedMessage create_encapsulated_message(std::span<const uint8_t> buf_ptr);
 }
diff --git a/include/sparrow_ipc/serialize_primitive_array.hpp b/include/sparrow_ipc/serialize_primitive_array.hpp
@@ -8,7 +8,6 @@
 #include "serialize.hpp"
 #include "utils.hpp"
 
-
 namespace sparrow_ipc
 {
     // TODO Use `arr` as const after fixing the issue upstream in sparrow::get_arrow_structures
@@ -62,11 +61,14 @@ namespace sparrow_ipc
         // I - Deserialize the Schema message
         std::optional<std::string> name;
         std::optional<std::vector<sparrow::metadata_pair>> metadata;
-        deserialize_schema_message(buf_ptr, current_offset, name, metadata);
+        deserialize_schema_message(std::span<const uint8_t>(buffer), current_offset, name, metadata);
 
         // II - Deserialize the RecordBatch message
         const uint32_t batch_meta_len = *(reinterpret_cast<const uint32_t*>(buf_ptr + current_offset));
-        const auto* record_batch = deserialize_record_batch_message(buf_ptr, current_offset);
+        const auto* record_batch = deserialize_record_batch_message(
+            std::span<const uint8_t>(buffer),
+            current_offset
+        );
 
         current_offset += utils::align_to_8(batch_meta_len);
         const uint8_t* body_ptr = buf_ptr + current_offset;
diff --git a/src/deserialize.cpp b/src/deserialize.cpp
@@ -11,15 +11,15 @@
 namespace sparrow_ipc
 {
     void deserialize_schema_message(
-        const uint8_t* buf_ptr,
+        std::span<const uint8_t> data,
         size_t& current_offset,
         std::optional<std::string>& name,
         std::optional<std::vector<sparrow::metadata_pair>>& metadata
     )
     {
-        const uint32_t schema_meta_len = *(reinterpret_cast<const uint32_t*>(buf_ptr + current_offset));
+        const uint32_t schema_meta_len = *(reinterpret_cast<const uint32_t*>(data.data() + current_offset));
         current_offset += sizeof(uint32_t);
-        const auto schema_message = org::apache::arrow::flatbuf::GetMessage(buf_ptr + current_offset);
+        const auto schema_message = org::apache::arrow::flatbuf::GetMessage(data.data() + current_offset);
         if (schema_message->header_type() != org::apache::arrow::flatbuf::MessageHeader::Schema)
         {
             throw std::runtime_error("Expected Schema message at the start of the buffer.");
@@ -56,10 +56,10 @@ namespace sparrow_ipc
     }
 
     const org::apache::arrow::flatbuf::RecordBatch*
-    deserialize_record_batch_message(const uint8_t* buf_ptr, size_t& current_offset)
+    deserialize_record_batch_message(std::span<const uint8_t> data, size_t& current_offset)
     {
         current_offset += sizeof(uint32_t);
-        const auto batch_message = org::apache::arrow::flatbuf::GetMessage(buf_ptr + current_offset);
+        const auto batch_message = org::apache::arrow::flatbuf::GetMessage(data.data() + current_offset);
         if (batch_message->header_type() != org::apache::arrow::flatbuf::MessageHeader::RecordBatch)
         {
             throw std::runtime_error("Expected RecordBatch message, but got a different type.");
@@ -225,7 +225,7 @@ namespace sparrow_ipc
         return arrays;
     }
 
-    std::vector<sparrow::record_batch> deserialize_stream(const uint8_t* buf_ptr)
+    std::vector<sparrow::record_batch> deserialize_stream(std::span<const uint8_t> data)
     {
         const org::apache::arrow::flatbuf::Schema* schema = nullptr;
         std::vector<sparrow::record_batch> record_batches;
@@ -234,7 +234,7 @@ namespace sparrow_ipc
         std::vector<sparrow::data_type> field_types;
         do
         {
-            const EncapsulatedMessage encapsulated_message = create_encapsulated_message(buf_ptr);
+            const EncapsulatedMessage encapsulated_message = create_encapsulated_message(data);
             const org::apache::arrow::flatbuf::Message* message = encapsulated_message.flat_buffer_message();
             switch (message->header_type())
             {
@@ -280,8 +280,8 @@ namespace sparrow_ipc
                     throw std::runtime_error("Unknown message header type.");
             }
             const size_t encapsulated_message_total_length = encapsulated_message.total_length();
-            buf_ptr += encapsulated_message_total_length;
-            if (is_end_of_stream(std::span<const uint8_t>{buf_ptr, 8}))
+            data = data.subspan(encapsulated_message_total_length);
+            if (is_end_of_stream(data.subspan(0, 8)))
             {
                 break;
             }
diff --git a/src/encapsulated_message.cpp b/src/encapsulated_message.cpp
@@ -7,21 +7,21 @@
 
 namespace sparrow_ipc
 {
-    EncapsulatedMessage::EncapsulatedMessage(const uint8_t* buf_ptr)
-        : m_buf_ptr(buf_ptr)
+    EncapsulatedMessage::EncapsulatedMessage(std::span<const uint8_t> data)
+        : m_data(data)
     {
     }
 
     const org::apache::arrow::flatbuf::Message* EncapsulatedMessage::flat_buffer_message() const
     {
-        const uint8_t* message_ptr = m_buf_ptr + (sizeof(uint32_t) * 2);  // 4 bytes continuation + 4 bytes
-                                                                          // metadata size
+        const uint8_t* message_ptr = m_data.data() + (sizeof(uint32_t) * 2);  // 4 bytes continuation + 4
+                                                                              // bytes metadata size
         return org::apache::arrow::flatbuf::GetMessage(message_ptr);
     }
 
     size_t EncapsulatedMessage::metadata_length() const
     {
-        return *(reinterpret_cast<const uint32_t*>(m_buf_ptr + sizeof(uint32_t)));
+        return *(reinterpret_cast<const uint32_t*>(m_data.data() + sizeof(uint32_t)));
     }
 
     [[nodiscard]] std::variant<
@@ -76,8 +76,7 @@ namespace sparrow_ipc
         const size_t offset = sizeof(uint32_t) * 2  // 4 bytes continuation + 4 bytes metadata size
                               + metadata_length();
         const size_t padded_offset = utils::align_to_8(offset);  // Round up to 8-byte boundary
-        const uint8_t* body_ptr = m_buf_ptr + padded_offset;
-        return {body_ptr, body_length()};
+        return m_data.subspan(padded_offset, body_length());
     }
 
     size_t EncapsulatedMessage::total_length() const
@@ -90,20 +89,20 @@ namespace sparrow_ipc
 
     std::span<const uint8_t> EncapsulatedMessage::as_span() const
     {
-        return {m_buf_ptr, total_length()};
+        return m_data;
     }
 
-    EncapsulatedMessage create_encapsulated_message(const uint8_t* buf_ptr)
+    EncapsulatedMessage create_encapsulated_message(std::span<const uint8_t> data)
     {
-        if (!buf_ptr)
+        if (!data.size() || data.size() < 8)
         {
-            throw std::invalid_argument("Buffer pointer cannot be null.");
+            throw std::invalid_argument("Buffer is too small to contain a valid message.");
         }
-        const std::span<const uint8_t> continuation_span(buf_ptr, 4);
+        const std::span<const uint8_t> continuation_span = data.subspan(0, 4);
         if (!is_continuation(continuation_span))
         {
             throw std::runtime_error("Buffer starts with continuation bytes, expected a valid message.");
         }
-        return {buf_ptr};
+        return {data};
     }
 }
diff --git a/src/serialize_null_array.cpp b/src/serialize_null_array.cpp
@@ -32,10 +32,13 @@ namespace sparrow_ipc
         // I - Deserialize the Schema message
         std::optional<std::string> name;
         std::optional<std::vector<sparrow::metadata_pair>> metadata;
-        deserialize_schema_message(buf_ptr, current_offset, name, metadata);
+        deserialize_schema_message(std::span<const uint8_t>(buffer), current_offset, name, metadata);
 
         // II - Deserialize the RecordBatch message
-        const auto* record_batch = deserialize_record_batch_message(buf_ptr, current_offset);
+        const auto* record_batch = deserialize_record_batch_message(
+            std::span<const uint8_t>(buffer),
+            current_offset
+        );
 
         // The body is empty, so we don't need to read any further.
         // Construct the null_array from the deserialized metadata.
diff --git a/tests/test_primitive_array_with_files.cpp b/tests/test_primitive_array_with_files.cpp
@@ -84,21 +84,26 @@ TEST_SUITE("Integration tests")
                 stream_file.close();
 
                 // Process the stream file
-                const auto record_batches_from_stream = sparrow_ipc::deserialize_stream(stream_data.data());
+                const auto record_batches_from_stream = sparrow_ipc::deserialize_stream(
+                    std::span<const uint8_t>(stream_data)
+                );
 
                 // Compare record batches
                 REQUIRE_EQ(record_batches_from_stream.size(), record_batches_from_json.size());
                 for (size_t i = 0; i < record_batches_from_stream.size(); ++i)
                 {
-                    for(size_t y = 0; y < record_batches_from_stream[i].nb_columns(); y++)
+                    for (size_t y = 0; y < record_batches_from_stream[i].nb_columns(); y++)
                     {
                         const auto& column_stream = record_batches_from_stream[i].get_column(y);
                         const auto& column_json = record_batches_from_json[i].get_column(y);
                         REQUIRE_EQ(column_stream.size(), column_json.size());
-                        for(size_t z = 0 ; z < column_json.size(); z++)
+                        for (size_t z = 0; z < column_json.size(); z++)
                         {
                             const auto col_name = column_stream.name().value_or("NA");
-                            INFO("Comparing batch " << i << ", column " << y << " named :"<< col_name <<" , row " << z);
+                            INFO(
+                                "Comparing batch " << i << ", column " << y << " named :" << col_name
+                                                   << " , row " << z
+                            );
                             const auto& column_stream_value = column_stream[z];
                             const auto& column_json_value = column_json[z];
                             CHECK_EQ(column_stream_value, column_json_value);

Original file line number	Diff line number	Diff line change
`@@ -9,7 +9,7 @@ namespace sparrow_ipc`
`9`	`9`	`{`
`10`	`10`	`public:`
`11`	`11`
`12`		`- EncapsulatedMessage(const uint8_t* buf_ptr);`
	`12`	`+ EncapsulatedMessage(std::span<const uint8_t> data);`
`13`	`13`
`14`	`14`	`[[nodiscard]] const org::apache::arrow::flatbuf::Message* flat_buffer_message() const;`
`15`	`15`
`@@ -36,8 +36,8 @@ namespace sparrow_ipc`
`36`	`36`
`37`	`37`	`private:`
`38`	`38`
`39`		`- const uint8_t* m_buf_ptr;`
	`39`	`+ std::span<const uint8_t> m_data;`
`40`	`40`	`};`
`41`	`41`
`42`		`- [[nodiscard]] EncapsulatedMessage create_encapsulated_message(const uint8_t* buf_ptr);`
	`42`	`+ [[nodiscard]] EncapsulatedMessage create_encapsulated_message(std::span<const uint8_t> buf_ptr);`
`43`	`43`	`}`
Original file line number	Diff line number	Diff line change
`@@ -11,15 +11,15 @@`
`11`	`11`	`namespace sparrow_ipc`
`12`	`12`	`{`
`13`	`13`	`void deserialize_schema_message(`
`14`		`- const uint8_t* buf_ptr,`
	`14`	`+ std::span<const uint8_t> data,`
`15`	`15`	`size_t& current_offset,`
`16`	`16`	`std::optional<std::string>& name,`
`17`	`17`	`std::optional<std::vector<sparrow::metadata_pair>>& metadata`
`18`	`18`	`)`
`19`	`19`	`{`
`20`		`- const uint32_t schema_meta_len = (reinterpret_cast<const uint32_t>(buf_ptr + current_offset));`
	`20`	`+ const uint32_t schema_meta_len = (reinterpret_cast<const uint32_t>(data.data() + current_offset));`
`21`	`21`	`current_offset += sizeof(uint32_t);`
`22`		`- const auto schema_message = org::apache::arrow::flatbuf::GetMessage(buf_ptr + current_offset);`
	`22`	`+ const auto schema_message = org::apache::arrow::flatbuf::GetMessage(data.data() + current_offset);`
`23`	`23`	`if (schema_message->header_type() != org::apache::arrow::flatbuf::MessageHeader::Schema)`
`24`	`24`	`{`
`25`	`25`	`throw std::runtime_error("Expected Schema message at the start of the buffer.");`
`@@ -56,10 +56,10 @@ namespace sparrow_ipc`
`56`	`56`	`}`
`57`	`57`
`58`	`58`	`const org::apache::arrow::flatbuf::RecordBatch*`
`59`		`- deserialize_record_batch_message(const uint8_t* buf_ptr, size_t& current_offset)`
	`59`	`+ deserialize_record_batch_message(std::span<const uint8_t> data, size_t& current_offset)`
`60`	`60`	`{`
`61`	`61`	`current_offset += sizeof(uint32_t);`
`62`		`- const auto batch_message = org::apache::arrow::flatbuf::GetMessage(buf_ptr + current_offset);`
	`62`	`+ const auto batch_message = org::apache::arrow::flatbuf::GetMessage(data.data() + current_offset);`
`63`	`63`	`if (batch_message->header_type() != org::apache::arrow::flatbuf::MessageHeader::RecordBatch)`
`64`	`64`	`{`
`65`	`65`	`throw std::runtime_error("Expected RecordBatch message, but got a different type.");`
`@@ -225,7 +225,7 @@ namespace sparrow_ipc`
`225`	`225`	`return arrays;`
`226`	`226`	`}`
`227`	`227`
`228`		`- std::vector<sparrow::record_batch> deserialize_stream(const uint8_t* buf_ptr)`
	`228`	`+ std::vector<sparrow::record_batch> deserialize_stream(std::span<const uint8_t> data)`
`229`	`229`	`{`
`230`	`230`	`const org::apache::arrow::flatbuf::Schema* schema = nullptr;`
`231`	`231`	`std::vector<sparrow::record_batch> record_batches;`
`@@ -234,7 +234,7 @@ namespace sparrow_ipc`
`234`	`234`	`std::vector<sparrow::data_type> field_types;`
`235`	`235`	`do`
`236`	`236`	`{`
`237`		`- const EncapsulatedMessage encapsulated_message = create_encapsulated_message(buf_ptr);`
	`237`	`+ const EncapsulatedMessage encapsulated_message = create_encapsulated_message(data);`
`238`	`238`	`const org::apache::arrow::flatbuf::Message* message = encapsulated_message.flat_buffer_message();`
`239`	`239`	`switch (message->header_type())`
`240`	`240`	`{`
`@@ -280,8 +280,8 @@ namespace sparrow_ipc`
`280`	`280`	`throw std::runtime_error("Unknown message header type.");`
`281`	`281`	`}`
`282`	`282`	`const size_t encapsulated_message_total_length = encapsulated_message.total_length();`
`283`		`- buf_ptr += encapsulated_message_total_length;`
`284`		`- if (is_end_of_stream(std::span<const uint8_t>{buf_ptr, 8}))`
	`283`	`+ data = data.subspan(encapsulated_message_total_length);`
	`284`	`+ if (is_end_of_stream(data.subspan(0, 8)))`
`285`	`285`	`{`
`286`	`286`	`break;`
`287`	`287`	`}`
Original file line number	Diff line number	Diff line change
`@@ -7,21 +7,21 @@`
`7`	`7`
`8`	`8`	`namespace sparrow_ipc`
`9`	`9`	`{`
`10`		`- EncapsulatedMessage::EncapsulatedMessage(const uint8_t* buf_ptr)`
`11`		`- : m_buf_ptr(buf_ptr)`
	`10`	`+ EncapsulatedMessage::EncapsulatedMessage(std::span<const uint8_t> data)`
	`11`	`+ : m_data(data)`
`12`	`12`	`{`
`13`	`13`	`}`
`14`	`14`
`15`	`15`	`const org::apache::arrow::flatbuf::Message* EncapsulatedMessage::flat_buffer_message() const`
`16`	`16`	`{`
`17`		`- const uint8_t* message_ptr = m_buf_ptr + (sizeof(uint32_t) * 2); // 4 bytes continuation + 4 bytes`
`18`		`- // metadata size`
	`17`	`+ const uint8_t* message_ptr = m_data.data() + (sizeof(uint32_t) * 2); // 4 bytes continuation + 4`
	`18`	`+ // bytes metadata size`
`19`	`19`	`return org::apache::arrow::flatbuf::GetMessage(message_ptr);`
`20`	`20`	`}`
`21`	`21`
`22`	`22`	`size_t EncapsulatedMessage::metadata_length() const`
`23`	`23`	`{`
`24`		`- return (reinterpret_cast<const uint32_t>(m_buf_ptr + sizeof(uint32_t)));`
	`24`	`+ return (reinterpret_cast<const uint32_t>(m_data.data() + sizeof(uint32_t)));`
`25`	`25`	`}`
`26`	`26`
`27`	`27`	`[[nodiscard]] std::variant<`
`@@ -76,8 +76,7 @@ namespace sparrow_ipc`
`76`	`76`	`const size_t offset = sizeof(uint32_t) * 2 // 4 bytes continuation + 4 bytes metadata size`
`77`	`77`	`+ metadata_length();`
`78`	`78`	`const size_t padded_offset = utils::align_to_8(offset); // Round up to 8-byte boundary`
`79`		`- const uint8_t* body_ptr = m_buf_ptr + padded_offset;`
`80`		`- return {body_ptr, body_length()};`
	`79`	`+ return m_data.subspan(padded_offset, body_length());`
`81`	`80`	`}`
`82`	`81`
`83`	`82`	`size_t EncapsulatedMessage::total_length() const`
`@@ -90,20 +89,20 @@ namespace sparrow_ipc`
`90`	`89`
`91`	`90`	`std::span<const uint8_t> EncapsulatedMessage::as_span() const`
`92`	`91`	`{`
`93`		`- return {m_buf_ptr, total_length()};`
	`92`	`+ return m_data;`
`94`	`93`	`}`
`95`	`94`
`96`		`- EncapsulatedMessage create_encapsulated_message(const uint8_t* buf_ptr)`
	`95`	`+ EncapsulatedMessage create_encapsulated_message(std::span<const uint8_t> data)`
`97`	`96`	`{`
`98`		`- if (!buf_ptr)`
	`97`	`+ if (!data.size() \|\| data.size() < 8)`
`99`	`98`	`{`
`100`		`- throw std::invalid_argument("Buffer pointer cannot be null.");`
	`99`	`+ throw std::invalid_argument("Buffer is too small to contain a valid message.");`
`101`	`100`	`}`
`102`		`- const std::span<const uint8_t> continuation_span(buf_ptr, 4);`
	`101`	`+ const std::span<const uint8_t> continuation_span = data.subspan(0, 4);`
`103`	`102`	`if (!is_continuation(continuation_span))`
`104`	`103`	`{`
`105`	`104`	`throw std::runtime_error("Buffer starts with continuation bytes, expected a valid message.");`
`106`	`105`	`}`
`107`		`- return {buf_ptr};`
	`106`	`+ return {data};`
`108`	`107`	`}`
`109`	`108`	`}`