Merge branch 'main' into k8s-presto

Resolve conflicts:
- Chart.yaml: Accept main's version (0.2.1-dev.0)
- values.yaml: Keep prestoWorker section, drop storage.storageClassName
  (host-path PVs removed on main)
- set-up-multi-{dedicated,shared}-test.sh: Combine Presto node support with
  main's get_image_helm_args
- guides-k8s-deployment.md: Keep Presto link references
- Accept deletion of all host-path PV/PVC files (query-scheduler-logs,
  query-worker-logs, query-worker-staged-streams, reducer-logs)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: junhaoliao

Cargo.lock

@@ -36,7 +36,5 @@ thiserror = "2.0.18"
 tokio = { version = "1.49.0", features = ["full"] }
 tower-http = { version = "0.6.8", features = ["cors"] }
 tracing = "0.1.44"
-tracing-appender = "0.2.4"
-tracing-subscriber = { version = "0.3.22", features = ["json", "env-filter", "fmt", "std"] }
 utoipa = { version = "5.4.0", features = ["axum_extras"] }
 utoipa-axum = "0.2.0"

components/api-server/Cargo.toml

@@ -1,11 +1,6 @@
 use anyhow::Context;
 use clap::Parser;
 use clp_rust_utils::{clp_config::package, serde::yaml};
-use tracing_appender::{
-    non_blocking::WorkerGuard,
-    rolling::{RollingFileAppender, Rotation},
-};
-use tracing_subscriber::{self, fmt::writer::MakeWriterExt};
 
 #[derive(Parser)]
 #[command(version, about = "API Server for CLP.")]
@@ -40,29 +35,6 @@ fn read_config_and_credentials(
     Ok((config, credentials))
 }
 
