Many small fixes on ArrowStream

include/sparrow/arrow_interface/arrow_array_stream.hpp

@@ -33,7 +33,6 @@
 
 #pragma once
 
-
 #include "sparrow/arrow_interface/arrow_array_stream/private_data.hpp"
 #include "sparrow/c_interface.hpp"
 #include "sparrow/c_stream_interface.hpp"
@@ -128,4 +127,7 @@ namespace sparrow
     SPARROW_API const char* get_last_error_from_arrow_array_stream(ArrowArrayStream* stream);
 
     SPARROW_API void fill_arrow_array_stream(ArrowArrayStream& stream);
-}
+
+    SPARROW_API ArrowArrayStream move_array_stream(ArrowArrayStream&& source);
+    SPARROW_API ArrowArrayStream move_array_stream(ArrowArrayStream& source);
+}

include/sparrow/arrow_interface/arrow_array_stream_proxy.hpp

@@ -101,7 +101,9 @@ namespace sparrow
         arrow_array_stream_proxy(const arrow_array_stream_proxy&) = delete;
         arrow_array_stream_proxy& operator=(const arrow_array_stream_proxy&) = delete;
 
+        SPARROW_API
         arrow_array_stream_proxy(arrow_array_stream_proxy&& other) noexcept;
+        SPARROW_API
         arrow_array_stream_proxy& operator=(arrow_array_stream_proxy&& other) noexcept;
 
         /**
@@ -111,6 +113,12 @@ namespace sparrow
          * released. This ensures proper cleanup of all Arrow C interface objects.
          */
         SPARROW_API ~arrow_array_stream_proxy();
+
+        /**
+         * Check whether the proxy has ownership of its internal `ArrowArrayStream`.
+         */
+        SPARROW_API bool owns_stream() const;
+
         /**
          * @brief Export the stream pointer.
          *
@@ -143,7 +151,7 @@ namespace sparrow
             requires layout<std::ranges::range_value_t<R>>
         void push(R&& arrays)
         {
-            arrow_array_stream_private_data& private_data = *get_private_data();
+            arrow_array_stream_private_data& private_data = get_private_data();
 
             // Check if we need to create schema from first array
             if (private_data.schema() == nullptr)
@@ -234,18 +242,11 @@ namespace sparrow
          */
         void throw_if_immutable() const;
 
-        /**
-         * @brief Gets the private data (const version).
-         *
-         * @return Const pointer to the stream's private data.
-         */
-        [[nodiscard]] SPARROW_API const arrow_array_stream_private_data* get_private_data() const;
-
         /**
          * @brief Gets the private data (mutable version).
          *
          * @return Mutable pointer to the stream's private data.
          */
-        [[nodiscard]] SPARROW_API arrow_array_stream_private_data* get_private_data();
+        [[nodiscard]] SPARROW_API arrow_array_stream_private_data& get_private_data();
     };
-}
+}

src/arrow_interface/arrow_array_stream.cpp

@@ -149,4 +149,16 @@ namespace sparrow
         stream.release = &release_arrow_array_stream;
         stream.private_data = new arrow_array_stream_private_data();
     }
-}
+
+    ArrowArrayStream move_array_stream(ArrowArrayStream&& source)
+    {
+        ArrowArrayStream target = source;
+        source = {};
+        return target;
+    }
+
+    ArrowArrayStream move_array_stream(ArrowArrayStream& source)
+    {
+        return move_array_stream(std::move(source));
+    }
+}

src/arrow_interface/arrow_array_stream_proxy.cpp

@@ -26,7 +26,7 @@ namespace sparrow
     }
 
     arrow_array_stream_proxy::arrow_array_stream_proxy(ArrowArrayStream&& stream)
-        : m_stream(stream)
+        : m_stream(move_array_stream(stream))
     {
     }
 
@@ -89,6 +89,11 @@ namespace sparrow
         }
     }
 
+    bool arrow_array_stream_proxy::owns_stream() const
+    {
+        return std::holds_alternative<ArrowArrayStream>(m_stream);
+    }
+
     ArrowArrayStream* arrow_array_stream_proxy::get_stream_ptr()
     {
         if (std::holds_alternative<ArrowArrayStream*>(m_stream))
@@ -113,16 +118,10 @@ namespace sparrow
         }
     }
 
