WIP: CREATE ASSERTION (skunkworks)

src/adapter/src/catalog.rs

@@ -1727,6 +1727,7 @@ pub(crate) fn comment_id_to_audit_object_type(id: CommentObjectId) -> ObjectType
         // Unreachable by construction: `COMMENT ON METRIC SINK` is rejected at parse, so no
         // metric-sink comment id is ever built. The arm exists only for exhaustiveness.
         CommentObjectId::MetricSink(_) => ObjectType::MetricSink,
+        CommentObjectId::Assertion(_) => ObjectType::Assertion,
         CommentObjectId::Index(_) => ObjectType::Index,
         CommentObjectId::Func(_) => ObjectType::Func,
         CommentObjectId::Connection(_) => ObjectType::Connection,
@@ -1769,6 +1770,7 @@ pub(crate) fn system_object_type_to_audit_object_type(
             mz_sql::catalog::ObjectType::Schema => ObjectType::Schema,
             mz_sql::catalog::ObjectType::Func => ObjectType::Func,
             mz_sql::catalog::ObjectType::NetworkPolicy => ObjectType::NetworkPolicy,
+            mz_sql::catalog::ObjectType::Assertion => ObjectType::Assertion,
         },
         SystemObjectType::System => ObjectType::System,
     }

src/adapter/src/catalog/apply.rs

@@ -2368,7 +2368,8 @@ fn sort_updates(updates: Vec<StateUpdate>) -> Vec<StateUpdate> {
                 CatalogItemType::View
                 | CatalogItemType::MaterializedView
                 | CatalogItemType::Index
-                | CatalogItemType::MetricSink => derived_items.push(update),
+                | CatalogItemType::MetricSink
+                | CatalogItemType::Assertion => derived_items.push(update),
                 CatalogItemType::Sink => sinks.push(update),
             }
         }

src/adapter/src/catalog/builtin_table_updates.rs

