components/clp-package-utils/clp_package_utils/controller.py around lines
882-903, the function currently declares -> str but returns a list and mutates
module-level lists; change the return annotation to -> list[str], create a local
copy services: list[str] = list(DOCKER_COMPOSE_BASE_SERVICES) (do not assign the
global), conditionally extend with DOCKER_COMPOSE_EXTENDED_SERVICES via services
+= DOCKER_COMPOSE_EXTENDED_SERVICES, remove the "results-cache" entry immutably
using a list comprehension (e.g. services = [s for s in services if s !=
"results-cache"]) instead of .remove(), and return the local services list so
globals remain unmodified.

In components/package-template/src/etc/clp-config.yaml around lines 17 to 18,
add a brief explanatory comment above the bundled key that states this list
controls which third‑party services CLP will deploy versus expect externally
(e.g., “Which third‑party services CLP should deploy; omit to use external
services instead”), so users understand the flag’s effect and when to remove it;
keep the comment short and placed immediately above the example list.

#!/bin/bash
# Create a GitHub issue to track the accidentally removed environment variable
gh issue create \
  --repo y-scope/clp \
  --title "Restore accidentally removed CLP_QUEUE_LOGS_DIR_HOST environment variable" \
  --body "## Description

During the refactoring in PR #1648, the \`CLP_QUEUE_LOGS_DIR_HOST\` environment variable was accidentally removed from the \`_set_up_env_for_queue\` method in \`components/clp-package-utils/clp_package_utils/controller.py\`.

## Context

- Reported by: @junhaoliao
- PR: https://github.com/y-scope/clp/pull/1648
- Comment: https://github.com/y-scope/clp/pull/1648#discussion_r2572694798

The environment variable needs to be restored to ensure proper queue component configuration." \
  --assignee junhaoliao

def resolve_host_path_in_container(host_path: pathlib.Path) -> pathlib.Path:
    """
    Resolves a host path and translates it into its container-mount equivalent absolute path. See
    `resolve_host_path` for details about the resolution.

    :param host_path: The host path.
    :return: The translated path (with /mnt/host prefix).
    """
    host_path = host_path.expanduser()

    if not host_path.is_absolute():
        pwd_host = os.getenv("CLP_PWD_HOST")
        if pwd_host is None:
            raise ValueError("Relative host path provided but CLP_PWD_HOST is not set.")
        host_path = pathlib.Path(pwd_host) / host_path
        host_path = host_path.absolute()

    resolved_path = _resolve_symlinks_in_path(host_path)
    if resolved_path is not None:
        return resolved_path

    return CONTAINER_DIR_FOR_HOST_ROOT / host_path.relative_to("/")

def get_archives_table_name(table_prefix: str, dataset: str | None) -> str:
    return _get_table_name(table_prefix, ARCHIVES_TABLE_SUFFIX, dataset)

def get_datasets_table_name(table_prefix: str) -> str:
    return _get_table_name(table_prefix, DATASETS_TABLE_SUFFIX, None)

def get_files_table_name(table_prefix: str, dataset: str | None) -> str:
    return _get_table_name(table_prefix, FILES_TABLE_SUFFIX, dataset)

def check_docker_dependencies(should_compose_project_be_running: bool, project_name: str):
    """
    Checks if Docker and Docker Compose are installed, and whether a Docker Compose project is
    running.

    :param should_compose_project_be_running:
    :param project_name: The Docker Compose project name to check.
    :raise DockerNotAvailableError: If any Docker dependency is not installed.
    :raise DockerComposeProjectNotRunningError: If the project isn't running when it should be.
    :raise DockerComposeProjectAlreadyRunningError: If the project is running when it shouldn't be.
    """
    try:
        subprocess.check_output(
            ["docker", "--version"],
            stderr=subprocess.STDOUT,
        )
    except subprocess.CalledProcessError as e:
        raise DockerNotAvailableError("docker is not installed or available on the path", e) from e

    is_running = _is_docker_compose_project_running(project_name)
    if should_compose_project_be_running and not is_running:
        raise DockerComposeProjectNotRunningError(project_name)
    if not should_compose_project_be_running and is_running:
        raise DockerComposeProjectAlreadyRunningError(project_name)