-    const arrow_array_stream_private_data* arrow_array_stream_proxy::get_private_data() const
-    {
-        throw_if_immutable();
-        return static_cast<const arrow_array_stream_private_data*>(get_stream_ptr()->private_data);
-    }
-
-    arrow_array_stream_private_data* arrow_array_stream_proxy::get_private_data()
+    arrow_array_stream_private_data& arrow_array_stream_proxy::get_private_data()
     {
         throw_if_immutable();
-        return static_cast<arrow_array_stream_private_data*>(get_stream_ptr()->private_data);
+        return *static_cast<arrow_array_stream_private_data*>(get_stream_ptr()->private_data);
     }
 
     ArrowArrayStream* arrow_array_stream_proxy::export_stream()
@@ -145,6 +144,11 @@ namespace sparrow
     std::optional<array> arrow_array_stream_proxy::pop()
     {
         ArrowArrayStream* stream = get_stream_ptr();
+        if (!stream)
+        {
+            throw std::runtime_error("ArrowArrayStream pointer is null");
+        }
+
         ArrowArray array{};
         if (int err = stream->get_next(stream, &array); err != 0)
         {
@@ -190,4 +194,4 @@ namespace sparrow
             throw std::runtime_error("ArrowArrayStream private data is not initialized");
         }
     }
-}
+}

test/test_arrow_array_stream_proxy.cpp

