add network policy create/alter/drop/comment

doc/user/content/sql/system-catalog/mz_internal.md

@@ -696,6 +696,16 @@ The `mz_network_policy_rules` table contains a row for each network policy rule.
 | `address`        | [`text`]   | The address the rule will take action on.                                                              |
 | `direction`      | [`text`]   | The direction of traffic the rule applies to. `ingress` is the only supported direction.               |
 
+## `mz_show_network_policiesjj`
+
+The `mz_show_show_network_policies` view contains a row for each network policy in the system.
+
+<!-- RELATION_SPEC mz_internal.mz_show_network_policies -->
+| Field       | Type     | Meaning                             |
+|-------------|----------|-------------------------------------|
+| `name`      | [`text`] | The name the network policy.        |
+| `comment`   | [`text`] | The comment on the network policy.  |
+
 ## `mz_show_all_privileges`
 
 The `mz_show_all_privileges` view contains a row for each privilege granted

doc/user/content/sql/types/mz_aclitem.md

@@ -26,17 +26,18 @@ is `<grantee>=<privileges>/<grantor>`.
 
 A list of all privileges and their abbreviations are below:
 
-| Privilege       | Description                                                                                    | Abbreviation      | Applicable Object Types                       |
-|-----------------|------------------------------------------------------------------------------------------------|-------------------|-----------------------------------------------|
-| `SELECT`        | Allows reading rows from an object.                                                            | r(”read”)         | Table, View, Materialized View, Source        |
-| `INSERT`        | Allows inserting into an object.                                                               | a(”append”)       | Table                                         |
-| `UPDATE`        | Allows updating an object (requires SELECT if a read is necessary).                            | w(”write”)        | Table                                         |
-| `DELETE`        | Allows deleting from an object (requires SELECT if a read is necessary).                       | d                 | Table                                         |
-| `CREATE`        | Allows creating a new object within another object.                                            | C                 | Database, Schema, Cluster                     |
-| `USAGE`         | Allows using an object or looking up members of an object.                                     | U                 | Database, Schema, Connection, Secret, Cluster |
-| `CREATEROLE`    | Allows creating, altering, deleting roles and the ability to grant and revoke role membership. | R("Role")         | System                                        |
-| `CREATEDB`      | Allows creating databases.                                                                     | B("dataBase")     | System                                        |
-| `CREATECLUSTER` | Allows creating clusters.                                                                      | N("compute Node") | System                                        |
+| Privilege             | Description                                                                                    | Abbreviation        | Applicable Object Types                       |
+|-----------------------|------------------------------------------------------------------------------------------------|---------------------|-----------------------------------------------|
+| `SELECT`              | Allows reading rows from an object.                                                            | r(”read”)           | Table, View, Materialized View, Source        |
+| `INSERT`              | Allows inserting into an object.                                                               | a(”append”)         | Table                                         |
+| `UPDATE`              | Allows updating an object (requires SELECT if a read is necessary).                            | w(”write”)          | Table                                         |
+| `DELETE`              | Allows deleting from an object (requires SELECT if a read is necessary).                       | d                   | Table                                         |
+| `CREATE`              | Allows creating a new object within another object.                                            | C                   | Database, Schema, Cluster                     |
+| `USAGE`               | Allows using an object or looking up members of an object.                                     | U                   | Database, Schema, Connection, Secret, Cluster |
+| `CREATEROLE`          | Allows creating, altering, deleting roles and the ability to grant and revoke role membership. | R("Role")           | System                                        |
+| `CREATEDB`            | Allows creating databases.                                                                     | B("dataBase")       | System                                        |
+| `CREATECLUSTER`       | Allows creating clusters.                                                                      | N("compute Node")   | System                                        |
+| `CREATENETWORKPOLICY` | Allows creating network policies.                                                              | P("network Policy") | System                                        |
 
 The `CREATEROLE` privilege is very powerful. It allows roles to grant and revoke membership in
 other roles, even if it doesn't have explicit membership in those roles. As a consequence, any role

doc/user/sql-grammar/sql-grammar.bnf

@@ -84,7 +84,7 @@ comment_on ::=
   'COMMENT ON' (
     'CLUSTER' | 'CLUSTER REPLICA' | 'COLUMN' | 'CONNECTION' | 'DATABASE' | 'FUNCTION' |
     'INDEX' | 'MATERIALIZED VIEW' | 'ROLE' | 'SCHEMA' | 'SECRET' | 'SINK' | 'SOURCE' |
-    'TABLE' | 'TYPE' | 'VIEW'
+    'TABLE' | 'TYPE' | 'VIEW' | 'NETWORK POLICY'
   ) object_name 'IS' ( string_literal | 'NULL' )
 commit ::=
   'COMMIT'

