Use ArrayBackedSet to replace std::set for index in segment

components/core/CMakeLists.txt

@@ -121,6 +121,8 @@ include(cmake/Modules/FindLibraryDependencies.cmake)
 FindDynamicLibraryDependencies(sqlite "${sqlite_DYNAMIC_LIBS}")
 
 set(SOURCE_FILES_clp
+        src/ArrayBackedPosIntSet.cpp
+        src/ArrayBackedPosIntSet.hpp
         src/clp/clp.cpp
         src/clp/CommandLineArguments.cpp
         src/clp/CommandLineArguments.hpp

components/core/src/ArrayBackedPosIntSet.cpp

@@ -0,0 +1 @@
+#include "ArrayBackedPosIntSet.hpp"

components/core/src/ArrayBackedPosIntSet.hpp

@@ -0,0 +1,199 @@
+#ifndef ARRAYBACKEDPOSINTSET_HPP
+#define ARRAYBACKEDPOSINTSET_HPP
+
+// C++ standard libraries
+#include <unordered_set>
+#include <vector>
+
+// spdlog
+#include <spdlog/spdlog.h>
+
+// Project headers
+#include "Defs.h"
+#include "streaming_compression/zstd/Compressor.hpp"
+#include "TraceableException.hpp"
+
+/**
+ * Template class of set implemented with vector<bool> for continuously increasing numeric value
+ * @tparam PosIntType
+ */
+template<typename PosIntType>
+class ArrayBackedPosIntSet {
+public:
+    // Types
+    class OperationFailed : public TraceableException {
+    public:
+        // Constructors
+        OperationFailed (ErrorCode error_code, const char* const filename, int line_number) : TraceableException(error_code, filename, line_number) {}
+
+        // Methods
+        const char* what () const noexcept override {
+            return "ArrayBackedPosIntSet operation failed";
+        }
+    };
+
+    // Constructors
+    ArrayBackedPosIntSet ();
+
+    explicit ArrayBackedPosIntSet (size_t initial_capacity);
+
+    // Methods
+    /**
+     * Gets the number of unique values in the set
+     */
+    size_t size () const { return m_size; }
+
+    /**
+     * Clears the set and restores its initial capacity
+     */
+    void clear ();
+
+    void insert (PosIntType value);
+
+    /**
+     * Inserts all values from the given set
+     * @param input_set
+     */
+    void insert_all (const ArrayBackedPosIntSet<PosIntType>& input_set);
+
+    /**
+     * Inserts all values from the given set
+     * @param input_set
+     */
+    void insert_all (const std::unordered_set<PosIntType>& input_set);
+
+    /**
+     * Inserts all values from the given vector
+     * @param input_vector
+     */
+    void insert_all (const std::vector<PosIntType>& input_vector);
+
+    /**
+     * Writes all values in the set into the given compressor
+     * @param compressor
+     */
+    void write_to_compressor (streaming_compression::Compressor& compressor) const;
+
+private:
+    // Methods
+    /**
+     * Increases the capacity of the bool array so that
+     * the given value becomes a valid index in the array
+     * @param value
+     */
+    void increase_capacity (size_t value);
+
+    // Variables
+    std::vector<bool> m_data;
+    size_t m_initial_capacity;
+
+    // The number of unique values in the set
+    size_t m_size;
+
+    // The largest value in the set
+    PosIntType m_largest_value;
+};
+
+template<typename PosIntType>
+ArrayBackedPosIntSet<PosIntType>::ArrayBackedPosIntSet () {
+    constexpr size_t cDefaultInitialCapacity = 1024;
+    m_initial_capacity = cDefaultInitialCapacity;
+    clear();
+}
+
+template<typename PosIntType>
+ArrayBackedPosIntSet<PosIntType>::ArrayBackedPosIntSet (size_t initial_capacity) {
+    m_initial_capacity = initial_capacity;
+    clear();
+}
+
+template<typename PosIntType>
+void ArrayBackedPosIntSet<PosIntType>::clear () {
+    m_data.clear();
+    m_data.resize(m_initial_capacity, false);
+    m_size = 0;
+    m_largest_value = 0;
+}
+
+template<typename PosIntType>
+void ArrayBackedPosIntSet<PosIntType>::insert (PosIntType value) {
+    if (value >= m_data.size()) {
+        increase_capacity(value);
+    }
+
+    // Add the value if it is not already in the set
+    if (false == m_data[value]) {
+        m_data[value] = true;
+        m_size++;
+
+        // Update the largest value if necessary
+        if (value > m_largest_value) {
+            m_largest_value = value;
+        }
+    }
+}
+
+template<typename PosIntType>
+void ArrayBackedPosIntSet<PosIntType>::insert_all (const ArrayBackedPosIntSet<PosIntType>& input_set) {
+    // Increase capacity if necessary
+    size_t input_set_largest_value = input_set.m_largest_value;
+    if (input_set_largest_value >= m_data.size()) {
+        increase_capacity(input_set_largest_value);
+    }
+
+    // Copy values from the input set
+    auto input_set_data = input_set.m_data;
+    for (auto value = 0; value <= input_set_largest_value; ++value) {
+        // Add a value only if
+        // - doesn't exist in this set
+        // - exists in the input set
+        if (false == m_data[value] && input_set_data[value]) {
+            m_data[value] = true;
+            m_size++;
+        }
+    }
+
+    // Update the largest value if necessary
+    if (input_set_largest_value > m_largest_value) {
+        m_largest_value = input_set_largest_value;
+    }
+}
+
+template<typename PosIntType>
+void ArrayBackedPosIntSet<PosIntType>::insert_all (const std::unordered_set<PosIntType>& input_set) {
+    for (const auto value : input_set) {
+        insert(value);
+    }
+}
+
+template<typename PosIntType>
+void ArrayBackedPosIntSet<PosIntType>::insert_all (const std::vector<PosIntType>& input_vector) {
+    for (const auto value : input_vector) {
+        insert(value);
+    }
+}
+
+template<typename PosIntType>
+void ArrayBackedPosIntSet<PosIntType>::write_to_compressor (streaming_compression::Compressor& compressor) const {
+    for (PosIntType value = 0; value <= m_largest_value; ++value) {
+        if (m_data[value]) {
+            compressor.write_numeric_value(value);
+        }
+    }
+}
+
+template<typename PosIntType>
+void ArrayBackedPosIntSet<PosIntType>::increase_capacity (size_t value) {
+    if (value < m_data.size()) {
+        SPDLOG_ERROR("Calling increase_capacity on value smaller than capacity.");
+        throw OperationFailed(ErrorCode_BadParam, __FILENAME__, __LINE__);
+    }
+    auto capacity = m_data.size();
+    do {
+        capacity += capacity / 2;
+    } while (capacity <= value);
+
+    m_data.resize(capacity, false);
+}
+
+#endif //ARRAYBACKEDPOSINTSET_HPP

components/core/src/DictionaryWriter.hpp

@@ -11,6 +11,7 @@
 #include <spdlog/spdlog.h>
 
 // Project headers
+#include "ArrayBackedPosIntSet.hpp"
 #include "Defs.h"
 #include "dictionary_utils.hpp"
 #include "FileWriter.hpp"
@@ -72,7 +73,7 @@ class DictionaryWriter {
      * @param segment_id
      * @param ids
      */
-    void index_segment (segment_id_t segment_id, const std::unordered_set<DictionaryIdType>& ids);
+    void index_segment (segment_id_t segment_id, const ArrayBackedPosIntSet<DictionaryIdType>& ids);
 
     /**
      * Gets the size of the dictionary when it is stored on disk
@@ -227,7 +228,7 @@ void DictionaryWriter<DictionaryIdType, EntryType>::open_and_preload (const std:
 }
 
 template <typename DictionaryIdType, typename EntryType>
-void DictionaryWriter<DictionaryIdType, EntryType>::index_segment (segment_id_t segment_id, const std::unordered_set<DictionaryIdType>& ids) {
+void DictionaryWriter<DictionaryIdType, EntryType>::index_segment (segment_id_t segment_id, const ArrayBackedPosIntSet<DictionaryIdType>& ids) {
     if (false == m_is_open) {
         throw OperationFailed(ErrorCode_NotInit, __FILENAME__, __LINE__);
     }
@@ -236,9 +237,7 @@ void DictionaryWriter<DictionaryIdType, EntryType>::index_segment (segment_id_t
 
     // NOTE: The IDs in `ids` are not validated to exist in this dictionary since we perform validation when loading the dictionary.
     m_segment_index_compressor.write_numeric_value<uint64_t>(ids.size());
-    for (auto id : ids) {
-        m_segment_index_compressor.write_numeric_value(id);
-    }
+    ids.write_to_compressor(m_segment_index_compressor);
 
     ++m_num_segments_in_index;
 

components/core/src/clp/FileCompressor.cpp

@@ -37,7 +37,7 @@ static void compute_and_add_empty_directories (const set<string>& directories, c
  * @param archive
  * @param file
  */
-static void write_message_to_encoded_file (const ParsedMessage& msg, streaming_archive::writer::Archive& archive, streaming_archive::writer::File* file);
+static void write_message_to_encoded_file (const ParsedMessage& msg, streaming_archive::writer::Archive& archive);
 
 static void compute_and_add_empty_directories (const set<string>& directories, const set<string>& parent_directories,
                                                const boost::filesystem::path& parent_path, streaming_archive::writer::Archive& archive)
@@ -69,12 +69,12 @@ static void compute_and_add_empty_directories (const set<string>& directories, c
     archive.add_empty_directories(empty_directories);
 }
 
-static void write_message_to_encoded_file (const ParsedMessage& msg, streaming_archive::writer::Archive& archive, streaming_archive::writer::File* file) {
+static void write_message_to_encoded_file (const ParsedMessage& msg, streaming_archive::writer::Archive& archive) {
     if (msg.has_ts_patt_changed()) {
-        archive.change_ts_pattern(*file, msg.get_ts_patt());
+        archive.change_ts_pattern(msg.get_ts_patt());
     }
 
-    archive.write_msg(*file, msg.get_ts(), msg.get_content(), msg.get_orig_num_bytes());
+    archive.write_msg(msg.get_ts(), msg.get_content(), msg.get_orig_num_bytes());
 }
 
 namespace clp {
@@ -117,32 +117,32 @@ namespace clp {
         m_parsed_message.clear();
 
         // Open compressed file
-        auto* file = create_and_open_file(archive_writer, path_for_compression, group_id, m_uuid_generator(), 0);
+        archive_writer.create_and_open_file(path_for_compression, group_id, m_uuid_generator(), 0);
 
         // Parse content from UTF-8 validation buffer
         size_t buf_pos = 0;
         while (m_message_parser.parse_next_message(false, m_utf8_validation_buf_length, m_utf8_validation_buf, buf_pos, m_parsed_message)) {
             if (archive_writer.get_data_size_of_dictionaries() >= target_data_size_of_dicts) {
-                split_file_and_archive(archive_user_config, path_for_compression, group_id, m_parsed_message.get_ts_patt(), archive_writer, file);
-            } else if (file->get_encoded_size_in_bytes() >= target_encoded_file_size) {
-                split_file(path_for_compression, group_id, m_parsed_message.get_ts_patt(), archive_writer, file);
+                split_file_and_archive(archive_user_config, path_for_compression, group_id, m_parsed_message.get_ts_patt(), archive_writer);
+            } else if (archive_writer.get_file().get_encoded_size_in_bytes() >= target_encoded_file_size) {
+                split_file(path_for_compression, group_id, m_parsed_message.get_ts_patt(), archive_writer);
             }
 
-            write_message_to_encoded_file(m_parsed_message, archive_writer, file);
+            write_message_to_encoded_file(m_parsed_message, archive_writer);
         }
 
         // Parse remaining content from file
         while (m_message_parser.parse_next_message(true, reader, m_parsed_message)) {
             if (archive_writer.get_data_size_of_dictionaries() >= target_data_size_of_dicts) {
-                split_file_and_archive(archive_user_config, path_for_compression, group_id, m_parsed_message.get_ts_patt(), archive_writer, file);
-            } else if (file->get_encoded_size_in_bytes() >= target_encoded_file_size) {
-                split_file(path_for_compression, group_id, m_parsed_message.get_ts_patt(), archive_writer, file);
+                split_file_and_archive(archive_user_config, path_for_compression, group_id, m_parsed_message.get_ts_patt(), archive_writer);
+            } else if (archive_writer.get_file().get_encoded_size_in_bytes() >= target_encoded_file_size) {
+                split_file(path_for_compression, group_id, m_parsed_message.get_ts_patt(), archive_writer);
             }
 
-            write_message_to_encoded_file(m_parsed_message, archive_writer, file);
+            write_message_to_encoded_file(m_parsed_message, archive_writer);
         }
 
-        close_file_and_mark_ready_for_segment(archive_writer, file);
+        close_file_and_append_to_segment(archive_writer);
     }
 
     bool FileCompressor::try_compressing_as_archive (size_t target_data_size_of_dicts, streaming_archive::writer::Archive::UserConfig& archive_user_config,

components/core/src/clp/utils.cpp

@@ -180,17 +180,9 @@ namespace clp {
         return all_paths_exist;
     }
 
-    streaming_archive::writer::File* create_and_open_file (streaming_archive::writer::Archive& archive_writer, const string& path_for_compression,
-                                                           group_id_t group_id, const boost::uuids::uuid& orig_file_id, size_t split_ix)
-    {
-        auto* file = archive_writer.create_file(path_for_compression, group_id, orig_file_id, split_ix);
-        archive_writer.open_file(*file);
-        return file;
-    }
-
-    void close_file_and_mark_ready_for_segment (streaming_archive::writer::Archive& archive_writer, streaming_archive::writer::File*& file) {
-        archive_writer.close_file(*file);
-        archive_writer.mark_file_ready_for_segment(file);
+    void close_file_and_append_to_segment (streaming_archive::writer::Archive& archive_writer) {
+        archive_writer.close_file();
+        archive_writer.append_file_to_segment();
     }
 
     void split_archive (streaming_archive::writer::Archive::UserConfig& archive_user_config, streaming_archive::writer::Archive& archive_writer) {
@@ -201,31 +193,32 @@ namespace clp {
     }
 
     void split_file (const string& path_for_compression, group_id_t group_id, const TimestampPattern* last_timestamp_pattern,
-                     streaming_archive::writer::Archive& archive_writer, streaming_archive::writer::File*& file)
+                     streaming_archive::writer::Archive& archive_writer)
     {
-        auto orig_file_id = file->get_orig_file_id();
-        auto split_ix = file->get_split_ix();
-        file->set_is_split(true);
-        close_file_and_mark_ready_for_segment(archive_writer, file);
+        const auto& encoded_file = archive_writer.get_file();
+        auto orig_file_id = encoded_file.get_orig_file_id();
+        auto split_ix = encoded_file.get_split_ix();
+        archive_writer.set_file_is_split(true);
+        close_file_and_append_to_segment(archive_writer);
 
-        file = create_and_open_file(archive_writer, path_for_compression, group_id, orig_file_id, ++split_ix);
+        archive_writer.create_and_open_file(path_for_compression, group_id, orig_file_id, ++split_ix);
         // Initialize the file's timestamp pattern to the previous split's pattern
-        archive_writer.change_ts_pattern(*file, last_timestamp_pattern);
+        archive_writer.change_ts_pattern(last_timestamp_pattern);
     }
 
     void split_file_and_archive (streaming_archive::writer::Archive::UserConfig& archive_user_config, const string& path_for_compression, group_id_t group_id,
-                                 const TimestampPattern* last_timestamp_pattern, streaming_archive::writer::Archive& archive_writer,
-                                 streaming_archive::writer::File*& file)
+                                 const TimestampPattern* last_timestamp_pattern, streaming_archive::writer::Archive& archive_writer)
     {
-        auto orig_file_id = file->get_orig_file_id();
-        auto split_ix = file->get_split_ix();
-        file->set_is_split(true);
-        close_file_and_mark_ready_for_segment(archive_writer, file);
+        const auto& encoded_file = archive_writer.get_file();
+        auto orig_file_id = encoded_file.get_orig_file_id();
+        auto split_ix = encoded_file.get_split_ix();
+        archive_writer.set_file_is_split(true);
+        close_file_and_append_to_segment(archive_writer);
 
         split_archive(archive_user_config, archive_writer);
 
-        file = create_and_open_file(archive_writer, path_for_compression, group_id, orig_file_id, ++split_ix);
+        archive_writer.create_and_open_file(path_for_compression, group_id, orig_file_id, ++split_ix);
         // Initialize the file's timestamp pattern to the previous split's pattern
-        archive_writer.change_ts_pattern(*file, last_timestamp_pattern);
+        archive_writer.change_ts_pattern(last_timestamp_pattern);
     }
 }