Merge remote-tracking branch 'upstream/main' into disable-log-order-no-range-index

Author: gibber9809

components/core/src/clp_s/ArchiveWriter.cpp

@@ -7,7 +7,6 @@
 #include <nlohmann/json.hpp>
 #include <spdlog/spdlog.h>
 
-#include "../clp/streaming_archive/Constants.hpp"
 #include "archive_constants.hpp"
 #include "Defs.hpp"
 #include "SchemaTree.hpp"
@@ -58,7 +57,7 @@ void ArchiveWriter::open(ArchiveWriterOption const& option) {
     m_array_dict->open(array_dict_path, m_compression_level, UINT64_MAX);
 }
 
-void ArchiveWriter::close() {
+auto ArchiveWriter::close(bool is_split) -> ArchiveStats {
     if (m_range_open) {
         if (auto const rc = close_current_range(); ErrorCodeSuccess != rc) {
             throw OperationFailed(rc, __FILENAME__, __LINE__);
@@ -87,15 +86,16 @@ void ArchiveWriter::close() {
         offset += original_size;
     }
 
+    nlohmann::json archive_range_index;
     if (m_single_file_archive) {
-        write_single_file_archive(files);
+        archive_range_index = write_single_file_archive(files);
     } else {
         FileWriter header_and_metadata_writer;
         header_and_metadata_writer.open(
                 m_archive_path + constants::cArchiveHeaderFile,
                 FileWriter::OpenMode::CreateForWriting
         );
-        write_archive_metadata(header_and_metadata_writer, files);
+        archive_range_index = write_archive_metadata(header_and_metadata_writer, files);
         size_t metadata_size = header_and_metadata_writer.get_pos() - sizeof(ArchiveHeader);
 
         m_compressed_size
@@ -107,8 +107,18 @@ void ArchiveWriter::close() {
         header_and_metadata_writer.close();
     }
 
+    ArchiveStats archive_stats{
+            m_id,
+            m_timestamp_dict.get_begin_timestamp(),
+            m_timestamp_dict.get_end_timestamp(),
+            m_uncompressed_size,
+            m_compressed_size,
+            archive_range_index,
+            is_split
+    };
     if (m_print_archive_stats) {
-        print_archive_stats();
+        std::cout << archive_stats.as_string() << '\n';
+        std::cout << std::flush;
     }
 
     m_id_to_schema_writer.clear();
@@ -123,14 +133,17 @@ void ArchiveWriter::close() {
     m_authoritative_timestamp_namespace.clear();
     m_matched_timestamp_prefix_length = 0ULL;
     m_matched_timestamp_prefix_node_id = constants::cRootNodeId;
+    return archive_stats;
 }
 
-void ArchiveWriter::write_single_file_archive(std::vector<ArchiveFileInfo> const& files) {
+auto ArchiveWriter::write_single_file_archive(std::vector<ArchiveFileInfo> const& files)
+        -> nlohmann::json {
     std::string single_file_archive_path = (std::filesystem::path(m_archives_dir) / m_id).string();
     FileWriter archive_writer;
     archive_writer.open(single_file_archive_path, FileWriter::OpenMode::CreateForWriting);
 
-    write_archive_metadata(archive_writer, files);
+    // Avoid brace initialization to avoid wrapping nlohmann::json return value in a JSON array
+    auto archive_range_index(write_archive_metadata(archive_writer, files));
     size_t metadata_section_size = archive_writer.get_pos() - sizeof(ArchiveHeader);
     write_archive_files(archive_writer, files);
     m_compressed_size = archive_writer.get_pos();
@@ -141,12 +154,13 @@ void ArchiveWriter::write_single_file_archive(std::vector<ArchiveFileInfo> const
     if (false == std::filesystem::remove(m_archive_path, ec)) {
         throw OperationFailed(ErrorCodeFileExists, __FILENAME__, __LINE__);
     }
+    return archive_range_index;
 }
 
-void ArchiveWriter::write_archive_metadata(
+auto ArchiveWriter::write_archive_metadata(
         FileWriter& archive_writer,
         std::vector<ArchiveFileInfo> const& files
-) {
+) -> nlohmann::json {
     archive_writer.seek_from_begin(sizeof(ArchiveHeader));
 
     ZstdCompressor compressor;
@@ -184,11 +198,15 @@ void ArchiveWriter::write_archive_metadata(
     compressor.write(encoded_timestamp_dict.data(), encoded_timestamp_dict.size());
 
     // Write range index
-    if (auto rc = m_range_index_writer.write(compressor); ErrorCodeSuccess != rc) {
+    nlohmann::json archive_range_index;
+    if (auto rc = m_range_index_writer.write(compressor, archive_range_index);
+        ErrorCodeSuccess != rc)
+    {
         throw OperationFailed(rc, __FILENAME__, __LINE__);
     }
 
     compressor.close();
+    return archive_range_index;
 }
 
 void ArchiveWriter::write_archive_files(
@@ -441,15 +459,4 @@ std::pair<size_t, size_t> ArchiveWriter::store_tables() {
 
     return {table_metadata_compressed_size, table_compressed_size};
 }
-
-auto ArchiveWriter::print_archive_stats() const -> void {
-    namespace Archive = clp::streaming_archive::cMetadataDB::Archive;
-    nlohmann::json json_msg
-            = {{Archive::Id, m_id},
-               {Archive::BeginTimestamp, m_timestamp_dict.get_begin_timestamp()},
-               {Archive::EndTimestamp, m_timestamp_dict.get_end_timestamp()},
-               {Archive::UncompressedSize, m_uncompressed_size},
-               {Archive::Size, m_compressed_size}};
-    std::cout << json_msg.dump(-1, ' ', true, nlohmann::json::error_handler_t::ignore) << std::endl;
-}
 }  // namespace clp_s

components/core/src/clp_s/ArchiveWriter.hpp

@@ -8,7 +8,9 @@
 
 #include <boost/uuid/uuid.hpp>
 #include <boost/uuid/uuid_io.hpp>
+#include <nlohmann/json.hpp>
 
+#include "../clp/streaming_archive/Constants.hpp"
 #include "archive_constants.hpp"
 #include "DictionaryWriter.hpp"
 #include "RangeIndexWriter.hpp"
@@ -31,6 +33,70 @@ struct ArchiveWriterOption {
     std::string authoritative_timestamp_namespace;
 };
 
+class ArchiveStats {
+public:
+    // Constructors
+    explicit ArchiveStats(
+            std::string id,
+            epochtime_t begin_timestamp,
+            epochtime_t end_timestamp,
+            size_t uncompressed_size,
+            size_t compressed_size,
+            nlohmann::json range_index,
+            bool is_split
+    )
+            : m_id{id},
+              m_begin_timestamp{begin_timestamp},
+              m_end_timestamp{end_timestamp},
+              m_uncompressed_size{uncompressed_size},
+              m_compressed_size{compressed_size},
+              m_range_index(std::move(range_index)),  // Avoid {} to prevent wrapping in JSON array.
+              m_is_split{is_split} {}
+
+    // Methods
+    /**
+     * @return The contents of `ArchiveStats` as a JSON object in a string.
+     */
+    [[nodiscard]] auto as_string() const -> std::string {
+        namespace Archive = clp::streaming_archive::cMetadataDB::Archive;
+        namespace File = clp::streaming_archive::cMetadataDB::File;
+        constexpr std::string_view cRangeIndex{"range_index"};
+
+        nlohmann::json json_msg
+                = {{Archive::Id, m_id},
+                   {Archive::BeginTimestamp, m_begin_timestamp},
+                   {Archive::EndTimestamp, m_end_timestamp},
+                   {Archive::UncompressedSize, m_uncompressed_size},
+                   {Archive::Size, m_compressed_size},
+                   {File::IsSplit, m_is_split},
+                   {cRangeIndex, m_range_index}};
+        return json_msg.dump(-1, ' ', false, nlohmann::json::error_handler_t::ignore);
+    }
+
+    auto get_id() const -> std::string const& { return m_id; }
+
+    auto get_begin_timestamp() const -> epochtime_t { return m_begin_timestamp; }
+
+    auto get_end_timestamp() const -> epochtime_t { return m_end_timestamp; }
+
+    auto get_uncompressed_size() const -> size_t { return m_uncompressed_size; }
+
+    auto get_compressed_size() const -> size_t { return m_compressed_size; }
+
+    auto get_range_index() const -> nlohmann::json const& { return m_range_index; }
+
+    auto get_is_split() const -> bool { return m_is_split; }
+
+private:
+    std::string m_id;
+    epochtime_t m_begin_timestamp{};
+    epochtime_t m_end_timestamp{};
+    size_t m_uncompressed_size{};
+    size_t m_compressed_size{};
+    nlohmann::json m_range_index;
+    bool m_is_split{};
+};
+
 class ArchiveWriter {
 public:
     class OperationFailed : public TraceableException {
@@ -80,9 +146,11 @@ class ArchiveWriter {
     void open(ArchiveWriterOption const& option);
 
     /**
-     * Closes the archive writer
+     * Closes the archive writer.
+     * @param is_split Whether the last file ingested into the archive is split.
+     * @return Statistics for the newly-written archive.
      */
-    void close();
+    [[nodiscard]] auto close(bool is_split = false) -> ArchiveStats;
 
     /**
      * Appends a message to the archive writer
@@ -220,16 +288,20 @@ class ArchiveWriter {
     /**
      * Writes the archive to a single file
      * @param files
+     * @return The archive range index as a JSON object.
      */
-    void write_single_file_archive(std::vector<ArchiveFileInfo> const& files);
+    [[nodiscard]] auto write_single_file_archive(std::vector<ArchiveFileInfo> const& files)
+            -> nlohmann::json;
 
     /**
      * Writes the metadata section of an archive.
      * @param archive_writer
      * @param files
+     * @return The archive range index as a JSON object.
      */
-    void
-    write_archive_metadata(FileWriter& archive_writer, std::vector<ArchiveFileInfo> const& files);
+    [[nodiscard]] auto
+    write_archive_metadata(FileWriter& archive_writer, std::vector<ArchiveFileInfo> const& files)
+            -> nlohmann::json;
 
     /**
      * Writes the file section of the single file archive
@@ -245,11 +317,6 @@ class ArchiveWriter {
      */
     void write_archive_header(FileWriter& archive_writer, size_t metadata_section_size);
 
-    /**
-     * Prints the archive's statistics (id, uncompressed size, compressed size, etc.)
-     */
-    auto print_archive_stats() const -> void;
-
     static constexpr size_t cReadBlockSize = 4 * 1024;
 
     size_t m_encoded_message_size{};

components/core/src/clp_s/CMakeLists.txt

@@ -232,6 +232,7 @@ target_link_libraries(
         PUBLIC
         absl::flat_hash_map
         clp_s::io
+        nlohmann_json::nlohmann_json
         simdjson::simdjson
         PRIVATE
         Boost::url
@@ -240,7 +241,6 @@ target_link_libraries(
         ${CURL_LIBRARIES}
         fmt::fmt
         msgpack-cxx
-        nlohmann_json::nlohmann_json
         spdlog::spdlog
 )