src/adapter/src/catalog.rs

@@ -1673,6 +1673,16 @@ impl SessionCatalog for ConnCatalog<'_> {
         }
     }
 
+    fn resolve_network_policy(
+        &self,
+        policy_name: &str,
+    ) -> Result<&dyn mz_sql::catalog::CatalogNetworkPolicy, SqlCatalogError> {
+        match self.state.try_get_network_policy_by_name(policy_name) {
+            Some(policy) => Ok(policy),
+            None => Err(SqlCatalogError::UnknownNetworkPolicy(policy_name.into())),
+        }
+    }
+
     fn try_get_role(&self, id: &RoleId) -> Option<&dyn CatalogRole> {
         Some(self.state.roles_by_id.get(id)?)
     }

src/adapter/src/catalog/state.rs

@@ -375,13 +375,10 @@ impl CatalogState {
                         seen,
                     ));
                 }
-                id @ ObjectId::Role(_) => {
-                    let unseen = seen.insert(id.clone());
-                    if unseen {
-                        dependents.push(id.clone());
-                    }
+                ObjectId::NetworkPolicy(id) => {
+                    dependents.extend_from_slice(&self.network_policy_dependents(*id, seen));
                 }
-                id @ ObjectId::NetworkPolicy(_) => {
+                id @ ObjectId::Role(_) => {
                     let unseen = seen.insert(id.clone());
                     if unseen {
                         dependents.push(id.clone());
@@ -399,7 +396,7 @@ impl CatalogState {
     /// itself.
     ///
     /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
-    /// earlier in the list than the roots. This is particularly userful for the order to drop
+    /// earlier in the list than the roots. This is particularly useful for the order to drop
     /// objects.
     fn cluster_dependents(
         &self,
@@ -430,7 +427,7 @@ impl CatalogState {
     /// itself.
     ///
     /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
-    /// earlier in the list than the roots. This is particularly userful for the order to drop
+    /// earlier in the list than the roots. This is particularly useful for the order to drop
     /// objects.
     pub(super) fn cluster_replica_dependents(
         &self,
@@ -451,7 +448,7 @@ impl CatalogState {
     /// itself.
     ///
     /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
-    /// earlier in the list than the roots. This is particularly userful for the order to drop
+    /// earlier in the list than the roots. This is particularly useful for the order to drop
     /// objects.
     fn database_dependents(
         &self,
@@ -481,7 +478,7 @@ impl CatalogState {
     /// itself.
     ///
     /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
-    /// earlier in the list than the roots. This is particularly userful for the order to drop
+    /// earlier in the list than the roots. This is particularly useful for the order to drop
     /// objects.
     fn schema_dependents(
         &self,
@@ -507,7 +504,7 @@ impl CatalogState {
     /// itself.
     ///
     /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
-    /// earlier in the list than the roots. This is particularly userful for the order to drop
+    /// earlier in the list than the roots. This is particularly useful for the order to drop
     /// objects.
     pub(super) fn item_dependents(
         &self,
@@ -533,6 +530,24 @@ impl CatalogState {
         dependents
     }
 
+    /// Returns all the IDs of all objects that depend on `network_policy_id`, including `network_policy_id`
+    /// itself.
+    ///
+    /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
+    /// earlier in the list than the roots. This is particularly useful for the order to drop
+    /// objects.
+    pub(super) fn network_policy_dependents(
+        &self,
+        network_policy_id: NetworkPolicyId,
+        _seen: &mut BTreeSet<ObjectId>,
+    ) -> Vec<ObjectId> {
+        let object_id = ObjectId::NetworkPolicy(network_policy_id);
+        // Currently network policies have no dependents
+        // when we add the ability for users or sources/sinks to have policies
+        // this method will need to be updated.
+        vec![object_id]
+    }
+
     /// Indicates whether the indicated item is considered stable or not.
     ///
     /// Only stable items can be used as dependencies of other catalog items.
@@ -721,6 +736,15 @@ impl CatalogState {
             .map(|id| &self.roles_by_id[id])
     }
 
+    pub(super) fn try_get_network_policy_by_name(
+        &self,
+        policy_name: &str,
+    ) -> Option<&NetworkPolicy> {
+        self.network_policies_by_name
+            .get(policy_name)
+            .map(|id| &self.network_policies_by_id[id])
+    }
+
     pub(crate) fn collect_role_membership(&self, id: &RoleId) -> BTreeSet<RoleId> {
         let mut membership = BTreeSet::new();
         let mut queue = VecDeque::from(vec![id]);

src/adapter/src/catalog/transact.rs

@@ -22,7 +22,7 @@ use mz_audit_log::{
     ObjectType, SchedulingDecisionsWithReasonsV1, VersionedEvent, VersionedStorageUsage,
 };
 use mz_catalog::builtin::BuiltinLog;
-use mz_catalog::durable::Transaction;
+use mz_catalog::durable::{NetworkPolicy, Transaction};
 use mz_catalog::memory::error::{AmbiguousRename, Error, ErrorKind};
 use mz_catalog::memory::objects::{
     CatalogItem, ClusterConfig, DataSourceDesc, SourceReferences, StateDiff, StateUpdate,
@@ -81,6 +81,12 @@ pub enum Op {
         attributes: RoleAttributes,
         vars: RoleVars,
     },
+    AlterNetworkPolicy {
+        id: NetworkPolicyId,
+        rules: Vec<NetworkPolicyRule>,
+        name: String,
+        owner_id: RoleId,
+    },
     CreateDatabase {
         name: String,
         owner_id: RoleId,
@@ -115,7 +121,6 @@ pub enum Op {
         owner_id: RoleId,
     },
     CreateNetworkPolicy {
-        id: NetworkPolicyId,
         rules: Vec<NetworkPolicyRule>,
         name: String,
         owner_id: RoleId,
@@ -655,6 +660,38 @@ impl Catalog {
 
                 info!("update role {name} ({id})");
             }
+            Op::AlterNetworkPolicy {
+                id,
+                rules,
+                name,
+                owner_id: _owner_id,
+            } => {
+                let existing_policy = state.get_network_policy(&id).clone();
+                let mut policy: NetworkPolicy = existing_policy.into();
+                policy.rules = rules;
+                if is_reserved_name(&name) {
+                    return Err(AdapterError::Catalog(Error::new(
+                        ErrorKind::ReservedNetworkPolicyName(name),
+                    )));
+                }
+                tx.update_network_policy(id, policy.clone())?;
+
+                CatalogState::add_to_audit_log(
+                    &state.system_configuration,
+                    oracle_write_ts,
+                    session,
+                    tx,
+                    audit_events,
+                    EventType::Alter,
+                    ObjectType::NetworkPolicy,
+                    EventDetails::IdNameV1(mz_audit_log::IdNameV1 {
+                        id: id.to_string(),
+                        name: name.clone(),
+                    }),
+                )?;
+
+                info!("update network policy {name} ({id})");
+            }
             Op::CreateDatabase { name, owner_id } => {
                 let database_owner_privileges = vec![rbac::owner_privilege(
                     mz_sql::catalog::ObjectType::Database,
@@ -1123,7 +1160,6 @@ impl Catalog {
                 }
             }
             Op::CreateNetworkPolicy {
-                id,
                 rules,
                 name,
                 owner_id,
@@ -1157,14 +1193,29 @@ impl Catalog {
                         .collect();
 
                 let temporary_oids: HashSet<_> = state.get_temporary_oids().collect();
-                tx.insert_user_network_policy(
-                    id,
-                    name,
+                let id = tx.insert_user_network_policy(
+                    name.clone(),
                     rules,
                     privileges,
                     owner_id,
                     &temporary_oids,
                 )?;
+
+                CatalogState::add_to_audit_log(
+                    &state.system_configuration,
+                    oracle_write_ts,
+                    session,
+                    tx,
+                    audit_events,
+                    EventType::Create,
+                    ObjectType::NetworkPolicy,
+                    EventDetails::IdNameV1(mz_audit_log::IdNameV1 {
+                        id: id.to_string(),
+                        name: name.clone(),
+                    }),
+                )?;
+
+                info!("created network policy {name} ({id})");
             }
             Op::Comment {
                 object_id,
@@ -1357,6 +1408,31 @@ impl Catalog {
                     info!("drop role {}", role.name());
                 }
 
+                // Drop any roles.
+                tx.remove_network_policies(&delta.network_policies)?;
+
+                for network_policy_id in delta.network_policies {
+                    let policy = state
+                        .network_policies_by_id
+                        .get(&network_policy_id)
+                        .expect("catalog out of sync");
+
+                    CatalogState::add_to_audit_log(
+                        &state.system_configuration,
+                        oracle_write_ts,
+                        session,
+                        tx,
+                        audit_events,
+                        EventType::Drop,
+                        ObjectType::NetworkPolicy,
+                        EventDetails::IdNameV1(mz_audit_log::IdNameV1 {
+                            id: policy.id.to_string(),
+                            name: policy.name.clone(),
+                        }),
+                    )?;
+                    info!("drop network policy {}", policy.name.clone());
+                }
+
                 // Drop any replicas.
                 let replicas = delta.replicas.keys().copied().collect();
                 tx.remove_cluster_replicas(&replicas)?;

src/adapter/src/command.rs

@@ -321,6 +321,8 @@ pub enum ExecuteResponse {
     CreatedContinualTask,
     /// The requested type was created.
     CreatedType,
+    /// The requested network policy was created.
+    CreatedNetworkPolicy,
     /// The requested prepared statement was removed.
     Deallocate { all: bool },
     /// The requested cursor was declared.
@@ -486,6 +488,7 @@ impl TryInto<ExecuteResponse> for ExecuteResponseKind {
             ExecuteResponseKind::CreatedMaterializedView => {
                 Ok(ExecuteResponse::CreatedMaterializedView)
             }
+            ExecuteResponseKind::CreatedNetworkPolicy => Ok(ExecuteResponse::CreatedNetworkPolicy),
             ExecuteResponseKind::CreatedContinualTask => Ok(ExecuteResponse::CreatedContinualTask),
             ExecuteResponseKind::CreatedType => Ok(ExecuteResponse::CreatedType),
             ExecuteResponseKind::Deallocate => Err(()),
@@ -550,6 +553,7 @@ impl ExecuteResponse {
             CreatedMaterializedView { .. } => Some("CREATE MATERIALIZED VIEW".into()),
             CreatedContinualTask { .. } => Some("CREATE CONTINUAL TASK".into()),
             CreatedType => Some("CREATE TYPE".into()),
+            CreatedNetworkPolicy => Some("CREATE NETWORKPOLICY".into()),
             Deallocate { all } => Some(format!("DEALLOCATE{}", if *all { " ALL" } else { "" })),
             DeclaredCursor => Some("DECLARE CURSOR".into()),
             Deleted(n) => Some(format!("DELETE {}", n)),
@@ -612,7 +616,8 @@ impl ExecuteResponse {
             | AlterConnection
             | AlterSource
             | AlterSink
-            | AlterTableAddColumn => &[AlteredObject],
+            | AlterTableAddColumn
+            | AlterNetworkPolicy => &[AlteredObject],
             AlterDefaultPrivileges => &[AlteredDefaultPrivileges],
             AlterSetCluster => &[AlteredObject],
             AlterRole => &[AlteredRole],
@@ -640,6 +645,7 @@ impl ExecuteResponse {
             CreateIndex => &[CreatedIndex],
             CreateType => &[CreatedType],
             PlanKind::Deallocate => &[ExecuteResponseKind::Deallocate],
+            CreateNetworkPolicy => &[CreatedNetworkPolicy],
             Declare => &[DeclaredCursor],
             DiscardTemp => &[DiscardedTemp],
             DiscardAll => &[DiscardedAll],

src/adapter/src/coord/catalog_serving.rs

@@ -72,6 +72,7 @@ pub fn auto_run_on_catalog_server<'a, 's, 'p>(
         | Plan::CreateDatabase(_)
         | Plan::CreateSchema(_)
         | Plan::CreateRole(_)
+        | Plan::CreateNetworkPolicy(_)
         | Plan::CreateCluster(_)
         | Plan::CreateClusterReplica(_)
         | Plan::CreateContinualTask(_)
@@ -106,6 +107,7 @@ pub fn auto_run_on_catalog_server<'a, 's, 'p>(
         | Plan::ExplainPushdown(_)
         | Plan::ExplainSinkSchema(_)
         | Plan::Insert(_)
+        | Plan::AlterNetworkPolicy(_)
         | Plan::AlterNoop(_)
         | Plan::AlterClusterRename(_)
         | Plan::AlterClusterSwap(_)

src/adapter/src/coord/command_handler.rs

@@ -659,6 +659,7 @@ impl Coordinator {
                     | Statement::AlterSystemResetAll(_)
                     | Statement::AlterSystemSet(_)
                     | Statement::AlterTableAddColumn(_)
+                    | Statement::AlterNetworkPolicy(_)
                     | Statement::CreateCluster(_)
                     | Statement::CreateClusterReplica(_)
                     | Statement::CreateConnection(_)
@@ -677,6 +678,7 @@ impl Coordinator {
                     | Statement::CreateType(_)
                     | Statement::CreateView(_)
                     | Statement::CreateWebhookSource(_)
+                    | Statement::CreateNetworkPolicy(_)
                     | Statement::Delete(_)
                     | Statement::DropObjects(_)
                     | Statement::DropOwned(_)