laminar_db/process_function/
registration.rs

1use std::sync::Arc;
2
3use datafusion::datasource::empty::EmptyTable;
4use laminar_connectors::connector::DeliveryGuarantee;
5use laminar_core::catalog::CatalogObjectKind;
6
7use super::{
8    NativeProcessFunction, ProcessFunctionDescriptor, ProcessFunctionInfo,
9    ProcessFunctionRegistration, ProcessHandler, ProcessRuntime,
10};
11use crate::db::{exact_table_reference, DbState, LaminarDB};
12use crate::error::DbError;
13
14impl LaminarDB {
15    /// Inspect process functions registered on this database instance.
16    #[must_use]
17    pub fn process_functions(&self) -> Vec<ProcessFunctionInfo> {
18        let manager = self.connector_manager.lock();
19        let mut functions = manager
20            .process_functions()
21            .values()
22            .map(|registration| ProcessFunctionInfo {
23                output_name: registration.output_name.clone(),
24                source_name: registration.source_name.clone(),
25                descriptor: registration.descriptor.clone(),
26            })
27            .collect::<Vec<_>>();
28        functions.sort_unstable_by(|left, right| left.output_name.cmp(&right.output_name));
29        functions
30    }
31
32    /// Register a trusted native keyed process function over one append-only source.
33    /// Registration is offline: call it after creating a local source and before `start()`. The
34    /// caller must register the same immutable implementation when constructing a replacement
35    /// database instance that restores an existing checkpoint.
36    ///
37    /// Local at-least-once execution requires checkpointing and a replayable append-only
38    /// connector that reproduces one channel in fixed replay batches with deterministic positions.
39    /// Only one logical source is admitted. Source and sink contracts are verified before startup I/O. A direct
40    /// in-memory source is available only with best-effort delivery.
41    /// Cluster execution admits at-least-once delivery with a splittable fixed-batch source.
42    /// Register the binding on every owner before catalog bootstrap, then include
43    /// `process_function_bootstrap_sql()` after the source in the ordered startup batch.
44    /// Process bindings cannot be changed through live catalog mutations.
45    /// Native code runs in the compute process and must be trusted and nonblocking.
46    ///
47    /// # Errors
48    /// Rejects incompatible schemas, identities, source modes, runtime modes, or resource limits.
49    pub async fn register_native_process_function(
50        &self,
51        output_name: &str,
52        source_name: &str,
53        descriptor: ProcessFunctionDescriptor,
54        handler: Arc<dyn NativeProcessFunction>,
55    ) -> Result<(), DbError> {
56        if descriptor.runtime != ProcessRuntime::NativeRust {
57            return Err(DbError::Unsupported(
58                "native registration requires the trusted native Rust runtime".into(),
59            ));
60        }
61        self.register_process_function(
62            output_name,
63            source_name,
64            descriptor,
65            ProcessHandler::Native(handler),
66        )
67        .await
68    }
69
70    /// Register a connected loopback Rust or Python worker. At-least-once Python requires a
71    /// replay-safe descriptor and the live binding supplied by `LocalPythonWorker` on Linux.
72    /// Its complete package must belong to the read-only root image; a standalone connected
73    /// client does not supply that binding. Undeclared Python supports best-effort delivery.
74    /// Source-order and cluster-bootstrap requirements match native registration.
75    /// The caller owns the worker process lifecycle and must keep it available until shutdown.
76    ///
77    /// # Errors
78    /// Rejects mismatched descriptors, unsupported modes, schemas, or resource limits.
79    #[cfg(feature = "process-remote")]
80    pub async fn register_remote_process_function(
81        &self,
82        output_name: &str,
83        source_name: &str,
84        descriptor: ProcessFunctionDescriptor,
85        client: Arc<super::remote::RemoteProcessClient>,
86    ) -> Result<(), DbError> {
87        if descriptor.runtime == ProcessRuntime::NativeRust {
88            return Err(DbError::InvalidOperation(
89                "remote registration requires a remote runtime".into(),
90            ));
91        }
92        if descriptor.to_manifest_json()? != client.descriptor().to_manifest_json()? {
93            return Err(DbError::InvalidOperation(
94                "connected process worker descriptor differs from registration".into(),
95            ));
96        }
97        self.register_process_function(
98            output_name,
99            source_name,
100            descriptor,
101            ProcessHandler::Remote(client),
102        )
103        .await
104    }
105
106    async fn register_process_function(
107        &self,
108        output_name: &str,
109        source_name: &str,
110        descriptor: ProcessFunctionDescriptor,
111        handler: ProcessHandler,
112    ) -> Result<(), DbError> {
113        let _topology = self.topology_ddl_lock.write().await;
114        self.ensure_topology_ddl_allowed("REGISTER PROCESS FUNCTION")?;
115        if DbState::load(&self.state) != DbState::Created {
116            return Err(DbError::Unsupported(
117                "process functions require registration before pipeline startup".into(),
118            ));
119        }
120        if self.config.delivery_guarantee == DeliveryGuarantee::ExactlyOnce {
121            return Err(DbError::Unsupported(
122                "process functions do not support exactly-once delivery".into(),
123            ));
124        }
125        if self.config.delivery_guarantee == DeliveryGuarantee::AtLeastOnce
126            && !handler.supports_replay(&descriptor)
127        {
128            return Err(DbError::Unsupported(
129                "at-least-once Python process functions require immutable dependency binding from a supervised replay-safe worker"
130                    .into(),
131            ));
132        }
133        if !valid_name(output_name) || !valid_name(source_name) || output_name == source_name {
134            return Err(DbError::InvalidOperation(
135                "process source and output require distinct lowercase SQL identifiers".into(),
136            ));
137        }
138        descriptor.to_manifest_json()?;
139        let registration = ProcessFunctionRegistration {
140            output_name: output_name.into(),
141            source_name: source_name.into(),
142            descriptor,
143            handler,
144        };
145        if self.is_cluster_runtime() {
146            if self.config.delivery_guarantee != DeliveryGuarantee::AtLeastOnce {
147                return Err(DbError::Unsupported(
148                    "cluster process functions require at-least-once delivery".into(),
149                ));
150            }
151            let mut manager = self.connector_manager.lock();
152            if manager.process_functions().contains_key(output_name) {
153                return Err(DbError::InvalidOperation(
154                    "process binding already exists".into(),
155                ));
156            }
157            manager.register_process_function(registration);
158            return Ok(());
159        }
160        self.install_process_function(&registration)
161    }
162
163    pub(crate) fn install_process_function(
164        &self,
165        registration: &ProcessFunctionRegistration,
166    ) -> Result<(), DbError> {
167        let ProcessFunctionRegistration {
168            output_name,
169            source_name,
170            descriptor,
171            ..
172        } = registration;
173        let source = self.catalog.get_source(source_name).ok_or_else(|| {
174            DbError::InvalidOperation(format!("process source '{source_name}' does not exist"))
175        })?;
176        self.validate_process_source_order(
177            output_name,
178            source_name,
179            self.connector_manager.lock().sources(),
180        )?;
181        if source.schema.as_ref() != descriptor.input_schema.as_ref() {
182            return Err(DbError::InvalidOperation(
183                "process descriptor input schema differs from its source".into(),
184            ));
185        }
186        if source.watermark_column.as_deref() != Some(descriptor.event_time_column.as_str())
187            || source
188                .is_processing_time
189                .load(std::sync::atomic::Ordering::Acquire)
190            || !source.primary_key.is_empty()
191        {
192            return Err(DbError::Unsupported(
193                "process source requires an append-only event-time watermark on the declared column"
194                    .into(),
195            ));
196        }
197        let reservation = self
198            .reserve_catalog_name(output_name, CatalogObjectKind::Stream, false)?
199            .ok_or_else(|| DbError::InvalidOperation("process output already exists".into()))?;
200        self.catalog.register_stream(output_name)?;
201        self.connector_manager
202            .lock()
203            .register_process_function(registration.clone());
204        self.ctx
205            .register_table(
206                exact_table_reference(output_name),
207                Arc::new(EmptyTable::new(Arc::clone(&descriptor.output_schema))),
208            )
209            .map_err(|error| {
210                DbError::Pipeline(format!(
211                    "could not register process output '{output_name}' for SQL planning: {error}"
212                ))
213            })?;
214        self.stream_schemas
215            .write()
216            .insert(output_name.clone(), Arc::clone(&descriptor.output_schema));
217        reservation.commit();
218        Ok(())
219    }
220}
221
222fn valid_name(name: &str) -> bool {
223    if name.len() > 128 {
224        return false;
225    }
226    let mut bytes = name.bytes();
227    matches!(bytes.next(), Some(b'a'..=b'z' | b'_'))
228        && bytes.all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_')
229}
laminar_db/process_function/registration.rs

laminar_db/process_function/
registration.rs
