feat: Implement Trino catalog and schema

Author: sbernauer

Cargo.lock

@@ -26,6 +26,7 @@ futures = { version = "0.3" }
 hyper = "1.4"
 indoc = "2.0"
 ldap3 = { version = "0.12", features = ["gssapi", "tls"] }
+md5 = "0.7"
 moka = { version = "0.12", features = ["future"] }
 native-tls = "0.2.12"
 pin-project = "1.1"

Cargo.toml

@@ -518,7 +518,7 @@ spec:
                         properties:
                           entryTimeToLive:
                             default: 1m
-                            description: How long metadata about each user should be cached for.
+                            description: How long responses should be cached for.
                             type: string
                         type: object
                     type: object
@@ -2449,7 +2449,7 @@ spec:
                         properties:
                           entryTimeToLive:
                             default: 1m
-                            description: How long metadata about each user should be cached for.
+                            description: How long responses should be cached for.
                             type: string
                         type: object
                     type: object

extra/crds.yaml

@@ -45,6 +45,8 @@ pub mod versioned {
     }
 }
 
+// We actually use it as serde(default) value, but Rust fails to see that
+#[allow(dead_code)]
 fn default_data_hub_env() -> String {
     "PROD".to_owned()
 }

rust/operator-binary/src/crd/resource_info_fetcher/mod.rs

@@ -17,6 +17,7 @@ axum.workspace = true
 clap.workspace = true
 futures.workspace = true
 hyper.workspace = true
+md5.workspace = true
 moka.workspace = true
 reqwest.workspace = true
 serde.workspace = true

rust/resource-info-fetcher/Cargo.toml

@@ -26,16 +26,21 @@ pub struct ResourceInfoRequest {
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Deserialize)]
 #[serde(rename_all = "camelCase")]
+// Soon there will be something else than Trino
+#[expect(clippy::enum_variant_names)]
 pub enum ResourceInfoRequestResource {
     TrinoTable {
         catalog: String,
         schema: String,
         table: String,
     },
-    // TrinoSchema {
-    //     catalog: String,
-    //     schema: String,
-    // },
+    TrinoSchema {
+        catalog: String,
+        schema: String,
+    },
+    TrinoCatalog {
+        catalog: String,
+    },
 }
 
 #[derive(Snafu, Debug)]

rust/resource-info-fetcher/src/api.rs

@@ -1,5 +1,6 @@
-use crate::api::{ResourceInfoRequest, ResourceInfoRequestResource};
+use std::collections::BTreeMap;
 
+use crate::api::{ResourceInfoRequest, ResourceInfoRequestResource};
 
 pub fn urn_for_request(request: &ResourceInfoRequest, env: &str) -> String {
     let stacklet = &request.stacklet;
@@ -13,5 +14,74 @@ pub fn urn_for_request(request: &ResourceInfoRequest, env: &str) -> String {
                 "urn:li:dataset:(urn:li:dataPlatform:{stacklet},{catalog}.{schema}.{table},{env})"
             )
         }
+        // Trino catalogs and schemas are modelled as DataHub `Container`s (subtypes `Database` and
+        // `Schema` respectively). Unlike datasets, their URN is not human-readable but a GUID derived
+        // from the container key - see `container_urn`.
+        ResourceInfoRequestResource::TrinoCatalog { catalog } => {
+            container_urn(&BTreeMap::from([
+                ("platform", stacklet.as_str()),
+                ("instance", env),
+                ("database", catalog.as_str()),
+            ]))
+        }
+        ResourceInfoRequestResource::TrinoSchema { catalog, schema } => {
+            container_urn(&BTreeMap::from([
+                ("platform", stacklet.as_str()),
+                ("instance", env),
+                ("database", catalog.as_str()),
+                ("schema", schema.as_str()),
+            ]))
+        }
+    }
+}
+
+/// Reproduces DataHub's `datahub_guid`: the container key is serialized to compact, key-sorted JSON
+/// and MD5-hashed. A [`BTreeMap`] yields sorted keys and `serde_json` emits no whitespace, which
+/// matches Python's `json.dumps(key, sort_keys=True, separators=(",", ":"))`.
+///
+/// Note that the SQL ingestion source sets `backcompat_env_as_instance`, so the configured `env`
+/// (e.g. `PROD`) ends up in the `instance` field and no `env` field is present in the key. The
+/// `platform` is the bare platform name (e.g. `trino`), *not* the `urn:li:dataPlatform:` form.
+fn container_urn(container_key: &BTreeMap<&str, &str>) -> String {
+    let key_json =
+        serde_json::to_string(container_key).expect("serializing a BTreeMap<&str, &str> cannot fail");
+    format!("urn:li:container:{:x}", md5::compute(key_json.as_bytes()))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Verified against a live DataHub: the schema `customer_analytics` in catalog `lakehouse`.
+    #[test]
+    fn trino_schema_urn_matches_data_hub() {
+        let request = ResourceInfoRequest {
+            stacklet: "trino".to_owned(),
+            resource: ResourceInfoRequestResource::TrinoSchema {
+                catalog: "lakehouse".to_owned(),
+                schema: "customer_analytics".to_owned(),
+            },
+        };
+
+        assert_eq!(
+            urn_for_request(&request, "PROD"),
+            "urn:li:container:c8531e5a52cacf56768d0bf77ca8787c"
+        );
+    }
+
+    /// Verified against a live DataHub: the catalog `lakehouse`.
+    #[test]
+    fn trino_catalog_urn_matches_data_hub() {
+        let request = ResourceInfoRequest {
+            stacklet: "trino".to_owned(),
+            resource: ResourceInfoRequestResource::TrinoCatalog {
+                catalog: "lakehouse".to_owned(),
+            },
+        };
+
+        assert_eq!(
+            urn_for_request(&request, "PROD"),
+            "urn:li:container:39967cd09b38e2d4736d1eb604cd5247"
+        );
     }
 }