feat(clp-s): Add the read-side implementation for the archive range index.

components/core/CMakeLists.txt

@@ -630,11 +630,14 @@ set(SOURCE_FILES_unitTest
         src/clp/version.hpp
         src/clp/WriterInterface.cpp
         src/clp/WriterInterface.hpp
+        tests/clp_s_test_utils.cpp
+        tests/clp_s_test_utils.hpp
         tests/LogSuppressor.hpp
         tests/TestOutputCleaner.hpp
         tests/test-BoundedReader.cpp
         tests/test-BufferedFileReader.cpp
         tests/test-clp_s-end_to_end.cpp
+        tests/test-clp_s-range_index.cpp
         tests/test-clp_s-search.cpp
         tests/test-EncodedVariableInterpreter.cpp
         tests/test-encoding_methods.cpp

components/core/src/clp_s/ArchiveReader.hpp

@@ -117,6 +117,10 @@ class ArchiveReader {
 
     std::shared_ptr<ReaderUtils::SchemaMap> get_schema_map() { return m_schema_map; }
 
+    auto get_range_index() const -> std::vector<RangeIndexEntry> const& {
+        return m_archive_reader_adaptor->get_range_index();
+    }
+
     /**
      * Writes decoded messages to a file.
      * @param writer

components/core/src/clp_s/ArchiveReaderAdaptor.cpp

@@ -6,15 +6,18 @@
 #include <optional>
 #include <string>
 #include <string_view>
+#include <utility>
 #include <vector>
 
 #include <msgpack.hpp>
+#include <nlohmann/json.hpp>
 #include <spdlog/spdlog.h>
 
 #include "../clp/BoundedReader.hpp"
 #include "../clp/FileReader.hpp"
 #include "archive_constants.hpp"
 #include "InputConfig.hpp"
+#include "RangeIndexWriter.hpp"
 #include "ReaderUtils.hpp"
 #include "SingleFileArchiveDefs.hpp"
 
@@ -97,6 +100,58 @@ ErrorCode ArchiveReaderAdaptor::try_read_archive_info(ZstdDecompressor& decompre
     return ErrorCodeSuccess;
 }
 
+auto ArchiveReaderAdaptor::try_read_range_index(ZstdDecompressor& decompressor, size_t size)
+        -> ErrorCode {
+    std::vector<char> buffer(size);
+    if (auto const rc = decompressor.try_read_exact_length(buffer.data(), buffer.size());
+        ErrorCodeSuccess != rc)
+    {
+        return rc;
+    }
+
+    auto range_index_json = nlohmann::json::from_msgpack(buffer.begin(), buffer.end(), true, false);
+    if (false == range_index_json.is_array()) {
+        return ErrorCodeCorrupt;
+    }
+
+    for (auto& range_index_entry : range_index_json) {
+        if (false == range_index_entry.contains(RangeIndexWriter::cStartIndexName)
+            || false == range_index_entry.at(RangeIndexWriter::cStartIndexName).is_number_integer())
+        {
+            return ErrorCodeCorrupt;
+        }
+        if (false == range_index_entry.contains(RangeIndexWriter::cEndIndexName)
+            || false == range_index_entry.at(RangeIndexWriter::cEndIndexName).is_number_integer())
+        {
+            return ErrorCodeCorrupt;
+        }
+        if (false == range_index_entry.contains(RangeIndexWriter::cMetadataFieldsName)
+            || false == range_index_entry.at(RangeIndexWriter::cMetadataFieldsName).is_object())
+        {
+            return ErrorCodeCorrupt;
+        }
+        size_t start_index{};
+        size_t end_index{};
+        try {
+            start_index = range_index_entry.at(RangeIndexWriter::cStartIndexName)
+                                  .template get<size_t>();
+            end_index
+                    = range_index_entry.at(RangeIndexWriter::cEndIndexName).template get<size_t>();
+        } catch (std::exception const&) {
+            return ErrorCodeCorrupt;
+        }
+        if (start_index > end_index) {
+            return ErrorCodeCorrupt;
+        }
+        m_range_index.emplace_back(
+                start_index,
+                end_index,
+                std::move(range_index_entry.at(RangeIndexWriter::cMetadataFieldsName))
+        );
+    }
+    return ErrorCodeSuccess;
+}
+
 auto
 ArchiveReaderAdaptor::try_read_unknown_metadata_packet(ZstdDecompressor& decompressor, size_t size)
         -> ErrorCode {
@@ -181,6 +236,9 @@ ErrorCode ArchiveReaderAdaptor::try_read_archive_metadata(ZstdDecompressor& deco
             case ArchiveMetadataPacketType::ArchiveInfo:
                 rc = try_read_archive_info(decompressor, packet_size);
                 break;
+            case ArchiveMetadataPacketType::RangeIndex:
+                rc = try_read_range_index(decompressor, packet_size);
+                break;
             default:
                 rc = try_read_unknown_metadata_packet(decompressor, packet_size);
                 break;

components/core/src/clp_s/ArchiveReaderAdaptor.hpp

@@ -1,10 +1,20 @@
 #ifndef CLP_S_ARCHIVEREADERADAPTOR_HPP
 #define CLP_S_ARCHIVEREADERADAPTOR_HPP
 
+#include <cstddef>
 #include <memory>
 #include <optional>
 #include <string>
 #include <string_view>
+#include <utility>
+#include <vector>
+
+// We use NOLINTNEXTLINE to satisfy clang-tidy here because while we don't use any symbols from
+// `nlohmann/json.hpp` directly this code does not compile without the definition of
+// `nlohmann::basic_json<>` found in the `nlohmann/json.hpp` header.
+// NOLINTNEXTLINE(misc-include-cleaner)
+#include <nlohmann/json.hpp>
+#include <nlohmann/json_fwd.hpp>
 
 #include "../clp/BoundedReader.hpp"
 #include "../clp/ReaderInterface.hpp"
@@ -15,6 +25,21 @@
 #include "ZstdDecompressor.hpp"
 
 namespace clp_s {
+/**
+ * RangeIndexEntry is a struct representing a single entry in the archive range index.
+ */
+struct RangeIndexEntry {
+    explicit RangeIndexEntry(size_t start_index, size_t end_index, nlohmann::json&& fields)
+            : start_index{start_index},
+              end_index{end_index},
+              // Note: brace initializer would make nlohmann wrap the fields object in an array.
+              fields(std::move(fields)) {}
+
+    size_t start_index;
+    size_t end_index;
+    nlohmann::json fields;
+};
+
 /**
  * ArchiveReaderAdaptor is an adaptor class which helps with reading single and multi-file archives
  * which exist on either S3 or a locally mounted file system.
@@ -62,6 +87,8 @@ class ArchiveReaderAdaptor {
 
     ArchiveHeader const& get_header() const { return m_archive_header; }
 
+    std::vector<RangeIndexEntry> const& get_range_index() const { return m_range_index; }
+
 private:
     /**
      * Tries to read an ArchiveFileInfo packet from the archive metadata.
@@ -90,6 +117,14 @@ class ArchiveReaderAdaptor {
      */
     ErrorCode try_read_archive_info(ZstdDecompressor& decompressor, size_t size);
 
+    /**
+     * Tries to read a RangeIndex packet from the archive metadata.
+     * @param decompressor
+     * @param size The number of decompressed bytes making up the packet.
+     * @return ErrorCodeSuccess on success or the relevant ErrorCode on failure.
+     */
+    auto try_read_range_index(ZstdDecompressor& decompressor, size_t size) -> ErrorCode;
+
     /**
      * Tries to read an unknown metadata packet from the archive metadata.
      * @param decompressor
@@ -140,6 +175,7 @@ class ArchiveReaderAdaptor {
     std::optional<std::string> m_current_reader_holder;
     std::shared_ptr<TimestampDictionaryReader> m_timestamp_dictionary;
     std::shared_ptr<clp::ReaderInterface> m_reader;
+    std::vector<RangeIndexEntry> m_range_index;
 };
 }  // namespace clp_s
 #endif  // CLP_S_ARCHIVEREADERADAPTOR_HPP

components/core/src/clp_s/indexer/CMakeLists.txt

@@ -90,6 +90,7 @@ target_link_libraries(indexer
         clp::string_utils
         date::date
         MariaDBClient::MariaDBClient
+        nlohmann_json::nlohmann_json
         OpenSSL::Crypto
         simdjson::simdjson
         spdlog::spdlog

components/core/tests/clp_s_test_utils.cpp

@@ -0,0 +1,54 @@
+#include "clp_s_test_utils.hpp"
+
+#include <filesystem>
+#include <string>
+
+#include <catch2/catch.hpp>
+
+#include "../src/clp_s/CommandLineArguments.hpp"
+#include "../src/clp_s/InputConfig.hpp"
+#include "../src/clp_s/JsonParser.hpp"
+
+void compress_archive(
+        std::string const& file_path,
+        std::string const& archive_directory,
+        bool single_file_archive,
+        bool structurize_arrays,
+        clp_s::CommandLineArguments::FileType file_type
+) {
+    constexpr auto cDefaultTargetEncodedSize{8ULL * 1024 * 1024 * 1024};  // 8 GiB
+    constexpr auto cDefaultMaxDocumentSize{512ULL * 1024 * 1024};  // 512 MiB
+    constexpr auto cDefaultMinTableSize{1ULL * 1024 * 1024};  // 1 MiB
+    constexpr auto cDefaultCompressionLevel{3};
+    constexpr auto cDefaultPrintArchiveStats{false};
+
+    std::filesystem::create_directory(archive_directory);
+    REQUIRE((std::filesystem::is_directory(archive_directory)));
+
+    clp_s::JsonParserOption parser_option{};
+    parser_option.input_paths.emplace_back(
+            clp_s::Path{.source = clp_s::InputSource::Filesystem, .path = file_path}
+    );
+    parser_option.archives_dir = archive_directory;
+    parser_option.target_encoded_size = cDefaultTargetEncodedSize;
+    parser_option.max_document_size = cDefaultMaxDocumentSize;
+    parser_option.min_table_size = cDefaultMinTableSize;
+    parser_option.compression_level = cDefaultCompressionLevel;
+    parser_option.print_archive_stats = cDefaultPrintArchiveStats;
+    parser_option.structurize_arrays = structurize_arrays;
+    parser_option.single_file_archive = single_file_archive;
+    parser_option.input_file_type = file_type;
+
+    clp_s::JsonParser parser{parser_option};
+    if (clp_s::CommandLineArguments::FileType::Json == file_type) {
+        REQUIRE(parser.parse());
+    } else if (clp_s::CommandLineArguments::FileType::KeyValueIr == file_type) {
+        REQUIRE(parser.parse_from_ir());
+    } else {
+        // This branch should be unreachable.
+        REQUIRE(false);
+    }
+    REQUIRE_NOTHROW(parser.store());
+
+    REQUIRE((false == std::filesystem::is_empty(archive_directory)));
+}

components/core/tests/clp_s_test_utils.hpp

@@ -0,0 +1,25 @@
+#ifndef CLP_S_TEST_UTILS_HPP
+#define CLP_S_TEST_UTILS_HPP
+#include <string>
+
+#include "../src/clp_s/CommandLineArguments.hpp"
+
+/**
+ * Compresses a file into an archive directory according to a given set of configuration options.
+ *
+ * This helper uses `REQUIRE...` statements to assert that compression was successful.
+ *
+ * @param file_path
+ * @param archive_directory
+ * @param single_file_archive
+ * @param structurize_arrays
+ * @param file_type
+ */
+void compress_archive(
+        std::string const& file_path,
+        std::string const& archive_directory,
+        bool single_file_archive,
+        bool structurize_arrays,
+        clp_s::CommandLineArguments::FileType file_type
+);
+#endif  // CLP_S_TEST_UTILS_HPP

components/core/tests/test-clp_s-end_to_end.cpp

@@ -8,9 +8,10 @@
 #include <catch2/catch.hpp>
 #include <fmt/format.h>
 
+#include "../src/clp_s/CommandLineArguments.hpp"
 #include "../src/clp_s/InputConfig.hpp"
 #include "../src/clp_s/JsonConstructor.hpp"
-#include "../src/clp_s/JsonParser.hpp"
+#include "clp_s_test_utils.hpp"
 #include "TestOutputCleaner.hpp"
 
 constexpr std::string_view cTestEndToEndArchiveDirectory{"test-end-to-end-archive"};
@@ -22,7 +23,6 @@ constexpr std::string_view cTestEndToEndInputFile{"test_no_floats_sorted.jsonl"}
 namespace {
 auto get_test_input_path_relative_to_tests_dir() -> std::filesystem::path;
 auto get_test_input_local_path() -> std::string;
-void compress(bool structurize_arrays, bool single_file_archive);
 auto extract() -> std::filesystem::path;
 void compare(std::filesystem::path const& extracted_json_path);
 
@@ -36,39 +36,6 @@ auto get_test_input_local_path() -> std::string {
     return (tests_dir / get_test_input_path_relative_to_tests_dir()).string();
 }
 
-void compress(bool structurize_arrays, bool single_file_archive) {
-    constexpr auto cDefaultTargetEncodedSize = 8ULL * 1024 * 1024 * 1024;  // 8 GiB
-    constexpr auto cDefaultMaxDocumentSize = 512ULL * 1024 * 1024;  // 512 MiB
-    constexpr auto cDefaultMinTableSize = 1ULL * 1024 * 1024;  // 1 MiB
-    constexpr auto cDefaultCompressionLevel = 3;
-    constexpr auto cDefaultPrintArchiveStats = false;
-
-    std::filesystem::create_directory(cTestEndToEndArchiveDirectory);
-    REQUIRE((std::filesystem::is_directory(cTestEndToEndArchiveDirectory)));
-
-    clp_s::JsonParserOption parser_option{};
-    parser_option.input_paths.emplace_back(
-            clp_s::Path{
-                    .source = clp_s::InputSource::Filesystem,
-                    .path = get_test_input_local_path()
-            }
-    );
-    parser_option.archives_dir = cTestEndToEndArchiveDirectory;
-    parser_option.target_encoded_size = cDefaultTargetEncodedSize;
-    parser_option.max_document_size = cDefaultMaxDocumentSize;
-    parser_option.min_table_size = cDefaultMinTableSize;
-    parser_option.compression_level = cDefaultCompressionLevel;
-    parser_option.print_archive_stats = cDefaultPrintArchiveStats;
-    parser_option.structurize_arrays = structurize_arrays;
-    parser_option.single_file_archive = single_file_archive;
-
-    clp_s::JsonParser parser{parser_option};
-    REQUIRE(parser.parse());
-    parser.store();
-
-    REQUIRE((false == std::filesystem::is_empty(cTestEndToEndArchiveDirectory)));
-}
-
 auto extract() -> std::filesystem::path {
     constexpr auto cDefaultOrdered = false;
     constexpr auto cDefaultTargetOrderedChunkSize = 0;
@@ -135,7 +102,13 @@ TEST_CASE("clp-s-compress-extract-no-floats", "[clp-s][end-to-end]") {
              std::string{cTestEndToEndOutputSortedJson}}
     };
 
-    compress(structurize_arrays, single_file_archive);
+    REQUIRE_NOTHROW(compress_archive(
+            get_test_input_local_path(),
+            std::string{cTestEndToEndArchiveDirectory},
+            single_file_archive,
+            structurize_arrays,
+            clp_s::CommandLineArguments::FileType::Json
+    ));
 
     auto extracted_json_path = extract();