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    /// Set the deployment profile.
290    ///
291    /// See [`Profile`] for the available tiers. [`Profile::Cluster`] also requires a cluster
292    /// controller at build time.
293    #[must_use]
294    pub fn profile(mut self, profile: Profile) -> Self {
295        self.profile = profile;
296        self.profile_explicit = true;
297        self
298    }
299
300    /// Set the object-store URL for durable checkpoints.
301    ///
302    /// Required when using [`Profile::Durable`] or
303    /// [`Profile::Cluster`].
304    #[must_use]
305    pub fn object_store_url(mut self, url: impl Into<String>) -> Self {
306        self.object_store_url = Some(url.into());
307        self
308    }
309
310    /// Set explicit credential/config overrides for the object store.
311    ///
312    /// Keys are backend-specific (e.g., `aws_access_key_id`, `aws_region`).
313    /// These supplement environment-variable-based credential resolution.
314    #[must_use]
315    pub fn object_store_options(mut self, opts: HashMap<String, String>) -> Self {
316        self.object_store_options = opts;
317        self
318    }
319
320    /// Set the end-to-end delivery guarantee for the pipeline.
321    #[must_use]
322    pub fn delivery_guarantee(
323        mut self,
324        guarantee: laminar_connectors::connector::DeliveryGuarantee,
325    ) -> Self {
326        self.config.delivery_guarantee = guarantee;
327        self.delivery_explicit = true;
328        self
329    }
330
331    /// Register a custom scalar UDF; available in SQL after `build()`.
332    #[must_use]
333    pub fn register_udf(mut self, udf: ScalarUDF) -> Self {
334        self.custom_udfs.push(udf);
335        self
336    }
337
338    /// Register a custom aggregate UDF; available in SQL after `build()`.
339    #[must_use]
340    pub fn register_udaf(mut self, udaf: AggregateUDF) -> Self {
341        self.custom_udafs.push(udaf);
342        self
343    }
344
345    /// Per-source Arrow-byte cap for embedded push queues (default 64 MiB).
346    /// Each source's snapshot history has an independent cap of the same size.
347    /// Typed handles are checked after Arrow conversion. Zero and values above
348    /// [`laminar_core::streaming::MAX_SOURCE_QUEUED_BYTES`] are rejected at build time.
349    #[must_use]
350    pub fn push_source_max_bytes(mut self, bytes: usize) -> Self {
351        self.config.push_source_max_bytes = bytes;
352        self
353    }
354
355    /// Source → coordinator channel capacity (default 64).
356    #[must_use]
357    pub fn pipeline_channel_capacity(mut self, capacity: usize) -> Self {
358        self.config.pipeline_channel_capacity = Some(capacity);
359        self
360    }
361
362    /// Shared connector FIFO Arrow-byte budget (default 64 MiB), including parked messages.
363    /// Also caps each producer's waiting batch; excludes decoder scratch and downstream buffers.
364    /// Zero and values above [`crate::MAX_SOURCE_QUEUE_BYTES`] are rejected at build time.
365    #[must_use]
366    pub fn source_queue_max_bytes(mut self, bytes: usize) -> Self {
367        self.config.source_queue_max_bytes = bytes;
368        self
369    }
370
371    /// Micro-batch coalescing window (default 5ms for connectors, 0 for embedded).
372    #[must_use]
373    pub fn pipeline_batch_window(mut self, window: std::time::Duration) -> Self {
374        self.config.pipeline_batch_window = Some(window);
375        self
376    }
377
378    /// Max drain time per cycle in nanoseconds (default 1ms).
379    #[must_use]
380    pub fn pipeline_drain_budget_ns(mut self, ns: u64) -> Self {
381        self.config.pipeline_drain_budget_ns = Some(ns);
382        self
383    }
384
385    /// Per-query execution budget in nanoseconds (default 8ms).
386    #[must_use]
387    pub fn pipeline_query_budget_ns(mut self, ns: u64) -> Self {
388        self.config.pipeline_query_budget_ns = Some(ns);
389        self
390    }
391
392    /// Shared per-DB limit for participating `DataFusion` reservations in bytes (default 256 MiB).
393    ///
394    /// Zero is rejected at build time. DB-owned contexts disable disk spilling.
395    /// This does not bound direct Arrow allocations, managed state, connector I/O or process RSS.
396    #[must_use]
397    pub fn datafusion_memory_limit_bytes(mut self, bytes: usize) -> Self {
398        self.config.datafusion_memory_limit_bytes = bytes;
399        self
400    }
401
402    /// Set the maximum live row count of each local reference table (default 1,000,000).
403    /// Zero is rejected at build time. Over-limit updates and snapshots fail atomically.
404    #[must_use]
405    pub fn reference_table_max_rows(mut self, rows: usize) -> Self {
406        self.config.reference_table_max_rows = rows;
407        self
408    }
409
410    /// Set each local reference table's retained-memory charge limit (default 256 MiB).
411    /// See [`LaminarConfig::reference_table_max_bytes`] for the accounting boundary.
412    /// Zero is rejected at build time.
413    #[must_use]
414    pub fn reference_table_max_bytes(mut self, bytes: usize) -> Self {
415        self.config.reference_table_max_bytes = bytes;
416        self
417    }
418
419    /// Set the live row limit for each local materialized view; must be nonzero.
420    /// Multisets count distinct rows. Defaults to 1,000,000.
421    #[must_use]
422    pub fn materialized_view_max_rows(mut self, rows: usize) -> Self {
423        self.config.materialized_view_max_rows = rows;
424        self
425    }
426
427    /// Set each local MV's retained-memory charge limit; must be nonzero.
428    /// See [`LaminarConfig::materialized_view_max_bytes`] for the accounting boundary.
429    #[must_use]
430    pub fn materialized_view_max_bytes(mut self, bytes: usize) -> Self {
431        self.config.materialized_view_max_bytes = bytes;
432        self
433    }
434
435    /// Set the byte budget of committed view state decoded for reads and cached by digest
436    /// (default 256 MiB). Zero disables the cache.
437    #[must_use]
438    pub fn materialized_view_read_cache_bytes(mut self, bytes: usize) -> Self {
439        self.config.materialized_view_read_cache_bytes = bytes;
440        self
441    }
442
443    /// Per-port operator input-buffer cap in batches (default 256).
444    #[must_use]
445    pub fn pipeline_max_input_buf_batches(mut self, batches: usize) -> Self {
446        self.config.pipeline_max_input_buf_batches = Some(batches);
447        self
448    }
449
450    /// Per-port retained Arrow-byte cap, including source priming (disabled by default).
451    /// Each fan-out port charges shared backing storage independently. An executed result
452    /// that cannot fit halts the pipeline before publication; it is never rerun in place.
453    #[must_use]
454    pub fn pipeline_max_input_buf_bytes(mut self, bytes: usize) -> Self {
455        self.config.pipeline_max_input_buf_bytes = Some(bytes);
456        self
457    }
458
459    /// Pipeline-wide charged-byte limit for managed operator working state.
460    ///
461    /// This execution-memory envelope is independent of checkpoint storage.
462    #[must_use]
463    pub fn pipeline_max_managed_state_bytes(mut self, bytes: usize) -> Self {
464        self.config.pipeline_max_managed_state_bytes = Some(bytes);
465        self
466    }
467
468    /// Retain temporal right-side history across idle periods for at least this duration.
469    #[must_use]
470    pub fn temporal_join_idle_history_retention(mut self, retention: std::time::Duration) -> Self {
471        self.config.temporal_join_idle_history_retention = Some(retention);
472        self
473    }
474
475    /// Mark inactive watermarked sources and input channels idle after this duration.
476    #[must_use]
477    pub fn source_idle_timeout(mut self, timeout: std::time::Duration) -> Self {
478        self.config.source_idle_timeout = Some(timeout);
479        self
480    }
481
482    /// Event timestamps farther ahead of wall clock do not advance source watermarks.
483    /// Zero disables the guard.
484    #[must_use]
485    pub fn event_time_max_future_skew(mut self, skew: std::time::Duration) -> Self {
486        self.config.event_time_max_future_skew = skew;
487        self
488    }
489
490    /// Backpressure policy (default `Backpressure`). `ShedOldest` is `BestEffort` only.
491    #[must_use]
492    pub fn pipeline_backpressure_policy(
493        mut self,
494        policy: crate::config::BackpressurePolicy,
495    ) -> Self {
496        self.config.pipeline_backpressure_policy = policy;
497        self
498    }
499
500    /// Auto-restart policy used when supervision is enabled.
501    #[must_use]
502    pub fn restart_policy(mut self, policy: crate::config::RestartPolicy) -> Self {
503        self.config.restart_policy = policy;
504        self
505    }
506
507    /// Register custom connectors; the callback runs after built-ins are wired and must propagate
508    /// registration errors. The registry is frozen before [`Self::build`] returns.
509    #[must_use]
510    pub fn register_connector(
511        mut self,
512        f: impl FnOnce(
513                &laminar_connectors::registry::ConnectorRegistry,
514            ) -> Result<(), laminar_connectors::error::ConnectorError>
515            + Send
516            + 'static,
517    ) -> Self {
518        self.connector_callbacks.push(Box::new(f));
519        self
520    }
521
522    /// Build the `LaminarDB` instance.
523    ///
524    /// # Errors
525    ///
526    /// Returns `DbError` if database creation fails.
527    // COMPAT: `build` remains async as part of the public API even when construction has no I/O.
528    #[allow(unknown_lints, clippy::unused_async, clippy::unused_async_trait_impl)]
529    pub async fn build(mut self) -> Result<Arc<LaminarDB>, DbError> {
530        let has_checkpoint_data_dir = self
531            .config
532            .checkpoint
533            .as_ref()
534            .and_then(|checkpoint| checkpoint.data_dir.as_ref())
535            .is_some();
536        if self.object_store_url.is_some() && has_checkpoint_data_dir {
537            return Err(DbError::Config(
538                "object_store_url conflicts with checkpoint.data_dir; configure exactly one checkpoint store"
539                    .into(),
540            ));
541        }
542        #[cfg(feature = "cluster")]
543        if self.has_cluster_checkpoint_store() && self.object_store_url.is_some() {
544            return Err(DbError::Config(
545                "verified cluster namespaces conflict with object_store_url; configure exactly one checkpoint object store"
546                    .into(),
547            ));
548        }
549        #[cfg(feature = "cluster")]
550        if self.has_cluster_checkpoint_store() && has_checkpoint_data_dir {
551            return Err(DbError::Config(
552                "verified cluster namespaces conflict with checkpoint.data_dir; configure exactly one checkpoint store"
553                    .into(),
554            ));
555        }
556
557        self.config.object_store_url = self.object_store_url.take();
558        self.config.object_store_options = std::mem::take(&mut self.object_store_options);
559        if let Some(url) = self.config.object_store_url.as_deref().filter(|url| {
560            laminar_core::storage_location::StorageProvider::detect_uri(url)
561                == Some(laminar_core::storage_location::StorageProvider::Local)
562        }) {
563            laminar_core::checkpoint::object_store_builder::file_url_path(url)
564                .map_err(|error| DbError::Config(format!("checkpoint storage URL: {error}")))?;
565        }
566
567        if !self.profile_explicit {
568            self.profile = Profile::from_config(&self.config, false);
569        }
570
571        let runtime_mode = self.resolve_runtime_mode()?;
572        if runtime_mode.is_cluster() && !self.profile_explicit {
573            self.profile = Profile::Cluster;
574        }
575        if runtime_mode.is_cluster() && !self.delivery_explicit {
576            self.config.delivery_guarantee =
577                laminar_connectors::connector::DeliveryGuarantee::AtLeastOnce;
578        }
579
580        self.profile
581            .validate_features()
582            .map_err(|e| DbError::Config(e.to_string()))?;
583        Self::validate_cluster_delivery(runtime_mode, self.config.delivery_guarantee)?;
584        #[cfg(feature = "cluster")]
585        if runtime_mode == RuntimeMode::Cluster {
586            if self.config.object_store_url.is_some() || has_checkpoint_data_dir {
587                return Err(DbError::Config(
588                    "[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"
589                        .into(),
590                ));
591            }
592            if !self.has_cluster_checkpoint_store() {
593                return Err(DbError::Config(
594                    "[LDB-0011] cluster runtime requires VerifiedClusterNamespaces from the shared checkpoint namespace proof"
595                        .into(),
596                ));
597            }
598            if let Some(namespaces) = &self.verified_cluster_namespaces {
599                let controller = self.cluster_controller.as_ref().ok_or_else(|| {
600                    DbError::Config("cluster namespace proof requires a cluster controller".into())
601                })?;
602                let expected = laminar_core::checkpoint::CheckpointParticipant {
603                    node_id: controller.instance_id().0,
604                    boot_incarnation: controller.recovery_incarnation(),
605                };
606                if namespaces.local_participant() != expected {
607                    let proved = namespaces.local_participant();
608                    return Err(DbError::Config(format!(
609                        "[LDB-0011] verified cluster namespaces belong to node {} boot {}, but the controller is node {} boot {}",
610                        proved.node_id,
611                        proved.boot_incarnation,
612                        expected.node_id,
613                        expected.boot_incarnation,
614                    )));
615                }
616            }
617        }
618        #[cfg(feature = "cluster")]
619        let profile_object_store = self.config.object_store_url.as_deref().or_else(|| {
620            self.has_cluster_checkpoint_store()
621                .then_some("verified-cluster-checkpoint-store")
622        });
623        #[cfg(not(feature = "cluster"))]
624        let profile_object_store = self.config.object_store_url.as_deref();
625        self.profile
626            .validate_config(&self.config, profile_object_store)
627            .map_err(|e| DbError::Config(e.to_string()))?;
628
629        if runtime_mode == RuntimeMode::Local
630            && self.config.delivery_guarantee
631                != laminar_connectors::connector::DeliveryGuarantee::BestEffort
632            && self.config.object_store_url.as_deref().is_some_and(|url| {
633                laminar_core::storage_location::StorageProvider::detect_uri(url)
634                    != Some(laminar_core::storage_location::StorageProvider::Local)
635            })
636        {
637            return Err(DbError::Config(
638                "[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"
639                    .into(),
640            ));
641        }
642
643        self.config.validate_backpressure_policy()?;
644        self.validate_vnode_topology(runtime_mode)?;
645        if let Some(key_groups) = self
646            .key_groups
647            .filter(|_| runtime_mode == RuntimeMode::Local && self.vnode_registry.is_none())
648        {
649            self.vnode_registry = Some(Arc::new(laminar_core::state::VnodeRegistry::single_owner(
650                u32::from(key_groups),
651                laminar_core::state::LOCAL_NODE_ID,
652            )));
653        }
654        #[cfg(feature = "cluster")]
655        self.bind_cluster_process_lease(runtime_mode)?;
656
657        self.profile.apply_defaults(&mut self.config);
658
659        #[cfg(feature = "cluster")]
660        let cluster_checkpoint_store = self.cluster_checkpoint_store();
661        let mut db = LaminarDB::open_with_config_and_vars_and_rules(
662            self.config,
663            self.config_vars,
664            &self.physical_optimizer_rules,
665            self.target_partitions,
666            runtime_mode,
667        )?;
668        for udf in self.custom_udfs {
669            db.register_custom_udf(udf)?;
670        }
671        for udaf in self.custom_udafs {
672            db.register_custom_udaf(udaf)?;
673        }
674        if let Some(runtime) = self.ai_runtime {
675            let handle = tokio::runtime::Handle::try_current().map_err(|_| {
676                DbError::InvalidOperation(
677                    "LaminarDB::build() with an AI runtime must run inside a Tokio runtime"
678                        .to_string(),
679                )
680            })?;
681            db.set_ai_runtime(runtime, handle);
682        }
683        for callback in self.connector_callbacks {
684            callback(db.connector_registry())?;
685        }
686        db.connector_registry().freeze();
687        #[cfg(feature = "cluster")]
688        if let Some(controller) = self.cluster_controller {
689            db.set_cluster_controller(controller)?;
690        }
691        #[cfg(feature = "cluster")]
692        if let Some(store) = cluster_checkpoint_store {
693            db.set_cluster_checkpoint_object_store(store)?;
694        }
695        #[cfg(feature = "cluster")]
696        if let Some(sender) = self.shuffle_sender {
697            db.set_shuffle_sender(sender);
698        }
699        #[cfg(feature = "cluster")]
700        if let Some(receiver) = self.shuffle_receiver {
701            db.set_shuffle_receiver(receiver);
702        }
703        #[cfg(feature = "cluster")]
704        if let Some(store) = self.decision_store {
705            db.set_decision_store(store);
706        }
707        #[cfg(feature = "cluster")]
708        if let Some(store) = self.assignment_snapshot_store {
709            db.set_assignment_snapshot_store(store);
710        }
711        #[cfg(feature = "cluster")]
712        if let Some(store) = self.catalog_manifest_store {
713            db.set_catalog_manifest_store(store);
714        }
715        if let Some(registry) = self.vnode_registry {
716            db.set_vnode_registry(registry);
717        }
718        Ok(Arc::new(db))
719    }
720
721    fn validate_vnode_topology(&self, runtime_mode: RuntimeMode) -> Result<(), DbError> {
722        let registry_key_groups = self
723            .vnode_registry
724            .as_ref()
725            .map(|registry| {
726                let registry_count = registry.vnode_count();
727                laminar_core::state::KeyGroupCount::try_from(registry_count).map_err(|_| {
728                    DbError::Config(format!(
729                        "vnode_registry count must be between 1 and {}, got {registry_count}",
730                        laminar_core::state::MAX_KEY_GROUP_COUNT
731                    ))
732                })
733            })
734            .transpose()?;
735        if let Some(registry) = self
736            .vnode_registry
737            .as_ref()
738            .filter(|_| runtime_mode == RuntimeMode::Local)
739        {
740            let assignment = registry.versioned_snapshot();
741            if let Some((vnode, owner)) = assignment
742                .owners()
743                .iter()
744                .copied()
745                .enumerate()
746                .find(|(_, owner)| *owner != laminar_core::state::LOCAL_NODE_ID)
747            {
748                return Err(DbError::Config(format!(
749                    "local vnode {vnode} must be owned by {}, got {owner}",
750                    laminar_core::state::LOCAL_NODE_ID.0
751                )));
752            }
753        }
754        if let (Some(configured), Some(registry)) = (self.key_groups, registry_key_groups) {
755            if configured != registry {
756                return Err(DbError::Config(format!(
757                    "configured key-group count {configured} does not match vnode_registry count {registry}"
758                )));
759            }
760        }
761        Ok(())
762    }
763
764    #[cfg(feature = "cluster")]
765    fn bind_cluster_process_lease(&self, runtime_mode: RuntimeMode) -> Result<(), DbError> {
766        if !runtime_mode.is_cluster() {
767            return Ok(());
768        }
769        let controller = self.cluster_controller.as_ref().ok_or_else(|| {
770            DbError::Config("cluster runtime requires a cluster controller".into())
771        })?;
772        let deadline = controller.process_lease_deadline().ok_or_else(|| {
773            DbError::Config(
774                "cluster runtime requires one shared process lease deadline before construction"
775                    .into(),
776            )
777        })?;
778        if !deadline.is_live() {
779            return Err(DbError::Config(
780                "cluster runtime process lease deadline is already expired".into(),
781            ));
782        }
783        match (&self.shuffle_sender, &self.shuffle_receiver) {
784            (Some(sender), Some(receiver)) => {
785                sender
786                    .bind_process_lease_deadline_pair(receiver, deadline)
787                    .map_err(|error| {
788                        DbError::Config(format!(
789                            "shuffle process lease does not match the controller: {error}"
790                        ))
791                    })?;
792            }
793            (None, None) => {}
794            _ => {
795                return Err(DbError::Config(
796                    "cluster shuffle sender and receiver must be installed together".into(),
797                ));
798            }
799        }
800        Ok(())
801    }
802
803    #[cfg(feature = "cluster")]
804    fn has_cluster_checkpoint_store(&self) -> bool {
805        let has_store = self.verified_cluster_namespaces.is_some();
806        #[cfg(test)]
807        let has_store = has_store || self.unchecked_cluster_checkpoint_store.is_some();
808        has_store
809    }
810
811    #[cfg(feature = "cluster")]
812    fn cluster_checkpoint_store(&self) -> Option<Arc<dyn object_store::ObjectStore>> {
813        let store = self
814            .verified_cluster_namespaces
815            .as_ref()
816            .map(laminar_core::cluster::control::VerifiedClusterNamespaces::checkpoint_store);
817        #[cfg(test)]
818        let store = store.or_else(|| self.unchecked_cluster_checkpoint_store.clone());
819        store
820    }
821
822    /// Resolve the distributed execution scope exactly once and reject partial or contradictory
823    /// cluster wiring before constructing the database.
824    fn resolve_runtime_mode(&self) -> Result<RuntimeMode, DbError> {
825        #[cfg(not(feature = "cluster"))]
826        {
827            if self.profile == Profile::Cluster {
828                return Err(DbError::Config(
829                    "Profile::Cluster requires a cluster controller and a cluster-enabled build"
830                        .into(),
831                ));
832            }
833            Ok(RuntimeMode::Local)
834        }
835
836        #[cfg(feature = "cluster")]
837        {
838            let has_controller = self.cluster_controller.is_some();
839            let has_cluster_only_handle = self.shuffle_sender.is_some()
840                || self.shuffle_receiver.is_some()
841                || self.decision_store.is_some()
842                || self.assignment_snapshot_store.is_some()
843                || self.catalog_manifest_store.is_some()
844                || self.has_cluster_checkpoint_store();
845
846            if has_cluster_only_handle && !has_controller {
847                return Err(DbError::Config(
848                    "cluster-only stores and shuffle handles require a cluster controller".into(),
849                ));
850            }
851            if self.profile == Profile::Cluster && !has_controller {
852                return Err(DbError::Config(
853                    "Profile::Cluster requires a cluster controller".into(),
854                ));
855            }
856            if self.profile_explicit && self.profile != Profile::Cluster && has_controller {
857                return Err(DbError::Config(format!(
858                    "profile {} cannot be combined with a cluster controller; select Profile::Cluster or omit the explicit profile",
859                    self.profile
860                )));
861            }
862
863            Ok(if has_controller {
864                RuntimeMode::Cluster
865            } else {
866                RuntimeMode::Local
867            })
868        }
869    }
870
871    /// Validate delivery semantics whose correctness depends on the runtime mode.
872    ///
873    /// Connector-specific exact-delivery certification is checked before connector I/O, when the
874    /// concrete source and sink contracts are available.
875    fn validate_cluster_delivery(
876        runtime_mode: RuntimeMode,
877        guarantee: laminar_connectors::connector::DeliveryGuarantee,
878    ) -> Result<(), DbError> {
879        use laminar_connectors::connector::DeliveryGuarantee;
880
881        if !runtime_mode.is_cluster() {
882            return Ok(());
883        }
884        match guarantee {
885            DeliveryGuarantee::BestEffort => Err(DbError::Config(
886                "cluster mode requires at_least_once delivery; best_effort has no defined \
887                 rebalance/state-loss contract"
888                    .into(),
889            )),
890            DeliveryGuarantee::AtLeastOnce | DeliveryGuarantee::ExactlyOnce => Ok(()),
891        }
892    }
893}
894
895impl Default for LaminarDbBuilder {
896    fn default() -> Self {
897        Self::new()
898    }
899}
900
901impl std::fmt::Debug for LaminarDbBuilder {
902    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
903        f.debug_struct("LaminarDbBuilder")
904            .field("config", &self.config)
905            .field("profile", &self.profile)
906            .field("profile_explicit", &self.profile_explicit)
907            .field("delivery_explicit", &self.delivery_explicit)
908            .field("key_groups", &self.key_groups)
909            .field("object_store_url", &self.object_store_url)
910            .field(
911                "object_store_options_count",
912                &self.object_store_options.len(),
913            )
914            .field("config_vars_count", &self.config_vars.len())
915            .field("connector_callbacks", &self.connector_callbacks.len())
916            .field("custom_udfs", &self.custom_udfs.len())
917            .field("custom_udafs", &self.custom_udafs.len())
918            .finish_non_exhaustive()
919    }
920}
921
922#[cfg(test)]
923mod tests;
laminar_db/builder/mod.rs

laminar_db/builder/
mod.rs