class DockerComposeProjectNotRunningError(DockerDependencyError):
    """Raised when a Docker Compose project is not running but should be."""

    def __init__(self, project_name: str) -> None:
        super().__init__(errno.ESRCH, f"Docker Compose project '{project_name}' is not running.")

class DockerDependencyError(OSError):
    """Base class for errors related to Docker dependencies."""

class BundledService(LowercaseStrEnum):
    DATABASE = auto()
    QUEUE = auto()
    REDIS = auto()
    RESULTS_CACHE = auto()

class ClpConfig(BaseModel):
    container_image_ref: NonEmptyStr | None = None

    logs_input: FsIngestionConfig | S3IngestionConfig = FsIngestionConfig()
    bundled: list[BundledServiceStr] = [
        BundledService.DATABASE,
        BundledService.QUEUE,
        BundledService.REDIS,
        BundledService.RESULTS_CACHE,
    ]

    package: Package = Package()
    database: Database = Database()
    queue: Queue = Queue()
    redis: Redis = Redis()
    reducer: Reducer = Reducer()
    results_cache: ResultsCache = ResultsCache()
    compression_scheduler: CompressionScheduler = CompressionScheduler()
    query_scheduler: QueryScheduler = QueryScheduler()
    compression_worker: CompressionWorker = CompressionWorker()
    query_worker: QueryWorker = QueryWorker()
    webui: WebUi = WebUi()
    garbage_collector: GarbageCollector = GarbageCollector()
    api_server: ApiServer = ApiServer()
    credentials_file_path: SerializablePath = CLP_DEFAULT_CREDENTIALS_FILE_PATH

    mcp_server: McpServer | None = None
    presto: Presto | None = None

    archive_output: ArchiveOutput = ArchiveOutput()
    stream_output: StreamOutput = StreamOutput()
    data_directory: SerializablePath = CLP_DEFAULT_DATA_DIRECTORY_PATH
    logs_directory: SerializablePath = CLP_DEFAULT_LOG_DIRECTORY_PATH
    tmp_directory: SerializablePath = CLP_DEFAULT_TMP_DIRECTORY_PATH
    aws_config_directory: SerializablePath | None = None

    _container_image_id_path: SerializablePath = PrivateAttr(
        default=CLP_PACKAGE_CONTAINER_IMAGE_ID_PATH
    )
    _version_file_path: SerializablePath = PrivateAttr(default=CLP_VERSION_FILE_PATH)

    @field_validator("aws_config_directory")
    @classmethod
    def expand_profile_user_home(cls, value: SerializablePath | None) -> SerializablePath | None:
        if value is not None:
            value = value.expanduser()
        return value

    def make_config_paths_absolute(self, clp_home: pathlib.Path):
        if StorageType.FS == self.logs_input.type:
            self.logs_input.make_config_paths_absolute(clp_home)
        self.credentials_file_path = make_config_path_absolute(clp_home, self.credentials_file_path)
        self.archive_output.storage.make_config_paths_absolute(clp_home)
        self.stream_output.storage.make_config_paths_absolute(clp_home)
        self.data_directory = make_config_path_absolute(clp_home, self.data_directory)
        self.logs_directory = make_config_path_absolute(clp_home, self.logs_directory)
        self.tmp_directory = make_config_path_absolute(clp_home, self.tmp_directory)
        self._container_image_id_path = make_config_path_absolute(
            clp_home, self._container_image_id_path
        )
        self._version_file_path = make_config_path_absolute(clp_home, self._version_file_path)

    def validate_logs_input_config(self, use_host_mount: bool = False):
        logs_input_type = self.logs_input.type
        if StorageType.FS == logs_input_type:
            # NOTE: This can't be a pydantic validator since input_logs_dir might be a
            # package-relative path that will only be resolved after pydantic validation
            input_logs_dir = self.logs_input.directory
            resolved_input_logs_dir = (
                resolve_host_path_in_container(input_logs_dir) if use_host_mount else input_logs_dir
            )
            if not resolved_input_logs_dir.exists():
                raise ValueError(f"logs_input.directory '{input_logs_dir}' doesn't exist.")
            if not resolved_input_logs_dir.is_dir():
                raise ValueError(f"logs_input.directory '{input_logs_dir}' is not a directory.")
        if StorageType.S3 == logs_input_type and StorageEngine.CLP_S != self.package.storage_engine:
            raise ValueError(
                f"logs_input.type = 's3' is only supported with package.storage_engine"
                f" = '{StorageEngine.CLP_S}'"
            )

    def validate_archive_output_config(self, use_host_mount: bool = False):
        if (
            StorageType.S3 == self.archive_output.storage.type
            and StorageEngine.CLP_S != self.package.storage_engine
        ):
            raise ValueError(
                f"archive_output.storage.type = 's3' is only supported with package.storage_engine"
                f" = '{StorageEngine.CLP_S}'"
            )
        archive_output_dir = self.archive_output.get_directory()
        resolved_archive_output_dir = (
            resolve_host_path_in_container(archive_output_dir)
            if use_host_mount
            else archive_output_dir
        )
        try:
            validate_path_could_be_dir(resolved_archive_output_dir)
        except ValueError as ex:
            raise ValueError(f"archive_output.storage's directory is invalid: {ex}")

    def validate_stream_output_config(self, use_host_mount: bool = False):
        if (
            StorageType.S3 == self.stream_output.storage.type
            and StorageEngine.CLP_S != self.package.storage_engine
        ):
            raise ValueError(
                f"stream_output.storage.type = 's3' is only supported with package.storage_engine"
                f" = '{StorageEngine.CLP_S}'"
            )
        stream_output_dir = self.stream_output.get_directory()
        resolved_stream_output_dir = (
            resolve_host_path_in_container(stream_output_dir)
            if use_host_mount
            else stream_output_dir
        )
        try:
            validate_path_could_be_dir(resolved_stream_output_dir)
        except ValueError as ex:
            raise ValueError(f"stream_output.storage's directory is invalid: {ex}")

    def validate_data_dir(self, use_host_mount: bool = False):
        data_dir = self.data_directory
        resolved_data_dir = resolve_host_path_in_container(data_dir) if use_host_mount else data_dir
        try:
            validate_path_could_be_dir(resolved_data_dir)
        except ValueError as ex:
            raise ValueError(f"data_directory is invalid: {ex}")

    def validate_logs_dir(self, use_host_mount: bool = False):
        logs_dir = self.logs_directory
        resolved_logs_dir = resolve_host_path_in_container(logs_dir) if use_host_mount else logs_dir
        try:
            validate_path_could_be_dir(resolved_logs_dir)
        except ValueError as ex:
            raise ValueError(f"logs_directory is invalid: {ex}")

    def validate_tmp_dir(self, use_host_mount: bool = False):
        tmp_dir = self.tmp_directory
        resolved_tmp_dir = resolve_host_path_in_container(tmp_dir) if use_host_mount else tmp_dir
        try:
            validate_path_could_be_dir(resolved_tmp_dir)
        except ValueError as ex:
            raise ValueError(f"tmp_directory is invalid: {ex}")

    def validate_aws_config_dir(self, use_host_mount: bool = False):
        profile_auth_used = False
        auth_configs = []

        if StorageType.S3 == self.logs_input.type:
            auth_configs.append(self.logs_input.aws_authentication)
        if StorageType.S3 == self.archive_output.storage.type:
            auth_configs.append(self.archive_output.storage.s3_config.aws_authentication)
        if StorageType.S3 == self.stream_output.storage.type:
            auth_configs.append(self.stream_output.storage.s3_config.aws_authentication)

        for auth in auth_configs:
            if AwsAuthType.profile == auth.type:
                profile_auth_used = True
                break

        if profile_auth_used:
            if self.aws_config_directory is None:
                raise ValueError(
                    "aws_config_directory must be set when using profile authentication"
                )
            resolved_aws_config_dir = (
                resolve_host_path_in_container(self.aws_config_directory)
                if use_host_mount
                else self.aws_config_directory
            )
            if not resolved_aws_config_dir.exists():
                raise ValueError(
                    f"aws_config_directory does not exist: '{self.aws_config_directory}'"
                )
        if not profile_auth_used and self.aws_config_directory is not None:
            raise ValueError(
                "aws_config_directory should not be set when profile authentication is not used"
            )

    def load_container_image_ref(self):
        if self.container_image_ref is not None:
            # Accept configured value for debug purposes
            return

        if self._container_image_id_path.exists():
            with open(self._container_image_id_path) as image_id_file:
                self.container_image_ref = image_id_file.read().strip()
        else:
            with open(self._version_file_path) as version_file:
                clp_package_version = version_file.read().strip()
            self.container_image_ref = f"ghcr.io/y-scope/clp/clp-package:{clp_package_version}"

    def get_shared_config_file_path(self) -> pathlib.Path:
        return self.logs_directory / CLP_SHARED_CONFIG_FILENAME

    def get_deployment_type(self) -> DeploymentType:
        if QueryEngine.PRESTO == self.package.query_engine:
            return DeploymentType.BASE
        return DeploymentType.FULL

    def dump_to_primitive_dict(self):
        custom_serialized_fields = {
            "database",
            "queue",
            "redis",
        }
        d = self.model_dump(exclude=custom_serialized_fields)
        for key in custom_serialized_fields:
            d[key] = getattr(self, key).dump_to_primitive_dict()

        return d

    @model_validator(mode="after")
    def validate_presto_config(self):
        query_engine = self.package.query_engine
        presto = self.presto
        if query_engine == QueryEngine.PRESTO and presto is None:
            raise ValueError(
                f"`presto` config must be non-null when query_engine is `{query_engine}`"
            )
        return self

    def transform_for_container(self):
        """
        Converts all relevant directories to absolute paths inside the container, and updates
        component hostnames and ports to their container service names and default ports.
        """
        self.data_directory = pathlib.Path("/") / CLP_DEFAULT_DATA_DIRECTORY_PATH
        self.logs_directory = pathlib.Path("/") / CLP_DEFAULT_LOG_DIRECTORY_PATH
        self.tmp_directory = pathlib.Path("/") / CLP_DEFAULT_TMP_DIRECTORY_PATH
        if self.aws_config_directory is not None:
            self.aws_config_directory = CONTAINER_AWS_CONFIG_DIRECTORY
        self.logs_input.transform_for_container()
        self.archive_output.storage.transform_for_container()
        self.stream_output.storage.transform_for_container()

        self.database.transform_for_container()
        self.queue.transform_for_container()
        self.redis.transform_for_container()
        self.results_cache.transform_for_container()
        self.query_scheduler.transform_for_container()
        self.reducer.transform_for_container()
        if self.package.query_engine == QueryEngine.PRESTO and self.presto is not None:
            self.presto.transform_for_container()

