Merge branch 'main' into package-image

Author: junhaoliao

.github/workflows/clp-s-generated-code-checks.yaml

@@ -24,6 +24,12 @@ jobs:
         with:
           submodules: "recursive"
 
+      - name: "Set up Java"
+        uses: "actions/setup-java@c5195efecf7bdfc987ee8bae7a71cb8b11521c00"
+        with:
+          distribution: "temurin"
+          java-version: "11"
+
       - name: "Install task"
         shell: "bash"
         run: "npm install -g @go-task/cli"
@@ -32,6 +38,10 @@ jobs:
         name: "Install coreutils (for md5sum)"
         run: "brew install coreutils"
 
+      - name: "Generate parsers"
+        shell: "bash"
+        run: "task clp-s-generate-parsers"
+
       - name: "Check if the generated parsers are the latest"
         shell: "bash"
         run:

components/clp-package-utils/clp_package_utils/scripts/dataset_manager.py

@@ -7,6 +7,8 @@
 
 from clp_py_utils.clp_config import (
     ARCHIVE_MANAGER_ACTION_NAME,
+    CLP_DB_PASS_ENV_VAR_NAME,
+    CLP_DB_USER_ENV_VAR_NAME,
     StorageEngine,
     StorageType,
 )
@@ -141,8 +143,12 @@ def main(argv: List[str]) -> int:
     if aws_mount:
         necessary_mounts.append(mounts.aws_config_dir)
 
+    extra_env_vars = {
+        CLP_DB_USER_ENV_VAR_NAME: clp_config.database.username,
+        CLP_DB_PASS_ENV_VAR_NAME: clp_config.database.password,
+    }
     container_start_cmd = generate_container_start_cmd(
-        container_name, necessary_mounts, clp_config.execution_container
+        container_name, necessary_mounts, clp_config.execution_container, extra_env_vars
     )
 
     if len(aws_env_vars) != 0:

components/clp-package-utils/clp_package_utils/scripts/native/compress.py

@@ -91,6 +91,10 @@ def handle_job_update(db, db_cursor, job_id, no_progress_reporting):
             # One or more tasks in the job has failed
             logger.error(f"Compression failed. {job_row['status_msg']}")
             break  # Done
+        if CompressionJobStatus.KILLED == job_status:
+            # The job is killed
+            logger.error(f"Compression killed. {job_row['status_msg']}")
+            break  # Done
 
         if CompressionJobStatus.RUNNING == job_status:
             if not no_progress_reporting:

components/clp-package-utils/clp_package_utils/scripts/native/dataset_manager.py

@@ -222,6 +222,7 @@ def main(argv: List[str]) -> int:
     try:
         clp_config = load_config_file(config_file_path, default_config_file_path, clp_home)
         clp_config.validate_logs_dir()
+        clp_config.database.load_credentials_from_env()
     except:
         logger.exception("Failed to load config.")
         return -1

components/clp-package-utils/clp_package_utils/scripts/native/utils.py

@@ -112,6 +112,7 @@ def wait_for_query_job(sql_adapter: SQL_Adapter, job_id: int) -> QueryJobStatus:
                 QueryJobStatus.SUCCEEDED,
                 QueryJobStatus.FAILED,
                 QueryJobStatus.CANCELLED,
+                QueryJobStatus.KILLED,
             ):
                 return new_status
 

components/clp-package-utils/clp_package_utils/scripts/start_clp.py

@@ -39,7 +39,7 @@
     get_files_table_name,
 )
 from clp_py_utils.s3_utils import generate_container_auth_options