-fn set_up_logging() -> anyhow::Result<WorkerGuard> {
-    let logs_directory =
-        std::env::var("CLP_LOGS_DIR").context("Expect `CLP_LOGS_DIR` environment variable.")?;
-    let logs_directory = std::path::Path::new(logs_directory.as_str());
-    let file_appender =
-        RollingFileAppender::new(Rotation::HOURLY, logs_directory, "api_server.log");
-    let (non_blocking_writer, guard) = tracing_appender::non_blocking(file_appender);
-    tracing_subscriber::fmt()
-        .event_format(
-            tracing_subscriber::fmt::format()
-                .with_level(true)
-                .with_target(false)
-                .with_file(true)
-                .with_line_number(true)
-                .json(),
-        )
-        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
-        .with_ansi(false)
-        .with_writer(std::io::stdout.and(non_blocking_writer))
-        .init();
-    Ok(guard)
-}
-
 async fn shutdown_signal() {
     let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
         .expect("failed to listen for SIGTERM");
@@ -79,7 +51,7 @@ async fn main() -> anyhow::Result<()> {
     let args = Args::parse();
 
     let (config, credentials) = read_config_and_credentials(&args)?;
-    let _guard = set_up_logging()?;
+    let _guard = clp_rust_utils::logging::set_up_logging("api_server.log");
 
     let api_server_config = config
         .api_server

components/api-server/src/bin/api_server.rs

@@ -28,9 +28,9 @@ pub struct QueryConfig {
     /// The search query as a KQL string.
     pub query_string: String,
 
-    /// The dataset to search within. If not provided, only `default` dataset will be searched.
+    /// The datasets to search within. If not provided, only `default` dataset will be searched.
     #[serde(default)]
-    pub dataset: Option<String>,
+    pub datasets: Option<Vec<String>>,
 
     /// The maximum number of results to return. Set to `0` for no limit.
     #[serde(default)]
@@ -58,7 +58,7 @@ pub struct QueryConfig {
 impl From<QueryConfig> for SearchJobConfig {
     fn from(value: QueryConfig) -> Self {
         Self {
-            dataset: value.dataset,
+            datasets: value.datasets,
             query_string: value.query_string,
             max_num_results: value.max_num_results,
             begin_timestamp: value.time_range_begin_millisecs,
@@ -128,10 +128,10 @@ impl Client {
     /// * Forwards [`sqlx::query::Query::execute`]'s return values on failure.
     pub async fn submit_query(&self, query_config: QueryConfig) -> Result<u64, ClientError> {
         let mut search_job_config: SearchJobConfig = query_config.into();
-        if search_job_config.dataset.is_none() {
-            search_job_config.dataset = match self.config.package.storage_engine {
+        if search_job_config.datasets.is_none() {
+            search_job_config.datasets = match self.config.package.storage_engine {
                 StorageEngine::Clp => None,
-                StorageEngine::ClpS => Some("default".to_owned()),
+                StorageEngine::ClpS => Some(vec!["default".to_owned()]),
             }
         }
         if search_job_config.max_num_results == 0 {
@@ -483,6 +483,52 @@ impl Client {
         Ok(mapped)
     }
 
+    /// Retrieves timestamp column names for a given dataset.
+    ///
+    /// # Returns
+    ///
+    /// A vector of timestamp column name strings on success.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if:
+    ///
+    /// * [`ClientError::InvalidDatasetName`] if the dataset name contains invalid characters.
+    /// * [`ClientError::DatasetNotFound`] if the dataset's column metadata table doesn't exist.
+    /// * Forwards [`sqlx::query::Query::fetch_all`]'s return values on failure.
+    pub async fn get_timestamp_column_names(
+        &self,
+        dataset_name: &str,
+    ) -> Result<Vec<String>, ClientError> {
+        // Must be kept in sync with `NodeType::Timestamp` in
+        // `components/core/src/clp_s/SchemaTree.hpp`.
+        const TIMESTAMP_NODE_TYPE: i8 = 14;
+        // MySQL error number for "Table doesn't exist".
+        const MYSQL_TABLE_NOT_FOUND: u16 = 1146;
+
+        if !clp_rust_utils::dataset::VALID_DATASET_NAME_REGEX.is_match(dataset_name) {
+            return Err(ClientError::InvalidDatasetName);
+        }
+        let table_name = format!("clp_{dataset_name}_column_metadata");
+        let names: Vec<String> =
+            sqlx::query_scalar(&format!("SELECT name FROM `{table_name}` WHERE type = ?"))
+                .bind(TIMESTAMP_NODE_TYPE)
+                .fetch_all(&self.sql_pool)
+                .await
+                .map_err(|err| {
+                    if let sqlx::Error::Database(db_err) = &err
+                        && let Some(mysql_err) =
+                            db_err.try_downcast_ref::<sqlx::mysql::MySqlDatabaseError>()
+                        && mysql_err.number() == MYSQL_TABLE_NOT_FOUND
+                    {
+                        return ClientError::DatasetNotFound(dataset_name.to_owned());
+                    }
+                    err.into()
+                })?;
+
+        Ok(names)
+    }
+
     /// # Returns
     ///
     /// A reference to the API server configuration.

components/api-server/src/client.rs

@@ -27,6 +27,12 @@ pub enum ClientError {
 
     #[error("Search job not found: {0}")]
     SearchJobNotFound(u64),
+
+    #[error("Invalid dataset name")]
+    InvalidDatasetName,
+
+    #[error("Dataset not found: {0}")]
+    DatasetNotFound(String),
 }
 
 /// Empty trait to mark errors that indicate malformed data.

components/api-server/src/error.rs

@@ -36,6 +36,10 @@ pub fn from_client(client: Client) -> Result<axum::Router, serde_json::Error> {
         .routes(routes!(query))
         .routes(routes!(query_results))
         .routes(routes!(cancel_query))
+        .route(
+            "/column_metadata/{dataset_name}/timestamp",
+            get(get_timestamp_column_names),
+        )
         .with_state(client)
         .split_for_parts();
     let api_json = api.to_json()?;
@@ -85,7 +89,7 @@ async fn health() -> String {
         content= QueryConfig,
         example = json!({
             "query_string": "*",
-            "dataset": "default",
+            "datasets": ["default"],
             "time_range_begin_millisecs": 0,
             "time_range_end_millisecs": 17_356_896,
             "ignore_case": true,
@@ -221,13 +225,33 @@ async fn cancel_query(
     }
 }
 
+async fn get_timestamp_column_names(
+    State(client): State<Client>,
+    Path(dataset_name): Path<String>,
+) -> Result<Json<Vec<String>>, HandlerError> {
+    let names = client
+        .get_timestamp_column_names(&dataset_name)
+        .await
+        .map_err(|err| {
+            tracing::error!(
+                "Failed to get timestamp column names for dataset '{}': {:?}",
+                dataset_name,
+                err
+            );
+            HandlerError::from(err)
+        })?;
+    Ok(Json(names))
+}
+
 /// Generic errors for request handlers.
 #[derive(Error, Debug)]
 enum HandlerError {
     #[error("Internal server error")]
     InternalServer,
     #[error("Not found")]
     NotFound,
+    #[error("Bad request: {0}")]
+    BadRequest(String),
 }
 
 impl From<axum::Error> for HandlerError {
@@ -239,7 +263,8 @@ impl From<axum::Error> for HandlerError {
 impl From<ClientError> for HandlerError {
     fn from(err: ClientError) -> Self {
         match err {
-            ClientError::SearchJobNotFound(_) => Self::NotFound,
+            ClientError::SearchJobNotFound(_) | ClientError::DatasetNotFound(_) => Self::NotFound,
+            ClientError::InvalidDatasetName => Self::BadRequest(format!("{err}")),
             _ => Self::InternalServer,
         }
     }
@@ -251,6 +276,7 @@ impl IntoResponse for HandlerError {
         match self {
             Self::NotFound => StatusCode::NOT_FOUND.into_response(),
             Self::InternalServer => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
+            Self::BadRequest(msg) => (StatusCode::BAD_REQUEST, msg).into_response(),
         }
     }
 }

components/api-server/src/routes.rs

@@ -60,7 +60,7 @@ async def submit_query(
         job_config = msgpack.packb(
             {
                 "begin_timestamp": begin_ts,
-                "dataset": CLP_DEFAULT_DATASET_NAME,
+                "datasets": [CLP_DEFAULT_DATASET_NAME],
                 "end_timestamp": end_ts,
                 "ignore_case": True,
                 "max_num_results": SEARCH_MAX_NUM_RESULTS,

components/clp-mcp-server/clp_mcp_server/clp_connector.py

@@ -1,15 +1,13 @@
 """CLP MCP Server entry point."""
 
 import ipaddress
-import logging
-import os
 import socket
 import sys
 from pathlib import Path
 
 import click
 from clp_py_utils.clp_config import ClpConfig, MCP_SERVER_COMPONENT_NAME
-from clp_py_utils.clp_logging import get_logger, get_logging_formatter, set_logging_level
+from clp_py_utils.clp_logging import configure_logging, get_logger
 from clp_py_utils.core import read_yaml_config_file
 from pydantic import ValidationError
 
@@ -38,12 +36,8 @@ def main(host: str, port: int, config_path: Path) -> int:
     :param config_path: The path to server's configuration file.
     :return: Exit code (0 for success, non-zero for failure).
     """
-    # Setup logging to file
-    log_file_path = Path(os.getenv("CLP_LOGS_DIR")) / "mcp_server.log"
-    logging_file_handler = logging.FileHandler(filename=log_file_path, encoding="utf-8")
-    logging_file_handler.setFormatter(get_logging_formatter())
-    logger.addHandler(logging_file_handler)
-    set_logging_level(logger, os.getenv("CLP_LOGGING_LEVEL"))
+    # Setup optional file logging and logging level.
+    configure_logging(logger, "mcp_server")
 
     exit_code = 0
 

components/clp-mcp-server/clp_mcp_server/clp_mcp_server.py

@@ -1,6 +1,6 @@
 [project]
 name = "clp-mcp-server"
-version = "0.9.1-dev"
+version = "0.10.1-dev"
 description = "MCP server for CLP"
 authors = [{name = "YScope Inc.", email = "dev@yscope.com"}]
 readme = "README.md"