feat(clp-s): Add an indexer process to populate column names and types of each table

components/core/src/clp_s/CMakeLists.txt

@@ -1,4 +1,5 @@
 add_subdirectory(search/kql)
+add_subdirectory(metadata_uploader)
 
 set(
         CLP_SOURCES

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

@@ -0,0 +1,101 @@
+set(
+        METADATA_UPLOADER_SOURCES
+        ../../clp/aws/AwsAuthenticationSigner.cpp
+        ../../clp/aws/AwsAuthenticationSigner.hpp
+        ../../clp/BoundedReader.cpp
+        ../../clp/BoundedReader.hpp
+        ../../clp/CurlDownloadHandler.cpp
+        ../../clp/CurlDownloadHandler.hpp
+        ../../clp/CurlEasyHandle.hpp
+        ../../clp/CurlGlobalInstance.cpp
+        ../../clp/CurlGlobalInstance.hpp
+        ../../clp/CurlOperationFailed.hpp
+        ../../clp/CurlStringList.hpp
+        ../../clp/database_utils.cpp
+        ../../clp/database_utils.hpp
+        ../../clp/FileReader.cpp
+        ../../clp/FileReader.hpp
+        ../../clp/GlobalMetadataDBConfig.cpp
+        ../../clp/GlobalMetadataDBConfig.hpp
+        ../../clp/hash_utils.cpp
+        ../../clp/hash_utils.hpp
+        ../../clp/MySQLDB.cpp
+        ../../clp/MySQLDB.hpp
+        ../../clp/MySQLParamBindings.cpp
+        ../../clp/MySQLParamBindings.hpp
+        ../../clp/MySQLPreparedStatement.cpp
+        ../../clp/MySQLPreparedStatement.hpp
+        ../../clp/NetworkReader.cpp
+        ../../clp/NetworkReader.hpp
+        ../../clp/ReaderInterface.cpp
+        ../../clp/ReaderInterface.hpp
+        ../../clp/Thread.cpp
+        ../../clp/Thread.hpp
+        ../ArchiveReader.cpp
+        ../ArchiveReader.hpp
+        ../ArchiveReaderAdaptor.cpp
+        ../ArchiveReaderAdaptor.hpp
+        ../ColumnReader.cpp
+        ../ColumnReader.hpp
+        ../DictionaryReader.hpp
+        ../DictionaryEntry.cpp
+        ../DictionaryEntry.hpp
+        ../FileReader.cpp
+        ../FileReader.hpp
+        ../FileWriter.cpp
+        ../FileWriter.hpp
+        ../InputConfig.cpp
+        ../InputConfig.hpp
+        ../PackedStreamReader.cpp
+        ../PackedStreamReader.hpp
+        ../ReaderUtils.cpp
+        ../ReaderUtils.hpp
+        ../SchemaReader.cpp
+        ../SchemaReader.hpp
+        ../SchemaTree.cpp
+        ../SchemaTree.hpp
+        ../TimestampDictionaryReader.cpp
+        ../TimestampDictionaryReader.hpp
+        ../TimestampEntry.cpp
+        ../TimestampEntry.hpp
+        ../TimestampPattern.cpp
+        ../TimestampPattern.hpp
+        ../Utils.cpp
+        ../Utils.hpp
+        ../VariableDecoder.cpp
+        ../VariableDecoder.hpp
+        ../ZstdCompressor.cpp
+        ../ZstdCompressor.hpp
+        ../ZstdDecompressor.cpp
+        ../ZstdDecompressor.hpp
+        CommandLineArguments.cpp
+        CommandLineArguments.hpp
+        metadata_uploader.cpp
+        MySQLTableMetadataDB.cpp
+        MySQLTableMetadataDB.hpp
+        TableMetadataManager.cpp
+        TableMetadataManager.hpp
+)
+
+add_executable(metadata-uploader ${METADATA_UPLOADER_SOURCES})
+target_compile_features(metadata-uploader PRIVATE cxx_std_20)
+target_include_directories(metadata-uploader PRIVATE "${PROJECT_SOURCE_DIR}/submodules")
+target_link_libraries(metadata-uploader
+        PRIVATE
+        absl::flat_hash_map
+        Boost::iostreams Boost::program_options Boost::url
+        ${CURL_LIBRARIES}
+        clp::string_utils
+        MariaDBClient::MariaDBClient
+        OpenSSL::Crypto
+        simdjson
+        spdlog::spdlog
+        yaml-cpp::yaml-cpp
+        ZStd::ZStd
+)
+# Put the built executable at the root of the build directory
+set_target_properties(
+        metadata-uploader
+        PROPERTIES
+        RUNTIME_OUTPUT_DIRECTORY "${PROJECT_BINARY_DIR}"
+)

components/core/src/clp_s/metadata_uploader/CommandLineArguments.cpp

@@ -0,0 +1,123 @@
+#include "CommandLineArguments.hpp"
+
+#include <iostream>
+
+#include <boost/program_options.hpp>
+#include <spdlog/spdlog.h>
+
+
+namespace po = boost::program_options;
+
+namespace clp_s::metadata_uploader {
+CommandLineArguments::ParsingResult
+CommandLineArguments::parse_arguments(int argc, char const** argv) {
+    // Print out basic usage if user doesn't specify any options
+    if (1 == argc) {
+        print_basic_usage();
+        return ParsingResult::Failure;
+    }
+
+    // Define general options
+    po::options_description general_options("General Options");
+    general_options.add_options()("help,h", "Print help");
+
+    // Define output options
+    po::options_description output_options("Output Options");
+    std::string metadata_db_config_file_path;
+    // clang-format off
+    output_options.add_options()(
+            "db-config-file",
+            po::value<std::string>(&metadata_db_config_file_path)->value_name("FILE")
+                ->default_value(metadata_db_config_file_path),
+            "Table metadata DB YAML config"
+    );
+    // clang-format on
+
+    // Define visible options
+    po::options_description visible_options;
+    visible_options.add(general_options);
+    visible_options.add(output_options);
+
+    // Define hidden positional options (not shown in Boost's program options help message)
+    po::options_description positional_options;
+    // clang-format off
+    positional_options.add_options()
+            ("archive-dir", po::value<std::string>(&m_archive_dir))
+            ("archive-id", po::value<std::string>(&m_archive_id));
+    // clang-format on
+    po::positional_options_description positional_options_description;
+    positional_options_description.add("archive-dir", 1);
+    positional_options_description.add("archive-id", 1);
+
+    // Aggregate all options
+    po::options_description all_options;
+    all_options.add(general_options);
+    all_options.add(output_options);
+    all_options.add(positional_options);
+
+    // Parse options
+    try {
+        // Parse options specified on the command line
+        po::parsed_options parsed = po::command_line_parser(argc, argv)
+                                            .options(all_options)
+                                            .positional(positional_options_description)
+                                            .run();
+        po::variables_map parsed_command_line_options;
+        store(parsed, parsed_command_line_options);
+
+        notify(parsed_command_line_options);
+
+        // Handle --help
+        if (parsed_command_line_options.count("help")) {
+            if (argc > 2) {
+                SPDLOG_WARN("Ignoring all options besides --help.");
+            }
+
+            print_basic_usage();
+
+            std::cerr << visible_options << std::endl;
+            return ParsingResult::InfoCommand;
+        }
+
+        // Validate required parameters
+        if (m_archive_dir.empty()) {
+            throw std::invalid_argument("ARCHIVE_DIR not specified or empty.");
+        }
+        if (m_archive_id.empty()) {
+            throw std::invalid_argument("ARCHIVE_ID not specified or empty.");
+        }
+        if (false == metadata_db_config_file_path.empty()) {
+            clp::GlobalMetadataDBConfig metadata_db_config;
+            try {
+                metadata_db_config.parse_config_file(metadata_db_config_file_path);
+            } catch (std::exception& e) {
+                SPDLOG_ERROR("Failed to validate metadata database config - {}.", e.what());
+                return ParsingResult::Failure;
+            }
+
+            if (clp::GlobalMetadataDBConfig::MetadataDBType::MySQL
+                != metadata_db_config.get_metadata_db_type())
+            {
+                SPDLOG_ERROR(
+                        "Invalid metadata database type for {}; only supported type is MySQL.",
+                        m_program_name
+                );
+                return ParsingResult::Failure;
+            }
+
+            m_metadata_db_config = std::move(metadata_db_config);
+        }
+    } catch (std::exception& e) {
+        SPDLOG_ERROR("{}", e.what());
+        print_basic_usage();
+        return ParsingResult::Failure;
+    }
+
+    return ParsingResult::Success;
+}
+
+void CommandLineArguments::print_basic_usage() const {
+    std::cerr << "Usage: " << get_program_name() << " [OPTIONS] ARCHIVE_DIR ARCHIVE_ID"
+              << std::endl;
+}
+}  // namespace clp_s::metadata_uploader

components/core/src/clp_s/metadata_uploader/CommandLineArguments.hpp

@@ -0,0 +1,50 @@
+#ifndef CLP_S_METADATA_UPLOADER_COMMANDLINEARGUMENTS_HPP
+#define CLP_S_METADATA_UPLOADER_COMMANDLINEARGUMENTS_HPP
+
+#include <string>
+
+#include "../../clp/GlobalMetadataDBConfig.hpp"
+
+namespace clp_s::metadata_uploader {
+/**
+ * Class to parse command line arguments
+ */
+class CommandLineArguments {
+public:
+    // Types
+    enum class ParsingResult {
+        Success = 0,
+        InfoCommand,
+        Failure
+    };
+
+    // Constructors
+    explicit CommandLineArguments(std::string const& program_name) : m_program_name(program_name) {}
+
+    // Methods
+    ParsingResult parse_arguments(int argc, char const* argv[]);
+
+    std::string const& get_program_name() const { return m_program_name; }
+
+    std::string const& get_archive_dir() const { return m_archive_dir; }
+
+    std::string const& get_archive_id() const { return m_archive_id; }
+
+    std::optional<clp::GlobalMetadataDBConfig> const& get_db_config() const {
+        return m_metadata_db_config;
+    }
+
+private:
+    // Methods
+    void print_basic_usage() const;
+
+    // Variables
+    std::string m_program_name;
+    std::string m_archive_dir;
+    std::string m_archive_id;
+
+    std::optional<clp::GlobalMetadataDBConfig> m_metadata_db_config;
+};
+}  // namespace clp_s::metadata_uploader
+
+#endif  // CLP_S_METADATA_UPLOADER_COMMANDLINEARGUMENTS_HPP

components/core/src/clp_s/metadata_uploader/MySQLTableMetadataDB.cpp

@@ -0,0 +1,98 @@
+#include "MySQLTableMetadataDB.hpp"
+
+#include <fmt/core.h>
+#include <spdlog/spdlog.h>
+
+#include "../../clp/database_utils.hpp"
+#include "../../clp/type_utils.hpp"
+
+enum class TableMetadataFieldIndexes : uint16_t {
+    Name = 0,
+    Type,
+    Length,
+};
+
+namespace clp_s::metadata_uploader {
+void MySQLTableMetadataDB::open() {
+    if (m_is_open) {
+        throw OperationFailed(ErrorCodeNotReady, __FILENAME__, __LINE__);
+    }
+
+    m_db.open(m_host, m_port, m_username, m_password, m_database_name);
+    m_is_open = true;
+}
+
+void MySQLTableMetadataDB::init(std::string const& table_name) {
+    if (false == m_is_open) {
+        throw OperationFailed(ErrorCodeNotReady, __FILENAME__, __LINE__);
+    }
+
+    m_db.execute_query(fmt::format(
+            "CREATE TABLE IF NOT EXISTS {}{} ("
+            "name VARCHAR(512) NOT NULL, "
+            "type BIGINT NOT NULL,"
+            "PRIMARY KEY (name, type)"
+            ")",
+            m_table_prefix,
+            table_name
+    ));
+
+    m_insert_field_statement.reset();
+
+    std::vector<std::string> table_metadata_field_names(
+            clp::enum_to_underlying_type(TableMetadataFieldIndexes::Length)
+    );
+    table_metadata_field_names[clp::enum_to_underlying_type(TableMetadataFieldIndexes::Name)]
+            = "name";
+    table_metadata_field_names[clp::enum_to_underlying_type(TableMetadataFieldIndexes::Type)]
+            = "type";
+    fmt::memory_buffer statement_buffer;
+    auto statement_buffer_ix = std::back_inserter(statement_buffer);
+
+    fmt::format_to(
+            statement_buffer_ix,
+            "INSERT IGNORE INTO {}{} ({}) VALUES ({})",
+            m_table_prefix,
+            table_name,
+            clp::get_field_names_sql(table_metadata_field_names),
+            clp::get_placeholders_sql(table_metadata_field_names.size())
+    );
+    SPDLOG_DEBUG("{:.{}}", statement_buffer.data(), statement_buffer.size());
+    m_insert_field_statement = std::make_unique<MySQLPreparedStatement>(
+            m_db.prepare_statement(statement_buffer.data(), statement_buffer.size())
+    );
+
+    m_is_init = true;
+}
+
+void MySQLTableMetadataDB::add_field(std::string const& field_name,
+                                     NodeType field_type) {
+    if (false == m_is_init) {
+        throw OperationFailed(ErrorCodeNotReady, __FILENAME__, __LINE__);
+    }
+
+    auto& statement_bindings = m_insert_field_statement->get_statement_bindings();
+    statement_bindings.bind_varchar(
+            clp::enum_to_underlying_type(TableMetadataFieldIndexes::Name),
+            field_name.c_str(),
+            field_name.length()
+    );
+
+    uint64_t field_type_value = static_cast<uint64_t>(field_type);
+    statement_bindings.bind_uint64(
+            clp::enum_to_underlying_type(TableMetadataFieldIndexes::Type),
+            field_type_value
+    );
+
+    if (false == m_insert_field_statement->execute()) {
+        throw OperationFailed(ErrorCodeFailure, __FILENAME__, __LINE__);
+    }
+}
+
+void MySQLTableMetadataDB::close() {
+    m_insert_field_statement.reset();
+    m_db.close();
+    m_is_open = false;
+    m_is_init = false;
+}
+}  // namespace clp_s::metadata_uploader