def dump_shared_container_config(container_clp_config: ClpConfig, clp_config: ClpConfig):
    """
    Dumps the given container config to `CLP_SHARED_CONFIG_FILENAME` in the logs directory, so that
    it's accessible in the container.

    :param container_clp_config:
    :param clp_config:
    """
    dump_container_config(container_clp_config, clp_config, CLP_SHARED_CONFIG_FILENAME)

def generate_docker_compose_container_config(clp_config: ClpConfig) -> ClpConfig:
    """
    Copies the given config and transforms mount paths and hosts for Docker Compose.

    :param clp_config:
    :return: The container config.
    """
    container_clp_config = clp_config.model_copy(deep=True)
    container_clp_config.transform_for_container()

    return container_clp_config

def get_clp_home():
    # Determine CLP_HOME from an environment variable or this script's path
    clp_home = None
    if "CLP_HOME" in os.environ:
        clp_home = pathlib.Path(os.environ["CLP_HOME"])
    else:
        for path in pathlib.Path(__file__).resolve().parents:
            if "lib" == path.name:
                clp_home = path.parent
                break

    if clp_home is None:
        raise ValueError("CLP_HOME is not set and could not be determined automatically.")
    if not clp_home.exists():
        raise ValueError("CLP_HOME set to nonexistent path.")

    return clp_home.resolve()

