laminar_db/db/topology/
planning.rs

1//! Local topology validation against an authority-audited parent. No durable operation is admitted.
2
3use std::collections::BTreeMap;
4use std::sync::atomic::Ordering;
5use std::sync::Arc;
6
7use laminar_core::checkpoint::PipelineIdentity;
8use laminar_core::cluster::control::{
9    CatalogManifest, CatalogManifestEntry, CatalogObjectKind, TopologyCatalogState, TopologyError,
10    TopologyVersion,
11};
12
13use crate::db::{DbState, LaminarDB, RuntimeMode};
14use crate::error::DbError;
15use crate::exchange_plan::ExchangePlan;
16use crate::operator::capability::{ManagedStateContract, OperatorStateClass};
17use crate::pipeline_identity::{
18    compatibility_digest, compatibility_identities, PipelineCompatibilityIdentities,
19    PipelineIdentityContext, PipelineRegistrations,
20};
21use crate::pipeline_lifecycle::PlannedTopologyGraph;
22
23const VALIDATION_FORMAT_VERSION: u16 = 3;
24pub(super) const MAX_VALIDATION_OBJECTS: usize = 256;
25const MAX_VALIDATION_STATEMENTS: usize = 64;
26const MAX_VALIDATION_SQL_BYTES: usize = 256 * 1024;
27
28pub use laminar_core::cluster::control::topology::{
29    ClusterTopologyObjectPlan, ClusterTopologyObjectTransition, ClusterTopologyValidation,
30    TopologyActivationRequirement, TopologyInitialization, TopologyValidationScope,
31};
32
33impl LaminarDB {
34    /// Compile a topology candidate separately from the active graph.
35    ///
36    /// Each entry is one typed CREATE, CREATE OR REPLACE or DROP SOURCE/STREAM/SINK without CASCADE.
37    /// Replacement preserves compatible state by default. Ordered DROP/CREATE explicitly resets
38    /// the affected objects to future-only input under new incarnations; reset dependents first.
39    /// Supports replayable sources, supported managed/stateless streams and durable sinks while
40    /// proving unchanged definitions remain compatible. New managed state starts empty at the cut;
41    /// input that can retract an unavailable prefix is rejected. No source is opened
42    /// or polled, no sink is opened or published, and no authority/checkpoint ID is written.
43    /// A successful response does not authorize SQL submission, restore or target activation.
44    ///
45    /// One compiler per process, 64 statements/256 KiB input, 256 parent/target object mappings, a 1 MiB
46    /// descriptor and a 30 second end-to-end deadline bound control-path resource use.
47    ///
48    /// # Errors
49    /// Rejects parent conflicts, unsupported mutations, uncertified execution or
50    /// connector contracts, divergent local definitions, malformed input and exceeded bounds.
51    pub async fn validate_cluster_topology_change(
52        &self,
53        expected_parent: TopologyVersion,
54        statements: &[String],
55    ) -> Result<ClusterTopologyValidation, DbError> {
56        self.plan_cluster_topology_change(expected_parent, statements)
57            .await
58            .map(|(report, _)| report)
59    }
60
61    pub(in crate::db) async fn plan_cluster_topology_change(
62        &self,
63        expected_parent: TopologyVersion,
64        statements: &[String],
65    ) -> Result<(ClusterTopologyValidation, CatalogManifest), DbError> {
66        validate_request_bounds(statements)?;
67        if !self.is_cluster_runtime() {
68            return Err(TopologyError::Unsupported(
69                "topology validation requires cluster mode".into(),
70            )
71            .into());
72        }
73        if !self.connector_manager.lock().process_functions().is_empty() {
74            return Err(TopologyError::Unsupported(
75                "process bindings are immutable; catalog changes require a new checkpoint namespace"
76                    .into(),
77            )
78            .into());
79        }
80        let _compiler = self
81            .topology_validation_lock
82            .try_lock()
83            .map_err(|_| TopologyError::PlanningBusy)?;
84        tokio::time::timeout(
85            std::time::Duration::from_secs(30),
86            Box::pin(self.validate_topology_candidate(expected_parent, statements)),
87        )
88        .await
89        .map_err(|_| TopologyError::PlanningTimedOut)?
90    }
91
92    async fn validate_topology_candidate(
93        &self,
94        expected_parent: TopologyVersion,
95        statements: &[String],
96    ) -> Result<(ClusterTopologyValidation, CatalogManifest), DbError> {
97        let active_inventory = {
98            let _catalog_read = self.topology_ddl_lock.read().await;
99            self.ensure_validation_catalog_available()?;
100            self.catalog_manifest_inventory()?
101        };
102        let store = self.catalog_manifest_store.lock().clone().ok_or_else(|| {
103            TopologyError::Protocol(
104                "topology validation requires a configured catalog authority".into(),
105            )
106        })?;
107        let (parent, state) = store.load_with_topology().await?.ok_or_else(|| {
108            TopologyError::Conflict("no sealed parent catalog; initialize and explicitly adopt the legacy baseline first".into())
109        })?;
110        let TopologyCatalogState::Versioned {
111            baseline,
112            committed,
113        } = &state
114        else {
115            return Err(TopologyError::Protocol(
116                "legacy catalog must be explicitly adopted before topology validation".into(),
117            )
118            .into());
119        };
120        if state.committed_version() != Some(expected_parent) {
121            return Err(TopologyError::Conflict(format!(
122                "expected parent {expected_parent:?}, authority has {:?}",
123                state.committed_version()
124            ))
125            .into());
126        }
127        if parent.entries.len() > MAX_VALIDATION_OBJECTS {
128            return Err(TopologyError::Unsupported(format!(
129                "candidate exceeds the {MAX_VALIDATION_OBJECTS} object planning bound"
130            ))
131            .into());
132        }
133        if active_inventory != parent.entries {
134            return Err(TopologyError::Conflict("local catalog is not the exact committed parent inventory; complete catalog replay before validation".into()).into());
135        }
136        let active_identities = self.topology_definition_identities().await?;
137        self.validate_bound_parent_pipeline(&active_identities.pipeline)
138            .await?;
139        let candidate = self.isolated_topology_catalog()?;
140        for entry in &parent.entries {
141            replay_entry(&candidate, entry).await?;
142        }
143        if candidate.reconcile_catalog_manifest_inventory(&parent)? != parent.entries {
144            return Err(TopologyError::Invalid(
145                "isolated parent replay changed its exact inventory or incarnations".into(),
146            )
147            .into());
148        }
149        let parent_identities = candidate.topology_definition_identities().await?;
150        if parent_identities != active_identities {
151            return Err(TopologyError::Conflict("resolved parent definitions differ from the live catalog; check non-secret environment/config references and connector contracts".into()).into());
152        }
153        let parent_graph = candidate.plan_topology_graph().await?;
154        let parent_objects = describe_catalog(
155            &candidate,
156            &parent,
157            &parent_identities,
158            &parent_graph,
159            &parent,
160        )?;
161        drop(parent_graph);
162        let retired = store.retired_topology_generations().await?;
163        let target = super::catalog_changes::apply_catalog_changes(
164            &candidate, &parent, statements, &retired,
165        )
166        .await?;
167        let target_identities = candidate.topology_definition_identities().await?;
168        let target_graph = candidate.plan_topology_graph().await?;
169        super::catalog_changes::validate_prepared_targets(&candidate, &target)?;
170        let objects = describe_catalog(
171            &candidate,
172            &target,
173            &target_identities,
174            &target_graph,
175            &parent,
176        )?;
177        if parent_identities.environment_sha256 != target_identities.environment_sha256 {
178            return Err(TopologyError::Unsupported(
179                "candidate changes the global state/routing/delivery ABI".into(),
180            )
181            .into());
182        }
183        let objects = classify_object_transitions(&parent_objects, objects, statements)?;
184        // No stale response can masquerade as validation against a newer parent. Normal lease
185        // renewals/checkpoints do not invalidate a read-only plan; the exact catalog must match.
186        if store.topology_state().await? != state
187            || self.topology_definition_identities().await? != active_identities
188        {
189            return Err(TopologyError::Conflict("parent authority or local definitions changed during validation; retry against the observed parent".into()).into());
190        }
191        self.ensure_validation_catalog_available()?;
192        self.validate_bound_parent_pipeline(&active_identities.pipeline)
193            .await?;
194        let target_version = expected_parent.successor()?;
195        let target_manifest = target.reference().map_err(TopologyError::from)?;
196        let mut report = ClusterTopologyValidation {
197            validation_format_version: VALIDATION_FORMAT_VERSION,
198            scope: TopologyValidationScope::LocalCandidatePlan,
199            deployment_id: baseline.deployment_id.clone(),
200            parent_version: expected_parent,
201            target_version,
202            parent_manifest: committed.as_ref().map_or_else(
203                || baseline.manifest.clone(),
204                |commit| commit.manifest.clone(),
205            ),
206            target_manifest,
207            parent_pipeline: parent_identities.pipeline,
208            target_pipeline: target_identities.pipeline,
209            environment_sha256: parent_identities.environment_sha256,
210            compatibility_sha256: String::new(),
211            statements: statements.to_vec(),
212            objects,
213            requires_processing_pause: true,
214            required_before_activation: vec![
215                TopologyActivationRequirement::ParticipantPlanAgreement,
216                TopologyActivationRequirement::ReconciledCheckpointCut,
217                TopologyActivationRequirement::DurableInitializationAndProgress,
218                TopologyActivationRequirement::ObservedActorRetirement,
219                TopologyActivationRequirement::AtomicTargetCommit,
220                TopologyActivationRequirement::InstalledTargetRelease,
221            ],
222        };
223        report.compatibility_sha256 = report.descriptor_digest()?;
224        report.validate_catalogs(&parent, &target)?;
225        Ok((report, target))
226    }
227
228    fn ensure_validation_catalog_available(&self) -> Result<(), DbError> {
229        if self.shutdown.load(Ordering::Acquire) {
230            return Err(DbError::Shutdown);
231        }
232        if !matches!(
233            DbState::load(&self.state),
234            DbState::Created | DbState::Running
235        ) {
236            return Err(TopologyError::Conflict("catalog validation requires a Created or Running database; finish recovery/startup first".into()).into());
237        }
238        self.ensure_catalog_cleanup_unfenced("topology validation")?;
239        if !self.custom_udfs.is_empty()
240            || !self.custom_udafs.is_empty()
241            || !self.physical_optimizer_rules.is_empty()
242        {
243            return Err(TopologyError::Unsupported(
244                "custom function or optimizer implementation identity is not bound by the migration descriptor"
245                    .into(),
246            )
247            .into());
248        }
249        Ok(())
250    }
251
252    pub(in crate::db) fn isolated_topology_catalog(&self) -> Result<LaminarDB, DbError> {
253        let mut candidate = Self::open_with_config_and_vars_and_rules(
254            self.config.clone(),
255            self.config_vars.as_ref().clone(),
256            &self.physical_optimizer_rules,
257            self.pipeline_target_partitions,
258            RuntimeMode::Cluster,
259        )?;
260        candidate.catalog = Arc::new(crate::catalog::SourceCatalog::for_topology_planning(
261            &self.config,
262        ));
263        candidate.connector_registry = Arc::clone(&self.connector_registry);
264        for registration in self.connector_manager.lock().process_functions().values() {
265            candidate
266                .connector_manager
267                .lock()
268                .register_process_function(registration.clone());
269        }
270        candidate.topology_planning_ownership_scope =
271            Some(self.has_cluster_query_ownership_scope());
272        *candidate.vnode_registry.lock() =
273            Some(Arc::new(laminar_core::state::VnodeRegistry::single_owner(
274                u32::from(self.checkpoint_key_groups()),
275                laminar_core::state::LOCAL_NODE_ID,
276            )));
277        Ok(candidate)
278    }
279
280    pub(crate) async fn topology_definition_identities(
281        &self,
282    ) -> Result<PipelineCompatibilityIdentities, DbError> {
283        let exchange_plan = self.cluster_exchange_plan().await?;
284        let vnode_count = self.checkpoint_key_groups().into();
285        let manager = self.connector_manager.lock();
286        compatibility_identities(&PipelineIdentityContext::new(
287            &self.config,
288            &self.catalog,
289            &self.connector_registry,
290            PipelineRegistrations::new(
291                manager.sources().values(),
292                manager.sinks().values(),
293                manager.streams().values(),
294                manager.tables().values(),
295            ),
296            &exchange_plan,
297            vnode_count,
298        ))
299    }
300
301    /// The registered streams' exchange plan exactly as cluster startup resolves it: cluster
302    /// interval admissions and no changelog source.
303    pub(crate) async fn cluster_exchange_plan(&self) -> Result<ExchangePlan, DbError> {
304        let (sources, sinks, streams) = {
305            let manager = self.connector_manager.lock();
306            (
307                manager.sources().clone(),
308                manager.sinks().clone(),
309                manager.streams().clone(),
310            )
311        };
312        let interval = self
313            .validate_persisted_interval_source_contracts(
314                &sources,
315                &sinks,
316                &streams,
317                RuntimeMode::Cluster,
318            )
319            .await?;
320        Ok(crate::pipeline_lifecycle::resolve_stream_output_schemas(
321            &self.ctx,
322            &streams,
323            &self.static_table_names(),
324            &interval.joins,
325            &rustc_hash::FxHashSet::default(),
326            &crate::exchange_plan::source_event_times(&self.catalog),
327        )
328        .await?
329        .exchange_plan)
330    }
331
332    pub(super) async fn validate_bound_parent_pipeline(
333        &self,
334        expected: &PipelineIdentity,
335    ) -> Result<(), DbError> {
336        if DbState::load(&self.state) != DbState::Running {
337            return Ok(());
338        }
339        let coordinator = self.coordinator.lock().await;
340        let bound = coordinator
341            .as_ref()
342            .ok_or_else(|| {
343                TopologyError::Conflict("running parent has no checkpoint coordinator".into())
344            })?
345            .bound_pipeline_identity()?;
346        if &bound != expected {
347            return Err(TopologyError::Conflict(
348                "running checkpoint pipeline identity differs from the planned parent".into(),
349            )
350            .into());
351        }
352        Ok(())
353    }
354}
355
356pub(super) fn validate_request_bounds(statements: &[String]) -> Result<(), DbError> {
357    if statements.is_empty() || statements.len() > MAX_VALIDATION_STATEMENTS {
358        return Err(TopologyError::Invalid(
359            "validation requires 1..=64 individual topology DDL statements".into(),
360        )
361        .into());
362    }
363    let bytes = statements
364        .iter()
365        .try_fold(0_usize, |total, sql| total.checked_add(sql.len()));
366    if bytes.is_none_or(|bytes| bytes > MAX_VALIDATION_SQL_BYTES) {
367        return Err(TopologyError::Invalid("validation SQL exceeds 256 KiB".into()).into());
368    }
369    Ok(())
370}
371
372pub(super) async fn replay_entry(
373    candidate: &LaminarDB,
374    entry: &CatalogManifestEntry,
375) -> Result<(), DbError> {
376    let statement = super::catalog_changes::parse_one_change(&entry.ddl)?;
377    let (name, kind, _) =
378        super::super::validate_cluster_catalog_create(candidate, &entry.ddl, &statement)?;
379    if name != entry.canonical_name || kind != entry.kind {
380        return Err(
381            TopologyError::Invalid("candidate DDL and catalog identity disagree".into()).into(),
382        );
383    }
384    crate::ddl::schema_resolution::RESOLVED_SCHEMA
385        .scope(
386            entry.schema_binding.clone(),
387            super::catalog_changes::apply_statement(candidate, &entry.ddl, &statement, &name),
388        )
389        .await
390}
391
392pub(super) fn describe_catalog(
393    candidate: &LaminarDB,
394    manifest: &CatalogManifest,
395    identities: &PipelineCompatibilityIdentities,
396    graph: &PlannedTopologyGraph,
397    parent: &CatalogManifest,
398) -> Result<BTreeMap<String, ClusterTopologyObjectPlan>, DbError> {
399    let manager = candidate.connector_manager.lock();
400    let mut objects = BTreeMap::<String, ClusterTopologyObjectPlan>::new();
401    for entry in &manifest.entries {
402        let name = &entry.canonical_name;
403        let preserved = parent.entries.iter().any(|old| {
404            &old.canonical_name == name
405                && old.kind == entry.kind
406                && old.catalog_generation == entry.catalog_generation
407        });
408        let dependencies = catalog_dependencies(&manager, entry)?;
409        let dependency_evidence: Vec<_> = dependencies.iter().map(|dependency| {
410            objects.get(dependency).map(|object| (&object.name, object.catalog_generation, &object.compatibility_sha256)).ok_or_else(|| TopologyError::Invalid(format!("'{name}' depends on '{dependency}' outside dependency-safe creation order")))
411        }).collect::<Result<_, _>>()?;
412        let stages = crate::sql_analysis::owned_stages(name, manager.streams().keys());
413        let definition_sha256 = object_definition(identities, graph, name, &stages)?;
414        let schema_sha256 = graph
415            .schemas
416            .get(name)
417            .map(|schema| {
418                crate::pipeline_identity::subscription_schema_fingerprint(schema)
419                    .map(laminar_core::checkpoint::SubscriptionDigest::to_hex)
420            })
421            .transpose()?;
422        let operator = graph.operators.get(name);
423        // Append views keep their rows in a log, not in a result node.
424        let view_result = (entry.kind == CatalogObjectKind::MaterializedView)
425            .then(|| laminar_core::mv::view_result_operator(name))
426            .filter(|operator| graph.operators.contains_key(operator));
427        let operators: Vec<_> = std::iter::once(name)
428            .chain(view_result.as_ref())
429            .chain(&stages)
430            .map(|operator| (operator, graph.operators.get(operator)))
431            .collect();
432        if let Some((missing, _)) = operators.iter().find(|(_, operator)| operator.is_none()) {
433            if matches!(
434                entry.kind,
435                CatalogObjectKind::Stream | CatalogObjectKind::MaterializedView
436            ) {
437                return Err(TopologyError::Invalid(format!(
438                    "stream '{missing}' is absent from the physical candidate graph"
439                ))
440                .into());
441            }
442        }
443        let state_contract = operators
444            .iter()
445            .find_map(|(_, operator)| operator.and_then(|capability| capability.managed_state))
446            .map(state_contract_name);
447        let seeded = if preserved {
448            Vec::new()
449        } else {
450            seeded_upstreams(name, entry.kind, &dependencies, &objects, graph)?
451        };
452        if !preserved {
453            for (operator_name, operator) in &operators {
454                validate_future_state_inputs(
455                    operator_name,
456                    *operator,
457                    &dependencies,
458                    &seeded,
459                    &objects,
460                    graph,
461                )?;
462            }
463        }
464        let identity = compatibility_digest(&(
465            "laminardb-topology-object-mapping-v1",
466            name,
467            entry.kind,
468            entry.catalog_generation,
469            &identities.environment_sha256,
470            &definition_sha256,
471            &schema_sha256,
472            operator,
473            graph.connector_sha256.get(name),
474            dependency_evidence,
475        ))?;
476        objects.insert(
477            name.clone(),
478            ClusterTopologyObjectPlan {
479                name: name.clone(),
480                kind: entry.kind,
481                catalog_generation: entry.catalog_generation,
482                transition: if preserved {
483                    ClusterTopologyObjectTransition::Preserve
484                } else {
485                    ClusterTopologyObjectTransition::AddFutureOnly
486                },
487                initialization: if preserved {
488                    TopologyInitialization::PreserveExactCut
489                } else if entry.kind == CatalogObjectKind::Source {
490                    TopologyInitialization::ResolveSourcePositionsOnce
491                } else if state_contract.is_some() && !seeded.is_empty() {
492                    TopologyInitialization::SeedFromUpstreamAtCut
493                } else if state_contract.is_some() {
494                    TopologyInitialization::EmptyManagedStateAtCut
495                } else {
496                    TopologyInitialization::FutureOnlyAtCut
497                },
498                definition_sha256,
499                compatibility_sha256: identity,
500                dependencies,
501                schema_sha256,
502                managed_state_contract: state_contract.map(str::to_owned),
503            },
504        );
505    }
506    Ok(objects)
507}
508
509fn catalog_dependencies(
510    manager: &crate::connector_manager::ConnectorManager,
511    entry: &CatalogManifestEntry,
512) -> Result<Vec<String>, DbError> {
513    let name = &entry.canonical_name;
514    let mut dependencies = match entry.kind {
515        CatalogObjectKind::Source => Vec::new(),
516        CatalogObjectKind::Sink => vec![manager
517            .sinks()
518            .get(name)
519            .ok_or_else(|| TopologyError::Invalid(format!("sink '{name}' has no registration")))?
520            .input
521            .clone()],
522        CatalogObjectKind::Stream | CatalogObjectKind::MaterializedView => {
523            // A split stream's inputs are those of its stages.
524            let mut references = std::collections::BTreeSet::new();
525            let mut pending = vec![name.clone()];
526            let mut visited = std::collections::BTreeSet::new();
527            while let Some(stream) = pending.pop() {
528                if !visited.insert(stream.clone()) {
529                    continue;
530                }
531                for reference in stream_references(manager, &stream)? {
532                    if crate::sql_analysis::stage_owner(&reference) == Some(name.as_str()) {
533                        pending.push(reference);
534                    } else {
535                        references.insert(reference);
536                    }
537                }
538            }
539            if references.is_empty() {
540                return Err(TopologyError::Unsupported(format!(
541                    "stream '{name}' has no explicit input activation boundary"
542                ))
543                .into());
544            }
545            references.into_iter().collect()
546        }
547        _ => {
548            return Err(TopologyError::Unsupported(format!(
549                "'{name}' has an unsupported catalog kind for topology validation"
550            ))
551            .into())
552        }
553    };
554    dependencies.sort_unstable();
555    Ok(dependencies)
556}
557
558fn stream_references(
559    manager: &crate::connector_manager::ConnectorManager,
560    name: &str,
561) -> Result<rustc_hash::FxHashSet<String>, DbError> {
562    let stream = manager
563        .streams()
564        .get(name)
565        .ok_or_else(|| TopologyError::Invalid(format!("stream '{name}' has no registration")))?;
566    let mut references = crate::sql_analysis::extract_table_references(&stream.query_sql);
567    for join in stream.join_config.iter().flatten() {
568        match join {
569            laminar_sql::translator::JoinOperatorConfig::StreamStream(config) => {
570                references.insert(config.left_table.clone());
571                references.insert(config.right_table.clone());
572            }
573            laminar_sql::translator::JoinOperatorConfig::Temporal(config) => {
574                references.insert(config.left_table.clone());
575                references.insert(config.right_table.clone());
576            }
577            laminar_sql::translator::JoinOperatorConfig::Lookup(_) => {
578                return Err(TopologyError::Unsupported(format!(
579                    "stream '{name}' has an unmapped join dependency contract"
580                ))
581                .into())
582            }
583        }
584    }
585    Ok(references)
586}
587
588/// A split stream's stages initialize, preserve and retire with it, so its definition binds theirs.
589fn object_definition(
590    identities: &PipelineCompatibilityIdentities,
591    graph: &PlannedTopologyGraph,
592    name: &str,
593    stages: &[String],
594) -> Result<String, DbError> {
595    let definition = |object: &str| {
596        identities.objects.get(object).ok_or_else(|| {
597            TopologyError::Invalid(format!(
598                "'{object}' is absent from canonical pipeline identity"
599            ))
600        })
601    };
602    let owner = definition(name)?;
603    if stages.is_empty() {
604        return Ok(owner.clone());
605    }
606    let stages = stages
607        .iter()
608        .map(|stage| Ok((stage, definition(stage)?, graph.operators.get(stage))))
609        .collect::<Result<Vec<_>, DbError>>()?;
610    compatibility_digest(&("laminardb-split-stream-definition-v1", owner, stages))
611}
612
613fn classify_object_transitions(
614    parent: &BTreeMap<String, ClusterTopologyObjectPlan>,
615    target: BTreeMap<String, ClusterTopologyObjectPlan>,
616    statements: &[String],
617) -> Result<Vec<ClusterTopologyObjectPlan>, DbError> {
618    let replacements = statements
619        .iter()
620        .filter_map(|sql| match super::catalog_changes::parse_one_change(sql) {
621            Ok(statement) if super::catalog_changes::is_replacement(&statement) => Some(
622                super::super::catalog_create_identity(&statement).and_then(|identity| {
623                    identity.map(|(name, _, _)| name).ok_or_else(|| {
624                        TopologyError::Invalid("replacement has no typed identity".into()).into()
625                    })
626                }),
627            ),
628            Ok(_) => None,
629            Err(error) => Some(Err(error)),
630        })
631        .collect::<Result<std::collections::BTreeSet<_>, DbError>>()?;
632    let mut retired = Vec::new();
633    for (name, previous) in parent {
634        match target.get(name) {
635            Some(current) if current == previous => {}
636            Some(current) if current.transition == ClusterTopologyObjectTransition::Preserve => {
637                // A stateless query can change future computation with an unchanged schema and
638                // dependency closure. Managed state and connectors require the exact definition.
639                let compatible = replacements.contains(name)
640                    && previous.kind == CatalogObjectKind::Stream
641                    && previous.managed_state_contract.is_none()
642                    && current.managed_state_contract.is_none()
643                    && current.schema_sha256 == previous.schema_sha256
644                    && current.dependencies == previous.dependencies
645                    && current.dependencies.iter().all(|dependency| {
646                        parent
647                            .get(dependency)
648                            .zip(target.get(dependency))
649                            .is_some_and(|(old, new)| {
650                                old.compatibility_sha256 == new.compatibility_sha256
651                            })
652                    });
653                if !compatible {
654                    return Err(TopologyError::Unsupported(format!(
655                        "preserved object '{name}' changed its resolved state, schema, connector or dependency contract; explicitly DROP/CREATE it and its affected dependents for a future-only reset"
656                    )).into());
657                }
658            }
659            _ => {
660                let mut removed = previous.clone();
661                removed.transition = ClusterTopologyObjectTransition::Remove;
662                removed.initialization = TopologyInitialization::RetireAtCut;
663                retired.push(removed);
664            }
665        }
666    }
667    retired.extend(target.into_values());
668    retired.sort_unstable_by(|left, right| {
669        (&left.name, left.catalog_generation).cmp(&(&right.name, right.catalog_generation))
670    });
671    Ok(retired)
672}
673
674/// The changelogs a new view reads that the migration preserves. The view takes their rows at
675/// the cut before any later change, so it can apply their retractions. Each must keep those
676/// rows: as a changelog aggregate, or as a view's result.
677fn seeded_upstreams<'a>(
678    view: &str,
679    kind: CatalogObjectKind,
680    dependencies: &'a [String],
681    objects: &BTreeMap<String, ClusterTopologyObjectPlan>,
682    graph: &PlannedTopologyGraph,
683) -> Result<Vec<&'a str>, DbError> {
684    if kind != CatalogObjectKind::MaterializedView {
685        return Ok(Vec::new());
686    }
687    let mut seeded = Vec::new();
688    for dependency in dependencies {
689        let Some(object) = objects.get(dependency).filter(|object| {
690            object.transition == ClusterTopologyObjectTransition::Preserve
691                && matches!(
692                    object.kind,
693                    CatalogObjectKind::Stream | CatalogObjectKind::MaterializedView
694                )
695        }) else {
696            continue;
697        };
698        if !graph.schemas.get(dependency).is_some_and(|schema| {
699            schema
700                .field_with_name(laminar_core::changelog::WEIGHT_COLUMN)
701                .is_ok()
702        }) {
703            continue;
704        }
705        let keeps_rows = graph.operators.get(dependency).is_some_and(|operator| {
706            operator.managed_state == Some(ManagedStateContract::SqlAggregateV1)
707        }) || (object.kind == CatalogObjectKind::MaterializedView
708            && graph
709                .operators
710                .contains_key(&laminar_core::mv::view_result_operator(dependency)));
711        if !keeps_rows {
712            return Err(TopologyError::Unsupported(format!(
713                "new view '{view}' reads changelog '{dependency}', which keeps no rows to seed it \
714                 from; read the aggregate or view that produces them"
715            ))
716            .into());
717        }
718        seeded.push(dependency.as_str());
719    }
720    Ok(seeded)
721}
722
723fn validate_future_state_inputs(
724    name: &str,
725    operator: Option<&crate::operator::capability::OperatorCapability>,
726    dependencies: &[String],
727    seeded: &[&str],
728    objects: &BTreeMap<String, ClusterTopologyObjectPlan>,
729    graph: &PlannedTopologyGraph,
730) -> Result<(), DbError> {
731    if operator.is_some_and(|capability| match capability.state_class {
732        OperatorStateClass::Stateless => capability.managed_state.is_some(),
733        OperatorStateClass::VnodeKeyed | OperatorStateClass::GlobalSingleton => {
734            capability.managed_state.is_none()
735        }
736        _ => true,
737    }) {
738        return Err(TopologyError::Unsupported(format!(
739            "new stream '{name}' has no supported empty managed-state initialization contract"
740        ))
741        .into());
742    }
743    if operator.is_none_or(|capability| capability.managed_state.is_none()) {
744        return Ok(());
745    }
746    let mut pending: Vec<_> = dependencies
747        .iter()
748        .map(String::as_str)
749        .filter(|dependency| !seeded.contains(dependency))
750        .collect();
751    let mut visited = std::collections::BTreeSet::new();
752    while let Some(dependency) = pending.pop() {
753        if !visited.insert(dependency) {
754            continue;
755        }
756        let object = objects.get(dependency).ok_or_else(|| {
757            TopologyError::Invalid("new managed state has an unresolved input".into())
758        })?;
759        if graph.source_input_modes.get(dependency)
760            == Some(&laminar_connectors::connector::SourceInputMode::FullChangelog)
761            || (object.transition == ClusterTopologyObjectTransition::Preserve
762                && graph.schemas.get(dependency).is_some_and(|schema| {
763                    schema
764                        .field_with_name(laminar_core::changelog::WEIGHT_COLUMN)
765                        .is_ok()
766                }))
767        {
768            return Err(TopologyError::Unsupported(format!(
769                "new managed stream '{name}' consumes changelog '{dependency}' without a cut baseline; empty state cannot apply retractions from before the cut"
770            )).into());
771        }
772        pending.extend(object.dependencies.iter().map(String::as_str));
773    }
774    Ok(())
775}
776
777const fn state_contract_name(contract: ManagedStateContract) -> &'static str {
778    match contract {
779        ManagedStateContract::SqlAggregateV1 => "sql_aggregate_v1",
780        ManagedStateContract::CoreWindowV1 => "core_window_v1",
781        ManagedStateContract::BoundedIntervalJoinV3 => "bounded_interval_join_v3",
782        ManagedStateContract::TemporalJoinV1 => "temporal_join_v1",
783        ManagedStateContract::ProcessFunctionV1 => "process_function_v1",
784        ManagedStateContract::ViewResultV1 => "view_result_v1",
785        #[cfg(test)]
786        ManagedStateContract::TestVnodeStateV1 => "test_vnode_state_v1",
787    }
788}
laminar_db/db/topology/planning.rs

laminar_db/db/topology/
planning.rs
