laminar_db/builder/
mod.rs

1//! Fluent builder for `LaminarDB` construction.
2#![allow(clippy::disallowed_types)] // cold path
3
4use std::collections::HashMap;
5use std::path::PathBuf;
6use std::sync::Arc;
7
8use datafusion_expr::{AggregateUDF, ScalarUDF};
9use laminar_core::streaming::{BackpressureStrategy, StreamCheckpointConfig};
10
11use crate::config::LaminarConfig;
12use crate::db::{LaminarDB, RuntimeMode};
13use crate::error::DbError;
14use crate::profile::Profile;
15
16/// Callback for registering custom connectors.
17type ConnectorCallback = Box<
18    dyn FnOnce(
19            &laminar_connectors::registry::ConnectorRegistry,
20        ) -> Result<(), laminar_connectors::error::ConnectorError>
21        + Send,
22>;
23
24/// Fluent builder for constructing a [`LaminarDB`] instance.
25///
26/// # Example
27///
28/// ```rust,ignore
29/// let db = LaminarDB::builder()
30///     .config_var("KAFKA_BROKERS", "localhost:9092")
31///     .buffer_size(131072)
32///     .build()
33///     .await?;
34/// ```
35pub struct LaminarDbBuilder {
36    config: LaminarConfig,
37    config_vars: HashMap<String, String>,
38    connector_callbacks: Vec<ConnectorCallback>,
39    profile: Profile,
40    profile_explicit: bool,
41    delivery_explicit: bool,
42    object_store_url: Option<String>,
43    object_store_options: HashMap<String, String>,
44    custom_udfs: Vec<ScalarUDF>,
45    custom_udafs: Vec<AggregateUDF>,
46    #[cfg(feature = "cluster")]
47    cluster_controller: Option<std::sync::Arc<laminar_core::cluster::control::ClusterController>>,
48    #[cfg(feature = "cluster")]
49    shuffle_sender: Option<std::sync::Arc<laminar_core::shuffle::ShuffleSender>>,
50    #[cfg(feature = "cluster")]
51    shuffle_receiver: Option<std::sync::Arc<laminar_core::shuffle::ShuffleReceiver>>,
52    #[cfg(feature = "cluster")]
53    decision_store: Option<std::sync::Arc<laminar_core::cluster::control::CheckpointDecisionStore>>,
54    #[cfg(feature = "cluster")]
55    assignment_snapshot_store:
56        Option<std::sync::Arc<laminar_core::cluster::control::AssignmentSnapshotStore>>,
57    #[cfg(feature = "cluster")]
58    catalog_manifest_store:
59        Option<std::sync::Arc<laminar_core::cluster::control::CatalogManifestStore>>,
60    #[cfg(feature = "cluster")]
61    verified_cluster_namespaces: Option<laminar_core::cluster::control::VerifiedClusterNamespaces>,
62    #[cfg(all(feature = "cluster", test))]
63    unchecked_cluster_checkpoint_store: Option<Arc<dyn object_store::ObjectStore>>,
64    vnode_registry: Option<std::sync::Arc<laminar_core::state::VnodeRegistry>>,
65    key_groups: Option<laminar_core::state::KeyGroupCount>,
66    physical_optimizer_rules: Vec<
67        std::sync::Arc<dyn datafusion::physical_optimizer::PhysicalOptimizerRule + Send + Sync>,
68    >,
69    target_partitions: Option<usize>,
70    ai_runtime: Option<std::sync::Arc<crate::ai::AiRuntime>>,
71}
72
73impl LaminarDbBuilder {
74    /// Create a new builder with default settings.
75    #[must_use]
76    pub fn new() -> Self {
77        Self {
78            config: LaminarConfig::default(),
79            config_vars: HashMap::new(),
80            connector_callbacks: Vec::new(),
81            profile: Profile::default(),
82            profile_explicit: false,
83            delivery_explicit: false,
84            object_store_url: None,
85            object_store_options: HashMap::new(),
86            custom_udfs: Vec::new(),
87            custom_udafs: Vec::new(),
88            #[cfg(feature = "cluster")]
89            cluster_controller: None,
90            #[cfg(feature = "cluster")]
91            shuffle_sender: None,
92            #[cfg(feature = "cluster")]
93            shuffle_receiver: None,
94            #[cfg(feature = "cluster")]
95            decision_store: None,
96            #[cfg(feature = "cluster")]
97            assignment_snapshot_store: None,
98            #[cfg(feature = "cluster")]
99            catalog_manifest_store: None,
100            #[cfg(feature = "cluster")]
101            verified_cluster_namespaces: None,
102            #[cfg(all(feature = "cluster", test))]
103            unchecked_cluster_checkpoint_store: None,
104            vnode_registry: None,
105            key_groups: None,
106            physical_optimizer_rules: Vec::new(),
107            target_partitions: None,
108            ai_runtime: None,
109        }
110    }
111
112    /// Install the AI subsystem; required for `ai_*` SQL functions.
113    #[must_use]
114    pub fn ai(mut self, runtime: std::sync::Arc<crate::ai::AiRuntime>) -> Self {
115        self.ai_runtime = Some(runtime);
116        self
117    }
118
119    /// Override `target_partitions`.
120    #[must_use]
121    pub fn target_partitions(mut self, n: usize) -> Self {
122        self.target_partitions = Some(n);
123        self
124    }
125
126    /// Register an additional `PhysicalOptimizerRule` on the session state.
127    #[must_use]
128    pub fn physical_optimizer_rule(
129        mut self,
130        rule: std::sync::Arc<
131            dyn datafusion::physical_optimizer::PhysicalOptimizerRule + Send + Sync,
132        >,
133    ) -> Self {
134        self.physical_optimizer_rules.push(rule);
135        self
136    }
137
138    /// Install the vnode topology and ownership registry.
139    #[must_use]
140    pub fn vnode_registry(
141        mut self,
142        registry: std::sync::Arc<laminar_core::state::VnodeRegistry>,
143    ) -> Self {
144        self.vnode_registry = Some(registry);
145        self
146    }
147
148    /// Set the stable key-group topology for a local runtime.
149    ///
150    /// When no registry is injected, the local node owns every configured key group.
151    #[must_use]
152    pub fn key_groups(mut self, key_groups: laminar_core::state::KeyGroupCount) -> Self {
153        self.key_groups = Some(key_groups);
154        self
155    }
156
157    /// Install the cluster control facade; selects cluster runtime semantics when the profile is
158    /// inferred. An explicitly selected profile must be [`Profile::Cluster`].
159    #[cfg(feature = "cluster")]
160    #[must_use]
161    pub fn cluster_controller(
162        mut self,
163        controller: std::sync::Arc<laminar_core::cluster::control::ClusterController>,
164    ) -> Self {
165        self.cluster_controller = Some(controller);
166        self
167    }
168
169    /// Install the outbound shuffle handle; pair with [`Self::shuffle_receiver`].
170    #[cfg(feature = "cluster")]
171    #[must_use]
172    pub fn shuffle_sender(
173        mut self,
174        sender: std::sync::Arc<laminar_core::shuffle::ShuffleSender>,
175    ) -> Self {
176        self.shuffle_sender = Some(sender);
177        self
178    }
179
180    /// Install the inbound shuffle handle; pair with [`Self::shuffle_sender`].
181    #[cfg(feature = "cluster")]
182    #[must_use]
183    pub fn shuffle_receiver(
184        mut self,
185        receiver: std::sync::Arc<laminar_core::shuffle::ShuffleReceiver>,
186    ) -> Self {
187        self.shuffle_receiver = Some(receiver);
188        self
189    }
190
191    /// Install the commit-marker store for cross-instance 2PC.
192    #[cfg(feature = "cluster")]
193    #[must_use]
194    pub fn decision_store(
195        mut self,
196        store: std::sync::Arc<laminar_core::cluster::control::CheckpointDecisionStore>,
197    ) -> Self {
198        self.decision_store = Some(store);
199        self
200    }
201
202    /// Install the assignment-snapshot store for dynamic rebalance.
203    #[cfg(feature = "cluster")]
204    #[must_use]
205    pub fn assignment_snapshot_store(
206        mut self,
207        store: std::sync::Arc<laminar_core::cluster::control::AssignmentSnapshotStore>,
208    ) -> Self {
209        self.assignment_snapshot_store = Some(store);
210        self
211    }
212
213    /// Install the catalog-manifest store for cluster-wide DDL replay.
214    #[cfg(feature = "cluster")]
215    #[must_use]
216    pub fn catalog_manifest_store(
217        mut self,
218        store: std::sync::Arc<laminar_core::cluster::control::CatalogManifestStore>,
219    ) -> Self {
220        self.catalog_manifest_store = Some(store);
221        self
222    }
223
224    /// Install the exact checkpoint handle admitted by the cluster namespace proof.
225    #[cfg(feature = "cluster")]
226    #[must_use]
227    pub fn verified_cluster_namespaces(
228        mut self,
229        namespaces: laminar_core::cluster::control::VerifiedClusterNamespaces,
230    ) -> Self {
231        self.verified_cluster_namespaces = Some(namespaces);
232        self
233    }
234
235    /// Install an unchecked checkpoint store in crate unit tests.
236    #[cfg(all(feature = "cluster", test))]
237    #[must_use]
238    pub(crate) fn cluster_checkpoint_object_store(
239        mut self,
240        store: Arc<dyn object_store::ObjectStore>,
241    ) -> Self {
242        self.unchecked_cluster_checkpoint_store = Some(store);
243        self
244    }
245
246    /// Set a config variable for `${VAR}` substitution in SQL.
247    #[must_use]
248    pub fn config_var(mut self, key: &str, value: &str) -> Self {
249        self.config_vars.insert(key.to_string(), value.to_string());
250        self
251    }
252
253    /// Set the bearer token presented when forwarding requests to the cluster
254    /// leader's HTTP API.
255    #[must_use]
256    pub fn http_auth_token(mut self, token: impl Into<String>) -> Self {
257        self.config.http_auth_token = Some(crate::config::SecretString::new(token));
258        self
259    }
260
261    /// Set the default buffer size for streaming channels.
262    #[must_use]
263    pub fn buffer_size(mut self, size: usize) -> Self {
264        self.config.default_buffer_size = size;
265        self
266    }
267
268    /// Set the default backpressure strategy.
269    #[must_use]
270    pub fn backpressure(mut self, strategy: BackpressureStrategy) -> Self {
271        self.config.default_backpressure = strategy;
272        self
273    }
274
275    /// Set the storage directory for WAL and checkpoints.
276    #[must_use]
277    pub fn storage_dir(mut self, path: impl Into<PathBuf>) -> Self {
278        self.config.storage_dir = Some(path.into());
279        self
280    }
281
282    /// Set checkpoint configuration.
283    #[must_use]
284    pub fn checkpoint(mut self, config: StreamCheckpointConfig) -> Self {
285        self.config.checkpoint = Some(config);
286        self
287    }
288
289    /// Select dirty-only changelog emission for keyed running aggregates.
290    #[must_use]
291    pub fn incremental_emit(mut self, enabled: bool) -> Self {
292        self.config.incremental_emit = enabled;
293        self
294    }
295
296    /// Set the deployment profile.
297    ///
298    /// See [`Profile`] for the available tiers. [`Profile::Cluster`] also requires a cluster
299    /// controller at build time.
300    #[must_use]
301    pub fn profile(mut self, profile: Profile) -> Self {
302        self.profile = profile;
303        self.profile_explicit = true;
304        self
305    }
306
307    /// Set the object-store URL for durable checkpoints.
308    ///
309    /// Required when using [`Profile::Durable`] or
310    /// [`Profile::Cluster`].
311    #[must_use]
312    pub fn object_store_url(mut self, url: impl Into<String>) -> Self {
313        self.object_store_url = Some(url.into());
314        self
315    }
316
317    /// Set explicit credential/config overrides for the object store.
318    ///
319    /// Keys are backend-specific (e.g., `aws_access_key_id`, `aws_region`).
320    /// These supplement environment-variable-based credential resolution.
321    #[must_use]
322    pub fn object_store_options(mut self, opts: HashMap<String, String>) -> Self {
323        self.object_store_options = opts;
324        self
325    }
326
327    /// Set the end-to-end delivery guarantee for the pipeline.
328    #[must_use]
329    pub fn delivery_guarantee(
330        mut self,
331        guarantee: laminar_connectors::connector::DeliveryGuarantee,
332    ) -> Self {
333        self.config.delivery_guarantee = guarantee;
334        self.delivery_explicit = true;
335        self
336    }
337
338    /// Register a custom scalar UDF; available in SQL after `build()`.
339    #[must_use]
340    pub fn register_udf(mut self, udf: ScalarUDF) -> Self {
341        self.custom_udfs.push(udf);
342        self
343    }
344
345    /// Register a custom aggregate UDF; available in SQL after `build()`.
346    #[must_use]
347    pub fn register_udaf(mut self, udaf: AggregateUDF) -> Self {
348        self.custom_udafs.push(udaf);
349        self
350    }
351
352    /// Source → coordinator channel capacity (default 64).
353    #[must_use]
354    pub fn pipeline_channel_capacity(mut self, capacity: usize) -> Self {
355        self.config.pipeline_channel_capacity = Some(capacity);
356        self
357    }
358
359    /// Micro-batch coalescing window (default 5ms for connectors, 0 for embedded).
360    #[must_use]
361    pub fn pipeline_batch_window(mut self, window: std::time::Duration) -> Self {
362        self.config.pipeline_batch_window = Some(window);
363        self
364    }
365
366    /// Max drain time per cycle in nanoseconds (default 1ms).
367    #[must_use]
368    pub fn pipeline_drain_budget_ns(mut self, ns: u64) -> Self {
369        self.config.pipeline_drain_budget_ns = Some(ns);
370        self
371    }
372
373    /// Per-query execution budget in nanoseconds (default 8ms).
374    #[must_use]
375    pub fn pipeline_query_budget_ns(mut self, ns: u64) -> Self {
376        self.config.pipeline_query_budget_ns = Some(ns);
377        self
378    }
379
380    /// Per-port operator input-buffer cap in batches (default 256).
381    #[must_use]
382    pub fn pipeline_max_input_buf_batches(mut self, batches: usize) -> Self {
383        self.config.pipeline_max_input_buf_batches = Some(batches);
384        self
385    }
386
387    /// Per-port operator input-buffer cap in bytes.
388    #[must_use]
389    pub fn pipeline_max_input_buf_bytes(mut self, bytes: usize) -> Self {
390        self.config.pipeline_max_input_buf_bytes = Some(bytes);
391        self
392    }
393
394    /// Pipeline-wide charged-byte limit for managed operator working state.
395    ///
396    /// This execution-memory envelope is independent of checkpoint storage.
397    #[must_use]
398    pub fn pipeline_max_managed_state_bytes(mut self, bytes: usize) -> Self {
399        self.config.pipeline_max_managed_state_bytes = Some(bytes);
400        self
401    }
402
403    /// Retain temporal right-side history across idle periods for at least this duration.
404    #[must_use]
405    pub fn temporal_join_idle_history_retention(mut self, retention: std::time::Duration) -> Self {
406        self.config.temporal_join_idle_history_retention = Some(retention);
407        self
408    }
409
410    /// Mark inactive watermarked sources and input channels idle after this duration.
411    #[must_use]
412    pub fn source_idle_timeout(mut self, timeout: std::time::Duration) -> Self {
413        self.config.source_idle_timeout = Some(timeout);
414        self
415    }
416
417    /// Event timestamps farther ahead of wall clock do not advance source watermarks.
418    /// Zero disables the guard.
419    #[must_use]
420    pub fn event_time_max_future_skew(mut self, skew: std::time::Duration) -> Self {
421        self.config.event_time_max_future_skew = skew;
422        self
423    }
424
425    /// Backpressure policy (default `Backpressure`).
426    #[must_use]
427    pub fn pipeline_backpressure_policy(
428        mut self,
429        policy: crate::config::BackpressurePolicy,
430    ) -> Self {
431        self.config.pipeline_backpressure_policy = policy;
432        self
433    }
434
435    /// Auto-restart policy used when supervision is enabled.
436    #[must_use]
437    pub fn restart_policy(mut self, policy: crate::config::RestartPolicy) -> Self {
438        self.config.restart_policy = policy;
439        self
440    }
441
442    /// Register custom connectors; the callback runs after built-ins are wired and must propagate
443    /// registration errors. The registry is frozen before [`Self::build`] returns.
444    #[must_use]
445    pub fn register_connector(
446        mut self,
447        f: impl FnOnce(
448                &laminar_connectors::registry::ConnectorRegistry,
449            ) -> Result<(), laminar_connectors::error::ConnectorError>
450            + Send
451            + 'static,
452    ) -> Self {
453        self.connector_callbacks.push(Box::new(f));
454        self
455    }
456
457    /// Build the `LaminarDB` instance.
458    ///
459    /// # Errors
460    ///
461    /// Returns `DbError` if database creation fails.
462    // COMPAT: `build` remains async as part of the public API even when construction has no I/O.
463    #[allow(unknown_lints, clippy::unused_async, clippy::unused_async_trait_impl)]
464    pub async fn build(mut self) -> Result<Arc<LaminarDB>, DbError> {
465        let has_checkpoint_data_dir = self
466            .config
467            .checkpoint
468            .as_ref()
469            .and_then(|checkpoint| checkpoint.data_dir.as_ref())
470            .is_some();
471        if self.object_store_url.is_some() && has_checkpoint_data_dir {
472            return Err(DbError::Config(
473                "object_store_url conflicts with checkpoint.data_dir; configure exactly one checkpoint store"
474                    .into(),
475            ));
476        }
477        #[cfg(feature = "cluster")]
478        if self.has_cluster_checkpoint_store() && self.object_store_url.is_some() {
479            return Err(DbError::Config(
480                "verified cluster namespaces conflict with object_store_url; configure exactly one checkpoint object store"
481                    .into(),
482            ));
483        }
484        #[cfg(feature = "cluster")]
485        if self.has_cluster_checkpoint_store() && has_checkpoint_data_dir {
486            return Err(DbError::Config(
487                "verified cluster namespaces conflict with checkpoint.data_dir; configure exactly one checkpoint store"
488                    .into(),
489            ));
490        }
491
492        self.config.object_store_url = self.object_store_url.take();
493        self.config.object_store_options = std::mem::take(&mut self.object_store_options);
494        if let Some(url) = self.config.object_store_url.as_deref().filter(|url| {
495            laminar_core::storage_location::StorageProvider::detect_uri(url)
496                == Some(laminar_core::storage_location::StorageProvider::Local)
497        }) {
498            laminar_core::checkpoint::object_store_builder::file_url_path(url)
499                .map_err(|error| DbError::Config(format!("checkpoint storage URL: {error}")))?;
500        }
501
502        if !self.profile_explicit {
503            self.profile = Profile::from_config(&self.config, false);
504        }
505
506        let runtime_mode = self.resolve_runtime_mode()?;
507        if runtime_mode.is_cluster() && !self.profile_explicit {
508            self.profile = Profile::Cluster;
509        }
510        if runtime_mode.is_cluster() && !self.delivery_explicit {
511            self.config.delivery_guarantee =
512                laminar_connectors::connector::DeliveryGuarantee::AtLeastOnce;
513        }
514
515        self.profile
516            .validate_features()
517            .map_err(|e| DbError::Config(e.to_string()))?;
518        Self::validate_cluster_delivery(runtime_mode, self.config.delivery_guarantee)?;
519        #[cfg(feature = "cluster")]
520        if runtime_mode == RuntimeMode::Cluster {
521            if self.config.object_store_url.is_some() || has_checkpoint_data_dir {
522                return Err(DbError::Config(
523                    "[LDB-0011] cluster checkpoint storage must be supplied by a successful shared namespace proof; object_store_url and checkpoint.data_dir are not cluster authorities"
524                        .into(),
525                ));
526            }
527            if !self.has_cluster_checkpoint_store() {
528                return Err(DbError::Config(
529                    "[LDB-0011] cluster runtime requires VerifiedClusterNamespaces from the shared checkpoint namespace proof"
530                        .into(),
531                ));
532            }
533            if let Some(namespaces) = &self.verified_cluster_namespaces {
534                let controller = self.cluster_controller.as_ref().ok_or_else(|| {
535                    DbError::Config("cluster namespace proof requires a cluster controller".into())
536                })?;
537                let expected = laminar_core::checkpoint::CheckpointParticipant {
538                    node_id: controller.instance_id().0,
539                    boot_incarnation: controller.recovery_incarnation(),
540                };
541                if namespaces.local_participant() != expected {
542                    let proved = namespaces.local_participant();
543                    return Err(DbError::Config(format!(
544                        "[LDB-0011] verified cluster namespaces belong to node {} boot {}, but the controller is node {} boot {}",
545                        proved.node_id,
546                        proved.boot_incarnation,
547                        expected.node_id,
548                        expected.boot_incarnation,
549                    )));
550                }
551            }
552        }
553        #[cfg(feature = "cluster")]
554        let profile_object_store = self.config.object_store_url.as_deref().or_else(|| {
555            self.has_cluster_checkpoint_store()
556                .then_some("verified-cluster-checkpoint-store")
557        });
558        #[cfg(not(feature = "cluster"))]
559        let profile_object_store = self.config.object_store_url.as_deref();
560        self.profile
561            .validate_config(&self.config, profile_object_store)
562            .map_err(|e| DbError::Config(e.to_string()))?;
563
564        if runtime_mode == RuntimeMode::Local
565            && self.config.delivery_guarantee
566                != laminar_connectors::connector::DeliveryGuarantee::BestEffort
567            && self.config.object_store_url.as_deref().is_some_and(|url| {
568                laminar_core::storage_location::StorageProvider::detect_uri(url)
569                    != Some(laminar_core::storage_location::StorageProvider::Local)
570            })
571        {
572            return Err(DbError::Config(
573                "[LDB-0014] a local replay-capable deployment with a shared cloud checkpoint namespace is not admitted until its writer lease is term-fenced; use a built-in or file:// local checkpoint directory, or best_effort delivery"
574                    .into(),
575            ));
576        }
577
578        Self::validate_backpressure(&self.config)?;
579        self.validate_vnode_topology(runtime_mode)?;
580        if let Some(key_groups) = self
581            .key_groups
582            .filter(|_| runtime_mode == RuntimeMode::Local && self.vnode_registry.is_none())
583        {
584            self.vnode_registry = Some(Arc::new(laminar_core::state::VnodeRegistry::single_owner(
585                u32::from(key_groups),
586                laminar_core::state::LOCAL_NODE_ID,
587            )));
588        }
589        #[cfg(feature = "cluster")]
590        self.bind_cluster_process_lease(runtime_mode)?;
591
592        self.profile.apply_defaults(&mut self.config);
593
594        #[cfg(feature = "cluster")]
595        let cluster_checkpoint_store = self.cluster_checkpoint_store();
596        let mut db = LaminarDB::open_with_config_and_vars_and_rules(
597            self.config,
598            self.config_vars,
599            &self.physical_optimizer_rules,
600            self.target_partitions,
601            runtime_mode,
602        )?;
603        for udf in self.custom_udfs {
604            db.register_custom_udf(udf)?;
605        }
606        for udaf in self.custom_udafs {
607            db.register_custom_udaf(udaf)?;
608        }
609        if let Some(runtime) = self.ai_runtime {
610            let handle = tokio::runtime::Handle::try_current().map_err(|_| {
611                DbError::InvalidOperation(
612                    "LaminarDB::build() with an AI runtime must run inside a Tokio runtime"
613                        .to_string(),
614                )
615            })?;
616            db.set_ai_runtime(runtime, handle);
617        }
618        for callback in self.connector_callbacks {
619            callback(db.connector_registry())?;
620        }
621        db.connector_registry().freeze();
622        #[cfg(feature = "cluster")]
623        if let Some(controller) = self.cluster_controller {
624            db.set_cluster_controller(controller)?;
625        }
626        #[cfg(feature = "cluster")]
627        if let Some(store) = cluster_checkpoint_store {
628            db.set_cluster_checkpoint_object_store(store)?;
629        }
630        #[cfg(feature = "cluster")]
631        if let Some(sender) = self.shuffle_sender {
632            db.set_shuffle_sender(sender);
633        }
634        #[cfg(feature = "cluster")]
635        if let Some(receiver) = self.shuffle_receiver {
636            db.set_shuffle_receiver(receiver);
637        }
638        #[cfg(feature = "cluster")]
639        if let Some(store) = self.decision_store {
640            db.set_decision_store(store);
641        }
642        #[cfg(feature = "cluster")]
643        if let Some(store) = self.assignment_snapshot_store {
644            db.set_assignment_snapshot_store(store);
645        }
646        #[cfg(feature = "cluster")]
647        if let Some(store) = self.catalog_manifest_store {
648            db.set_catalog_manifest_store(store);
649        }
650        if let Some(registry) = self.vnode_registry {
651            db.set_vnode_registry(registry);
652        }
653        Ok(Arc::new(db))
654    }
655
656    fn validate_vnode_topology(&self, runtime_mode: RuntimeMode) -> Result<(), DbError> {
657        let registry_key_groups = self
658            .vnode_registry
659            .as_ref()
660            .map(|registry| {
661                let registry_count = registry.vnode_count();
662                laminar_core::state::KeyGroupCount::try_from(registry_count).map_err(|_| {
663                    DbError::Config(format!(
664                        "vnode_registry count must be between 1 and {}, got {registry_count}",
665                        laminar_core::state::MAX_KEY_GROUP_COUNT
666                    ))
667                })
668            })
669            .transpose()?;
670        if let Some(registry) = self
671            .vnode_registry
672            .as_ref()
673            .filter(|_| runtime_mode == RuntimeMode::Local)
674        {
675            let assignment = registry.versioned_snapshot();
676            if let Some((vnode, owner)) = assignment
677                .owners()
678                .iter()
679                .copied()
680                .enumerate()
681                .find(|(_, owner)| *owner != laminar_core::state::LOCAL_NODE_ID)
682            {
683                return Err(DbError::Config(format!(
684                    "local vnode {vnode} must be owned by {}, got {owner}",
685                    laminar_core::state::LOCAL_NODE_ID.0
686                )));
687            }
688        }
689        if let (Some(configured), Some(registry)) = (self.key_groups, registry_key_groups) {
690            if configured != registry {
691                return Err(DbError::Config(format!(
692                    "configured key-group count {configured} does not match vnode_registry count {registry}"
693                )));
694            }
695        }
696        Ok(())
697    }
698
699    #[cfg(feature = "cluster")]
700    fn bind_cluster_process_lease(&self, runtime_mode: RuntimeMode) -> Result<(), DbError> {
701        if !runtime_mode.is_cluster() {
702            return Ok(());
703        }
704        let controller = self.cluster_controller.as_ref().ok_or_else(|| {
705            DbError::Config("cluster runtime requires a cluster controller".into())
706        })?;
707        let deadline = controller.process_lease_deadline().ok_or_else(|| {
708            DbError::Config(
709                "cluster runtime requires one shared process lease deadline before construction"
710                    .into(),
711            )
712        })?;
713        if !deadline.is_live() {
714            return Err(DbError::Config(
715                "cluster runtime process lease deadline is already expired".into(),
716            ));
717        }
718        match (&self.shuffle_sender, &self.shuffle_receiver) {
719            (Some(sender), Some(receiver)) => {
720                sender
721                    .bind_process_lease_deadline_pair(receiver, deadline)
722                    .map_err(|error| {
723                        DbError::Config(format!(
724                            "shuffle process lease does not match the controller: {error}"
725                        ))
726                    })?;
727            }
728            (None, None) => {}
729            _ => {
730                return Err(DbError::Config(
731                    "cluster shuffle sender and receiver must be installed together".into(),
732                ));
733            }
734        }
735        Ok(())
736    }
737
738    fn validate_backpressure(config: &LaminarConfig) -> Result<(), DbError> {
739        use crate::config::BackpressurePolicy;
740        use laminar_connectors::connector::DeliveryGuarantee;
741
742        let policy = config.pipeline_backpressure_policy;
743        if policy == BackpressurePolicy::Backpressure {
744            return Ok(());
745        }
746
747        let has_count_cap = config.pipeline_max_input_buf_batches.is_none_or(|c| c > 0);
748        let has_byte_cap = config.pipeline_max_input_buf_bytes.is_some_and(|b| b > 0);
749        if !has_count_cap && !has_byte_cap {
750            return Err(DbError::Config(format!(
751                "backpressure_policy={policy:?} requires at least one of \
752                 pipeline_max_input_buf_batches (>0) or pipeline_max_input_buf_bytes"
753            )));
754        }
755
756        if policy == BackpressurePolicy::ShedOldest
757            && config.delivery_guarantee == DeliveryGuarantee::ExactlyOnce
758        {
759            return Err(DbError::Config(
760                "ShedOldest drops data; it is incompatible with exactly-once \
761                 delivery. Use Backpressure or Fail, or downgrade the guarantee."
762                    .into(),
763            ));
764        }
765        Ok(())
766    }
767
768    #[cfg(feature = "cluster")]
769    fn has_cluster_checkpoint_store(&self) -> bool {
770        let has_store = self.verified_cluster_namespaces.is_some();
771        #[cfg(test)]
772        let has_store = has_store || self.unchecked_cluster_checkpoint_store.is_some();
773        has_store
774    }
775
776    #[cfg(feature = "cluster")]
777    fn cluster_checkpoint_store(&self) -> Option<Arc<dyn object_store::ObjectStore>> {
778        let store = self
779            .verified_cluster_namespaces
780            .as_ref()
781            .map(laminar_core::cluster::control::VerifiedClusterNamespaces::checkpoint_store);
782        #[cfg(test)]
783        let store = store.or_else(|| self.unchecked_cluster_checkpoint_store.clone());
784        store
785    }
786
787    /// Resolve the distributed execution scope exactly once and reject partial or contradictory
788    /// cluster wiring before constructing the database.
789    fn resolve_runtime_mode(&self) -> Result<RuntimeMode, DbError> {
790        #[cfg(not(feature = "cluster"))]
791        {
792            if self.profile == Profile::Cluster {
793                return Err(DbError::Config(
794                    "Profile::Cluster requires a cluster controller and a cluster-enabled build"
795                        .into(),
796                ));
797            }
798            Ok(RuntimeMode::Local)
799        }
800
801        #[cfg(feature = "cluster")]
802        {
803            let has_controller = self.cluster_controller.is_some();
804            let has_cluster_only_handle = self.shuffle_sender.is_some()
805                || self.shuffle_receiver.is_some()
806                || self.decision_store.is_some()
807                || self.assignment_snapshot_store.is_some()
808                || self.catalog_manifest_store.is_some()
809                || self.has_cluster_checkpoint_store();
810
811            if has_cluster_only_handle && !has_controller {
812                return Err(DbError::Config(
813                    "cluster-only stores and shuffle handles require a cluster controller".into(),
814                ));
815            }
816            if self.profile == Profile::Cluster && !has_controller {
817                return Err(DbError::Config(
818                    "Profile::Cluster requires a cluster controller".into(),
819                ));
820            }
821            if self.profile_explicit && self.profile != Profile::Cluster && has_controller {
822                return Err(DbError::Config(format!(
823                    "profile {} cannot be combined with a cluster controller; select Profile::Cluster or omit the explicit profile",
824                    self.profile
825                )));
826            }
827
828            Ok(if has_controller {
829                RuntimeMode::Cluster
830            } else {
831                RuntimeMode::Local
832            })
833        }
834    }
835
836    /// Validate delivery semantics whose correctness depends on the runtime mode.
837    ///
838    /// Connector-specific exact-delivery certification is checked before connector I/O, when the
839    /// concrete source and sink contracts are available.
840    fn validate_cluster_delivery(
841        runtime_mode: RuntimeMode,
842        guarantee: laminar_connectors::connector::DeliveryGuarantee,
843    ) -> Result<(), DbError> {
844        use laminar_connectors::connector::DeliveryGuarantee;
845
846        if !runtime_mode.is_cluster() {
847            return Ok(());
848        }
849        match guarantee {
850            DeliveryGuarantee::BestEffort => Err(DbError::Config(
851                "cluster mode requires at_least_once delivery; best_effort has no defined \
852                 rebalance/state-loss contract"
853                    .into(),
854            )),
855            DeliveryGuarantee::AtLeastOnce | DeliveryGuarantee::ExactlyOnce => Ok(()),
856        }
857    }
858}
859
860impl Default for LaminarDbBuilder {
861    fn default() -> Self {
862        Self::new()
863    }
864}
865
866impl std::fmt::Debug for LaminarDbBuilder {
867    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
868        f.debug_struct("LaminarDbBuilder")
869            .field("config", &self.config)
870            .field("profile", &self.profile)
871            .field("profile_explicit", &self.profile_explicit)
872            .field("delivery_explicit", &self.delivery_explicit)
873            .field("key_groups", &self.key_groups)
874            .field("object_store_url", &self.object_store_url)
875            .field(
876                "object_store_options_count",
877                &self.object_store_options.len(),
878            )
879            .field("config_vars_count", &self.config_vars.len())
880            .field("connector_callbacks", &self.connector_callbacks.len())
881            .field("custom_udfs", &self.custom_udfs.len())
882            .field("custom_udafs", &self.custom_udafs.len())
883            .finish_non_exhaustive()
884    }
885}
886
887#[cfg(test)]
888mod tests;
laminar_db/builder/mod.rs

laminar_db/builder/
mod.rs
