wip

Author: Alex-PLACET

include/sparrow_ipc/deserialize.hpp

@@ -6,42 +6,10 @@
 
 #include <sparrow/record_batch.hpp>
 
-#include "Message_generated.h"
-#include "Schema_generated.h"
 #include "sparrow_ipc/config/config.hpp"
-#include "sparrow_ipc/encapsulated_message.hpp"
-#include "sparrow_ipc/metadata.hpp"
 
 namespace sparrow_ipc
 {
-    /**
-     * @brief Deserializes arrays from an Apache Arrow RecordBatch using the provided schema.
-     *
-     * This function processes each field in the schema and deserializes the corresponding
-     * data from the RecordBatch into sparrow::array objects. It handles various Arrow data
-     * types including primitive types (bool, integers, floating point), binary data, and
-     * string data with their respective size variants.
-     *
-     * @param record_batch The Apache Arrow FlatBuffer RecordBatch containing the serialized data
-     * @param schema The Apache Arrow FlatBuffer Schema defining the structure and types of the data
-     * @param encapsulated_message The message containing the binary data buffers
-     * @param field_metadata Metadata for each field
-     *
-     * @return std::vector<sparrow::array> A vector of deserialized arrays, one for each field in the schema
-     *
-     * @throws std::runtime_error If an unsupported data type, integer bit width, or floating point precision
-     * is encountered
-     *
-     * The function maintains a buffer index that is incremented as it processes each field
-     * to correctly map data buffers to their corresponding arrays.
-     */
-    [[nodiscard]] SPARROW_IPC_API std::vector<sparrow::array> get_arrays_from_record_batch(
-        const org::apache::arrow::flatbuf::RecordBatch& record_batch,
-        const org::apache::arrow::flatbuf::Schema& schema,
-        const encapsulated_message& encapsulated_message,
-        const std::vector<std::optional<std::vector<sparrow::metadata_pair>>>& field_metadata
-    );
-
     /**
      * @brief Deserializes an Arrow IPC stream from binary data into a vector of record batches.
      *

include/sparrow_ipc/serialize.hpp

@@ -3,8 +3,6 @@
 #include <ranges>
 
 #include <sparrow/record_batch.hpp>
-
-#include "Message_generated.h"
 #include "sparrow_ipc/any_output_stream.hpp"
 #include "sparrow_ipc/compression.hpp"
 #include "sparrow_ipc/config/config.hpp"

include/sparrow_ipc/serialize_utils.hpp

@@ -5,7 +5,6 @@
 
 #include <sparrow/record_batch.hpp>
 
-#include "Message_generated.h"
 #include "sparrow_ipc/any_output_stream.hpp"
 #include "sparrow_ipc/compression.hpp"
 #include "sparrow_ipc/config/config.hpp"

integration_tests/CMakeLists.txt

@@ -1,13 +1,44 @@
 cmake_minimum_required(VERSION 3.28)
 
+# Create the integration_tools library
+add_library(integration_tools src/integration_tools.cpp)
+
+target_link_libraries(integration_tools
+    PUBLIC
+        sparrow-ipc
+        sparrow::sparrow
+        sparrow::json_reader
+)
+
+target_include_directories(integration_tools
+    PUBLIC
+        ${CMAKE_CURRENT_SOURCE_DIR}/include
+    PRIVATE
+        ${CMAKE_SOURCE_DIR}/include
+        ${CMAKE_BINARY_DIR}/generated
+)
+
+set_target_properties(integration_tools
+    PROPERTIES
+        CXX_STANDARD 20
+        CXX_STANDARD_REQUIRED ON
+        CXX_EXTENSIONS OFF
+        FOLDER "integration_tests"
+)
+
+add_dependencies(integration_tools generate_flatbuffers_headers)
+
 # Create executable for arrow_file_to_stream integration test
 add_executable(arrow_file_to_stream arrow_file_to_stream.cpp)
 
 target_link_libraries(arrow_file_to_stream
     PRIVATE
-    sparrow-ipc
-    sparrow::sparrow
-    sparrow::json_reader
+        integration_tools
+)
+
+target_include_directories(arrow_file_to_stream
+    PRIVATE
+        ${CMAKE_CURRENT_SOURCE_DIR}/include
 )
 
 set_target_properties(arrow_file_to_stream
@@ -23,21 +54,19 @@ set_target_properties(arrow_file_to_stream
         INSTALL_RPATH "$ORIGIN"
 )
 
-target_include_directories(arrow_file_to_stream
-    PRIVATE
-    ${CMAKE_SOURCE_DIR}/include
-    ${CMAKE_BINARY_DIR}/generated
-)
-
 add_dependencies(arrow_file_to_stream generate_flatbuffers_headers)
 
 # Create executable for arrow_stream_to_file integration test
 add_executable(arrow_stream_to_file arrow_stream_to_file.cpp)
 
 target_link_libraries(arrow_stream_to_file
     PRIVATE
-    sparrow-ipc
-    sparrow::sparrow
+        integration_tools
+)
+
+target_include_directories(arrow_stream_to_file
+    PRIVATE
+        ${CMAKE_CURRENT_SOURCE_DIR}/include
 )
 
 set_target_properties(arrow_stream_to_file
@@ -53,22 +82,19 @@ set_target_properties(arrow_stream_to_file
         INSTALL_RPATH "$ORIGIN"
 )
 
-target_include_directories(arrow_stream_to_file
-    PRIVATE
-    ${CMAKE_SOURCE_DIR}/include
-    ${CMAKE_BINARY_DIR}/generated
-)
-
 add_dependencies(arrow_stream_to_file generate_flatbuffers_headers)
 
 # Create executable for arrow_json_to_file integration test
 add_executable(arrow_json_to_file arrow_json_to_file.cpp)
 
 target_link_libraries(arrow_json_to_file
     PRIVATE
-    sparrow-ipc
-    sparrow::sparrow
-    sparrow::json_reader
+        integration_tools
+)
+
+target_include_directories(arrow_json_to_file
+    PRIVATE
+        ${CMAKE_CURRENT_SOURCE_DIR}/include
 )
 
 set_target_properties(arrow_json_to_file
@@ -84,22 +110,19 @@ set_target_properties(arrow_json_to_file
         INSTALL_RPATH "$ORIGIN"
 )
 
-target_include_directories(arrow_json_to_file
-    PRIVATE
-    ${CMAKE_SOURCE_DIR}/include
-    ${CMAKE_BINARY_DIR}/generated
-)
-
 add_dependencies(arrow_json_to_file generate_flatbuffers_headers)
 
 # Create executable for arrow_validate integration test
 add_executable(arrow_validate arrow_validate.cpp)
 
 target_link_libraries(arrow_validate
     PRIVATE
-        sparrow-ipc
-        sparrow::sparrow
-        sparrow::json_reader
+        integration_tools
+)
+
+target_include_directories(arrow_validate
+    PRIVATE
+        ${CMAKE_CURRENT_SOURCE_DIR}/include
 )
 
 set_target_properties(arrow_validate
@@ -115,46 +138,25 @@ set_target_properties(arrow_validate
         INSTALL_RPATH "$ORIGIN"
 )
 
-target_include_directories(arrow_validate
-    PRIVATE
-    ${CMAKE_SOURCE_DIR}/include
-    ${CMAKE_BINARY_DIR}/generated
-)
-
 add_dependencies(arrow_validate generate_flatbuffers_headers)
 
 # Create test executable for integration tools
 add_executable(test_integration_tools main.cpp test_integration_tools.cpp)
 
 target_link_libraries(test_integration_tools
     PRIVATE
-    sparrow-ipc
-    sparrow::sparrow
-    sparrow::json_reader
-    doctest::doctest
-    arrow-testing-data
-)
-
-target_compile_definitions(test_integration_tools
-    PRIVATE
-    INTEGRATION_TOOLS_DIR="${CMAKE_CURRENT_BINARY_DIR}"
-)
-
-set_target_properties(test_integration_tools
-    PROPERTIES
-    CXX_STANDARD 20
-    CXX_STANDARD_REQUIRED ON
-    CXX_EXTENSIONS OFF
+        integration_tools
+        doctest::doctest
+        arrow-testing-data
 )
 
 target_include_directories(test_integration_tools
     PRIVATE
-    ${CMAKE_SOURCE_DIR}/include
-    ${CMAKE_BINARY_DIR}/generated
+        ${CMAKE_CURRENT_SOURCE_DIR}/include
+        ${CMAKE_SOURCE_DIR}/include
+        ${CMAKE_BINARY_DIR}/generated
 )
 
-add_dependencies(test_integration_tools generate_flatbuffers_headers arrow_file_to_stream arrow_stream_to_file arrow_json_to_file arrow_validate)
-
 # Register with CTest
 enable_testing()
 add_test(NAME integration_tools_test COMMAND test_integration_tools)

integration_tests/arrow_file_to_stream.cpp

@@ -1,16 +1,9 @@
 #include <cstdlib>
 #include <filesystem>
-#include <fstream>
 #include <iostream>
 #include <vector>
 
-#include <nlohmann/json.hpp>
-#include <sparrow/record_batch.hpp>
-
-#include "sparrow/json_reader/json_parser.hpp"
-
-#include <sparrow_ipc/memory_output_stream.hpp>
-#include <sparrow_ipc/stream_file_serializer.hpp>
+#include "integration_tools.hpp"
 
 /**
  * @brief Reads a JSON file containing record batches and outputs the serialized Arrow IPC stream to stdout.
@@ -39,68 +32,8 @@ int main(int argc, char* argv[])
 
     try
     {
-        // Check if the JSON file exists
-        if (!std::filesystem::exists(json_path))
-        {
-            std::cerr << "Error: File not found: " << json_path << "\n";
-            return EXIT_FAILURE;
-        }
-
-        // Open and parse the JSON file
-        std::ifstream json_file(json_path);
-        if (!json_file.is_open())
-        {
-            std::cerr << "Error: Could not open file: " << json_path << "\n";
-            return EXIT_FAILURE;
-        }
-
-        nlohmann::json json_data;
-        try
-        {
-            json_data = nlohmann::json::parse(json_file);
-        }
-        catch (const nlohmann::json::parse_error& e)
-        {
-            std::cerr << "Error: Failed to parse JSON file: " << e.what() << "\n";
-            return EXIT_FAILURE;
-        }
-        json_file.close();
-
-        // Get the number of batches
-        if (!json_data.contains("batches") || !json_data["batches"].is_array())
-        {
-            std::cerr << "Error: JSON file does not contain a 'batches' array.\n";
-            return EXIT_FAILURE;
-        }
-
-        const size_t num_batches = json_data["batches"].size();
-
-        // Parse all record batches from JSON
-        std::vector<sparrow::record_batch> record_batches;
-        record_batches.reserve(num_batches);
-
-        for (size_t batch_idx = 0; batch_idx < num_batches; ++batch_idx)
-        {
-            try
-            {
-                record_batches.emplace_back(
-                    sparrow::json_reader::build_record_batch_from_json(json_data, batch_idx)
-                );
-            }
-            catch (const std::exception& e)
-            {
-                std::cerr << "Error: Failed to build record batch " << batch_idx << ": " << e.what()
-                          << "\n";
-                return EXIT_FAILURE;
-            }
-        }
-
-        // Serialize record batches to Arrow IPC stream format
-        std::vector<uint8_t> stream_data;
-        sparrow_ipc::memory_output_stream stream(stream_data);
-        sparrow_ipc::stream_file_serializer serializer(stream);
-        serializer << record_batches;
-        serializer.end();
+        // Convert JSON file to stream using the library
+        std::vector<uint8_t> stream_data = integration_tools::json_file_to_stream(json_path);
 
         // Write the binary stream to stdout
         std::cout.write(reinterpret_cast<const char*>(stream_data.data()), stream_data.size());