@@ -170,7 +170,10 @@ impl CatalogState {
             | CatalogItem::Index(_)
             | CatalogItem::Log(_)
             | CatalogItem::Secret(_)
-            | CatalogItem::MetricSink(_) => vec![],
+            | CatalogItem::MetricSink(_)
+            // Stage 1 assertions have no dedicated builtin-table row; they are
+            // read out of mz_catalog_raw via create_sql.
+            | CatalogItem::Assertion(_) => vec![],
             // Connection details (mz_kafka_connections, mz_ssh_tunnel_connections,
             // mz_aws_connections, mz_aws_privatelink_connections) are now derived
             // from the persisted create_sql by materialized views over

src/adapter/src/catalog/command.rs

@@ -136,6 +136,7 @@ impl CatalogResultCommand {
         match show {
             ShowStatement::ShowCreateView(_)
             | ShowStatement::ShowCreateMaterializedView(_)
+            | ShowStatement::ShowCreateAssertion(_)
             | ShowStatement::ShowCreateSource(_)
             | ShowStatement::ShowCreateTable(_)
             | ShowStatement::ShowCreateSink(_)

src/adapter/src/catalog/consistency.rs

@@ -270,6 +270,7 @@ impl CatalogState {
                 | CommentObjectId::Source(item_id)
                 | CommentObjectId::Sink(item_id)
                 | CommentObjectId::MetricSink(item_id)
+                | CommentObjectId::Assertion(item_id)
                 | CommentObjectId::Index(item_id)
                 | CommentObjectId::Func(item_id)
                 | CommentObjectId::Connection(item_id)

src/adapter/src/catalog/open.rs

@@ -972,6 +972,7 @@ fn add_new_remove_old_builtin_items_migration(
             CatalogItemType::MaterializedView => CommentObjectId::MaterializedView(id),
             CatalogItemType::Sink
             | CatalogItemType::MetricSink
+            | CatalogItemType::Assertion
             | CatalogItemType::Index
             | CatalogItemType::Type
             | CatalogItemType::Func

src/adapter/src/catalog/state.rs

@@ -30,10 +30,10 @@ use mz_catalog::config::{AwsPrincipalContext, ClusterReplicaSizeMap};
 use mz_catalog::expr_cache::{LocalExpressions, latest_item_version};
 use mz_catalog::memory::error::{Error, ErrorKind};
 use mz_catalog::memory::objects::{
-    CatalogCollectionEntry, CatalogEntry, CatalogItem, Cluster, ClusterReplica, CommentsMap,
-    Connection, DataSourceDesc, Database, DefaultPrivileges, Index, MaterializedView, MetricSink,
-    NetworkPolicy, Role, RoleAuth, Schema, Secret, Sink, Source, SourceReferences, Table,
-    TableDataSource, Type, View,
+    Assertion, CatalogCollectionEntry, CatalogEntry, CatalogItem, Cluster, ClusterReplica,
+    CommentsMap, Connection, DataSourceDesc, Database, DefaultPrivileges, Index, MaterializedView,
+    MetricSink, NetworkPolicy, Role, RoleAuth, Schema, Secret, Sink, Source, SourceReferences,
+    Table, TableDataSource, Type, View,
 };
 use mz_controller::clusters::{
     ManagedReplicaLocation, ReplicaAllocation, ReplicaLocation, UnmanagedReplicaLocation,
@@ -74,9 +74,9 @@ use mz_sql::names::{
     ResolvedDatabaseSpecifier, ResolvedIds, SchemaId, SchemaSpecifier, SystemObjectId,
 };
 use mz_sql::plan::{
-    CreateConnectionPlan, CreateIndexPlan, CreateMaterializedViewPlan, CreateMetricSinkPlan,
-    CreateSecretPlan, CreateSinkPlan, CreateSourcePlan, CreateTablePlan, CreateTypePlan,
-    CreateViewPlan, Params, Plan, PlanContext,
+    CreateAssertionPlan, CreateConnectionPlan, CreateIndexPlan, CreateMaterializedViewPlan,
+    CreateMetricSinkPlan, CreateSecretPlan, CreateSinkPlan, CreateSourcePlan, CreateTablePlan,
+    CreateTypePlan, CreateViewPlan, Params, Plan, PlanContext,
 };
 use mz_sql::rbac;
 use mz_sql::session::metadata::SessionMetadata;
@@ -624,7 +624,10 @@ impl CatalogState {
                 | CatalogItem::Source(_)
                 | CatalogItem::Type(_)
                 | CatalogItem::Func(_)
-                | CatalogItem::Secret(_) => (),
+                | CatalogItem::Secret(_)
+                // Stage 1 assertions ship no dataflow, so they contribute no
+                // introspection dependencies.
+                | CatalogItem::Assertion(_) => (),
             }
         }
 
@@ -1814,6 +1817,29 @@ impl CatalogState {
                     dataflow_metainfo: None,
                 })
             }
+            Plan::CreateAssertion(CreateAssertionPlan { assertion, .. }) => {
+                // Stage 1: an assertion is a declaration-only catalog object.
+                // We keep the unoptimized violations query (HIR) and the
+                // dependencies it references, but do not optimize it or render a
+                // dataflow; that is deferred to Stage 2.
+                let raw_expr = Arc::new(assertion.expr);
+                // Resolve all item dependencies from the HIR expression, matching
+                // how views/materialized views derive theirs.
+                let dependencies = raw_expr
+                    .depends_on()
+                    .into_iter()
+                    .map(|gid| self.get_entry_by_global_id(&gid).id())
+                    .collect();
+                CatalogItem::Assertion(Assertion {
+                    create_sql: assertion.create_sql,
+                    global_id,
+                    raw_expr,
+                    resolved_ids,
+                    dependencies: DependencyIds(dependencies),
+                    cluster_id: assertion.cluster_id,
+                    target_replica: assertion.target_replica,
+                })
+            }
             Plan::CreateSink(CreateSinkPlan {
                 sink,
                 with_snapshot,
@@ -2195,6 +2221,7 @@ impl CatalogState {
             | CatalogItemType::Source
             | CatalogItemType::Sink
             | CatalogItemType::MetricSink
+            | CatalogItemType::Assertion
             | CatalogItemType::View
             | CatalogItemType::MaterializedView
             | CatalogItemType::Index
@@ -3023,6 +3050,7 @@ impl CatalogState {
             | CommentObjectId::Source(id)
             | CommentObjectId::Sink(id)
             | CommentObjectId::MetricSink(id)
+            | CommentObjectId::Assertion(id)
             | CommentObjectId::Index(id)
             | CommentObjectId::Func(id)
             | CommentObjectId::Connection(id)
@@ -3053,6 +3081,7 @@ impl CatalogState {
             | CommentObjectId::Source(id)
             | CommentObjectId::Sink(id)
             | CommentObjectId::MetricSink(id)
+            | CommentObjectId::Assertion(id)
             | CommentObjectId::Index(id)
             | CommentObjectId::Func(id)
             | CommentObjectId::Connection(id)

src/adapter/src/catalog/timeline.rs

@@ -94,6 +94,7 @@ impl Catalog {
                         CatalogItem::View(_)
                         | CatalogItem::Sink(_)
                         | CatalogItem::MetricSink(_)
+                        | CatalogItem::Assertion(_)
                         | CatalogItem::Type(_)
                         | CatalogItem::Func(_)
                         | CatalogItem::Secret(_)
@@ -232,6 +233,7 @@ impl Catalog {
                     }
                     CatalogItem::Sink(_)
                     | CatalogItem::MetricSink(_)
+                    | CatalogItem::Assertion(_)
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)
                     | CatalogItem::Secret(_)

src/adapter/src/catalog/transact.rs

@@ -906,7 +906,9 @@ impl Catalog {
                     | CatalogItem::Func(_)
                     | CatalogItem::Secret(_)
                     | CatalogItem::Connection(_)
-                    | CatalogItem::MetricSink(_) => {}
+                    | CatalogItem::MetricSink(_)
+                    // Stage 1 assertions store no locally-optimized expression.
+                    | CatalogItem::Assertion(_) => {}
                 }
             }
         }
@@ -1729,7 +1731,10 @@ impl Catalog {
                     | CatalogItem::Connection(_)
                     // Metric sinks write to the replica's metrics registry, never to persist,
                     // so there is no storage collection to create.
-                    | CatalogItem::MetricSink(_) => (),
+                    | CatalogItem::MetricSink(_)
+                    // Stage 1 assertions declare an invariant but ship no dataflow and
+                    // allocate no storage collection.
+                    | CatalogItem::Assertion(_) => (),
                 }
 
                 let system_user = session.map_or(false, |s| s.user().is_system_user());
@@ -1914,7 +1919,8 @@ impl Catalog {
                         | CatalogItem::Func(_)
                         | CatalogItem::Secret(_)
                         | CatalogItem::Connection(_)
-                        | CatalogItem::MetricSink(_) => EventDetails::IdFullNameV1(IdFullNameV1 {
+                        | CatalogItem::MetricSink(_)
+                        | CatalogItem::Assertion(_) => EventDetails::IdFullNameV1(IdFullNameV1 {
                             id: id.to_string(),
                             name,
                         }),

src/adapter/src/command.rs

@@ -704,6 +704,8 @@ pub enum ExecuteResponse {
     CreatedClusterReplica,
     /// The requested index was created.
     CreatedIndex,
+    /// The requested assertion was created.
+    CreatedAssertion,
     /// The requested metric sink was created.
     CreatedMetricSink,
     /// The requested introspection subscribe was created.
@@ -886,6 +888,7 @@ impl TryInto<ExecuteResponse> for ExecuteResponseKind {
                 Ok(ExecuteResponse::CreatedClusterReplica)
             }
             ExecuteResponseKind::CreatedIndex => Ok(ExecuteResponse::CreatedIndex),
+            ExecuteResponseKind::CreatedAssertion => Ok(ExecuteResponse::CreatedAssertion),
             ExecuteResponseKind::CreatedMetricSink => Ok(ExecuteResponse::CreatedMetricSink),
             ExecuteResponseKind::CreatedSecret => Ok(ExecuteResponse::CreatedSecret),
             ExecuteResponseKind::CreatedSink => Ok(ExecuteResponse::CreatedSink),
@@ -951,6 +954,7 @@ impl ExecuteResponse {
             CreatedCluster { .. } => Some("CREATE CLUSTER".into()),
             CreatedClusterReplica { .. } => Some("CREATE CLUSTER REPLICA".into()),
             CreatedIndex { .. } => Some("CREATE INDEX".into()),
+            CreatedAssertion { .. } => Some("CREATE ASSERTION".into()),
             CreatedMetricSink { .. } => Some("CREATE METRIC SINK".into()),
             CreatedSecret { .. } => Some("CREATE SECRET".into()),
             CreatedSink { .. } => Some("CREATE SINK".into()),
@@ -1050,6 +1054,7 @@ impl ExecuteResponse {
             CreateTable => &[CreatedTable],
             CreateView => &[CreatedView],
             CreateMaterializedView => &[CreatedMaterializedView],
+            CreateAssertion => &[CreatedAssertion],
             CreateIndex => &[CreatedIndex],
             CreateMetricSink => &[CreatedMetricSink],
             CreateType => &[CreatedType],

src/adapter/src/coord.rs

@@ -2806,7 +2806,10 @@ impl Coordinator {
                 CatalogItem::Log(_)
                 | CatalogItem::Type(_)
                 | CatalogItem::Func(_)
-                | CatalogItem::Secret(_) => {}
+                | CatalogItem::Secret(_)
+                // Stage 1 assertions ship no dataflow and no storage collection,
+                // so there is nothing to bootstrap.
+                | CatalogItem::Assertion(_) => {}
             }
         }
 
@@ -3389,7 +3392,9 @@ impl Coordinator {
                 | CatalogItem::Connection(_)
                 // Nothing to bootstrap: a metric sink has no storage collection, it publishes
                 // into the replica's metrics registry.
-                | CatalogItem::MetricSink(_) => (),
+                | CatalogItem::MetricSink(_)
+                // Stage 1 assertions have no storage collection.
+                | CatalogItem::Assertion(_) => (),
             }
         }
 
@@ -3898,7 +3903,9 @@ impl Coordinator {
                 | CatalogItem::Type(_)
                 | CatalogItem::Func(_)
                 | CatalogItem::Secret(_)
-                | CatalogItem::Connection(_) => (),
+                | CatalogItem::Connection(_)
+                // Stage 1 assertions ship no dataflow, so there is no plan to build.
+                | CatalogItem::Assertion(_) => (),
             }
         }
 
@@ -3931,7 +3938,9 @@ impl Coordinator {
                 | CatalogItem::Type(_)
                 | CatalogItem::Func(_)
                 | CatalogItem::Secret(_)
-                | CatalogItem::Connection(_) => continue,
+                | CatalogItem::Connection(_)
+                // Stage 1 assertions ship no dataflow, so there is no as-of to compute.
+                | CatalogItem::Assertion(_) => continue,
             };
             if let Some(plan) = self.catalog.try_get_physical_plan(&gid) {
                 catalog_ids.push(gid);

src/adapter/src/coord/appends.rs

@@ -1224,6 +1224,7 @@ pub(crate) fn waiting_on_startup_appends(
         | Plan::CreateTable(_)
         | Plan::CreateView(_)
         | Plan::CreateMaterializedView(_)
+        | Plan::CreateAssertion(_)
         | Plan::CreateIndex(_)
         | Plan::CreateMetricSink(_)
         | Plan::CreateType(_)

src/adapter/src/coord/catalog_implications.rs

@@ -1700,7 +1700,8 @@ impl Coordinator {
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)
                     | CatalogItem::Secret(_)
-                    | CatalogItem::MetricSink(_) => {
+                    | CatalogItem::MetricSink(_)
+                    | CatalogItem::Assertion(_) => {
                         // Other item types don't have connection dependencies
                         // that need updating.
                     }
@@ -1939,6 +1940,9 @@ impl CatalogImplication {
                 CatalogItem::Log(_) => {}
                 CatalogItem::Type(_) => {}
                 CatalogItem::Func(_) => {}
+                // Stage 1 assertions have no storage/compute controller
+                // implications: no storage collection, no dataflow.
+                CatalogItem::Assertion(_) => {}
             },
             ParsedStateUpdateKind::Cluster {
                 durable_cluster: _,

src/adapter/src/coord/catalog_serving.rs

@@ -90,6 +90,7 @@ pub fn auto_run_on_catalog_server<'a, 's, 'p>(
         | Plan::CreateTable(_)
         | Plan::CreateView(_)
         | Plan::CreateMaterializedView(_)
+        | Plan::CreateAssertion(_)
         | Plan::CreateIndex(_)
         | Plan::CreateMetricSink(_)
         | Plan::CreateType(_)

src/adapter/src/coord/command_handler.rs

@@ -1492,6 +1492,7 @@ impl Coordinator {
                     | Statement::CreateDatabase(_)
                     | Statement::CreateIndex(_)
                     | Statement::CreateMaterializedView(_)
+                    | Statement::CreateAssertion(_)
                     | Statement::CreateRole(_)
                     | Statement::CreateSchema(_)
                     | Statement::CreateSecret(_)

src/adapter/src/coord/ddl.rs

@@ -1344,7 +1344,8 @@ impl Coordinator {
                         | CatalogItem::Index(_)
                         | CatalogItem::Type(_)
                         | CatalogItem::Func(_)
-                        | CatalogItem::MetricSink(_) => {}
+                        | CatalogItem::MetricSink(_)
+                        | CatalogItem::Assertion(_) => {}
                     }
                 }
                 Op::DropObjects(drop_object_infos) => {
@@ -1416,7 +1417,8 @@ impl Coordinator {
                                     | CatalogItem::Index(_)
                                     | CatalogItem::Type(_)
                                     | CatalogItem::Func(_)
-                                    | CatalogItem::MetricSink(_) => {}
+                                    | CatalogItem::MetricSink(_)
+                                    | CatalogItem::Assertion(_) => {}
                                 }
                             }
                         }
@@ -1447,7 +1449,8 @@ impl Coordinator {
                     | CatalogItem::Index(_)
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)
-                    | CatalogItem::MetricSink(_) => {}
+                    | CatalogItem::MetricSink(_)
+                    | CatalogItem::Assertion(_) => {}
                 },
                 Op::AlterRole { .. }
                 | Op::AlterRetainHistory { .. }

src/adapter/src/coord/indexes.rs

@@ -70,6 +70,7 @@ impl DataflowBuilder<'_> {
                     }
                     CatalogItem::Sink(_)
                     | CatalogItem::MetricSink(_)
+                    | CatalogItem::Assertion(_)
                     | CatalogItem::Index(_)
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)

src/adapter/src/coord/read_then_write.rs

@@ -328,8 +328,8 @@ pub(crate) fn validate_read_then_write_dependencies(
             DependencyPolicy::UserDml => match item_type {
                 typ @ (Func | View | MaterializedView) => id.is_user() || matches!(typ, Func),
                 Source | Secret | Connection => false,
-                // Cannot select from sinks or indexes.
-                Sink | MetricSink | Index => unreachable!(),
+                // Cannot select from sinks, indexes, or (in Stage 1) assertions.
+                Sink | MetricSink | Assertion | Index => unreachable!(),
                 Table => id.is_user() && is_writable_table,
                 Type => true,
             },

src/adapter/src/coord/sequencer.rs

@@ -276,6 +276,12 @@ impl Coordinator {
                     self.sequence_create_materialized_view(ctx, plan, resolved_ids)
                         .await;
                 }
+                Plan::CreateAssertion(plan) => {
+                    let result = self
+                        .sequence_create_assertion(ctx.session(), plan, resolved_ids)
+                        .await;
+                    ctx.retire(result);
+                }
                 Plan::CreateIndex(plan) => {
                     self.sequence_create_index(ctx, plan, resolved_ids).await;
                 }

src/adapter/src/coord/sequencer/inner.rs

@@ -125,6 +125,7 @@ type RtrTimestampFuture = BoxFuture<'static, Result<Timestamp, StorageError>>;
 
 mod cluster;
 mod copy_from;
+mod create_assertion;
 mod create_index;
 mod create_materialized_view;
 mod create_metric_sink;