@@ -74,24 +74,127 @@ namespace sparrow
 
     TEST_SUITE("arrow_array_stream_proxy")
     {
-        TEST_CASE("constructor - default")
+        TEST_CASE("constructor")
         {
-            arrow_array_stream_proxy proxy;
-            ArrowArrayStream* aas = proxy.export_stream();
-            REQUIRE_NE(aas, nullptr);
-            aas->release(aas);
-            delete aas;
+            SUBCASE("default")
+            {
+                arrow_array_stream_proxy proxy;
+                ArrowArrayStream* aas = proxy.export_stream();
+                REQUIRE_NE(aas, nullptr);
+                aas->release(aas);
+                delete aas;
+            }
+
+            SUBCASE("pointer")
+            {
+                ArrowArrayStream* stream = new ArrowArrayStream;
+                fill_arrow_array_stream(*stream);
+                arrow_array_stream_proxy proxy(stream);
+                ArrowArrayStream* aas = proxy.export_stream();
+                REQUIRE_NE(aas, nullptr);
+                aas->release(aas);
+                delete aas;
+            }
+
+            SUBCASE("move")
+            {
+                ArrowArrayStream stream{};
+                fill_arrow_array_stream(stream);
+                arrow_array_stream_proxy proxy(std::move(stream));
+                REQUIRE_EQ(stream.private_data, nullptr);
+                REQUIRE_EQ(stream.release, nullptr);
+                ArrowArrayStream* aas = proxy.export_stream();
+                REQUIRE_NE(aas, nullptr);
+                aas->release(aas);
+                delete aas;
+            }
         }
 
-        TEST_CASE("constructor - from existing stream")
+        TEST_CASE("move semantics")
         {
-            ArrowArrayStream* stream = new ArrowArrayStream;
-            fill_arrow_array_stream(*stream);
-            arrow_array_stream_proxy proxy(stream);
-            ArrowArrayStream* aas = proxy.export_stream();
-            REQUIRE_NE(aas, nullptr);
-            aas->release(aas);
-            delete aas;
+            SUBCASE("move constructor")
+            {
+                {
+                    ArrowArrayStream stream{};
+                    fill_arrow_array_stream(stream);
+                    arrow_array_stream_proxy src(std::move(stream));
+                    arrow_array_stream_proxy dst(std::move(src));
+
+                    auto* src_str = src.export_stream();
+                    auto* dst_str = dst.export_stream();
+                    REQUIRE_EQ(src_str->release, nullptr);
+                    REQUIRE_NE(dst_str->release, nullptr);
+                    delete src_str;
+                    delete dst_str;
+                }
+
+                {
+                    ArrowArrayStream stream{};
+                    fill_arrow_array_stream(stream);
+                    arrow_array_stream_proxy src(&stream);
+                    arrow_array_stream_proxy dst(std::move(src));
+
+                    auto* src_str = src.export_stream();
+                    auto* dst_str = dst.export_stream();
+                    REQUIRE_EQ(src_str, nullptr);
+                    REQUIRE_NE(dst_str, nullptr);
+                }
+            }
+
+            SUBCASE("move assignment")
+            {
+                {
+                    auto test_array = make_test_primitive_array<int32_t>(10);
+                    ArrowArrayStream stream;
+                    fill_arrow_array_stream(stream);
+                    arrow_array_stream_proxy src(std::move(stream));
+                    src.push(std::move(test_array));
+
+                    ArrowArrayStream stream2;
+                    fill_arrow_array_stream(stream2);
+                    arrow_array_stream_proxy dst(std::move(stream2));
+                    dst = std::move(src);
+
+                    auto* src_str = src.export_stream();
+                    REQUIRE_EQ(src_str->release, nullptr);
+                    auto dst_arr = dst.pop();
+                    REQUIRE(dst_arr.has_value());
+                    delete src_str;
+                }
+
+                {
+                    auto test_array = make_test_primitive_array<int32_t>(10);
+                    ArrowArrayStream stream;
+                    fill_arrow_array_stream(stream);
+                    arrow_array_stream_proxy src(&stream);
+                    src.push(std::move(test_array));
+
+                    ArrowArrayStream stream2;
+                    fill_arrow_array_stream(stream2);
+                    arrow_array_stream_proxy dst(&stream2);
+                    dst = std::move(src);
+
+                    auto* src_str = src.export_stream();
+                    REQUIRE_EQ(src_str, nullptr);
+                    auto dst_arr = dst.pop();
+                    REQUIRE(dst_arr.has_value());
+                }
+            }
+        }
+
+        TEST_CASE("owns_stream")
+        {
+            {
+                arrow_array_stream_proxy proxy{};
+                REQUIRE(proxy.owns_stream());
+            }
+
+            {
+                ArrowArrayStream stream{};
+                fill_arrow_array_stream(stream);
+                arrow_array_stream_proxy proxy(&stream);
+                REQUIRE(!proxy.owns_stream());
+            }
         }
 
         TEST_CASE("export_stream")
@@ -110,7 +213,16 @@ namespace sparrow
 
         TEST_CASE("push and pop - single int32 array")
         {
-            SUBCASE("single array")
+            SUBCASE("empty stream")
+            {
+                ArrowArrayStream stream{};
+                fill_arrow_array_stream(stream);
+                arrow_array_stream_proxy src(&stream);
+                arrow_array_stream_proxy dst(std::move(src));
+                CHECK_THROWS_AS(src.pop(), std::runtime_error);
+            }
+
+            SUBCASE("single int32 array")
             {
                 arrow_array_stream_proxy proxy;
                 auto test_array = make_test_primitive_array<int32_t>(10);
@@ -119,35 +231,35 @@ namespace sparrow
                 REQUIRE(result.has_value());
                 CHECK_EQ(result->size(), 10);
             }
-        }
-
-        TEST_CASE("push and pop - multiple arrays")
-        {
-            arrow_array_stream_proxy proxy;
-
-            // Create and push multiple arrays (schema created from first array)
-            std::vector<primitive_array<int32_t>> arrays;
-            arrays.push_back(make_test_primitive_array<int32_t>(5, 0));
-            arrays.push_back(make_test_primitive_array<int32_t>(7, 10));
-            arrays.push_back(make_test_primitive_array<int32_t>(3, 20));
 
-            for (auto& arr : arrays)
+            SUBCASE("multiple arrays")
             {
-                proxy.push(std::move(arr));
-            }
-
-            // Pop all arrays
-            const auto result1 = proxy.pop();
-            REQUIRE(result1.has_value());
-            CHECK_EQ(result1->size(), 5);
-
-            auto result2 = proxy.pop();
-            REQUIRE(result2.has_value());
-            CHECK_EQ(result2->size(), 7);
+                arrow_array_stream_proxy proxy;
 
-            const auto result3 = proxy.pop();
-            REQUIRE(result3.has_value());
-            CHECK_EQ(result3->size(), 3);
+                // Create and push multiple arrays (schema created from first array)
+                std::vector<primitive_array<int32_t>> arrays;
+                arrays.push_back(make_test_primitive_array<int32_t>(5, 0));
+                arrays.push_back(make_test_primitive_array<int32_t>(7, 10));
+                arrays.push_back(make_test_primitive_array<int32_t>(3, 20));
+
+                for (auto& arr : arrays)
+                {
+                    proxy.push(std::move(arr));
+                }
+
+                // Pop all arrays
+                const auto result1 = proxy.pop();
+                REQUIRE(result1.has_value());
+                CHECK_EQ(result1->size(), 5);
+
+                auto result2 = proxy.pop();
+                REQUIRE(result2.has_value());
+                CHECK_EQ(result2->size(), 7);
+
+                const auto result3 = proxy.pop();
+                REQUIRE(result3.has_value());
+                CHECK_EQ(result3->size(), 3);
+            }
         }
 
         TEST_CASE("end of stream")