Added incremental record_batch API (#345)

* Added incremental record_batch API

* Added missing break

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: JohanMabille

include/sparrow/record_batch.hpp

@@ -156,18 +156,36 @@ namespace sparrow
          */
         SPARROW_API struct_array extract_struct_array();
 
+        /**
+         * Appends the array \ref column to the record batch, and maps it with
+         * \ref name.
+         *
+         * @param name The name of the column to append.
+         * @param column The array to append.
+         */
+        SPARROW_API void add_column(name_type name, array column);
+
+        /**
+         * Appends the array \ref column to the record batch, and maps it to
+         * its internal name. \ref column must have a name.
+         *
+         * @param column The array to append.
+         */
+        SPARROW_API void add_column(array column);
+
     private:
 
         template <class U, class R>
         [[nodiscard]] std::vector<U> to_vector(R&& range) const;
 
-        SPARROW_API void init_array_map();
+        SPARROW_API void update_array_map_cache() const;
 
         [[nodiscard]] SPARROW_API bool check_consistency() const;
 
         std::vector<name_type> m_name_list;
         std::vector<array> m_array_list;
-        std::unordered_map<name_type, array*> m_array_map;
+        mutable std::unordered_map<name_type, const array*> m_array_map;
+        mutable bool m_dirty_map = true;
     };
 
     /**
@@ -192,8 +210,7 @@ namespace sparrow
         : m_name_list(to_vector<name_type>(std::move(names)))
         , m_array_list(to_vector<array>(std::move(columns)))
     {
-        init_array_map();
-        SPARROW_ASSERT_TRUE(check_consistency());
+        update_array_map_cache();
     }
 
     namespace detail
@@ -217,8 +234,7 @@ namespace sparrow
         : m_name_list(detail::get_names(columns))
         , m_array_list(to_vector<array>(std::move(columns)))
     {
-        init_array_map();
-        SPARROW_ASSERT_TRUE(check_consistency());
+        update_array_map_cache();
     }
 
     template <class U, class R>

src/record_batch.cpp

@@ -31,8 +31,7 @@ namespace sparrow
             m_array_list.push_back(std::move(array));
         }
 
-        init_array_map();
-        SPARROW_ASSERT_TRUE(check_consistency());
+        update_array_map_cache();
     }
 
     record_batch::record_batch(struct_array&& arr)
@@ -50,22 +49,21 @@ namespace sparrow
             m_name_list.push_back(std::string(arr.name().value()));
             m_array_list.push_back(std::move(arr));
         }
-        init_array_map();
-        SPARROW_ASSERT_TRUE(check_consistency());
+        update_array_map_cache();
     }
 
     record_batch::record_batch(const record_batch& rhs)
         : m_name_list(rhs.m_name_list)
         , m_array_list(rhs.m_array_list)
     {
-        init_array_map();
+        update_array_map_cache();
     }
 
     record_batch& record_batch::operator=(const record_batch& rhs)
     {
         m_name_list = rhs.m_name_list;
         m_array_list = rhs.m_array_list;
-        init_array_map();
+        update_array_map_cache();
         return *this;
     }
 
@@ -81,6 +79,7 @@ namespace sparrow
 
     bool record_batch::contains_column(const name_type& name) const
     {
+        update_array_map_cache();
         return m_array_map.contains(name);
     }
 
@@ -92,6 +91,7 @@ namespace sparrow
 
     const array& record_batch::get_column(const name_type& name) const
     {
+        update_array_map_cache();
         const auto iter = m_array_map.find(name);
         if (iter == m_array_map.end())
         {
@@ -126,14 +126,40 @@ namespace sparrow
         return struct_array(std::move(m_array_list));
     }
 
-    void record_batch::init_array_map()
+    void record_batch::add_column(name_type name, array column)
     {
-        m_array_map.clear();
-        m_array_map.reserve(m_name_list.size());
-        for (size_t i = 0; i < m_name_list.size(); ++i)
+        m_name_list.push_back(std::move(name));
+        m_array_list.push_back(std::move(column));
+        m_dirty_map = true;
+    }
+
+    void record_batch::add_column(array column)
+    {
+        auto opt_col_name = column.name();
+        SPARROW_ASSERT_TRUE(opt_col_name.has_value());
+        std::string name(opt_col_name.value());
+        add_column(std::move(name), std::move(column));
+    }
+
+    void record_batch::update_array_map_cache() const
+    {
+        if (!m_dirty_map)
         {
-            m_array_map.try_emplace(m_name_list[i], &(m_array_list[i]));
+            return;
         }
+
+        // Columns can only be appened, so update the map
+        // in reverse order and stops when it finds a name
+        // already contained in it.
+        for (std::size_t i = m_name_list.size(); i != 0; --i)
+        {
+            if (!m_array_map.try_emplace(m_name_list[i - 1], &(m_array_list[i - 1])).second)
+            {
+                break;
+            }
+        }
+        m_dirty_map = false;
+        SPARROW_ASSERT_TRUE(check_consistency());
     }
 
     bool record_batch::check_consistency() const

test/test_record_batch.cpp

@@ -196,6 +196,39 @@ namespace sparrow
             CHECK_EQ(extr, control);
         }
 
+        TEST_CASE("add_column")
+        {
+            auto record = make_record_batch(col_size);
+            primitive_array<std::int32_t> pr3(
+                std::ranges::iota_view{std::int32_t(3), 3 + std::int32_t(col_size)},
+                "column3"
+            );
+
+            auto ctrl = pr3;
+
+            record.add_column(array(std::move(pr3)));
+            std::vector<std::string> ctrl_name_list = make_name_list();
+            ctrl_name_list.push_back("column3");
+            std::vector<std::string> name_list(record.names().begin(), record.names().end());
+            CHECK_EQ(name_list, ctrl_name_list);
+
+            const auto& col3 = record.get_column(3);
+            bool res = col3.visit(
+                [&ctrl]<typename T>(const T& arg)
+                {
+                    if constexpr (std::same_as<primitive_array<std::int32_t>, T>)
+                    {
+                        return arg == ctrl;
+                    }
+                    else
+                    {
+                        return false;
+                    }
+                }
+            );
+            CHECK(res);
+        }
+
 #if defined(__cpp_lib_format)
         TEST_CASE("formatter")
         {