feat: Allow tweaking catalog name in Trino

CHANGELOG.md

@@ -10,6 +10,7 @@ All notable changes to this project will be documented in this file.
 - Added support for the [PostgreSQL connector](https://trino.io/docs/current/connector/postgresql.html) using the new generic database connection mechanism.
   Previously, users had to use the `generic` connector ([#883]).
 - Added support for Trino 481 ([#900]).
+- Add a new `.spec.name.inferred.replaceHyphensWithUnderscores` field on TrinoCatalog, which allows tweaking the catalog name in Trino ([#903]).
 
 ### Changed
 
@@ -43,6 +44,7 @@ All notable changes to this project will be documented in this file.
 [#895]: https://github.com/stackabletech/trino-operator/pull/895
 [#897]: https://github.com/stackabletech/trino-operator/pull/897
 [#900]: https://github.com/stackabletech/trino-operator/pull/900
+[#903]: https://github.com/stackabletech/trino-operator/pull/903
 
 ## [26.3.0] - 2026-03-16
 

docs/modules/trino/pages/usage-guide/catalogs/index.adoc

@@ -105,6 +105,29 @@ In this case the `hive` and `iceberg` catalogs will be used as they both match t
 
 A `TrinoCluster` can, once created, detect and use new catalogs that have been subsequently created with a matching label. This also means that it is possible to reuse a `TrinoCatalog` within multiple `TrinoClusters`.
 
+=== Catalog name tweaking
+
+By default the name of the catalog in Trino is inferred from the `.metadata.name` of the TrinoCatalog object.
+This ensures that no catalog names clash, as there can only be one TrinoCatalog with a given name.
+
+One inconvenience is that you need to quote catalogs (or schemas and tables for that matter) containing `-` in Trino, while `_` is fine.
+As Kubernetes doesn't allow `_` in the object names, we offer a feature to replace `-` with `_`, which allows you to use valid Kubernetes names, but keeps the convenience of using `_` in catalog names.
+
+In order to replace `-` with `_`, use the setting `name.inferred.replaceHyphensWithUnderscores`.
+The Trino catalog will be called `my_postgres`:
+
+[source,yaml]
+----
+kind: TrinoCatalog
+metadata:
+  name: my-postgres
+spec:
+  name:
+    inferred:
+      replaceHyphensWithUnderscores: true
+# ...
+----
+
 === Generic fallback connector
 
 Trino supports lots of different connectors and we can not cover all the available connectors.

extra/crds.yaml

@@ -4227,6 +4227,34 @@ spec:
                 items:
                   type: string
                 type: array
+              name:
+                default:
+                  inferred:
+                    replaceHyphensWithUnderscores: false
+                description: The name of the catalog
+                oneOf:
+                - required:
+                  - inferred
+                properties:
+                  inferred:
+                    description: |-
+                      Infer the catalog name from the `.metadata.name` of the TrinoCatalog resource.
+
+                      This ensures that no catalog names clash, as there can only be one TrinoCatalog with a
+                      given name.
+                    properties:
+                      replaceHyphensWithUnderscores:
+                        default: false
+                        description: |-
+                          Whether hyphens (`-`) in the name of the catalog should be replaced by underscores (`_`).
+
+                          This is recommended because Kubernetes only allows `a-z` and `-`, while Trino
+                          requires quoting for catalogs containing `-` characters. This mechanism allows
+                          you to use valid Kubernetes names, but keeps the convenience of using `_` in
+                          catalog names.
+                        type: boolean
+                    type: object
+                type: object
             required:
             - connector
             type: object

rust/operator-binary/Cargo.toml

@@ -9,6 +9,12 @@ repository.workspace = true
 publish = false
 build = "build.rs"
 
+[features]
+# The macro attributed_string_type is used in this operator. It produces test
+# code if the feature "test-support" is set. This feature is defined here to
+# suppress a Clippy warning.
+test-support = []
+
 [dependencies]
 stackable-operator = { workspace = true, features = ["test-support"] }
 

rust/operator-binary/src/catalog/black_hole.rs

@@ -2,20 +2,22 @@ use async_trait::async_trait;
 use stackable_operator::{client::Client, v2::types::kubernetes::NamespaceName};
 
 use super::{FromTrinoCatalogError, ToCatalogConfig, config::CatalogConfig};
-use crate::crd::catalog::black_hole::BlackHoleConnector;
+use crate::{
+    controller::dereference::TrinoCatalogName, crd::catalog::black_hole::BlackHoleConnector,
+};
 
 pub const CONNECTOR_NAME: &str = "blackhole";
 
 #[async_trait]
 impl ToCatalogConfig for BlackHoleConnector {
     async fn to_catalog_config(
         &self,
-        catalog_name: &str,
+        catalog_name: &TrinoCatalogName,
         _catalog_namespace: &NamespaceName,
         _client: &Client,
         _trino_version: u16,
     ) -> Result<CatalogConfig, FromTrinoCatalogError> {
         // No additional properties needed
-        Ok(CatalogConfig::new(catalog_name.to_string(), CONNECTOR_NAME))
+        Ok(CatalogConfig::new(catalog_name, CONNECTOR_NAME))
     }
 }

rust/operator-binary/src/catalog/commons.rs

@@ -19,6 +19,7 @@ use super::{
 };
 use crate::{
     config,
+    controller::dereference::TrinoCatalogName,
     crd::{
         CONFIG_DIR_NAME,
         catalog::commons::{HdfsConnection, MetastoreConnection},
@@ -30,7 +31,7 @@ impl ExtendCatalogConfig for MetastoreConnection {
     async fn extend_catalog_config(
         &self,
         catalog_config: &mut CatalogConfig,
-        catalog_name: &str,
+        catalog_name: &TrinoCatalogName,
         catalog_namespace: &NamespaceName,
         client: &Client,
         _trino_version: u16,
@@ -72,7 +73,7 @@ impl ExtendCatalogConfig for s3::v1alpha1::InlineConnectionOrReference {
     async fn extend_catalog_config(
         &self,
         catalog_config: &mut CatalogConfig,
-        _catalog_name: &str,
+        _catalog_name: &TrinoCatalogName,
         catalog_namespace: &NamespaceName,
         client: &Client,
         _trino_version: u16,
@@ -117,7 +118,7 @@ impl ExtendCatalogConfig for HdfsConnection {
     async fn extend_catalog_config(
         &self,
         catalog_config: &mut CatalogConfig,
-        catalog_name: &str,
+        catalog_name: &TrinoCatalogName,
         _catalog_namespace: &NamespaceName,
         _client: &Client,
         _trino_version: u16,

rust/operator-binary/src/catalog/config.rs

@@ -5,17 +5,19 @@ use stackable_operator::{
     k8s_openapi::api::core::v1::{
         ConfigMapKeySelector, EnvVar, EnvVarSource, SecretKeySelector, Volume, VolumeMount,
     },
-    kube::Resource,
     v2::types::kubernetes::NamespaceName,
 };
 
 use super::{FromTrinoCatalogError, ToCatalogConfig};
-use crate::crd::catalog::{TrinoCatalogConnector, v1alpha1};
+use crate::{
+    controller::dereference::TrinoCatalogName,
+    crd::catalog::{TrinoCatalogConnector, v1alpha1},
+};
 
 #[derive(Clone, Debug)]
 pub struct CatalogConfig {
     /// Name of the catalog
-    pub name: String,
+    pub name: TrinoCatalogName,
 
     /// Properties of the catalog
     pub properties: BTreeMap<String, String>,
@@ -39,9 +41,9 @@ pub struct CatalogConfig {
 }
 
 impl CatalogConfig {
-    pub fn new(name: impl Into<String>, connector_name: impl Into<String>) -> Self {
+    pub fn new(name: &TrinoCatalogName, connector_name: impl Into<String>) -> Self {
         let mut config = CatalogConfig {
-            name: name.into(),
+            name: name.clone(),
             properties: BTreeMap::new(),
             env_bindings: Vec::new(),
             load_env_from_files: BTreeMap::new(),
@@ -105,17 +107,12 @@ impl CatalogConfig {
     }
 
     pub async fn from_catalog(
+        catalog_name: &TrinoCatalogName,
         catalog: &v1alpha1::TrinoCatalog,
         client: &Client,
         catalog_namespace: &NamespaceName,
         trino_version: u16,
     ) -> Result<CatalogConfig, FromTrinoCatalogError> {
-        let catalog_name = catalog
-            .meta()
-            .name
-            .clone()
-            .ok_or(FromTrinoCatalogError::InvalidCatalogSpec)?;
-
         let to_catalog_config: &dyn ToCatalogConfig = match &catalog.spec.connector {
             TrinoCatalogConnector::BlackHole(black_hole_connector) => black_hole_connector,
             TrinoCatalogConnector::DeltaLake(delta_lake_connector) => delta_lake_connector,
@@ -128,7 +125,7 @@ impl CatalogConfig {
             TrinoCatalogConnector::Tpch(tpch_connector) => tpch_connector,
         };
         let mut catalog_config = to_catalog_config
-            .to_catalog_config(&catalog_name, catalog_namespace, client, trino_version)
+            .to_catalog_config(catalog_name, catalog_namespace, client, trino_version)
             .await?;
 
         catalog_config
@@ -138,7 +135,7 @@ impl CatalogConfig {
         for removal in &catalog.spec.config_removals {
             if catalog_config.properties.remove(removal).is_none() {
                 tracing::warn!(
-                    catalog.name = catalog_name,
+                    catalog.name = %catalog_name,
                     property = removal,
                     "You asked to remove a non-existing config property from a catalog"
                 );
@@ -149,8 +146,8 @@ impl CatalogConfig {
     }
 }
 
-fn calculate_env_name(catalog: impl Into<String>, property: impl Into<String>) -> String {
-    let catalog = catalog.into().replace(['.', '-'], "_");
+fn calculate_env_name(catalog_name: &TrinoCatalogName, property: impl Into<String>) -> String {
+    let catalog = catalog_name.to_string().replace(['.', '-'], "_");
     let property = property.into().replace(['.', '-'], "_");
     format!("CATALOG_{catalog}_{property}").to_uppercase()
 }

rust/operator-binary/src/catalog/delta_lake.rs

@@ -2,20 +2,22 @@ use async_trait::async_trait;
 use stackable_operator::{client::Client, v2::types::kubernetes::NamespaceName};
 
 use super::{ExtendCatalogConfig, FromTrinoCatalogError, ToCatalogConfig, config::CatalogConfig};
-use crate::crd::catalog::delta_lake::DeltaLakeConnector;
+use crate::{
+    controller::dereference::TrinoCatalogName, crd::catalog::delta_lake::DeltaLakeConnector,
+};
 
 pub const CONNECTOR_NAME: &str = "delta_lake";
 
 #[async_trait]
 impl ToCatalogConfig for DeltaLakeConnector {
     async fn to_catalog_config(
         &self,
-        catalog_name: &str,
+        catalog_name: &TrinoCatalogName,
         catalog_namespace: &NamespaceName,
         client: &Client,
         trino_version: u16,
     ) -> Result<CatalogConfig, FromTrinoCatalogError> {
-        let mut config = CatalogConfig::new(catalog_name.to_string(), CONNECTOR_NAME);
+        let mut config = CatalogConfig::new(catalog_name, CONNECTOR_NAME);
 
         // No authorization checks are enforced at the catalog level.
         // We don't want the delta connector to prevent users from dropping tables.

rust/operator-binary/src/catalog/generic.rs

@@ -2,19 +2,22 @@ use async_trait::async_trait;
 use stackable_operator::{client::Client, v2::types::kubernetes::NamespaceName};
 
 use super::{FromTrinoCatalogError, ToCatalogConfig, config::CatalogConfig};
-use crate::crd::catalog::generic::{GenericConnector, Property};
+use crate::{
+    controller::dereference::TrinoCatalogName,
+    crd::catalog::generic::{GenericConnector, Property},
+};
 
 #[async_trait]
 impl ToCatalogConfig for GenericConnector {
     async fn to_catalog_config(
         &self,
-        catalog_name: &str,
+        catalog_name: &TrinoCatalogName,
         _catalog_namespace: &NamespaceName,
         _client: &Client,
         _trino_version: u16,
     ) -> Result<CatalogConfig, FromTrinoCatalogError> {
         let connector_name = &self.connector_name;
-        let mut config = CatalogConfig::new(catalog_name.to_string(), connector_name);
+        let mut config = CatalogConfig::new(catalog_name, connector_name);
 
         for (property_name, property) in &self.properties {
             match property {

rust/operator-binary/src/catalog/google_sheet.rs

@@ -6,20 +6,23 @@ use stackable_operator::{
 };
 
 use super::{FromTrinoCatalogError, ToCatalogConfig, config::CatalogConfig};
-use crate::crd::{CONFIG_DIR_NAME, catalog::google_sheet::GoogleSheetConnector};
+use crate::{
+    controller::dereference::TrinoCatalogName,
+    crd::{CONFIG_DIR_NAME, catalog::google_sheet::GoogleSheetConnector},
+};
 
 pub const CONNECTOR_NAME: &str = "gsheets";
 
 #[async_trait]
 impl ToCatalogConfig for GoogleSheetConnector {
     async fn to_catalog_config(
         &self,
-        catalog_name: &str,
+        catalog_name: &TrinoCatalogName,
         _catalog_namespace: &NamespaceName,
         _client: &Client,
         _trino_version: u16,
     ) -> Result<CatalogConfig, FromTrinoCatalogError> {
-        let mut config = CatalogConfig::new(catalog_name.to_string(), CONNECTOR_NAME);
+        let mut config = CatalogConfig::new(catalog_name, CONNECTOR_NAME);
 
         let volume_name = format!("{catalog_name}-google-sheets-credentials");
         let google_sheets_credentials_dir =

rust/operator-binary/src/catalog/hive.rs

@@ -2,20 +2,20 @@ use async_trait::async_trait;
 use stackable_operator::{client::Client, v2::types::kubernetes::NamespaceName};
 
 use super::{ExtendCatalogConfig, FromTrinoCatalogError, ToCatalogConfig, config::CatalogConfig};
-use crate::crd::catalog::hive::HiveConnector;
+use crate::{controller::dereference::TrinoCatalogName, crd::catalog::hive::HiveConnector};
 
 pub const CONNECTOR_NAME: &str = "hive";
 
 #[async_trait]
 impl ToCatalogConfig for HiveConnector {
     async fn to_catalog_config(
         &self,
-        catalog_name: &str,
+        catalog_name: &TrinoCatalogName,
         catalog_namespace: &NamespaceName,
         client: &Client,
         trino_version: u16,
     ) -> Result<CatalogConfig, FromTrinoCatalogError> {
-        let mut config = CatalogConfig::new(catalog_name.to_string(), CONNECTOR_NAME);
+        let mut config = CatalogConfig::new(catalog_name, CONNECTOR_NAME);
 
         // No authorization checks are enforced at the catalog level.
         // We don't want the hive connector to prevent users from dropping tables.

rust/operator-binary/src/catalog/iceberg.rs

@@ -2,20 +2,20 @@ use async_trait::async_trait;
 use stackable_operator::{client::Client, v2::types::kubernetes::NamespaceName};
 
 use super::{ExtendCatalogConfig, FromTrinoCatalogError, ToCatalogConfig, config::CatalogConfig};
-use crate::crd::catalog::iceberg::IcebergConnector;
+use crate::{controller::dereference::TrinoCatalogName, crd::catalog::iceberg::IcebergConnector};
 
 pub const CONNECTOR_NAME: &str = "iceberg";
 
 #[async_trait]
 impl ToCatalogConfig for IcebergConnector {
     async fn to_catalog_config(
         &self,
-        catalog_name: &str,
+        catalog_name: &TrinoCatalogName,
         catalog_namespace: &NamespaceName,
         client: &Client,
         trino_version: u16,
     ) -> Result<CatalogConfig, FromTrinoCatalogError> {
-        let mut config = CatalogConfig::new(catalog_name.to_string(), CONNECTOR_NAME);
+        let mut config = CatalogConfig::new(catalog_name, CONNECTOR_NAME);
 
         // No authorization checks are enforced at the catalog level.
         // We don't want the iceberg connector to prevent users from dropping tables.

rust/operator-binary/src/catalog/mod.rs

@@ -14,6 +14,8 @@ use async_trait::async_trait;
 use snafu::Snafu;
 use stackable_operator::{client::Client, v2::types::kubernetes::NamespaceName};
 
+use crate::controller::dereference::TrinoCatalogName;
+
 use self::config::CatalogConfig;
 
 #[derive(Debug, Snafu)]
@@ -62,7 +64,7 @@ pub enum FromTrinoCatalogError {
 pub trait ToCatalogConfig {
     async fn to_catalog_config(
         &self,
-        catalog_name: &str,
+        catalog_name: &TrinoCatalogName,
         catalog_namespace: &NamespaceName,
         client: &Client,
         trino_version: u16,
@@ -74,7 +76,7 @@ pub trait ExtendCatalogConfig {
     async fn extend_catalog_config(
         &self,
         catalog_config: &mut CatalogConfig,
-        catalog_name: &str,
+        catalog_name: &TrinoCatalogName,
         catalog_namespace: &NamespaceName,
         client: &Client,
         trino_version: u16,