feat(clp-s): Add the write-side implementation for the archive range index.

components/core/CMakeLists.txt

@@ -317,6 +317,8 @@ set(SOURCE_FILES_clp_s_unitTest
     src/clp_s/JsonParser.hpp
     src/clp_s/PackedStreamReader.cpp
     src/clp_s/PackedStreamReader.hpp
+    src/clp_s/RangeIndexWriter.cpp
+    src/clp_s/RangeIndexWriter.hpp
     src/clp_s/ReaderUtils.cpp
     src/clp_s/ReaderUtils.hpp
     src/clp_s/Schema.cpp

components/core/src/clp_s/ArchiveReaderAdaptor.cpp

@@ -97,6 +97,13 @@ ErrorCode ArchiveReaderAdaptor::try_read_archive_info(ZstdDecompressor& decompre
     return ErrorCodeSuccess;
 }
 
+auto
+ArchiveReaderAdaptor::try_read_unknown_metadata_packet(ZstdDecompressor& decompressor, size_t size)
+        -> ErrorCode {
+    std::vector<char> buffer(size);
+    return decompressor.try_read_exact_length(buffer.data(), buffer.size());
+}
+
 ErrorCode ArchiveReaderAdaptor::load_archive_metadata() {
     constexpr size_t cDecompressorFileReadBufferCapacity = 64 * 1024;
     m_reader = try_create_reader_at_header();
@@ -175,6 +182,7 @@ ErrorCode ArchiveReaderAdaptor::try_read_archive_metadata(ZstdDecompressor& deco
                 rc = try_read_archive_info(decompressor, packet_size);
                 break;
             default:
+                rc = try_read_unknown_metadata_packet(decompressor, packet_size);
                 break;
         }
         if (ErrorCodeSuccess != rc) {

components/core/src/clp_s/ArchiveReaderAdaptor.hpp

@@ -90,6 +90,14 @@ class ArchiveReaderAdaptor {
      */
     ErrorCode try_read_archive_info(ZstdDecompressor& decompressor, size_t size);
 
+    /**
+     * Tries to read an unknown metadata packet from the archive metadata.
+     * @param decompressor
+     * @param size The number of decompressed bytes making up the packet.
+     * @return ErrorCodeSuccess on success or the relevant ErrorCode on failure.
+     */
+    auto try_read_unknown_metadata_packet(ZstdDecompressor& decompressor, size_t size) -> ErrorCode;
+
     /**
      * Tries to create a reader for the archive header.
      * @return A ReaderInterface opened and pointing to the archive header on success.

components/core/src/clp_s/ArchiveWriter.cpp

@@ -58,6 +58,11 @@ void ArchiveWriter::open(ArchiveWriterOption const& option) {
 }
 
 void ArchiveWriter::close() {
+    if (m_range_open) {
+        if (auto const rc = close_current_range(); ErrorCodeSuccess != rc) {
+            throw OperationFailed(rc, __FILENAME__, __LINE__);
+        }
+    }
     auto var_dict_compressed_size = m_var_dict->close();
     auto log_dict_compressed_size = m_log_dict->close();
     auto array_dict_compressed_size = m_array_dict->close();
@@ -145,8 +150,12 @@ void ArchiveWriter::write_archive_metadata(
 
     ZstdCompressor compressor;
     compressor.open(archive_writer, m_compression_level);
-    uint8_t const num_packets{3U};
-    compressor.write_numeric_value(num_packets);
+    uint8_t num_optional_packets{0ULL};
+    if (false == m_range_index_writer.empty()) {
+        ++num_optional_packets;
+    }
+    uint8_t const num_constant_packets{3U};
+    compressor.write_numeric_value<uint8_t>(num_constant_packets + num_optional_packets);
 
     // Write archive info
     ArchiveInfoPacket archive_info{.num_segments = 1};
@@ -173,6 +182,11 @@ void ArchiveWriter::write_archive_metadata(
     compressor.write_numeric_value(static_cast<uint32_t>(encoded_timestamp_dict.size()));
     compressor.write(encoded_timestamp_dict.data(), encoded_timestamp_dict.size());
 
+    // Write range index
+    if (auto rc = m_range_index_writer.write(compressor); ErrorCodeSuccess != rc) {
+        throw OperationFailed(rc, __FILENAME__, __LINE__);
+    }
+
     compressor.close();
 }
 

components/core/src/clp_s/ArchiveWriter.hpp

@@ -1,6 +1,8 @@
 #ifndef CLP_S_ARCHIVEWRITER_HPP
 #define CLP_S_ARCHIVEWRITER_HPP
 
+#include <optional>
+#include <string>
 #include <string_view>
 #include <utility>
 
@@ -9,6 +11,7 @@
 
 #include "archive_constants.hpp"
 #include "DictionaryWriter.hpp"
+#include "RangeIndexWriter.hpp"
 #include "Schema.hpp"
 #include "SchemaMap.hpp"
 #include "SchemaTree.hpp"
@@ -165,6 +168,39 @@ class ArchiveWriter {
      */
     size_t get_data_size();
 
+    /**
+     * Adds a metadata key value pair to the current range in the range index, opening a range if no
+     * range currently exists.
+     * @param key
+     * @param value
+     * @return ErrorCodeSuccess on success or the relevant error code on failure.
+     */
+    template <typename T>
+    [[nodiscard]] auto add_field_to_current_range(std::string const& key, T const& value)
+            -> ErrorCode {
+        if (false == m_range_open) {
+            if (auto const rc = m_range_index_writer.open_range(m_next_log_event_id);
+                ErrorCodeSuccess != rc)
+            {
+                return rc;
+            }
+            m_range_open = true;
+        }
+        return m_range_index_writer.add_value_to_range(key, value);
+    }
+
+    /**
+     * Closes the currently open range in the range index.
+     * @return ErrorCodeSuccess on success or the relevant error code on failure.
+     */
+    [[nodiscard]] auto close_current_range() -> ErrorCode {
+        auto const rc = m_range_index_writer.close_range(m_next_log_event_id);
+        if (ErrorCodeSuccess == rc) {
+            m_range_open = false;
+        }
+        return rc;
+    }
+
 private:
     /**
      * Initializes the schema writer
@@ -250,6 +286,9 @@ class ArchiveWriter {
     FileWriter m_table_metadata_file_writer;
     ZstdCompressor m_tables_compressor;
     ZstdCompressor m_table_metadata_compressor;
+
+    RangeIndexWriter m_range_index_writer;
+    bool m_range_open{false};
 };
 }  // namespace clp_s
 

components/core/src/clp_s/CMakeLists.txt

@@ -112,6 +112,8 @@ set(
         PackedStreamReader.cpp
         PackedStreamReader.hpp
         ParsedMessage.hpp
+        RangeIndexWriter.cpp
+        RangeIndexWriter.hpp
         ReaderUtils.cpp
         ReaderUtils.hpp
         Schema.cpp

components/core/src/clp_s/JsonParser.cpp

@@ -9,13 +9,15 @@
 #include <vector>
 
 #include <absl/container/flat_hash_map.h>
+#include <boost/uuid/uuid_io.hpp>
 #include <curl/curl.h>
 #include <simdjson.h>
 #include <spdlog/spdlog.h>
 
 #include "../clp/ffi/ir_stream/decoding_methods.hpp"
 #include "../clp/ffi/ir_stream/Deserializer.hpp"
 #include "../clp/ffi/ir_stream/IrUnitType.hpp"
+#include "../clp/ffi/ir_stream/protocol_constants.hpp"
 #include "../clp/ffi/KeyValuePairLogEvent.hpp"
 #include "../clp/ffi/SchemaTree.hpp"
 #include "../clp/ffi/Value.hpp"
@@ -480,6 +482,7 @@ void JsonParser::parse_line(ondemand::value line, int32_t parent_node_id, std::s
 }
 
 bool JsonParser::parse() {
+    auto archive_creator_id = boost::uuids::to_string(m_generator());
     for (auto const& path : m_input_paths) {
         auto reader{ReaderUtils::try_create_reader(path, m_network_auth)};
         if (nullptr == reader) {
@@ -504,14 +507,59 @@ bool JsonParser::parse() {
         size_t bytes_consumed_up_to_prev_archive = 0;
         size_t bytes_consumed_up_to_prev_record = 0;
 
+        size_t file_split_number{0ULL};
         int32_t log_event_idx_node_id{};
-        auto add_log_event_idx_node = [&]() {
+        auto initialize_fields_for_archive = [&]() -> bool {
             if (m_record_log_order) {
                 log_event_idx_node_id
                         = add_metadata_field(constants::cLogEventIdxName, NodeType::Integer);
             }
+            if (auto const rc = m_archive_writer->add_field_to_current_range(
+                        std::string{constants::range_index::cFilename},
+                        path.path
+                );
+                ErrorCodeSuccess != rc)
+            {
+                SPDLOG_ERROR(
+                        "Failed to add metadata field \"{}\" ({})",
+                        constants::range_index::cFilename,
+                        static_cast<int64_t>(rc)
+                );
+                return false;
+            }
+            if (auto const rc = m_archive_writer->add_field_to_current_range(
+                        std::string{constants::range_index::cFileSplitNumber},
+                        file_split_number
+                );
+                ErrorCodeSuccess != rc)
+            {
+                SPDLOG_ERROR(
+                        "Failed to add metadata field \"{}\" ({})",
+                        constants::range_index::cFileSplitNumber,
+                        static_cast<int64_t>(rc)
+                );
+                return false;
+            }
+            if (auto const rc = m_archive_writer->add_field_to_current_range(
+                        std::string{constants::range_index::cArchiveCreatorId},
+                        archive_creator_id
+                );
+                ErrorCodeSuccess != rc)
+            {
+                SPDLOG_ERROR(
+                        "Failed to add metadata field \"{}\" ({})",
+                        constants::range_index::cArchiveCreatorId,
+                        static_cast<int64_t>(rc)
+                );
+                return false;
+            }
+            return true;
         };
-        add_log_event_idx_node();
+        if (false == initialize_fields_for_archive()) {
+            m_archive_writer->close();
+            return false;
+        }
+        auto update_fields_after_archive_split = [&]() { ++file_split_number; };
 
         while (json_file_iterator.get_json(json_it)) {
             m_current_schema.clear();
@@ -571,7 +619,11 @@ bool JsonParser::parse() {
                 );
                 bytes_consumed_up_to_prev_archive = bytes_consumed_up_to_prev_record;
                 split_archive();
-                add_log_event_idx_node();
+                update_fields_after_archive_split();
+                if (false == initialize_fields_for_archive()) {
+                    m_archive_writer->close();
+                    return false;
+                }
             }
 
             m_current_parsed_message.clear();
@@ -603,6 +655,12 @@ bool JsonParser::parse() {
             m_archive_writer->close();
             return false;
         }
+
+        if (auto const rc = m_archive_writer->close_current_range(); ErrorCodeSuccess != rc) {
+            SPDLOG_ERROR("Failed to close metadata range: {}", static_cast<int64_t>(rc));
+            m_archive_writer->close();
+            return false;
+        }
     }
     return true;
 }
@@ -887,6 +945,7 @@ void JsonParser::parse_kv_log_event(KeyValuePairLogEvent const& kv) {
 
 auto JsonParser::parse_from_ir() -> bool {
     constexpr size_t cDecompressorReadBufferCapacity{64 * 1024};  // 64 KB
+    auto archive_creator_id = boost::uuids::to_string(m_generator());
     for (auto const& path : m_input_paths) {
         auto reader{ReaderUtils::try_create_reader(path, m_network_auth)};
         if (nullptr == reader) {
@@ -917,14 +976,83 @@ auto JsonParser::parse_from_ir() -> bool {
         auto& deserializer = deserializer_result.value();
         auto& ir_unit_handler{deserializer.get_ir_unit_handler()};
 
+        size_t file_split_number{0ULL};
         int32_t log_event_idx_node_id{};
-        auto add_log_event_idx_node = [&]() {
+        auto initialize_fields_for_archive = [&]() -> bool {
             if (m_record_log_order) {
                 log_event_idx_node_id
                         = add_metadata_field(constants::cLogEventIdxName, NodeType::Integer);
             }
+            if (auto const rc = m_archive_writer->add_field_to_current_range(
+                        std::string{constants::range_index::cFilename},
+                        path.path
+                );
+                ErrorCodeSuccess != rc)
+            {
+                SPDLOG_ERROR(
+                        "Failed to add metadata field \"{}\" ({})",
+                        constants::range_index::cFilename,
+                        static_cast<int64_t>(rc)
+                );
+                return false;
+            }
+            if (auto const rc = m_archive_writer->add_field_to_current_range(
+                        std::string{constants::range_index::cFileSplitNumber},
+                        file_split_number
+                );
+                ErrorCodeSuccess != rc)
+            {
+                SPDLOG_ERROR(
+                        "Failed to add metadata field \"{}\" ({})",
+                        constants::range_index::cFileSplitNumber,
+                        static_cast<int64_t>(rc)
+                );
+                return false;
+            }
+            if (auto const rc = m_archive_writer->add_field_to_current_range(
+                        std::string{constants::range_index::cArchiveCreatorId},
+                        archive_creator_id
+                );
+                ErrorCodeSuccess != rc)
+            {
+                SPDLOG_ERROR(
+                        "Failed to add metadata field \"{}\" ({})",
+                        constants::range_index::cArchiveCreatorId,
+                        static_cast<int64_t>(rc)
+                );
+                return false;
+            }
+            auto const& metadata = deserializer.get_metadata();
+            if (metadata.contains(clp::ffi::ir_stream::cProtocol::Metadata::UserDefinedMetadataKey))
+            {
+                for (auto const& [metadata_key, metadata_value] :
+                     metadata.at(clp::ffi::ir_stream::cProtocol::Metadata::UserDefinedMetadataKey)
+                             .items())
+                {
+                    if (auto const rc = m_archive_writer->add_field_to_current_range(
+                                metadata_key,
+                                metadata_value
+                        );
+                        ErrorCodeSuccess != rc)
+                    {
+                        SPDLOG_ERROR(
+                                "Failed to add metadata field \"{}\" ({})",
+                                metadata_key,
+                                static_cast<int64_t>(rc)
+                        );
+                        return false;
+                    }
+                }
+            }
+            return true;
         };
-        add_log_event_idx_node();
+        if (false == initialize_fields_for_archive()) {
+            decompressor.close();
+            m_archive_writer->close();
+            return false;
+        }
+        auto update_fields_after_archive_split = [&]() { ++file_split_number; };
+
         while (true) {
             auto const kv_log_event_result{deserializer.deserialize_next_ir_unit(decompressor)};
 
@@ -979,7 +1107,12 @@ auto JsonParser::parse_from_ir() -> bool {
                     m_archive_writer->increment_uncompressed_size(curr_pos - last_pos);
                     last_pos = curr_pos;
                     split_archive();
-                    add_log_event_idx_node();
+                    update_fields_after_archive_split();
+                    if (false == initialize_fields_for_archive()) {
+                        m_archive_writer->close();
+                        decompressor.close();
+                        return false;
+                    }
                 }
 
                 ir_unit_handler.clear();
@@ -1004,6 +1137,11 @@ auto JsonParser::parse_from_ir() -> bool {
         curr_pos = decompressor.get_pos();
         m_archive_writer->increment_uncompressed_size(curr_pos - last_pos);
         decompressor.close();
+        if (auto const rc = m_archive_writer->close_current_range(); ErrorCodeSuccess != rc) {
+            SPDLOG_ERROR("Failed to close metadata range: {}", static_cast<int64_t>(rc));
+            m_archive_writer->close();
+            return false;
+        }
     }
     return true;
 }