wip

Author: Alex-PLACET

cmake/external_dependencies.cmake

@@ -85,4 +85,48 @@ if(SPARROW_IPC_BUILD_TESTS)
         GIT_REPOSITORY https://github.com/doctest/doctest.git
         TAG v2.4.12
     )
+    
+    message(STATUS "📦 Fetching arrow-testing")
+    cmake_policy(PUSH)
+    cmake_policy(SET CMP0174 NEW)  # Suppress warning about FetchContent_Declare GIT_REPOSITORY
+    # Fetch arrow-testing data (no CMake build needed)
+    FetchContent_Declare(
+        arrow-testing
+        GIT_REPOSITORY https://github.com/apache/arrow-testing.git
+        GIT_SHALLOW TRUE
+        # CONFIGURE_COMMAND ""
+        # BUILD_COMMAND ""
+        # INSTALL_COMMAND ""
+    )
+    FetchContent_MakeAvailable(arrow-testing)
+    cmake_policy(POP)
+    
+    # Create interface library for easy access to test data
+    add_library(arrow-testing-data INTERFACE)
+    message(STATUS "Arrow testing data directory: ${arrow-testing_SOURCE_DIR}")
+    target_compile_definitions(arrow-testing-data INTERFACE
+        ARROW_TESTING_DATA_DIR="${arrow-testing_SOURCE_DIR}"
+    )
+    message(STATUS "\t✅ Fetched arrow-testing")
+
+    # Iterate over all the files in the arrow-testing-data source directiory. When it's a gz, extract in place.
+    file(GLOB_RECURSE arrow_testing_data_targz_files CONFIGURE_DEPENDS
+        "${arrow-testing_SOURCE_DIR}/data/arrow-ipc-stream/integration/1.0.0-littleendian/*.json.gz"
+    )
+    foreach(file_path IN LISTS arrow_testing_data_targz_files)
+            cmake_path(GET file_path PARENT_PATH parent_dir)
+            cmake_path(GET file_path STEM filename)
+            set(destination_file_path "${parent_dir}/${filename}.json")
+            if(EXISTS "${destination_file_path}")
+                message(VERBOSE "File already extracted: ${destination_file_path}")
+            else()
+                message(STATUS "Extracting ${file_path}")
+                if(WIN32)
+                    execute_process(COMMAND powershell -Command "$i=\"${file_path}\"; $o=\"${destination_file_path}\"; [IO.Compression.GZipStream]::new([IO.File]::OpenRead($i),[IO.Compression.CompressionMode]::Decompress).CopyTo([IO.File]::Create($o))")
+                else()
+                    execute_process(COMMAND gunzip -kf "${file_path}")
+                endif()
+            endif()
+    endforeach()
+
 endif()

include/sparrow_ipc/encapsulated_message.hpp

@@ -39,5 +39,6 @@ namespace sparrow_ipc
         std::span<const uint8_t> m_data;
     };
 
-    [[nodiscard]] EncapsulatedMessage create_encapsulated_message(std::span<const uint8_t> buf_ptr);
+    [[nodiscard]] std::pair<EncapsulatedMessage, std::span<const uint8_t>>
+    extract_encapsulated_message(std::span<const uint8_t> buf_ptr);
 }

src/deserialize.cpp

@@ -234,7 +234,7 @@ namespace sparrow_ipc
         std::vector<sparrow::data_type> field_types;
         do
         {
-            const EncapsulatedMessage encapsulated_message = create_encapsulated_message(data);
+            const auto [encapsulated_message, rest] = extract_encapsulated_message(data);
             const org::apache::arrow::flatbuf::Message* message = encapsulated_message.flat_buffer_message();
             switch (message->header_type())
             {
@@ -279,8 +279,7 @@ namespace sparrow_ipc
                 default:
                     throw std::runtime_error("Unknown message header type.");
             }
-            const size_t encapsulated_message_total_length = encapsulated_message.total_length();
-            data = data.subspan(encapsulated_message_total_length);
+            data = rest;
             if (is_end_of_stream(data.subspan(0, 8)))
             {
                 break;

src/encapsulated_message.cpp

@@ -92,7 +92,8 @@ namespace sparrow_ipc
         return m_data;
     }
 
-    EncapsulatedMessage create_encapsulated_message(std::span<const uint8_t> data)
+    std::pair<EncapsulatedMessage, std::span<const uint8_t>>
+    extract_encapsulated_message(std::span<const uint8_t> data)
     {
         if (!data.size() || data.size() < 8)
         {
@@ -103,6 +104,8 @@ namespace sparrow_ipc
         {
             throw std::runtime_error("Buffer starts with continuation bytes, expected a valid message.");
         }
-        return {data};
+        EncapsulatedMessage message(data);
+        std::span<const uint8_t> rest = data.subspan(message.total_length());
+        return {std::move(message), std::move(rest)};
     }
 }

tests/CMakeLists.txt

@@ -2,8 +2,7 @@ cmake_minimum_required(VERSION 3.28)
 
 set(test_target "test_sparrow_ipc_lib")
 
-set(
-    SPARROW_IPC_TESTS_SRC
+set(SPARROW_IPC_TESTS_SRC
     include/sparrow_ipc_tests_helpers.hpp
     # TODO move all the files below under src?
     main.cpp
@@ -21,8 +20,11 @@ target_link_libraries(${test_target}
         sparrow-ipc
         doctest::doctest
         sparrow::json_reader
+        arrow-testing-data
 )
 
+set(APACHE_ARROW_TESTING_DIR $<TARGET_PROPERTY:arrow-testing,INTERFACE_INCLUDE_DIRECTORIES>)
+
 target_compile_definitions(${test_target}
     PRIVATE
         TESTS_RESOURCES_FILES_PATH="${CMAKE_CURRENT_SOURCE_DIR}/resources/")

tests/test_primitive_array_with_files.cpp

@@ -15,7 +15,10 @@
 #include "sparrow_ipc/deserialize.hpp"
 
 
-const std::filesystem::path tests_resources_files_path = TESTS_RESOURCES_FILES_PATH;
+const std::filesystem::path arrow_testing_data_dir = ARROW_TESTING_DATA_DIR;
+
+const std::filesystem::path tests_resources_files_path = arrow_testing_data_dir / "data" / "arrow-ipc-stream"
+                                                         / "integration" / "1.0.0-littleendian";
 
 const std::vector<std::filesystem::path> files_paths_to_test = {
     tests_resources_files_path / "generated_primitive",