-from job_orchestration.scheduler.constants import QueueName
+from job_orchestration.scheduler.constants import SchedulerType
 from pydantic import BaseModel
 
 from clp_package_utils.general import (
@@ -649,7 +649,7 @@ def start_compression_worker(
     mounts: CLPDockerMounts,
 ):
     celery_method = "job_orchestration.executor.compress"
-    celery_route = f"{QueueName.COMPRESSION}"
+    celery_route = SchedulerType.COMPRESSION
     compression_worker_mounts = [mounts.archives_output_dir]
     generic_start_worker(
         COMPRESSION_WORKER_COMPONENT_NAME,
@@ -674,7 +674,7 @@ def start_query_worker(
     mounts: CLPDockerMounts,
 ):
     celery_method = "job_orchestration.executor.query"
-    celery_route = f"{QueueName.QUERY}"
+    celery_route = SchedulerType.QUERY
 
     query_worker_mounts = [mounts.stream_output_dir]
     if StorageType.FS == clp_config.archive_output.storage.type:

components/core/src/clp/ffi/ir_stream/Deserializer.hpp

@@ -1,6 +1,7 @@
 #ifndef CLP_FFI_IR_STREAM_DESERIALIZER_HPP
 #define CLP_FFI_IR_STREAM_DESERIALIZER_HPP
 
+#include <cstddef>
 #include <cstdint>
 #include <memory>
 #include <string>
@@ -196,6 +197,7 @@ class Deserializer {
     IrUnitHandlerType m_ir_unit_handler;
     bool m_is_complete{false};
     [[no_unique_address]] QueryHandlerType m_query_handler;
+    size_t m_next_log_event_idx{0};
 };
 
 /**
@@ -304,6 +306,9 @@ auto Deserializer<IrUnitHandler, QueryHandlerType>::deserialize_next_ir_unit(
                     m_utc_offset
             ))};
 
+            auto const log_event_idx{m_next_log_event_idx};
+            m_next_log_event_idx += 1;
+
             if constexpr (search::IsNonEmptyQueryHandler<QueryHandlerType>::value) {
                 if (search::AstEvaluationResult::True
                     != YSTDLIB_ERROR_HANDLING_TRYX(
@@ -314,7 +319,9 @@ auto Deserializer<IrUnitHandler, QueryHandlerType>::deserialize_next_ir_unit(
                 }
             }
 
-            if (auto const err{m_ir_unit_handler.handle_log_event(std::move(log_event))};
+            if (auto const err{
+                        m_ir_unit_handler.handle_log_event(std::move(log_event), log_event_idx)
+                };
                 IRErrorCode::IRErrorCode_Success != err)
             {
                 return ir_error_code_to_errc(err);

components/core/src/clp/ffi/ir_stream/IrUnitHandlerReq.hpp

@@ -2,6 +2,7 @@
 #define CLP_FFI_IR_STREAM_IRUNITHANDLERREQ_HPP
 
 #include <concepts>
+#include <cstddef>
 #include <memory>
 #include <utility>
 
@@ -23,6 +24,7 @@ template <typename IrUnitHandlerType>
 concept IrUnitHandlerInterfaceReq = requires(
         IrUnitHandlerType handler,
         KeyValuePairLogEvent&& log_event,
+        size_t log_event_idx,
         bool is_auto_generated,
         UtcOffset utc_offset_old,
         UtcOffset utc_offset_new,
@@ -32,10 +34,11 @@ concept IrUnitHandlerInterfaceReq = requires(
     /**
      * Handles a log event IR unit.
      * @param log_event The deserialized result from IR deserializer.
+     * @param log_event_idx The log event index of `log_event` in the stream.
      * @return IRErrorCode::Success on success, user-defined error code on failures.
      */
     {
-        handler.handle_log_event(std::forward<KeyValuePairLogEvent &&>(log_event))
+        handler.handle_log_event(std::forward<KeyValuePairLogEvent &&>(log_event), log_event_idx)
     } -> std::same_as<IRErrorCode>;
 
     /**

components/core/src/clp/ffi/ir_stream/search/test/test_deserializer_integration.cpp

@@ -1,3 +1,4 @@
+#include <cstddef>
 #include <cstdint>
 #include <memory>
 #include <sstream>
@@ -40,7 +41,9 @@ using JsonPair = std::pair<nlohmann::json, nlohmann::json>;
  */
 class IrUnitHandler {
 public:
-    [[nodiscard]] auto handle_log_event(KeyValuePairLogEvent&& log_event) -> IRErrorCode {
+    [[nodiscard]] auto
+    handle_log_event(KeyValuePairLogEvent&& log_event, [[maybe_unused]] size_t log_event_idx)
+            -> IRErrorCode {
         m_deserialized_log_events.emplace_back(std::move(log_event));
         return IRErrorCode::IRErrorCode_Success;
     }

components/core/src/clp_s/JsonParser.cpp

@@ -1,5 +1,6 @@
 #include "JsonParser.hpp"
 
+#include <cstddef>
 #include <cstdint>
 #include <iostream>
 #include <memory>
@@ -45,7 +46,9 @@ namespace clp_s {
  */
 class IrUnitHandler {
 public:
-    [[nodiscard]] auto handle_log_event(KeyValuePairLogEvent&& log_event) -> IRErrorCode {
+    [[nodiscard]] auto
+    handle_log_event(KeyValuePairLogEvent&& log_event, [[maybe_unused]] size_t log_event_idx)
+            -> IRErrorCode {
         m_deserialized_log_event.emplace(std::move(log_event));
         return IRErrorCode::IRErrorCode_Success;
     }