Skip to content

feat(clp-s): Expose archive range index metadata in printed archive statistics; Forward archive-level statistics to c++ writers. #1005

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gibber9809 merged 13 commits into from
Jun 23, 2025
Merged

feat(clp-s): Expose archive range index metadata in printed archive statistics; Forward archive-level statistics to c++ writers. #1005

Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
8e300fa
Refactor clp-s archive statistics information and include file-level …
gibber9809 Jun 12, 2025
c425c98
Make archive statistics available at boundary of write interface
gibber9809 Jun 13, 2025
8ff5131
Add test for forwarded ArchiveStats
gibber9809 Jun 13, 2025
263d51f
Address rabbit comment
gibber9809 Jun 13, 2025
20e7e6a
Rename metadata to range_index in archive stats
gibber9809 Jun 17, 2025
2791749
Address rabbit comments
gibber9809 Jun 17, 2025
8e5f865
Merge remote-tracking branch 'upstream/main' into improve-archive-stats
gibber9809 Jun 17, 2025
50ebcaf
Address rabbit comment
gibber9809 Jun 18, 2025
9afcff0
Update compression task to expect utf-8 archive stats
gibber9809 Jun 18, 2025
9c91ee0
Merge branch 'main' into improve-archive-stats
gibber9809 Jun 20, 2025
249a5b4
Address review comments
gibber9809 Jun 23, 2025
f8b5f74
Update components/core/src/clp_s/ArchiveWriter.hpp
gibber9809 Jun 23, 2025
6ccfa03
Address review comments
gibber9809 Jun 23, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 27 additions & 21 deletions components/core/src/clp_s/ArchiveWriter.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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__);
Expand Down Expand Up @@ -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
Expand All @@ -107,8 +107,17 @@ 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() << std::endl;

@kirkrodrigues kirkrodrigues Jun 23, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
std::cout << archive_stats.as_string() << std::endl;
std::cout << archive_stats.as_string() << '\n';

To resolve clang-tidy violation.

}

m_id_to_schema_writer.clear();
Expand All @@ -123,14 +132,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();
Expand All @@ -141,12 +153,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;
Expand Down Expand Up @@ -184,11 +197,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(
Expand Down Expand Up @@ -441,15 +458,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
66 changes: 56 additions & 10 deletions components/core/src/clp_s/ArchiveWriter.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -31,6 +33,49 @@ struct ArchiveWriterOption {
std::string authoritative_timestamp_namespace;
};

struct ArchiveStats {
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
)
: id{id},
begin_timestamp{begin_timestamp},
end_timestamp{end_timestamp},
uncompressed_size{uncompressed_size},
compressed_size{compressed_size},
range_index(std::move(range_index)), // Avoid {} to prevent wrapping in JSON array.
is_split{is_split} {}

Comment thread
gibber9809 marked this conversation as resolved.
Resolved
[[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, id},
{Archive::BeginTimestamp, begin_timestamp},
{Archive::EndTimestamp, end_timestamp},
{Archive::UncompressedSize, uncompressed_size},
{Archive::Size, compressed_size},
{File::IsSplit, is_split},
{cRangeIndex, range_index}};
return json_msg.dump(-1, ' ', false, nlohmann::json::error_handler_t::ignore);
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved

std::string const id;
epochtime_t const begin_timestamp{};
epochtime_t const end_timestamp{};
size_t const uncompressed_size{};
size_t const compressed_size{};
nlohmann::json const range_index;
bool const is_split{};

@kirkrodrigues kirkrodrigues Jun 23, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess the reason you're const qualifying these variables is so that they're not independently modified by a caller? clang-tidy understandably has concerns about const-qualified members, so I wonder if we should just convert ArchiveStats to a full-fledged class so that these members are private? Is any caller actually using these members directly?

@gibber9809 gibber9809 Jun 23, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, this was a rabbit suggestion. I agree it makes sense to just turn it into a full class at this point though.

};

class ArchiveWriter {
public:
class OperationFailed : public TraceableException {
Expand Down Expand Up @@ -80,9 +125,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.

@wraymo wraymo Jun 17, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess every archive except the last one is a split? Why do we need this information?

@gibber9809 gibber9809 Jun 17, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it could be helpful for error handling in some scenarios -- e.g. if some file gets split across two archives and compression fails while the second archive is being compressed we can use this flag to tell that the last file in the first archive hasn't been fully ingested and handle the error appropriately.

@wraymo wraymo Jun 18, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How would looking at the printed ArchiveStats help in this case? As I understand it, regardless of whether an error occurs or not, the last archive is never a split.

@gibber9809 gibber9809 Jun 18, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Say a user is ingesting files A and B and we produce two archives, where B is split between the first and the second archive. Consider the scenario where the second archive fails compression such that no archive stats get printed for the second archive.

In this scenario, without the is_split flag, the user can see that the first archive contains the files "A" and "B", but has no reliable way of determining that "B" is not fully ingested. However, if they do have the "is_split" flag then they know that they should expect at least one more archive containing the rest of "B".

Right now, since we still close/print archive stats for failed archives it's not as useful, but if we improve the error handling in JsonConstructor I think it could be pretty useful. The c++ code-path doesn't have this issue since we're only forwarding archive stats for the succesful archives.

@wraymo wraymo Jun 18, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we have a direct way to figure out which file is split across two archives if one archive contains multiple files

@wraymo wraymo Jun 18, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh the range index is part of the statistics

* @return Statistics for the newly-written archive.
*/
void close();
[[nodiscard]] auto close(bool is_split = false) -> ArchiveStats;

/**
* Appends a message to the archive writer
Expand Down Expand Up @@ -220,16 +267,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;

Comment thread
gibber9809 marked this conversation as resolved.
Resolved
/**
* Writes the file section of the single file archive
Expand All @@ -245,11 +296,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{};
Expand Down
2 changes: 1 addition & 1 deletion components/core/src/clp_s/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,6 +232,7 @@ target_link_libraries(
PUBLIC
absl::flat_hash_map
clp_s::io
nlohmann_json::nlohmann_json
simdjson::simdjson
PRIVATE
Boost::url
Expand All @@ -240,7 +241,6 @@ target_link_libraries(
${CURL_LIBRARIES}
fmt::fmt
msgpack-cxx
nlohmann_json::nlohmann_json
spdlog::spdlog
)

Expand Down
Loading