clp-s: Implement table packing

components/core/src/clp_s/ArchiveReader.cpp

@@ -27,8 +27,8 @@ void ArchiveReader::open(string_view archives_dir, string_view archive_id) {
     m_schema_tree = ReaderUtils::read_schema_tree(archive_path_str);
     m_schema_map = ReaderUtils::read_schemas(archive_path_str);
 
-    m_tables_file_reader.open(archive_path_str + constants::cArchiveTablesFile);
     m_table_metadata_file_reader.open(archive_path_str + constants::cArchiveTableMetadataFile);
+    m_stream_reader.open_packed_streams(archive_path_str + constants::cArchiveTablesFile);
 }
 
 void ArchiveReader::read_metadata() {
@@ -38,46 +38,74 @@ void ArchiveReader::read_metadata() {
             cDecompressorFileReadBufferCapacity
     );
 
+    m_stream_reader.read_metadata(m_table_metadata_decompressor);
+
     size_t num_schemas;
     if (auto error = m_table_metadata_decompressor.try_read_numeric_value(num_schemas);
         ErrorCodeSuccess != error)
     {
         throw OperationFailed(error, __FILENAME__, __LINE__);
     }
 
-    for (size_t i = 0; i < num_schemas; i++) {
+    bool prev_metadata_initialized{false};
+    SchemaReader::SchemaMetadata prev_metadata{};
+    int32_t prev_schema_id{};
+    for (size_t i = 0; i < num_schemas; ++i) {
+        size_t stream_id;
+        size_t stream_offset;
         int32_t schema_id;
-        uint64_t num_messages;
-        size_t table_offset;
-        size_t uncompressed_size;
+        size_t num_messages;
 
-        if (auto error = m_table_metadata_decompressor.try_read_numeric_value(schema_id);
+        if (auto error = m_table_metadata_decompressor.try_read_numeric_value(stream_id);
             ErrorCodeSuccess != error)
         {
             throw OperationFailed(error, __FILENAME__, __LINE__);
         }
 
-        if (auto error = m_table_metadata_decompressor.try_read_numeric_value(num_messages);
+        if (auto error = m_table_metadata_decompressor.try_read_numeric_value(stream_offset);
             ErrorCodeSuccess != error)
         {
             throw OperationFailed(error, __FILENAME__, __LINE__);
         }
 
-        if (auto error = m_table_metadata_decompressor.try_read_numeric_value(table_offset);
+        if (stream_offset > m_stream_reader.get_uncompressed_stream_size(stream_id)) {
+            throw OperationFailed(ErrorCodeCorrupt, __FILENAME__, __LINE__);
+        }
+
+        if (auto error = m_table_metadata_decompressor.try_read_numeric_value(schema_id);
             ErrorCodeSuccess != error)
         {
             throw OperationFailed(error, __FILENAME__, __LINE__);
         }
 
-        if (auto error = m_table_metadata_decompressor.try_read_numeric_value(uncompressed_size);
+        if (auto error = m_table_metadata_decompressor.try_read_numeric_value(num_messages);
             ErrorCodeSuccess != error)
         {
             throw OperationFailed(error, __FILENAME__, __LINE__);
         }
 
-        m_id_to_table_metadata[schema_id] = {num_messages, table_offset, uncompressed_size};
+        if (prev_metadata_initialized) {
+            size_t uncompressed_size{0};
+            if (stream_id != prev_metadata.stream_id) {
+                uncompressed_size
+                        = m_stream_reader.get_uncompressed_stream_size(prev_metadata.stream_id)
+                          - prev_metadata.stream_offset;
+            } else {
+                uncompressed_size = stream_offset - prev_metadata.stream_offset;
+            }
+            prev_metadata.uncompressed_size = uncompressed_size;
+            m_id_to_schema_metadata[prev_schema_id] = prev_metadata;
+        } else {
+            prev_metadata_initialized = true;
+        }
+        prev_metadata = {stream_id, stream_offset, num_messages, 0};
+        prev_schema_id = schema_id;
         m_schema_ids.push_back(schema_id);
     }
+    prev_metadata.uncompressed_size
+            = m_stream_reader.get_uncompressed_stream_size(prev_metadata.stream_id)
+              - prev_metadata.stream_offset;
+    m_id_to_schema_metadata[prev_schema_id] = prev_metadata;
     m_table_metadata_decompressor.close();
 }
 
@@ -89,14 +117,12 @@ void ArchiveReader::read_dictionaries_and_metadata() {
     read_metadata();
 }
 
-SchemaReader& ArchiveReader::read_table(
+SchemaReader& ArchiveReader::read_schema_table(
         int32_t schema_id,
         bool should_extract_timestamp,
         bool should_marshal_records
 ) {
-    constexpr size_t cDecompressorFileReadBufferCapacity = 64 * 1024;  // 64 KB
-
-    if (m_id_to_table_metadata.count(schema_id) == 0) {
+    if (m_id_to_schema_metadata.count(schema_id) == 0) {
         throw OperationFailed(ErrorCodeFileNotFound, __FILENAME__, __LINE__);
     }
 
@@ -107,30 +133,26 @@ SchemaReader& ArchiveReader::read_table(
             should_marshal_records
     );
 
-    m_tables_file_reader.try_seek_from_begin(m_id_to_table_metadata[schema_id].offset);
-    m_tables_decompressor.open(m_tables_file_reader, cDecompressorFileReadBufferCapacity);
-    m_schema_reader.load(
-            m_tables_decompressor,
-            m_id_to_table_metadata[schema_id].uncompressed_size
-    );
-    m_tables_decompressor.close_for_reuse();
+    auto& schema_metadata = m_id_to_schema_metadata[schema_id];
+    auto stream_buffer = read_stream(schema_metadata.stream_id, true);
+    m_schema_reader
+            .load(stream_buffer, schema_metadata.stream_offset, schema_metadata.uncompressed_size);
     return m_schema_reader;
 }
 
 std::vector<std::shared_ptr<SchemaReader>> ArchiveReader::read_all_tables() {
-    constexpr size_t cDecompressorFileReadBufferCapacity = 64 * 1024;  // 64 KB
-
     std::vector<std::shared_ptr<SchemaReader>> readers;
-    readers.reserve(m_id_to_table_metadata.size());
-    for (auto const& [id, table_metadata] : m_id_to_table_metadata) {
+    readers.reserve(m_id_to_schema_metadata.size());
+    for (auto schema_id : m_schema_ids) {
         auto schema_reader = std::make_shared<SchemaReader>();
-        initialize_schema_reader(*schema_reader, id, true, true);
-
-        m_tables_file_reader.try_seek_from_begin(table_metadata.offset);
-        m_tables_decompressor.open(m_tables_file_reader, cDecompressorFileReadBufferCapacity);
-        schema_reader->load(m_tables_decompressor, table_metadata.uncompressed_size);
-        m_tables_decompressor.close_for_reuse();
-
+        initialize_schema_reader(*schema_reader, schema_id, true, true);
+        auto& schema_metadata = m_id_to_schema_metadata[schema_id];
+        auto stream_buffer = read_stream(schema_metadata.stream_id, false);
+        schema_reader->load(
+                stream_buffer,
+                schema_metadata.stream_offset,
+                schema_metadata.uncompressed_size
+        );
         readers.push_back(std::move(schema_reader));
     }
     return readers;
@@ -237,7 +259,7 @@ void ArchiveReader::initialize_schema_reader(
             m_schema_tree,
             schema_id,
             schema.get_ordered_schema_view(),
-            m_id_to_table_metadata[schema_id].num_messages,
+            m_id_to_schema_metadata[schema_id].num_messages,
             should_marshal_records
     );
     auto timestamp_column_ids = m_timestamp_dict->get_authoritative_timestamp_column_ids();
@@ -284,9 +306,8 @@ void ArchiveReader::initialize_schema_reader(
 
 void ArchiveReader::store(FileWriter& writer) {
     std::string message;
-
-    for (auto& [id, table_metadata] : m_id_to_table_metadata) {
-        auto& schema_reader = read_table(id, false, true);
+    for (auto schema_id : m_schema_ids) {
+        auto& schema_reader = read_schema_table(schema_id, false, true);
         while (schema_reader.get_next_message(message)) {
             writer.write(message.c_str(), message.length());
         }
@@ -304,11 +325,28 @@ void ArchiveReader::close() {
     m_array_dict->close();
     m_timestamp_dict->close();
 
-    m_tables_file_reader.close();
+    m_stream_reader.close();
     m_table_metadata_file_reader.close();
 
-    m_id_to_table_metadata.clear();
+    m_id_to_schema_metadata.clear();
     m_schema_ids.clear();
+    m_cur_stream_id = 0;
+    m_stream_buffer.reset();
+    m_stream_buffer_size = 0ULL;
 }
 
+std::shared_ptr<char[]> ArchiveReader::read_stream(size_t stream_id, bool reuse_buffer) {
+    if (nullptr != m_stream_buffer && m_cur_stream_id == stream_id) {
+        return m_stream_buffer;
+    }
+
+    if (false == reuse_buffer) {
+        m_stream_buffer.reset();
+        m_stream_buffer_size = 0;
+    }
+
+    m_stream_reader.read_stream(stream_id, m_stream_buffer, m_stream_buffer_size);
+    m_cur_stream_id = stream_id;
+    return m_stream_buffer;
+}
 }  // namespace clp_s

components/core/src/clp_s/ArchiveReader.hpp

@@ -10,6 +10,7 @@
 #include <boost/filesystem.hpp>
 
 #include "DictionaryReader.hpp"
+#include "PackedStreamReader.hpp"
 #include "ReaderUtils.hpp"
 #include "SchemaReader.hpp"
 #include "TimestampDictionaryReader.hpp"
@@ -91,8 +92,11 @@ class ArchiveReader {
      * @param should_marshal_records
      * @return the schema reader
      */
-    SchemaReader&
-    read_table(int32_t schema_id, bool should_extract_timestamp, bool should_marshal_records);
+    SchemaReader& read_schema_table(
+            int32_t schema_id,
+            bool should_extract_timestamp,
+            bool should_marshal_records
+    );
 
     /**
      * Loads all of the tables in the archive and returns SchemaReaders for them.
@@ -171,6 +175,18 @@ class ArchiveReader {
             bool should_marshal_records
     );
 
+    /**
+     * Reads a table with given ID from the table reader. If read_stream is called multiple times in
+     * a row for the same stream_id a cached buffer is returned. This function allows the caller to
+     * ask for the same buffer to be reused to read multiple different tables: this can save memory
+     * allocations, but can only be used when tables are read one at a time.
+     * @param stream_id
+     * @param reuse_buffer when true the same buffer is reused across invocations, overwriting data
+     * returned previous calls to read_stream
+     * @return a buffer containing the decompressed stream identified by stream_id
+     */
+    std::shared_ptr<char[]> read_stream(size_t stream_id, bool reuse_buffer);
+
     bool m_is_open;
     std::string m_archive_id;
     std::shared_ptr<VariableDictionaryReader> m_var_dict;
@@ -181,13 +197,15 @@ class ArchiveReader {
     std::shared_ptr<SchemaTree> m_schema_tree;
     std::shared_ptr<ReaderUtils::SchemaMap> m_schema_map;
     std::vector<int32_t> m_schema_ids;
-    std::map<int32_t, SchemaReader::TableMetadata> m_id_to_table_metadata;
+    std::map<int32_t, SchemaReader::SchemaMetadata> m_id_to_schema_metadata;
 
-    FileReader m_tables_file_reader;
+    PackedStreamReader m_stream_reader;
     FileReader m_table_metadata_file_reader;
-    ZstdDecompressor m_tables_decompressor;
     ZstdDecompressor m_table_metadata_decompressor;
     SchemaReader m_schema_reader;
+    std::shared_ptr<char[]> m_stream_buffer{};
+    size_t m_stream_buffer_size{0ULL};
+    size_t m_cur_stream_id{0ULL};
 };
 }  // namespace clp_s
 

components/core/src/clp_s/ArchiveWriter.cpp

@@ -1,5 +1,8 @@
 #include "ArchiveWriter.hpp"
 
+#include <algorithm>
+#include <tuple>
+
 #include <json/single_include/nlohmann/json.hpp>
 
 #include "archive_constants.hpp"
@@ -11,6 +14,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_min_table_size = option.min_table_size;
     auto archive_path = boost::filesystem::path(option.archives_dir) / m_id;
 
     boost::system::error_code boost_error_code;
@@ -138,19 +142,82 @@ size_t ArchiveWriter::store_tables() {
             FileWriter::OpenMode::CreateForWriting
     );
     m_table_metadata_compressor.open(m_table_metadata_file_writer, m_compression_level);
-    m_table_metadata_compressor.write_numeric_value(m_id_to_schema_writer.size());
-    for (auto& i : m_id_to_schema_writer) {
-        m_table_metadata_compressor.write_numeric_value(i.first);
-        m_table_metadata_compressor.write_numeric_value(i.second->get_num_messages());
-        m_table_metadata_compressor.write_numeric_value(m_tables_file_writer.get_pos());
 
-        m_tables_compressor.open(m_tables_file_writer, m_compression_level);
-        size_t uncompressed_size = i.second->store(m_tables_compressor);
-        m_tables_compressor.close();
-        delete i.second;
+    /**
+     * Packed stream metadata schema
+     * # num packed streams <64 bit>
+     * # [offset into file <64 bit> uncompressed size <64 bit>]+
+     * # num schemas <64 bit>
+     * # [stream id <64 bit> offset into stream <64 bit> schema id <32 bit> num messages <64 bit>]+
+     *
+     * Schema tables are packed into a series of compression streams. Each of those compression
+     * streams is identified by a 64 bit stream id. In the first half of the metadata we identify
+     * how many streams there are, and the offset into the file where each compression stream can
+     * be found. In the second half of the metadata we record how many schema tables there are,
+     * which compression stream they belong to, the offset into that compression stream where
+     * they can be found, and how many messages that schema table contains.
+     *
+     * We buffer the first half of the metadata in the "stream_metadata" vector, and the second half
+     * of the metadata in the "schema_metadata" vector as we compress the tables. The metadata is
+     * flushed once all of the schema tables have been compressed.
+     */
+    using schema_map_it = decltype(m_id_to_schema_writer)::iterator;
+    std::vector<schema_map_it> schemas;
+    std::vector<std::tuple<size_t, size_t>> stream_metadata;
+    std::vector<std::tuple<size_t, size_t, int32_t, size_t>> schema_metadata;
+
+    schema_metadata.reserve(m_id_to_schema_writer.size());
+    schemas.reserve(m_id_to_schema_writer.size());
+    for (auto it = m_id_to_schema_writer.begin(); it != m_id_to_schema_writer.end(); ++it) {
+        schemas.push_back(it);
+    }
+    auto comp = [](schema_map_it const& lhs, schema_map_it const& rhs) -> bool {
+        return lhs->second->get_total_uncompressed_size()
+               > rhs->second->get_total_uncompressed_size();
+    };
+    std::sort(schemas.begin(), schemas.end(), comp);
+
+    size_t current_stream_offset = 0;
+    size_t current_stream_id = 0;
+    size_t current_table_file_offset = 0;
+    m_tables_compressor.open(m_tables_file_writer, m_compression_level);
+    for (auto it : schemas) {
+        it->second->store(m_tables_compressor);
+        schema_metadata.emplace_back(
+                current_stream_id,
+                current_stream_offset,
+                it->first,
+                it->second->get_num_messages()
+        );
+        current_stream_offset += it->second->get_total_uncompressed_size();
+        delete it->second;
+
+        if (current_stream_offset > m_min_table_size || schemas.size() == schema_metadata.size()) {
+            stream_metadata.emplace_back(current_table_file_offset, current_stream_offset);
+            m_tables_compressor.close();
+            current_stream_offset = 0;
+            ++current_stream_id;
+            current_table_file_offset = m_tables_file_writer.get_pos();
 
+            if (schemas.size() != schema_metadata.size()) {
+                m_tables_compressor.open(m_tables_file_writer, m_compression_level);
+            }
+        }
+    }
+
+    m_table_metadata_compressor.write_numeric_value(stream_metadata.size());
+    for (auto& [file_offset, uncompressed_size] : stream_metadata) {
+        m_table_metadata_compressor.write_numeric_value(file_offset);
         m_table_metadata_compressor.write_numeric_value(uncompressed_size);
     }
+
+    m_table_metadata_compressor.write_numeric_value(schema_metadata.size());
+    for (auto& [stream_id, stream_offset, schema_id, num_messages] : schema_metadata) {
+        m_table_metadata_compressor.write_numeric_value(stream_id);
+        m_table_metadata_compressor.write_numeric_value(stream_offset);
+        m_table_metadata_compressor.write_numeric_value(schema_id);
+        m_table_metadata_compressor.write_numeric_value(num_messages);
+    }
     m_table_metadata_compressor.close();
 
     compressed_size += m_table_metadata_file_writer.get_pos();

components/core/src/clp_s/ArchiveWriter.hpp

@@ -21,6 +21,7 @@ struct ArchiveWriterOption {
     std::string archives_dir;
     int compression_level;
     bool print_archive_stats;
+    size_t min_table_size;
 };
 
 class ArchiveWriter {
@@ -159,6 +160,7 @@ class ArchiveWriter {
     std::shared_ptr<clp::GlobalMySQLMetadataDB> m_metadata_db;
     int m_compression_level{};
     bool m_print_archive_stats{};
+    size_t m_min_table_size{};
 
     SchemaMap m_schema_map;
     SchemaTree m_schema_tree;