From 578aa1e11ac919f9b604d8f9a56ed40528dce646 Mon Sep 17 00:00:00 2001 From: gibber9809 Date: Fri, 17 Oct 2025 16:03:04 +0000 Subject: [PATCH 1/4] Add support for detecting unstructured logtext inputs. --- components/core/cmake/Options/options.cmake | 1 + components/core/src/clp_s/CMakeLists.txt | 1 + components/core/src/clp_s/InputConfig.cpp | 59 +++++++++++++++++++++ components/core/src/clp_s/InputConfig.hpp | 1 + 4 files changed, 62 insertions(+) diff --git a/components/core/cmake/Options/options.cmake b/components/core/cmake/Options/options.cmake index 1c8c327360..3802b3dde4 100644 --- a/components/core/cmake/Options/options.cmake +++ b/components/core/cmake/Options/options.cmake @@ -270,6 +270,7 @@ function(set_clp_s_io_dependencies) set_clp_need_flags( CLP_NEED_BOOST CLP_NEED_FMT + CLP_NEED_SIMDJSON CLP_NEED_SPDLOG CLP_NEED_ZSTD ) diff --git a/components/core/src/clp_s/CMakeLists.txt b/components/core/src/clp_s/CMakeLists.txt index b768488565..bb45ea8f4f 100644 --- a/components/core/src/clp_s/CMakeLists.txt +++ b/components/core/src/clp_s/CMakeLists.txt @@ -214,6 +214,7 @@ if(CLP_BUILD_CLP_S_IO) Boost::iostreams Boost::url clp_s::clp_dependencies fmt::fmt + simdjson::simdjson spdlog::spdlog ${zstd_TARGET} ) diff --git a/components/core/src/clp_s/InputConfig.cpp b/components/core/src/clp_s/InputConfig.cpp index e94c8bb8db..b87c114ad4 100644 --- a/components/core/src/clp_s/InputConfig.cpp +++ b/components/core/src/clp_s/InputConfig.cpp @@ -7,9 +7,11 @@ #include #include #include +#include #include #include +#include #include #include "../clp/aws/AwsAuthenticationSigner.hpp" @@ -21,6 +23,7 @@ #include "../clp/spdlog_with_specializations.hpp" #include "../clp/streaming_compression/Decompressor.hpp" #include "../clp/streaming_compression/zstd/Decompressor.hpp" +#include "../clp/utf8_utils.hpp" #include "Utils.hpp" namespace clp_s { @@ -134,6 +137,14 @@ auto could_be_kvir(char const* peek_buf, size_t peek_size) -> bool; */ auto could_be_json(char const* peek_buf, size_t peek_size) -> bool; +/** + * Checks if an input contains logtext, based on the first few bytes of data from the input. + * @param peek_buf A pointer to a buffer containing peeked data from the start of an input stream. + * @param peek_size The number of bytes of peeked data in the buffer. + * @return Whether the input could be logtext. + */ +auto could_be_logtext(char const* peek_buf, size_t peek_size) -> bool; + auto try_create_file_reader(std::string_view const file_path) -> std::shared_ptr { try { @@ -242,6 +253,49 @@ auto could_be_json(char const* peek_buf, size_t peek_size) -> bool { return false; } +auto could_be_logtext(char const* peek_buf, size_t peek_size) -> bool { + constexpr size_t cMaxUtf8CodepointBytes = 4ULL; + // Full 4-byte character followed by truncated 3-bytes of 4-byte character. + constexpr size_t cMaxRunWithoutFullUtf8Codepoint = 2 * cMaxUtf8CodepointBytes - 1ULL; + + size_t cur_byte{ + peek_size < cMaxRunWithoutFullUtf8Codepoint + ? 0ULL + : (peek_size - cMaxRunWithoutFullUtf8Codepoint) + }; + + std::optional legal_last_byte_index{std::nullopt}; + auto mark_last_legal_character = [&](size_t remaining_bytes_in_char) { + auto const last_byte_in_char = cur_byte + remaining_bytes_in_char; + if (last_byte_in_char < peek_size) { + legal_last_byte_index = last_byte_in_char; + } + cur_byte = last_byte_in_char; + }; + + // Find the last complete UTF-8 character if it exists. We don't bother validating continuation + // bytes, since they will be checked in the final validation step anyway. + for (; cur_byte < peek_size; ++cur_byte) { + uint8_t const c{static_cast(peek_buf[cur_byte])}; + if ((clp::cFourByteUtf8CharHeaderMask & c) == clp::cFourByteUtf8CharHeader) { + mark_last_legal_character(3ULL); + } else if ((clp::cThreeByteUtf8CharHeaderMask & c) == clp::cThreeByteUtf8CharHeader) { + mark_last_legal_character(2ULL); + } else if ((clp::cTwoByteUtf8CharHeaderMask & c) == clp::cThreeByteUtf8CharHeader) { + mark_last_legal_character(1ULL); + } else if (clp::utf8_utils_internal::is_ascii_char(c)) { + mark_last_legal_character(0ULL); + } + } + + // We were not able to find a complete UTF-8 character, so the input is definitely not UTF-8. + if (false == legal_last_byte_index.has_value()) { + return false; + } + + return simdjson::validate_utf8(peek_buf, legal_last_byte_index.value() + 1ULL); +} + auto peek_start_and_deduce_type(std::shared_ptr& reader) -> FileType { char const* peek_buf{}; size_t peek_size{}; @@ -262,6 +316,10 @@ auto peek_start_and_deduce_type(std::shared_ptr& reader) -> return FileType::Json; } + if (could_be_logtext(peek_buf, peek_size)) { + return FileType::LogText; + } + return FileType::Unknown; } } // namespace @@ -307,6 +365,7 @@ auto try_create_reader(Path const& path, NetworkAuthOption const& network_auth) switch (type) { case FileType::Json: case FileType::KeyValueIr: + case FileType::LogText: return {std::move(readers), type}; case FileType::Zstd: { readers.emplace_back( diff --git a/components/core/src/clp_s/InputConfig.hpp b/components/core/src/clp_s/InputConfig.hpp index 45c880f06e..15562d8807 100644 --- a/components/core/src/clp_s/InputConfig.hpp +++ b/components/core/src/clp_s/InputConfig.hpp @@ -23,6 +23,7 @@ constexpr char cAwsSessionTokenEnvVar[] = "AWS_SESSION_TOKEN"; enum class FileType : uint8_t { Json = 0, KeyValueIr, + LogText, Zstd, Unknown }; From 2ed2c39aa64e93eb6831789c6bd411705da26397 Mon Sep 17 00:00:00 2001 From: gibber9809 Date: Fri, 17 Oct 2025 16:03:45 +0000 Subject: [PATCH 2/4] Explicitly reject regular logtext as input to compression in clp-s. --- components/core/src/clp_s/JsonParser.cpp | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/components/core/src/clp_s/JsonParser.cpp b/components/core/src/clp_s/JsonParser.cpp index 2d45581842..8fbf597b18 100644 --- a/components/core/src/clp_s/JsonParser.cpp +++ b/components/core/src/clp_s/JsonParser.cpp @@ -637,6 +637,13 @@ bool JsonParser::ingest() { case FileType::KeyValueIr: ingestion_successful = ingest_kvir(nested_readers.back(), path, archive_creator_id); break; + case FileType::LogText: + SPDLOG_ERROR( + "Direct ingestion of unstructured logtext is not supported from input {}", + path.path + ); + std::ignore = m_archive_writer->close(); + return false; case FileType::Zstd: case FileType::Unknown: default: { From 0f5b069e7a06206ff922f77576874491ea0fd66f Mon Sep 17 00:00:00 2001 From: gibber9809 Date: Fri, 17 Oct 2025 17:51:14 +0000 Subject: [PATCH 3/4] Fix typo. --- components/core/src/clp_s/InputConfig.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/core/src/clp_s/InputConfig.cpp b/components/core/src/clp_s/InputConfig.cpp index b87c114ad4..cf3421ef79 100644 --- a/components/core/src/clp_s/InputConfig.cpp +++ b/components/core/src/clp_s/InputConfig.cpp @@ -281,7 +281,7 @@ auto could_be_logtext(char const* peek_buf, size_t peek_size) -> bool { mark_last_legal_character(3ULL); } else if ((clp::cThreeByteUtf8CharHeaderMask & c) == clp::cThreeByteUtf8CharHeader) { mark_last_legal_character(2ULL); - } else if ((clp::cTwoByteUtf8CharHeaderMask & c) == clp::cThreeByteUtf8CharHeader) { + } else if ((clp::cTwoByteUtf8CharHeaderMask & c) == clp::cTwoByteUtf8CharHeader) { mark_last_legal_character(1ULL); } else if (clp::utf8_utils_internal::is_ascii_char(c)) { mark_last_legal_character(0ULL); From 9641345b31f553bec3a2f64096ab7cbf83aebe15 Mon Sep 17 00:00:00 2001 From: gibber9809 Date: Mon, 20 Oct 2025 18:20:41 +0000 Subject: [PATCH 4/4] Update comments in could_be_logtext per review suggestion. --- components/core/src/clp_s/InputConfig.cpp | 24 +++++++++++++++++++---- 1 file changed, 20 insertions(+), 4 deletions(-) diff --git a/components/core/src/clp_s/InputConfig.cpp b/components/core/src/clp_s/InputConfig.cpp index cf3421ef79..d4c892a14f 100644 --- a/components/core/src/clp_s/InputConfig.cpp +++ b/components/core/src/clp_s/InputConfig.cpp @@ -253,9 +253,28 @@ auto could_be_json(char const* peek_buf, size_t peek_size) -> bool { return false; } +/** + * This function decides whether the buffer could be logtext based on whether it contains valid, but + * potentially truncated, UTF-8 data. + * + * To check for valid UTF-8 while accounting for truncation we need to: + * 1. Find the last complete UTF-8 codepoint in the buffer + * 2. Validate that all of the data in the buffer including the last complete codepoint is valid + * UTF-8. + * + * For the first step we can always find the last complete codepoint by scanning the last 7 bytes of + * the buffer. This is because in the worst case, the stream terminates with a 4-byte codepoint + * followed by another 4-byte codepoint with its last byte truncated. The approach, then, is to + * scan the last 7 bytes of the buffer to locate the last complete codepoint. Since we use full + * UTF-8 validation in the second step, this scan only needs to look for UTF-8 header bytes without + * validating continuation bytes. If no complete codepoint can be found, we know that the input is + * not valid UTF-8, otherwise we continue on to validating the entire buffer through the end of the + * last complete codepoint. + * + * For the second step, we simply use an off-the-shelf fast UTF-8 validator. + */ auto could_be_logtext(char const* peek_buf, size_t peek_size) -> bool { constexpr size_t cMaxUtf8CodepointBytes = 4ULL; - // Full 4-byte character followed by truncated 3-bytes of 4-byte character. constexpr size_t cMaxRunWithoutFullUtf8Codepoint = 2 * cMaxUtf8CodepointBytes - 1ULL; size_t cur_byte{ @@ -273,8 +292,6 @@ auto could_be_logtext(char const* peek_buf, size_t peek_size) -> bool { cur_byte = last_byte_in_char; }; - // Find the last complete UTF-8 character if it exists. We don't bother validating continuation - // bytes, since they will be checked in the final validation step anyway. for (; cur_byte < peek_size; ++cur_byte) { uint8_t const c{static_cast(peek_buf[cur_byte])}; if ((clp::cFourByteUtf8CharHeaderMask & c) == clp::cFourByteUtf8CharHeader) { @@ -288,7 +305,6 @@ auto could_be_logtext(char const* peek_buf, size_t peek_size) -> bool { } } - // We were not able to find a complete UTF-8 character, so the input is definitely not UTF-8. if (false == legal_last_byte_index.has_value()) { return false; }