feat(CLP-JSON): Add Messages and Files stats display to CLP-JSON Web UI Ingest page.

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`)

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 {

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<size_t>(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;

components/core/src/clp_s/ArchiveWriter.hpp

@@ -47,15 +47,19 @@ 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},
               m_end_timestamp{end_timestamp},
               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 {
@@ -288,6 +300,8 @@ class ArchiveWriter {
         return rc;
     }
 
+    void increment_num_files() { ++m_num_files; }
+
 private:
     /**
      * Initializes the schema writer
@@ -375,6 +389,7 @@ class ArchiveWriter {
 
     RangeIndexWriter m_range_index_writer;
     bool m_range_open{false};
+    size_t m_num_files{};
 };
 }  // namespace clp_s
 

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;
         }

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",
     ]

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 (
-            <div className={styles["detailsGrid"]}>
-                <div className={styles["timeRange"]}>
-                    <TimeRange
-                        beginDate={dayjs.utc(details.begin_timestamp)}
-                        endDate={dayjs.utc(details.end_timestamp)}
-                        isLoading={isPending}/>
-                </div>
-                <Messages
-                    isLoading={isPending}
-                    numMessages={details.num_messages}/>
-                <Files
-                    isLoading={isPending}
-                    numFiles={details.num_files}/>
-            </div>
-        );
-    }
-
     return (
-        <div>
-            <TimeRange
-                beginDate={dayjs.utc(details.begin_timestamp)}
-                endDate={dayjs.utc(details.end_timestamp)}
-                isLoading={isPending}/>
+        <div className={styles["detailsGrid"]}>
+            <div className={styles["timeRange"]}>
+                <TimeRange
+                    beginDate={dayjs.utc(details.begin_timestamp)}
+                    endDate={dayjs.utc(details.end_timestamp)}
+                    isLoading={isPending}/>
+            </div>
+            <Messages
+                isLoading={isPending}
+                numMessages={details.num_messages}/>
+            <Files
+                isLoading={isPending}
+                numFiles={details.num_files}/>
         </div>
     );
 };

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
   `;
 };
 

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",
 }
 
 /**