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

components/core/src/clp_s/ArchiveWriter.cpp

@@ -11,6 +11,7 @@ void ArchiveWriter::open(ArchiveWriterOption const& option) {
     m_id = boost::uuids::to_string(option.id);
     m_compression_level = option.compression_level;
     m_print_archive_stats = option.print_archive_stats;
+    m_single_file_archive = option.single_file_archive;
     auto archive_path = boost::filesystem::path(option.archives_dir) / m_id;
 
     boost::system::error_code boost_error_code;
@@ -43,13 +44,37 @@ void ArchiveWriter::open(ArchiveWriterOption const& option) {
 }
 
 void ArchiveWriter::close() {
-    m_compressed_size += m_var_dict->close();
-    m_compressed_size += m_log_dict->close();
-    m_compressed_size += m_array_dict->close();
-    m_compressed_size += m_timestamp_dict->close();
-    m_compressed_size += m_schema_tree.store(m_archive_path, m_compression_level);
-    m_compressed_size += m_schema_map.store(m_archive_path, m_compression_level);
-    m_compressed_size += store_tables();
+    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();
+    auto timestamp_dict_compressed_size = m_timestamp_dict->close();
+    auto schema_tree_compressed_size = m_schema_tree.store(m_archive_path, m_compression_level);
+    auto schema_map_compressed_size = m_schema_map.store(m_archive_path, m_compression_level);
+    auto [table_metadata_compressed_size, table_compressed_size] = store_tables();
+
+    if (m_single_file_archive) {
+        std::vector<ArchiveFileInfo> files{
+                {constants::cArchiveSchemaTreeFile, schema_tree_compressed_size},
+                {constants::cArchiveSchemaMapFile, schema_map_compressed_size},
+                {constants::cArchiveVarDictFile, var_dict_compressed_size},
+                {constants::cArchiveLogDictFile, log_dict_compressed_size},
+                {constants::cArchiveArrayDictFile, array_dict_compressed_size},
+                {constants::cArchiveTableMetadataFile, table_metadata_compressed_size},
+                {constants::cArchiveTablesFile, table_compressed_size}
+        };
+        uint64_t offset = 0;
+        for (auto& file : files) {
+            uint64_t original_size = file.o;
+            file.o = offset;
+            offset += original_size;
+        }
+        write_single_file_archive(files, timestamp_dict_compressed_size);
+    } else {
+        m_compressed_size = var_dict_compressed_size + log_dict_compressed_size
+                            + array_dict_compressed_size + timestamp_dict_compressed_size
+                            + schema_tree_compressed_size + schema_map_compressed_size
+                            + table_metadata_compressed_size + table_compressed_size;
+    }
 
     if (m_metadata_db) {
         update_metadata_db();
@@ -67,6 +92,113 @@ void ArchiveWriter::close() {
     m_compressed_size = 0UL;
 }
 
+void ArchiveWriter::write_single_file_archive(
+        std::vector<ArchiveFileInfo> const& files,
+        size_t timestamp_dict_compressed_size
+) {
+    std::string archive_path = m_archive_path + constants::cArchiveFile;
+    FileWriter archive_writer;
+    archive_writer.open(archive_path, FileWriter::OpenMode::CreateForWriting);
+
+    write_archive_metadata(archive_writer, files, timestamp_dict_compressed_size);
+    size_t metadata_section_size = archive_writer.get_pos() - sizeof(ArchiveHeader);
+    write_archive_files(archive_writer, files);
+    m_compressed_size = archive_writer.get_pos();
+    write_archive_header(archive_writer, metadata_section_size);
+
+    archive_writer.close();
+}
+
+void ArchiveWriter::write_archive_metadata(
+        FileWriter& archive_writer,
+        std::vector<ArchiveFileInfo> const& files,
+        size_t timestamp_dict_compressed_size
+) {
+    archive_writer.seek_from_begin(sizeof(ArchiveHeader));
+
+    ZstdCompressor compressor;
+    compressor.open(archive_writer, m_compression_level);
+    compressor.write_numeric_value(3);  // Number of packets
+
+    // Write archive info
+    ArchiveInfoPacket archive_info{.num_segments = 1};
+    std::stringstream msgpack_buffer;
+    msgpack::pack(msgpack_buffer, archive_info);
+    std::string archive_info_str = msgpack_buffer.str();
+    compressor.write_numeric_value(ArchiveMetadataPacketType::ArchiveInfo);
+    compressor.write_numeric_value(archive_info_str.size());
+    compressor.write_string(archive_info_str);
+
+    // Write archive file info
+    ArchiveFileInfoPacket archive_file_info{.files{files}};
+    msgpack_buffer.clear();
+    msgpack::pack(msgpack_buffer, archive_file_info);
+    std::string archive_file_info_str = msgpack_buffer.str();
+    compressor.write_numeric_value(ArchiveMetadataPacketType::ArchiveFileInfo);
+    compressor.write_numeric_value(archive_file_info_str.size());
+    compressor.write_string(archive_file_info_str);
+
+    // Write timestamp dictionary
+    compressor.write_numeric_value(ArchiveMetadataPacketType::TimestampDictionary);
+    compressor.write_numeric_value(timestamp_dict_compressed_size);
+    std::string timestamp_dict_path = m_archive_path + constants::cArchiveTimestampDictFile;
+    FileReader timestamp_dict_reader;
+    timestamp_dict_reader.open(timestamp_dict_path);
+    char read_buffer[cReadBlockSize];
+    while (true) {
+        size_t num_bytes_read{0};
+        ErrorCode error_code
+                = timestamp_dict_reader.try_read(read_buffer, cReadBlockSize, num_bytes_read);
+        if (ErrorCodeSuccess != error_code) {
+            break;
+        }
+        compressor.write(read_buffer, num_bytes_read);
+    }
+    timestamp_dict_reader.close();
+    std::filesystem::remove(timestamp_dict_path);
+    compressor.close();
+}
+
+void ArchiveWriter::write_archive_files(
+        FileWriter& archive_writer,
+        std::vector<ArchiveFileInfo> const& files
+) {
+    for (auto const& file : files) {
+        std::string file_path = m_archive_path + file.n;
+        FileReader reader;
+        reader.open(file_path);
+        char read_buffer[cReadBlockSize];
+        while (true) {
+            size_t num_bytes_read{0};
+            ErrorCode const error_code
+                    = reader.try_read(read_buffer, cReadBlockSize, num_bytes_read);
+            if (ErrorCodeSuccess != error_code) {
+                break;
+            }
+            archive_writer.write(read_buffer, num_bytes_read);
+        }
+        reader.close();
+        boost::filesystem::remove(file_path);
+    }
+}
+
+void ArchiveWriter::write_archive_header(FileWriter& archive_writer, size_t metadata_section_size) {
+    ArchiveHeader header{
+            .magic_number{0},
+            .version
+            = (cArchiveMajorVersion << 24) | (cArchiveMinorVersion << 16) | cArchivePatchVersion,
+            .uncompressed_size = m_uncompressed_size,
+            .compressed_size = m_compressed_size,
+            .reserved_padding{0},
+            .metadata_section_size = static_cast<uint32_t>(metadata_section_size),
+            .compression_type = static_cast<uint16_t>(ArchiveCompressionType::Zstd),
+            .padding = 0
+    };
+    std::memcpy(&header.magic_number, "ARCHIVES", sizeof(header.magic_number));
+    archive_writer.seek_from_begin(0);
+    archive_writer.write(reinterpret_cast<char const*>(&header), sizeof(header));
+}
+
 void ArchiveWriter::append_message(
         int32_t schema_id,
         Schema const& schema,
@@ -127,8 +259,7 @@ void ArchiveWriter::initialize_schema_writer(SchemaWriter* writer, Schema const&
     }
 }
 
-size_t ArchiveWriter::store_tables() {
-    size_t compressed_size = 0;
+std::pair<size_t, size_t> ArchiveWriter::store_tables() {
     m_tables_file_writer.open(
             m_archive_path + constants::cArchiveTablesFile,
             FileWriter::OpenMode::CreateForWriting
@@ -153,13 +284,13 @@ size_t ArchiveWriter::store_tables() {
     }
     m_table_metadata_compressor.close();
 
-    compressed_size += m_table_metadata_file_writer.get_pos();
-    compressed_size += m_tables_file_writer.get_pos();
+    auto table_metadata_compressed_size = m_table_metadata_file_writer.get_pos();
+    auto table_compressed_size = m_tables_file_writer.get_pos();
 
     m_table_metadata_file_writer.close();
     m_tables_file_writer.close();
 
-    return compressed_size;
+    return {table_metadata_compressed_size, table_compressed_size};
 }
 
 void ArchiveWriter::update_metadata_db() {

components/core/src/clp_s/ArchiveWriter.hpp

@@ -13,6 +13,7 @@
 #include "SchemaMap.hpp"
 #include "SchemaTree.hpp"
 #include "SchemaWriter.hpp"
+#include "SingleFileArchiveDefs.hpp"
 #include "TimestampDictionaryWriter.hpp"
 
 namespace clp_s {
@@ -21,6 +22,7 @@ struct ArchiveWriterOption {
     std::string archives_dir;
     int compression_level;
     bool print_archive_stats;
+    bool single_file_archive;
 };
 
 class ArchiveWriter {
@@ -109,13 +111,16 @@ class ArchiveWriter {
     }
 
     /**
-     * Increments the size of the compressed data written to the archive
+     * Increments the size of the original (uncompressed) logs ingested into the archive. This size
+     * tracks the raw input size before any encoding or compression.
      * @param size
      */
     void increment_uncompressed_size(size_t size) { m_uncompressed_size += size; }
 
     /**
-     * @return Size of the uncompressed data written to the archive
+     * @return The total size of the encoded (uncompressed) data written to the archive. This
+     * reflects the size of the data after encoding but before compression.
+     * TODO: Add the size of schema tree, schema map and timestamp dictionary
      */
     size_t get_data_size();
 
@@ -128,10 +133,48 @@ class ArchiveWriter {
     void initialize_schema_writer(SchemaWriter* writer, Schema const& schema);
 
     /**
-     * Stores the tables
-     * @return Size of the compressed data in bytes
+     * Compresses and stores the tables.
+     * @return A pair containing:
+     *         - The size of the compressed table metadata in bytes.
+     *         - The size of the compressed tables in bytes.
      */
-    [[nodiscard]] size_t store_tables();
+    [[nodiscard]] std::pair<size_t, size_t> store_tables();
+
+    /**
+     * Writes the archive to a single file
+     * @param files
+     * @param timestamp_dict_compressed_size
+     */
+    void write_single_file_archive(
+            std::vector<ArchiveFileInfo> const& files,
+            size_t timestamp_dict_compressed_size
+    );
+
+    /**
+     * Writes the metadata section of the single file archive
+     * @param archive_writer
+     * @param files
+     * @param timestamp_dict_compressed_size
+     */
+    void write_archive_metadata(
+            FileWriter& archive_writer,
+            std::vector<ArchiveFileInfo> const& files,
+            size_t timestamp_dict_compressed_size
+    );
+
+    /**
+     * Writes the file section of the single file archive
+     * @param archive_writer
+     * @param files
+     */
+    void write_archive_files(FileWriter& archive_writer, std::vector<ArchiveFileInfo> const& files);
+
+    /**
+     * Writes the header section of the single file archive
+     * @param archive_writer
+     * @param metadata_section_size
+     */
+    void write_archive_header(FileWriter& archive_writer, size_t metadata_section_size);
 
     /**
      * Updates the metadata db with the archive's metadata (id, size, timestamp ranges, etc.)
@@ -143,6 +186,8 @@ class ArchiveWriter {
      */
     void print_archive_stats();
 
+    static constexpr size_t cReadBlockSize = 4 * 1024;
+
     size_t m_encoded_message_size{};
     size_t m_uncompressed_size{};
     size_t m_compressed_size{};
@@ -159,6 +204,7 @@ class ArchiveWriter {
     std::shared_ptr<clp::GlobalMySQLMetadataDB> m_metadata_db;
     int m_compression_level{};
     bool m_print_archive_stats{};
+    bool m_single_file_archive{};
 
     SchemaMap m_schema_map;
     SchemaTree m_schema_tree;

components/core/src/clp_s/CommandLineArguments.cpp

@@ -185,6 +185,10 @@ CommandLineArguments::parse_arguments(int argc, char const** argv) {
                     "print-archive-stats",
                     po::bool_switch(&m_print_archive_stats),
                     "Print statistics (json) about the archive after it's compressed."
+            )(
+                    "single-file-archive",
+                    po::bool_switch(&m_single_file_archive),
+                    "Create a single archive file instead of multiple files."
             )(
                     "structurize-arrays",
                     po::bool_switch(&m_structurize_arrays),

components/core/src/clp_s/CommandLineArguments.hpp

@@ -102,6 +102,8 @@ class CommandLineArguments {
 
     OutputHandlerType get_output_handler_type() const { return m_output_handler_type; }
 
+    bool get_single_file_archive() const { return m_single_file_archive; }
+
     bool get_structurize_arrays() const { return m_structurize_arrays; }
 
     bool get_ordered_decompression() const { return m_ordered_decompression; }
@@ -172,6 +174,7 @@ class CommandLineArguments {
     size_t m_target_encoded_size{8ULL * 1024 * 1024 * 1024};  // 8 GiB
     bool m_print_archive_stats{false};
     size_t m_max_document_size{512ULL * 1024 * 1024};  // 512 MB
+    bool m_single_file_archive{false};
     bool m_structurize_arrays{false};
     bool m_ordered_decompression{false};
     size_t m_ordered_chunk_size{0};

components/core/src/clp_s/JsonParser.cpp

@@ -31,6 +31,7 @@ JsonParser::JsonParser(JsonParserOption const& option)
     m_archive_options.archives_dir = option.archives_dir;
     m_archive_options.compression_level = option.compression_level;
     m_archive_options.print_archive_stats = option.print_archive_stats;
+    m_archive_options.single_file_archive = option.single_file_archive;
     m_archive_options.id = m_generator();
 
     m_archive_writer = std::make_unique<ArchiveWriter>(option.metadata_db);

components/core/src/clp_s/JsonParser.hpp

@@ -34,6 +34,7 @@ struct JsonParserOption {
     size_t max_document_size;
     int compression_level;
     bool print_archive_stats;
+    bool single_file_archive;
     bool structurize_arrays;
     std::shared_ptr<clp::GlobalMySQLMetadataDB> metadata_db;
 };

components/core/src/clp_s/SingleFileArchiveDefs.hpp

@@ -0,0 +1,56 @@
+#ifndef CLP_S_ARCHIVEDEFS_HPP
+#define CLP_S_ARCHIVEDEFS_HPP
+
+#include <string>
+
+#include "msgpack.hpp"
+
+namespace clp_s {
+// define the version
+constexpr uint8_t cArchiveMajorVersion = 0;
+constexpr uint8_t cArchiveMinorVersion = 2;
+constexpr uint16_t cArchivePatchVersion = 0;
+
+struct ArchiveHeader {
+    uint8_t magic_number[8];
+    uint32_t version;
+    uint64_t uncompressed_size;
+    uint64_t compressed_size;
+    uint64_t reserved_padding[4];
+    uint32_t metadata_section_size;
+    uint16_t compression_type;
+    uint16_t padding;
+};
+
+enum class ArchiveCompressionType : uint16_t {
+    Zstd = 0,
+};
+
+enum struct ArchiveMetadataPacketType : uint8_t {
+    ArchiveInfo = 0,
+    ArchiveFileInfo = 1,
+    TimestampDictionary = 2,
+};
+
+struct ArchiveInfoPacket {
+    uint64_t num_segments;
+    // TODO: Add more fields in the future
+
+    MSGPACK_DEFINE(num_segments);
+};
+
+struct ArchiveFileInfo {
+    std::string n;  // name
+    uint64_t o;  // offset
+
+    MSGPACK_DEFINE(n, o);
+};
+
+struct ArchiveFileInfoPacket {
+    std::vector<ArchiveFileInfo> files;
+
+    MSGPACK_DEFINE(files);
+};
+}  // namespace clp_s
+
+#endif  // CLP_S_ARCHIVEDEFS_HPP