feat(clp-core): Add the write path for single-file archives.

components/core/CMakeLists.txt

@@ -573,6 +573,9 @@ set(SOURCE_FILES_unitTest
         src/clp/streaming_archive/reader/Segment.hpp
         src/clp/streaming_archive/reader/SegmentManager.cpp
         src/clp/streaming_archive/reader/SegmentManager.hpp
+        src/clp/streaming_archive/single_file_archive/Defs.hpp
+        src/clp/streaming_archive/single_file_archive/writer.cpp
+        src/clp/streaming_archive/single_file_archive/writer.hpp
         src/clp/streaming_archive/writer/Archive.cpp
         src/clp/streaming_archive/writer/Archive.hpp
         src/clp/streaming_archive/writer/File.cpp

components/core/src/clp/clp/CMakeLists.txt

@@ -108,6 +108,9 @@ set(
         ../streaming_archive/reader/Segment.hpp
         ../streaming_archive/reader/SegmentManager.cpp
         ../streaming_archive/reader/SegmentManager.hpp
+        ../streaming_archive/single_file_archive/Defs.hpp
+        ../streaming_archive/single_file_archive/writer.cpp
+        ../streaming_archive/single_file_archive/writer.hpp
         ../streaming_archive/writer/Archive.cpp
         ../streaming_archive/writer/Archive.hpp
         ../streaming_archive/writer/File.cpp

components/core/src/clp/clp/CommandLineArguments.cpp

@@ -373,6 +373,10 @@ CommandLineArguments::parse_arguments(int argc, char const* argv[]) {
                             ->default_value(m_schema_file_path),
                     "Path to a schema file. If not specified, heuristics are used to determine "
                     "dictionary variables. See README-Schema.md for details."
+            )(
+                    "single-file-archive",
+                    po::bool_switch(&m_single_file_archive),
+                    "Output archive as a single-file archive"
             );
 
             po::options_description all_compression_options;

components/core/src/clp/clp/CommandLineArguments.hpp

@@ -23,6 +23,7 @@ class CommandLineArguments : public CommandLineArgumentsBase {
     explicit CommandLineArguments(std::string const& program_name)
             : CommandLineArgumentsBase(program_name),
               m_show_progress(false),
+              m_single_file_archive(false),
               m_sort_input_files(true),
               m_print_archive_stats_progress(false),
               m_target_segment_uncompressed_size(1L * 1024 * 1024 * 1024),
@@ -45,6 +46,8 @@ class CommandLineArguments : public CommandLineArgumentsBase {
 
     bool show_progress() const { return m_show_progress; }
 
+    [[nodiscard]] auto single_file_archive() const -> bool { return m_single_file_archive; }
+
     bool sort_input_files() const { return m_sort_input_files; }
 
     bool print_archive_stats_progress() const { return m_print_archive_stats_progress; }
@@ -92,6 +95,7 @@ class CommandLineArguments : public CommandLineArgumentsBase {
     std::string m_output_dir;
     std::string m_schema_file_path;
     bool m_show_progress;
+    bool m_single_file_archive;
     bool m_print_archive_stats_progress;
     size_t m_target_encoded_file_size;
     size_t m_target_segment_uncompressed_size;

components/core/src/clp/clp/compression.cpp

@@ -107,6 +107,7 @@ bool compress(
     archive_user_config.global_metadata_db = global_metadata_db.get();
     archive_user_config.print_archive_stats_progress
             = command_line_args.print_archive_stats_progress();
+    archive_user_config.use_single_file_archive = command_line_args.single_file_archive();
 
     // Open Archive
     streaming_archive::writer::Archive archive_writer;
@@ -135,7 +136,8 @@ bool compress(
         );
     }
     for (auto it = files_to_compress.cbegin(); it != files_to_compress.cend(); ++it) {
-        if (archive_writer.get_data_size_of_dictionaries() >= target_data_size_of_dictionaries) {
+        if (archive_writer.get_data_size_of_dictionaries() >= target_data_size_of_dictionaries)
+        {
             split_archive(archive_user_config, archive_writer);
         }
         if (false
@@ -163,7 +165,8 @@ bool compress(
          file_group_id_comparator);
     // Compress grouped files
     for (auto const& file_to_compress : grouped_files_to_compress) {
-        if (archive_writer.get_data_size_of_dictionaries() >= target_data_size_of_dictionaries) {
+        if (archive_writer.get_data_size_of_dictionaries() >= target_data_size_of_dictionaries)
+        {
             split_archive(archive_user_config, archive_writer);
         }
         if (false

components/core/src/clp/streaming_archive/single_file_archive/Defs.hpp

@@ -0,0 +1,61 @@
+#ifndef CLP_STREAMING_ARCHIVE_SINGLE_FILE_ARCHIVE_DEFS_HPP
+#define CLP_STREAMING_ARCHIVE_SINGLE_FILE_ARCHIVE_DEFS_HPP
+
+#include <cstdint>
+#include <string>
+
+#include "../../Defs.h"
+#include "../ArchiveMetadata.hpp"
+#include "../Constants.hpp"
+#include "msgpack.hpp"
+
+namespace clp::streaming_archive::single_file_archive {
+
+using single_file_archive_format_version_t = uint32_t;
+
+// Single file archive version.
+constexpr uint8_t cVersionMajor{0};
+constexpr uint8_t cVersionMinor{1};
+constexpr uint16_t cVersionPatch{1};
+constexpr single_file_archive_format_version_t cVersion{cVersionMajor << 24 | cVersionMinor << 16 | cVersionPatch};
+
+static constexpr size_t cNumMagicNumberChars{4};
+static constexpr std::array<uint8_t, cNumMagicNumberChars>
+        cUnstructuredSfaMagicNumber{'Y', 'C', 'L', 'P'};
+static constexpr std::string_view cUnstructuredSfaExtension{".clp"};
+
+static constexpr size_t cNumStaticFiles{6};
+constexpr std::array<char const*, cNumStaticFiles> cStaticArchiveFileNames{
+        cMetadataFileName,
+        cMetadataDBFileName,
+        cLogTypeDictFilename,
+        cLogTypeSegmentIndexFilename,
+        cVarDictFilename,
+        cVarSegmentIndexFilename
+};
+
+static constexpr size_t cNumUnused{6};
+
+struct __attribute__((packed)) SingleFileArchiveHeader {
+    std::array<uint8_t, cNumMagicNumberChars> magic;
+    single_file_archive_format_version_t version;
+    uint64_t metadata_size;
+    std::array<uint64_t, cNumUnused> unused;
+};
+
+struct FileInfo {
+    std::string name;
+    uint64_t offset;
+    // Variables are renamed when serialized to match single-file archive specification.
+    MSGPACK_DEFINE_MAP(MSGPACK_NVP("n", name), MSGPACK_NVP("o", offset));
+};
+
+struct SingleFileArchiveMetadata {
+    std::vector<FileInfo> archive_files;
+    ArchiveMetadata archive_metadata;
+    uint64_t num_segments;
+    MSGPACK_DEFINE_MAP(archive_files, archive_metadata, num_segments);
+};
+}  // namespace clp::streaming_archive::single_file_archive
+
+#endif  // CLP_STREAMING_ARCHIVE_SINGLE_FILE_ARCHIVE_DEFS_HPP

components/core/src/clp/streaming_archive/single_file_archive/writer.cpp

@@ -0,0 +1,236 @@
+#include "writer.hpp"
+
+#include <cstddef>
+#include <cstring>
+#include <filesystem>
+#include <sstream>
+#include <vector>
+
+#include <fmt/core.h>
+#include <msgpack.hpp>
+#include <spdlog.h>
+
+#include "../../Defs.h"
+#include "../../ErrorCode.hpp"
+#include "../../FileReader.hpp"
+#include "../../FileWriter.hpp"
+#include "../../TraceableException.hpp"
+#include "../ArchiveMetadata.hpp"
+#include "../Constants.hpp"
+#include "Defs.hpp"
+
+namespace clp::streaming_archive::single_file_archive {
+
+namespace {
+constexpr size_t cReadBlockSize = 4096;
+
+/**
+ * Gets the size of a file specified by `file_path` and adds it to file section `offset`.
+ * @param file_path
+ * @param[out] offset File section offset for the single-file archive. The returned offset
+ * represents the starting position of the next file in single-file archive.
+ * @throws OperationFailed if error getting file size.
+ */
+auto get_file_size_and_update_offset(std::filesystem::path const& file_path, uint64_t& offset)
+        -> void;
+
+/**
+ * Generates metadata for the file section of a single-file archive. The metadata consists
+ * of a list of file names and their corresponding starting offsets.
+ *
+ * @param multi_file_archive_path
+ * @param next_segment_id
+ * @return Vector containing a `FileInfo` struct for every file in the multi-file archive.
+ * @throws Propagates `update_offset`'s exceptions.
+ */
+[[nodiscard]] auto
+get_file_infos(std::filesystem::path const& multi_file_archive_path, segment_id_t next_segment_id)
+        -> std::vector<FileInfo>;
+
+/**
+ * Combines file section metadata, multi-file archive metadata, and the number of segments into
+ * single-file archive metadata. Once combined, serializes the metadata into MsgPack format.
+ *
+ * @param multi_file_archive_metadata
+ * @param multi_file_archive_path
+ * @param next_segment_id
+ * @return Packed metadata.
+ */
+[[nodiscard]] auto pack_single_file_archive_metadata(
+        ArchiveMetadata const& multi_file_archive_metadata,
+        std::filesystem::path const& multi_file_archive_path,
+        segment_id_t next_segment_id
+) -> std::stringstream;
+
+/**
+ * Writes single-file archive header.
+ *
+ * @param single_file_archive_writer
+ * @param packed_metadata_size
+ */
+auto write_archive_header(FileWriter& single_file_archive_writer, size_t packed_metadata_size) -> void;
+
+/**
+ * Reads the content of a file and writes it to the single-file archive.
+ * @param file_path
+ * @param single_file_archive_writer
+ * @throws OperationFailed if reading the file fails.
+ */
+auto write_archive_file(std::string const& file_path, FileWriter& single_file_archive_writer) -> void;
+
+/**
+ * Iterates over files in the multi-file archive and copies their contents to the single-file archive.
+ * Skips metadata file, which should be handled by `write_archive_metadata` method.
+ *
+ * @param single_file_archive_writer
+ * @param multi_file_archive_path
+ * @param next_segment_id
+ * @throws Propagates `update_offset`'s exceptions.
+ */
+auto write_archive_files(
+        FileWriter& single_file_archive_writer,
+        std::filesystem::path const& multi_file_archive_path,
+        segment_id_t next_segment_id
+) -> void;
+
+auto get_file_size_and_update_offset(std::filesystem::path const& file_path, uint64_t& offset)
+        -> void {
+    try {
+        auto size = std::filesystem::file_size(file_path);
+        offset += size;
+    } catch (std::filesystem::filesystem_error const& e) {
+        throw OperationFailed(
+                ErrorCode_Failure,
+                __FILENAME__,
+                __LINE__,
+                fmt::format("Failed to get file size: {}", e.what())
+        );
+    }
+}
+
+auto
+get_file_infos(std::filesystem::path const& multi_file_archive_path, segment_id_t next_segment_id)
+        -> std::vector<FileInfo> {
+    std::vector<FileInfo> files;
+    uint64_t offset = 0;
+
+    for (auto const& static_archive_file_name : cStaticArchiveFileNames) {
+        files.emplace_back(FileInfo{std::string(static_archive_file_name), offset});
+        get_file_size_and_update_offset(multi_file_archive_path / static_archive_file_name, offset);
+    }
+
+    std::filesystem::path segment_dir_path = multi_file_archive_path / cSegmentsDirname;
+
+    for (size_t i = 0; i < next_segment_id; ++i) {
+        auto const segment_id = std::to_string(i);
+        files.emplace_back(FileInfo{segment_id, offset});
+        get_file_size_and_update_offset(segment_dir_path / segment_id, offset);
+    }
+
+    // Add sentinel indicating total size of all files.
+    files.emplace_back(FileInfo{"", offset});
+
+    return files;
+}
+
+auto pack_single_file_archive_metadata(
+        ArchiveMetadata const& multi_file_archive_metadata,
+        std::filesystem::path const& multi_file_archive_path,
+        segment_id_t next_segment_id
+) -> std::stringstream {
+    SingleFileArchiveMetadata single_file_archive{
+            .archive_files = get_file_infos(multi_file_archive_path, next_segment_id),
+            .archive_metadata = multi_file_archive_metadata,
+            .num_segments = next_segment_id,
+    };
+
+    std::stringstream buf;
+    msgpack::pack(buf, single_file_archive);
+
+    return buf;
+}
+
+auto write_archive_header(FileWriter& single_file_archive_writer, size_t packed_metadata_size) -> void {
+    SingleFileArchiveHeader header{
+            .magic = cUnstructuredSfaMagicNumber,
+            .version = cVersion,
+            .metadata_size = packed_metadata_size,
+            .unused{}
+    };
+
+    single_file_archive_writer.write(reinterpret_cast<char const*>(&header), sizeof(header));
+}
+
+auto write_archive_file(std::filesystem::path const& file_path, FileWriter& single_file_archive_writer)
+        -> void {
+    FileReader reader(file_path.string());
+    std::array<char, cReadBlockSize> read_buffer{};
+    while (true) {
+        size_t num_bytes_read{};
+        ErrorCode const error_code
+                = reader.try_read(read_buffer.data(), cReadBlockSize, num_bytes_read);
+        if (ErrorCode_EndOfFile == error_code) {
+            break;
+        }
+        if (ErrorCode_Success != error_code) {
+            throw OperationFailed(error_code, __FILENAME__, __LINE__);
+        }
+        single_file_archive_writer.write(read_buffer.data(), num_bytes_read);
+    }
+}
+
+auto write_archive_files(
+        FileWriter& single_file_archive_writer,
+        std::filesystem::path const& multi_file_archive_path,
+        segment_id_t next_segment_id
+) -> void {
+    for (auto const& static_archive_file_name : cStaticArchiveFileNames) {
+        std::filesystem::path static_archive_file_path
+                = multi_file_archive_path / static_archive_file_name;
+        write_archive_file(static_archive_file_path, single_file_archive_writer);
+    }
+
+    std::filesystem::path segment_dir_path = multi_file_archive_path / cSegmentsDirname;
+    for (size_t i = 0; i < next_segment_id; ++i) {
+        std::filesystem::path segment_path = segment_dir_path / std::to_string(i);
+        write_archive_file(segment_path, single_file_archive_writer);
+    }
+}
+}  // namespace
+
+auto write_single_file_archive(
+        ArchiveMetadata const& multi_file_archive_metadata,
+        std::filesystem::path const& multi_file_archive_path,
+        segment_id_t next_segment_id
+) -> void {
+    FileWriter single_file_archive_writer;
+    std::filesystem::path single_file_archive_path
+            = multi_file_archive_path.string()
+              + std::string(single_file_archive::cUnstructuredSfaExtension);
+
+    if (std::filesystem::exists(single_file_archive_path)) {
+        throw OperationFailed(ErrorCode_Failure, __FILENAME__, __LINE__);
+    }
+
+    single_file_archive_writer.open(
+            single_file_archive_path.string(),
+            FileWriter::OpenMode::CREATE_FOR_WRITING
+    );
+
+    auto const packed_metadata = pack_single_file_archive_metadata(
+            multi_file_archive_metadata,
+            multi_file_archive_path,
+            next_segment_id
+    );
+
+    write_archive_header(single_file_archive_writer, packed_metadata.str().size());
+    write_archive_files(single_file_archive_writer, multi_file_archive_path, next_segment_id);
+
+    single_file_archive_writer.close();
+    try {
+        std::filesystem::remove_all(multi_file_archive_path);
+    } catch (std::filesystem::filesystem_error& e) {
+        throw OperationFailed(ErrorCode_Failure, __FILENAME__, __LINE__);
+    }
+}
+}  // namespace clp::streaming_archive::single_file_archive