Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
28 changes: 20 additions & 8 deletions components/core/src/clp_s/ArchiveReader.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
#include <utility>
#include <vector>

#include <nlohmann/json_fwd.hpp>
#include <ystdlib/error_handling/Result.hpp>

#include <clp_s/ArchiveReaderAdaptor.hpp>
Expand Down Expand Up @@ -160,8 +161,8 @@ class ArchiveReader {
void close();

/**
* @return The schema ids in the archive. It also defines the order that tables should be read
* in to avoid seeking backwards.
* @return The schema ids in the archive. It also defines the order that tables should be
* read in to avoid seeking backwards.
*/
[[nodiscard]] std::vector<int32_t> const& get_schema_ids() const { return m_schema_ids; }

Expand All @@ -182,6 +183,16 @@ class ArchiveReader {
return get_header().has_deprecated_timestamp_format();
}

/**
* @param log_event_idx
* @return The file-level metadata associated with the record at `log_event_idx`.
* @throws ArchiveReaderAdaptor::OperationFailed when `log_event_idx` cannot be mapped to
* any metadata.
*/
[[nodiscard]] auto get_metadata_for_log_event(int64_t log_event_idx) -> nlohmann::json const& {
Copy link
Contributor

Choose a reason for hiding this comment

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

Should log event simply take an unsigned type like uint64_t?

return m_archive_reader_adaptor->get_metadata_for_log_event(log_event_idx);
}

private:
/**
* Reads archive metadata and prepares the archive reader for subsequent archive reads.
Expand Down Expand Up @@ -239,13 +250,14 @@ class ArchiveReader {
);

/**
* Reads a table with given ID from the packed stream 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.
* Reads a table with given ID from the packed stream 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
* @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);
Expand Down
12 changes: 12 additions & 0 deletions components/core/src/clp_s/ArchiveReaderAdaptor.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,9 @@ auto ArchiveReaderAdaptor::try_read_range_index(ZstdDecompressor& decompressor,
end_index,
std::move(range_index_entry.at(RangeIndexWriter::cMetadataFieldsName))
);
if (start_index != end_index) {
m_non_empty_range_metadata_map.emplace(end_index, m_range_index.back().fields);
}
Comment on lines +171 to +173
Copy link
Contributor

@Bill-hbrhbr Bill-hbrhbr Mar 13, 2026

Choose a reason for hiding this comment

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

I guess I've always had the question if the range index accepts files with empty ranges, or gaps between adjacent ranges.

}
return ErrorCodeSuccess;
}
Expand Down Expand Up @@ -362,4 +365,13 @@ void ArchiveReaderAdaptor::checkin_reader_for_section(std::string_view section)

m_current_reader_holder.reset();
}

auto ArchiveReaderAdaptor::get_metadata_for_log_event(int64_t log_event_idx)
-> nlohmann::json const& {
auto const it{m_non_empty_range_metadata_map.upper_bound(log_event_idx)};
if (m_non_empty_range_metadata_map.end() == it || log_event_idx < 0) {
Comment on lines +371 to +372
Copy link
Contributor

Choose a reason for hiding this comment

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

Logic seems like you don't really need to look up anything if the log even idx is negative, which makes me feel more strongly about using uint64_t

throw OperationFailed(ErrorCodeBadParam, __FILENAME__, __LINE__);
}
return it->second;
}
} // namespace clp_s
8 changes: 8 additions & 0 deletions components/core/src/clp_s/ArchiveReaderAdaptor.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,13 @@ class ArchiveReaderAdaptor {

std::vector<RangeIndexEntry> const& get_range_index() const { return m_range_index; }

/**
* @param log_event_idx
* @return The file-level metadata associated with the record at `log_event_idx`.
* @throws OperationFailed when `log_event_idx` cannot be mapped to any metadata.
*/
[[nodiscard]] auto get_metadata_for_log_event(int64_t log_event_idx) -> nlohmann::json const&;

private:
/**
* Tries to read an ArchiveFileInfo packet from the archive metadata.
Expand Down Expand Up @@ -188,6 +195,7 @@ class ArchiveReaderAdaptor {
std::shared_ptr<TimestampDictionaryReader> m_timestamp_dictionary;
std::shared_ptr<clp::ReaderInterface> m_reader;
std::vector<RangeIndexEntry> m_range_index;
std::map<int64_t, nlohmann::json> m_non_empty_range_metadata_map;
};
} // namespace clp_s
#endif // CLP_S_ARCHIVEREADERADAPTOR_HPP
13 changes: 13 additions & 0 deletions components/core/tests/test-clp_s-range_index.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
#include "../src/clp/type_utils.hpp"
#include "../src/clp_s/archive_constants.hpp"
#include "../src/clp_s/ArchiveReader.hpp"
#include "../src/clp_s/ArchiveReaderAdaptor.hpp"
#include "../src/clp_s/ArchiveWriter.hpp"
#include "../src/clp_s/InputConfig.hpp"
#include "../src/clp_s/RangeIndexWriter.hpp"
Expand All @@ -33,6 +34,7 @@ constexpr std::string_view cTestRangeIndexInputFile{"test_no_floats_sorted.jsonl
constexpr std::string_view cTestRangeIndexIRInputFile{"test_no_floats_sorted.ir"};
constexpr std::string_view cTestRangeIndexIRMetadataKey{"test_key"};
constexpr std::string_view cTestRangeIndexIRMetadataValue{"test_value"};
constexpr size_t cNumRecordsInInputFile{4};

namespace {
auto get_test_input_path_relative_to_tests_dir() -> std::filesystem::path;
Expand Down Expand Up @@ -138,6 +140,17 @@ void read_and_check_archive_metadata(bool from_ir) {
REQUIRE_NOTHROW(archive_reader.open(archive_path, clp_s::NetworkAuthOption{}));
auto const& range_index = archive_reader.get_range_index();
check_archive_range_index(range_index, from_ir);
for (size_t i{}; i < cNumRecordsInInputFile; ++i) {
REQUIRE(archive_reader.get_metadata_for_log_event(i) == range_index.at(0).fields);
}
REQUIRE_THROWS_AS(
archive_reader.get_metadata_for_log_event(-1),
clp_s::ArchiveReaderAdaptor::OperationFailed
);
REQUIRE_THROWS_AS(
archive_reader.get_metadata_for_log_event(cNumRecordsInInputFile),
clp_s::ArchiveReaderAdaptor::OperationFailed
);
REQUIRE_NOTHROW(archive_reader.close());
}
}
Expand Down
Loading