From abffa6a79bebaa39628c1d8f464ee54a3d76fd35 Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Sat, 23 May 2026 00:52:00 -0400 Subject: [PATCH 1/2] feat(CLP-JSON): Add Messages and Files stats display to CLP-JSON Web UI Ingest page. --- .../clp_py_utils/clp_metadata_db_utils.py | 2 + .../src/clp/streaming_archive/Constants.hpp | 2 + components/core/src/clp_s/ArchiveWriter.cpp | 5 ++- components/core/src/clp_s/ArchiveWriter.hpp | 18 +++++++- .../executor/compress/compression_task.py | 2 + .../src/pages/IngestPage/Details/index.tsx | 37 ++++++---------- .../src/pages/IngestPage/Details/sql.ts | 43 +++++-------------- .../client/src/pages/IngestPage/sqlConfig.ts | 2 + 8 files changed, 52 insertions(+), 59 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py index ac9f26efc1..0d8f7966bb 100644 --- a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py +++ b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py @@ -32,6 +32,8 @@ def _create_archives_table(db_cursor, archives_table_name: str) -> None: `size` BIGINT NOT NULL, `creator_id` VARCHAR(64) NOT NULL, `creation_ix` INT NOT NULL, + `num_messages` BIGINT NULL DEFAULT NULL, + `num_files` BIGINT NULL DEFAULT NULL, KEY `archives_creation_order` (`creator_id`,`creation_ix`) USING BTREE, UNIQUE KEY `archive_id` (`id`) USING BTREE, PRIMARY KEY (`pagination_id`) diff --git a/components/core/src/clp/streaming_archive/Constants.hpp b/components/core/src/clp/streaming_archive/Constants.hpp index d5a162b931..63d29fffc7 100644 --- a/components/core/src/clp/streaming_archive/Constants.hpp +++ b/components/core/src/clp/streaming_archive/Constants.hpp @@ -34,6 +34,8 @@ constexpr char UncompressedSize[] = "uncompressed_size"; constexpr char Size[] = "size"; constexpr char CreatorId[] = "creator_id"; constexpr char CreationIx[] = "creation_ix"; +constexpr char NumMessages[] = "num_messages"; +constexpr char NumFiles[] = "num_files"; } // namespace Archive namespace File { diff --git a/components/core/src/clp_s/ArchiveWriter.cpp b/components/core/src/clp_s/ArchiveWriter.cpp index 50886ad99b..a785a961ea 100644 --- a/components/core/src/clp_s/ArchiveWriter.cpp +++ b/components/core/src/clp_s/ArchiveWriter.cpp @@ -120,7 +120,9 @@ auto ArchiveWriter::close(bool is_split) -> ArchiveStats { m_uncompressed_size, m_compressed_size, archive_range_index, - is_split + is_split, + static_cast(m_next_log_event_id), + m_num_files }; if (m_print_archive_stats) { std::cout << archive_stats.as_string() << '\n'; @@ -135,6 +137,7 @@ auto ArchiveWriter::close(bool is_split) -> ArchiveStats { m_uncompressed_size = 0UL; m_compressed_size = 0UL; m_next_log_event_id = 0; + m_num_files = 0; m_authoritative_timestamp.clear(); m_authoritative_timestamp_namespace.clear(); m_matched_timestamp_prefix_length = 0ULL; diff --git a/components/core/src/clp_s/ArchiveWriter.hpp b/components/core/src/clp_s/ArchiveWriter.hpp index 452df9fe2f..dba53e9ecd 100644 --- a/components/core/src/clp_s/ArchiveWriter.hpp +++ b/components/core/src/clp_s/ArchiveWriter.hpp @@ -47,7 +47,9 @@ class ArchiveStats { size_t uncompressed_size, size_t compressed_size, nlohmann::json range_index, - bool is_split + bool is_split, + size_t num_messages, + size_t num_files ) : m_id{id}, m_begin_timestamp{begin_timestamp}, @@ -55,7 +57,9 @@ class ArchiveStats { m_uncompressed_size{uncompressed_size}, m_compressed_size{compressed_size}, m_range_index(std::move(range_index)), // Avoid {} to prevent wrapping in JSON array. - m_is_split{is_split} {} + m_is_split{is_split}, + m_num_messages{num_messages}, + m_num_files{num_files} {} // Methods /** @@ -72,6 +76,8 @@ class ArchiveStats { {Archive::EndTimestamp, m_end_timestamp}, {Archive::UncompressedSize, m_uncompressed_size}, {Archive::Size, m_compressed_size}, + {Archive::NumMessages, m_num_messages}, + {Archive::NumFiles, m_num_files}, {File::IsSplit, m_is_split}, {cRangeIndex, m_range_index}}; return json_msg.dump(-1, ' ', false, nlohmann::json::error_handler_t::ignore); @@ -91,6 +97,10 @@ class ArchiveStats { [[nodiscard]] auto get_is_split() const -> bool { return m_is_split; } + [[nodiscard]] auto get_num_messages() const -> size_t { return m_num_messages; } + + [[nodiscard]] auto get_num_files() const -> size_t { return m_num_files; } + private: std::string m_id; epochtime_t m_begin_timestamp{}; @@ -99,6 +109,8 @@ class ArchiveStats { size_t m_compressed_size{}; nlohmann::json m_range_index; bool m_is_split{}; + size_t m_num_messages{}; + size_t m_num_files{}; }; class ArchiveWriter { @@ -272,6 +284,7 @@ class ArchiveWriter { return rc; } m_range_open = true; + ++m_num_files; } return m_range_index_writer.add_value_to_range(key, value); } @@ -375,6 +388,7 @@ class ArchiveWriter { RangeIndexWriter m_range_index_writer; bool m_range_open{false}; + size_t m_num_files{}; }; } // namespace clp_s diff --git a/components/job-orchestration/job_orchestration/executor/compress/compression_task.py b/components/job-orchestration/job_orchestration/executor/compress/compression_task.py index 13e5ea1361..5584cbe672 100644 --- a/components/job-orchestration/job_orchestration/executor/compress/compression_task.py +++ b/components/job-orchestration/job_orchestration/executor/compress/compression_task.py @@ -106,6 +106,8 @@ def update_archive_metadata( "begin_timestamp", "end_timestamp", "id", + "num_messages", + "num_files", "size", "uncompressed_size", ] diff --git a/components/webui/client/src/pages/IngestPage/Details/index.tsx b/components/webui/client/src/pages/IngestPage/Details/index.tsx index bce7950c66..29ee368117 100644 --- a/components/webui/client/src/pages/IngestPage/Details/index.tsx +++ b/components/webui/client/src/pages/IngestPage/Details/index.tsx @@ -42,31 +42,20 @@ const Details = () => { enabled: CLP_STORAGE_ENGINES.CLP === SETTINGS_STORAGE_ENGINE || isSuccessDatasetNames, }); - if (CLP_STORAGE_ENGINES.CLP === SETTINGS_STORAGE_ENGINE) { - return ( -
-
- -
- - -
- ); - } - return ( -
- +
+
+ +
+ +
); }; diff --git a/components/webui/client/src/pages/IngestPage/Details/sql.ts b/components/webui/client/src/pages/IngestPage/Details/sql.ts index b2faead0ce..44851648c8 100644 --- a/components/webui/client/src/pages/IngestPage/Details/sql.ts +++ b/components/webui/client/src/pages/IngestPage/Details/sql.ts @@ -70,43 +70,22 @@ FROM const buildMultiDatasetDetailsSql = (datasetNames: string[]): string => { const archiveQueries = datasetNames.map((name) => ` SELECT - MIN(${CLP_ARCHIVES_TABLE_COLUMN_NAMES.BEGIN_TIMESTAMP}) AS begin_timestamp, - MAX(${CLP_ARCHIVES_TABLE_COLUMN_NAMES.END_TIMESTAMP}) AS end_timestamp + ${CLP_ARCHIVES_TABLE_COLUMN_NAMES.BEGIN_TIMESTAMP}, + ${CLP_ARCHIVES_TABLE_COLUMN_NAMES.END_TIMESTAMP}, + ${CLP_ARCHIVES_TABLE_COLUMN_NAMES.NUM_MESSAGES}, + ${CLP_ARCHIVES_TABLE_COLUMN_NAMES.NUM_FILES} FROM ${settings.SqlDbClpTablePrefix}${name}_${SqlTableSuffix.ARCHIVES} `); - const fileQueries = datasetNames.map((name) => ` - SELECT - COUNT(DISTINCT ${CLP_FILES_TABLE_COLUMN_NAMES.ORIG_FILE_ID}) AS num_files, - CAST( - COALESCE(SUM(${CLP_FILES_TABLE_COLUMN_NAMES.NUM_MESSAGES}), 0) AS UNSIGNED - ) AS num_messages - FROM ${settings.SqlDbClpTablePrefix}${name}_${SqlTableSuffix.FILES} - `); - return ` SELECT - a.begin_timestamp, - a.end_timestamp, - b.num_files, - b.num_messages - FROM - ( - SELECT - MIN(begin_timestamp) AS begin_timestamp, - MAX(end_timestamp) AS end_timestamp - FROM ( - ${archiveQueries.join("\nUNION ALL\n")} - ) AS archives_combined - ) a, - ( - SELECT - SUM(num_files) AS num_files, - SUM(num_messages) AS num_messages - FROM ( - ${fileQueries.join("\nUNION ALL\n")} - ) AS files_combined - ) b; + MIN(begin_timestamp) AS begin_timestamp, + MAX(end_timestamp) AS end_timestamp, + CAST(COALESCE(SUM(num_messages), 0) AS UNSIGNED) AS num_messages, + CAST(COALESCE(SUM(num_files), 0) AS UNSIGNED) AS num_files + FROM ( + ${archiveQueries.join("\nUNION ALL\n")} + ) AS archives_combined `; }; diff --git a/components/webui/client/src/pages/IngestPage/sqlConfig.ts b/components/webui/client/src/pages/IngestPage/sqlConfig.ts index 3a8ec48be1..e8157e8681 100644 --- a/components/webui/client/src/pages/IngestPage/sqlConfig.ts +++ b/components/webui/client/src/pages/IngestPage/sqlConfig.ts @@ -6,6 +6,8 @@ enum CLP_ARCHIVES_TABLE_COLUMN_NAMES { END_TIMESTAMP = "end_timestamp", UNCOMPRESSED_SIZE = "uncompressed_size", SIZE = "size", + NUM_MESSAGES = "num_messages", + NUM_FILES = "num_files", } /** From 6e735773a262160fa5648c486ace3734c8db0c29 Mon Sep 17 00:00:00 2001 From: Junhao Liao Date: Sat, 23 May 2026 02:58:21 -0400 Subject: [PATCH 2/2] fix(clp-s): Count distinct input files instead of range splits for `num_files` stat. Previously, `m_num_files` was incremented in `add_field_to_current_range()` every time a new range was opened, which over-counted files when a single large input file was split across multiple archives. Now the increment is moved to the callers (`ingest_json`/`ingest_kvier`) where `file_split_number` is available, so only the first split (split 0) of each input file is counted. --- components/core/src/clp_s/ArchiveWriter.hpp | 3 ++- components/core/src/clp_s/JsonParser.cpp | 6 ++++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/components/core/src/clp_s/ArchiveWriter.hpp b/components/core/src/clp_s/ArchiveWriter.hpp index dba53e9ecd..49be0e4a5b 100644 --- a/components/core/src/clp_s/ArchiveWriter.hpp +++ b/components/core/src/clp_s/ArchiveWriter.hpp @@ -284,7 +284,6 @@ class ArchiveWriter { return rc; } m_range_open = true; - ++m_num_files; } return m_range_index_writer.add_value_to_range(key, value); } @@ -301,6 +300,8 @@ class ArchiveWriter { return rc; } + void increment_num_files() { ++m_num_files; } + private: /** * Initializes the schema writer diff --git a/components/core/src/clp_s/JsonParser.cpp b/components/core/src/clp_s/JsonParser.cpp index 69e9766bbc..378d680cc8 100644 --- a/components/core/src/clp_s/JsonParser.cpp +++ b/components/core/src/clp_s/JsonParser.cpp @@ -723,6 +723,9 @@ auto JsonParser::ingest_json( size_t file_split_number{0ULL}; int32_t log_event_idx_node_id{}; auto initialize_fields_for_archive = [&]() -> bool { + if (0 == file_split_number) { + m_archive_writer->increment_num_files(); + } if (false == m_record_log_order) { return true; } @@ -891,6 +894,9 @@ auto JsonParser::ingest_kvir( size_t file_split_number{0ULL}; int32_t log_event_idx_node_id{}; auto initialize_fields_for_archive = [&]() -> bool { + if (0 == file_split_number) { + m_archive_writer->increment_num_files(); + } if (false == m_record_log_order) { return true; }