def is_retention_period_configured(clp_config: ClpConfig) -> bool:
    if clp_config.archive_output.retention_period is not None:
        return True

    if clp_config.results_cache.retention_period is not None:
        return True

    return False

def validate_db_config(
    clp_config: ClpConfig,
    component_config: pathlib.Path,
    data_dir: pathlib.Path,
    logs_dir: pathlib.Path,
):
    resolved_component_config = resolve_host_path_in_container(component_config)
    if not resolved_component_config.exists():
        raise ValueError(f"{DB_COMPONENT_NAME} configuration file missing: '{component_config}'.")
    _validate_data_directory(data_dir, DB_COMPONENT_NAME)
    _validate_log_directory(logs_dir, DB_COMPONENT_NAME)

    validate_port(f"{DB_COMPONENT_NAME}.port", clp_config.database.host, clp_config.database.port)

def validate_webui_config(
    clp_config: ClpConfig,
    client_settings_json_path: pathlib.Path,
    server_settings_json_path: pathlib.Path,
):
    for path in [client_settings_json_path, server_settings_json_path]:
        resolved_path = resolve_host_path_in_container(path)
        if not resolved_path.exists():
            raise ValueError(f"{WEBUI_COMPONENT_NAME} {path} is not a valid path to settings.json")

    validate_port(f"{WEBUI_COMPONENT_NAME}.port", clp_config.webui.host, clp_config.webui.port)

