laminar_db/process_function/
catalog.rs

1//! Process bindings are sealed as typed stream DDL; code is supplied by the deployment.
2
3use super::ProcessFunctionRegistration;
4use crate::db::DbState;
5use crate::{DbError, DdlInfo, ExecuteResult, LaminarDB};
6
7impl ProcessFunctionRegistration {
8    fn catalog_ddl(&self) -> Result<String, DbError> {
9        let manifest = String::from_utf8(self.descriptor.to_manifest_json()?)
10            .map_err(|error| DbError::InvalidOperation(error.to_string()))?;
11        Ok(format!(
12            "CREATE STREAM {} AS SELECT * FROM laminar_process('{}', '{}')",
13            self.output_name,
14            self.source_name,
15            manifest.replace('\'', "''"),
16        ))
17    }
18}
19
20impl LaminarDB {
21    /// Return the canonical invocation DDL for a registered cluster process binding.
22    /// Include this statement after its source and before its consumers in
23    /// [`Self::execute_cluster_bootstrap_batch`]. Every owner must register the same immutable
24    /// implementation before bootstrapping or replaying that catalog. SQL never loads code.
25    ///
26    /// # Errors
27    /// Rejects an unknown binding or an incompatible manifest.
28    #[cfg(feature = "cluster")]
29    pub fn process_function_bootstrap_sql(&self, output_name: &str) -> Result<String, DbError> {
30        self.connector_manager
31            .lock()
32            .process_functions()
33            .get(output_name)
34            .ok_or_else(|| DbError::InvalidOperation("process binding does not exist".into()))?
35            .catalog_ddl()
36    }
37
38    pub(crate) fn create_bound_process_stream(
39        &self,
40        sql: &str,
41        output_name: &str,
42    ) -> Result<Option<ExecuteResult>, DbError> {
43        let registration = self
44            .connector_manager
45            .lock()
46            .process_functions()
47            .get(output_name)
48            .cloned();
49        let Some(registration) = registration else {
50            return Ok(None);
51        };
52        if !self.is_cluster_runtime() || DbState::load(&self.state) != DbState::Created {
53            return Err(DbError::Unsupported(
54                "process invocation DDL requires cluster startup bootstrap".into(),
55            ));
56        }
57        if crate::pipeline_identity::canonical_sql(sql)
58            != crate::pipeline_identity::canonical_sql(&registration.catalog_ddl()?)
59        {
60            return Err(DbError::Unsupported(
61                "process stream DDL differs from its immutable deployment binding".into(),
62            ));
63        }
64        self.install_process_function(&registration)?;
65        Ok(Some(ExecuteResult::Ddl(DdlInfo {
66            statement_type: "CREATE STREAM".into(),
67            object_name: output_name.into(),
68            #[cfg(feature = "cluster")]
69            topology_operation: None,
70            applied: true,
71        })))
72    }
73}
74
75#[cfg(all(test, feature = "cluster"))]
76mod tests {
77    use super::*;
78    use crate::process_function::tests::{descriptor, AccountActivity};
79    use laminar_connectors::connector::DeliveryGuarantee;
80    use laminar_core::cluster::control::{ClusterController, InMemoryKv};
81    use laminar_core::cluster::discovery::NodeId;
82    use std::sync::Arc;
83
84    async fn database(delivery: DeliveryGuarantee) -> Arc<LaminarDB> {
85        let (_, membership) = tokio::sync::watch::channel(Vec::new());
86        let controller = Arc::new(ClusterController::new(
87            NodeId(7),
88            Arc::new(InMemoryKv::new(NodeId(7))),
89            None,
90            membership,
91        ));
92        controller
93            .set_process_lease_deadline(Arc::new(
94                laminar_core::cluster::control::LeaseDeadline::live_for(
95                    std::time::Duration::from_secs(30),
96                ),
97            ))
98            .unwrap();
99        LaminarDB::builder()
100            .cluster_controller(controller)
101            .cluster_checkpoint_object_store(Arc::new(object_store::memory::InMemory::new()))
102            .delivery_guarantee(delivery)
103            .build()
104            .await
105            .unwrap()
106    }
107
108    #[tokio::test]
109    async fn cluster_process_registration_requires_at_least_once() {
110        let db = database(DeliveryGuarantee::ExactlyOnce).await;
111        assert!(db
112            .register_native_process_function(
113                "activity",
114                "events",
115                descriptor(),
116                Arc::new(AccountActivity)
117            )
118            .await
119            .unwrap_err()
120            .to_string()
121            .contains("delivery"));
122        assert!(db.process_functions().is_empty());
123        db.shutdown().await.unwrap();
124    }
125
126    #[tokio::test]
127    async fn process_catalog_binding_rejects_reordering_and_package_drift() {
128        let db = database(DeliveryGuarantee::AtLeastOnce).await;
129        let mut binding = descriptor();
130        binding.function_id = "account'activity".into();
131        db.register_native_process_function(
132            "activity",
133            "events",
134            binding,
135            Arc::new(AccountActivity),
136        )
137        .await
138        .unwrap();
139        let ddl = db.process_function_bootstrap_sql("activity").unwrap();
140        assert_eq!(laminar_sql::parse_streaming_sql(&ddl).unwrap().len(), 1);
141        for changed in [
142            format!("{ddl} WHERE amount > 0"),
143            format!("{ddl} LIMIT 1"),
144            ddl.replace("account''activity", "another_activity"),
145        ] {
146            assert!(db
147                .create_bound_process_stream(&changed, "activity")
148                .unwrap_err()
149                .to_string()
150                .contains("immutable deployment binding"));
151            assert!(db.catalog.get_stream_entry("activity").is_none());
152        }
153        assert!(db
154            .register_native_process_function(
155                "activity",
156                "events",
157                descriptor(),
158                Arc::new(AccountActivity)
159            )
160            .await
161            .is_err());
162        assert!(db.process_function_bootstrap_sql("unknown").is_err());
163        assert!(db
164            .validate_cluster_topology_change(
165                laminar_core::cluster::control::TopologyVersion::LEGACY_BASELINE,
166                &["DROP STREAM activity".into()],
167            )
168            .await
169            .unwrap_err()
170            .to_string()
171            .contains("process bindings are immutable"));
172        db.shutdown().await.unwrap();
173    }
174}
laminar_db/process_function/catalog.rs

laminar_db/process_function/
catalog.rs