def get_deployment_type(self) -> DeploymentType:
        if QueryEngine.PRESTO == self.package.query_engine:
            return DeploymentType.BASE
        return DeploymentType.FULL

Learnt from: junhaoliao
Repo: y-scope/clp PR: 1178
File: components/clp-package-utils/clp_package_utils/controller.py:315-315
Timestamp: 2025-10-17T19:59:25.596Z
Learning: In components/clp-package-utils/clp_package_utils/controller.py, worker log directories (compression_worker, query_worker, reducer) created via `mkdir()` do not need `_chown_paths_if_root()` calls because directories are created with the same owner as the script caller. This differs from infrastructure service directories (database, queue, Redis, results cache) which do require explicit ownership changes.

Learnt from: junhaoliao
Repo: y-scope/clp PR: 1136
File: components/webui/client/src/pages/IngestPage/Details/sql.ts:1-1
Timestamp: 2025-07-29T21:00:07.757Z
Learning: User junhaoliao requested creating a GitHub issue to track server-side SQL query error handling improvements, specifically to prevent uncaught query errors from causing 500 errors to reach the client.

Learnt from: junhaoliao
Repo: y-scope/clp PR: 1501
File: tools/deployment/presto-clp/scripts/init.py:10-13
Timestamp: 2025-10-27T07:07:37.901Z
Learning: In `tools/deployment/presto-clp/scripts/init.py`, the `DATABASE_COMPONENT_NAME` and `DATABASE_DEFAULT_PORT` constants are intentionally duplicated from `clp_py_utils.clp_config` because `clp_py_utils` is not installed in the Presto init script's runtime environment. The two flows are separate and this duplication is documented. There are plans to merge these flows after a future release.