mz_catalog/durable/
initialize.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10use std::collections::BTreeMap;
11use std::iter;
12use std::str::FromStr;
13use std::sync::LazyLock;
14use std::time::Duration;
15
16use base64::prelude::*;
17use ipnet::IpNet;
18use itertools::max;
19use mz_audit_log::{CreateOrDropClusterReplicaReasonV1, EventV1, VersionedEvent};
20use mz_controller::clusters::ReplicaLogging;
21use mz_controller_types::{ClusterId, ReplicaId};
22use mz_ore::collections::HashSet;
23use mz_ore::now::EpochMillis;
24use mz_persist_types::ShardId;
25use mz_pgrepr::oid::{
26    FIRST_USER_OID, NETWORK_POLICIES_DEFAULT_POLICY_OID, ROLE_PUBLIC_OID,
27    SCHEMA_INFORMATION_SCHEMA_OID, SCHEMA_MZ_CATALOG_OID, SCHEMA_MZ_CATALOG_UNSTABLE_OID,
28    SCHEMA_MZ_INTERNAL_OID, SCHEMA_MZ_INTROSPECTION_OID, SCHEMA_MZ_UNSAFE_OID,
29    SCHEMA_PG_CATALOG_OID,
30};
31use mz_repr::adt::mz_acl_item::{AclMode, MzAclItem};
32use mz_repr::network_policy_id::NetworkPolicyId;
33use mz_repr::role_id::RoleId;
34use mz_sql::catalog::{
35    DefaultPrivilegeAclItem, DefaultPrivilegeObject, ObjectType, RoleAttributesRaw, RoleMembership,
36    RoleVars, SystemObjectType,
37};
38use mz_sql::names::{
39    DatabaseId, ObjectId, PUBLIC_ROLE_NAME, ResolvedDatabaseSpecifier, SchemaId, SchemaSpecifier,
40};
41use mz_sql::plan::{NetworkPolicyRule, PolicyAddress};
42use mz_sql::rbac;
43use mz_sql::session::user::{MZ_SUPPORT_ROLE_ID, MZ_SYSTEM_ROLE_ID};
44
45use crate::builtin::BUILTIN_ROLES;
46use crate::durable::upgrade::CATALOG_VERSION;
47use crate::durable::{
48    AUDIT_LOG_ID_ALLOC_KEY, BUILTIN_MIGRATION_SHARD_KEY, BootstrapArgs,
49    CATALOG_CONTENT_VERSION_KEY, CatalogError, ClusterConfig, ClusterVariant,
50    ClusterVariantManaged, DATABASE_ID_ALLOC_KEY, DefaultPrivilege, EXPRESSION_CACHE_SHARD_KEY,
51    MOCK_AUTHENTICATION_NONCE_KEY, OID_ALLOC_KEY, ReplicaConfig, ReplicaLocation, Role,
52    SCHEMA_ID_ALLOC_KEY, STORAGE_USAGE_ID_ALLOC_KEY, SYSTEM_CLUSTER_ID_ALLOC_KEY,
53    SYSTEM_CLUSTER_REPLICA_SIZE_ID_ALLOC_KEY, SYSTEM_REPLICA_ID_ALLOC_KEY, Schema, Transaction,
54    USER_CLUSTER_ID_ALLOC_KEY, USER_CLUSTER_REPLICA_SIZE_ID_ALLOC_KEY,
55    USER_NETWORK_POLICY_ID_ALLOC_KEY, USER_REPLICA_ID_ALLOC_KEY, USER_ROLE_ID_ALLOC_KEY,
56};
57
58/// The key within the "config" Collection that stores the version of the catalog.
59pub const USER_VERSION_KEY: &str = "user_version";
60/// The key within the "config" collection that stores whether the remote configuration was
61/// synchronized at least once.
62pub(crate) const SYSTEM_CONFIG_SYNCED_KEY: &str = "system_config_synced";
63
64/// The key used within the "config" collection where we store a mirror of the
65/// `with_0dt_deployment_max_wait` "system var" value. This is mirrored so that
66/// we can toggle the flag with LaunchDarkly, but use it in boot before
67/// LaunchDarkly is available.
68///
69/// NOTE: Weird prefix because we can't start with a `0`.
70pub(crate) const WITH_0DT_DEPLOYMENT_MAX_WAIT: &str = "with_0dt_deployment_max_wait";
71
72/// The key used within the "config" collection where we store a mirror of the
73/// `with_0dt_deployment_ddl_check_interval` "system var" value. This is
74/// mirrored so that we can toggle the flag with LaunchDarkly, but use it in
75/// boot before LaunchDarkly is available.
76///
77/// NOTE: Weird prefix because we can't start with a `0`.
78pub(crate) const WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL: &str =
79    "with_0dt_deployment_ddl_check_interval";
80
81/// The key used within the "config" collection where we store a mirror of the
82/// `enable_0dt_deployment_panic_after_timeout` "system var" value. This is
83/// mirrored so that we can toggle the flag with LaunchDarkly, but use it in
84/// boot before LaunchDarkly is available.
85pub(crate) const ENABLE_0DT_DEPLOYMENT_PANIC_AFTER_TIMEOUT: &str =
86    "enable_0dt_deployment_panic_after_timeout";
87
88const USER_ID_ALLOC_KEY: &str = "user";
89const SYSTEM_ID_ALLOC_KEY: &str = "system";
90
91const DEFAULT_USER_CLUSTER_ID: ClusterId = ClusterId::User(1);
92const DEFAULT_USER_CLUSTER_NAME: &str = "quickstart";
93
94const DEFAULT_USER_REPLICA_ID: ReplicaId = ReplicaId::User(1);
95
96const MATERIALIZE_DATABASE_ID_VAL: u64 = 1;
97const MATERIALIZE_DATABASE_ID: DatabaseId = DatabaseId::User(MATERIALIZE_DATABASE_ID_VAL);
98
99const MZ_CATALOG_SCHEMA_ID: u64 = 1;
100const PG_CATALOG_SCHEMA_ID: u64 = 2;
101const PUBLIC_SCHEMA_ID: u64 = 3;
102const MZ_INTERNAL_SCHEMA_ID: u64 = 4;
103const INFORMATION_SCHEMA_ID: u64 = 5;
104pub const MZ_UNSAFE_SCHEMA_ID: u64 = 6;
105pub const MZ_CATALOG_UNSTABLE_SCHEMA_ID: u64 = 7;
106pub const MZ_INTROSPECTION_SCHEMA_ID: u64 = 8;
107
108const DEFAULT_ALLOCATOR_ID: u64 = 1;
109
110pub const DEFAULT_USER_NETWORK_POLICY_ID: NetworkPolicyId = NetworkPolicyId::User(1);
111pub const DEFAULT_USER_NETWORK_POLICY_NAME: &str = "default";
112pub const DEFAULT_USER_NETWORK_POLICY_RULES: &[(
113    &str,
114    mz_sql::plan::NetworkPolicyRuleAction,
115    mz_sql::plan::NetworkPolicyRuleDirection,
116    &str,
117)] = &[(
118    "open_ingress",
119    mz_sql::plan::NetworkPolicyRuleAction::Allow,
120    mz_sql::plan::NetworkPolicyRuleDirection::Ingress,
121    "0.0.0.0/0",
122)];
123
124static DEFAULT_USER_NETWORK_POLICY_PRIVILEGES: LazyLock<Vec<MzAclItem>> = LazyLock::new(|| {
125    vec![rbac::owner_privilege(
126        ObjectType::NetworkPolicy,
127        MZ_SYSTEM_ROLE_ID,
128    )]
129});
130
131static SYSTEM_SCHEMA_PRIVILEGES: LazyLock<Vec<MzAclItem>> = LazyLock::new(|| {
132    vec![
133        rbac::default_builtin_object_privilege(mz_sql::catalog::ObjectType::Schema),
134        MzAclItem {
135            grantee: MZ_SUPPORT_ROLE_ID,
136            grantor: MZ_SYSTEM_ROLE_ID,
137            acl_mode: AclMode::USAGE,
138        },
139        rbac::owner_privilege(mz_sql::catalog::ObjectType::Schema, MZ_SYSTEM_ROLE_ID),
140    ]
141});
142
143static MZ_CATALOG_SCHEMA: LazyLock<Schema> = LazyLock::new(|| Schema {
144    id: SchemaId::System(MZ_CATALOG_SCHEMA_ID),
145    oid: SCHEMA_MZ_CATALOG_OID,
146    database_id: None,
147    name: "mz_catalog".to_string(),
148    owner_id: MZ_SYSTEM_ROLE_ID,
149    privileges: SYSTEM_SCHEMA_PRIVILEGES.clone(),
150});
151static PG_CATALOG_SCHEMA: LazyLock<Schema> = LazyLock::new(|| Schema {
152    id: SchemaId::System(PG_CATALOG_SCHEMA_ID),
153    oid: SCHEMA_PG_CATALOG_OID,
154    database_id: None,
155    name: "pg_catalog".to_string(),
156    owner_id: MZ_SYSTEM_ROLE_ID,
157    privileges: SYSTEM_SCHEMA_PRIVILEGES.clone(),
158});
159static MZ_INTERNAL_SCHEMA: LazyLock<Schema> = LazyLock::new(|| Schema {
160    id: SchemaId::System(MZ_INTERNAL_SCHEMA_ID),
161    oid: SCHEMA_MZ_INTERNAL_OID,
162    database_id: None,
163    name: "mz_internal".to_string(),
164    owner_id: MZ_SYSTEM_ROLE_ID,
165    privileges: SYSTEM_SCHEMA_PRIVILEGES.clone(),
166});
167static INFORMATION_SCHEMA: LazyLock<Schema> = LazyLock::new(|| Schema {
168    id: SchemaId::System(INFORMATION_SCHEMA_ID),
169    oid: SCHEMA_INFORMATION_SCHEMA_OID,
170    database_id: None,
171    name: "information_schema".to_string(),
172    owner_id: MZ_SYSTEM_ROLE_ID,
173    privileges: SYSTEM_SCHEMA_PRIVILEGES.clone(),
174});
175static MZ_UNSAFE_SCHEMA: LazyLock<Schema> = LazyLock::new(|| Schema {
176    id: SchemaId::System(MZ_UNSAFE_SCHEMA_ID),
177    oid: SCHEMA_MZ_UNSAFE_OID,
178    database_id: None,
179    name: "mz_unsafe".to_string(),
180    owner_id: MZ_SYSTEM_ROLE_ID,
181    privileges: SYSTEM_SCHEMA_PRIVILEGES.clone(),
182});
183static MZ_CATALOG_UNSTABLE_SCHEMA: LazyLock<Schema> = LazyLock::new(|| Schema {
184    id: SchemaId::System(MZ_CATALOG_UNSTABLE_SCHEMA_ID),
185    oid: SCHEMA_MZ_CATALOG_UNSTABLE_OID,
186    database_id: None,
187    name: "mz_catalog_unstable".to_string(),
188    owner_id: MZ_SYSTEM_ROLE_ID,
189    privileges: SYSTEM_SCHEMA_PRIVILEGES.clone(),
190});
191static MZ_INTROSPECTION_SCHEMA: LazyLock<Schema> = LazyLock::new(|| Schema {
192    id: SchemaId::System(MZ_INTROSPECTION_SCHEMA_ID),
193    oid: SCHEMA_MZ_INTROSPECTION_OID,
194    database_id: None,
195    name: "mz_introspection".to_string(),
196    owner_id: MZ_SYSTEM_ROLE_ID,
197    privileges: SYSTEM_SCHEMA_PRIVILEGES.clone(),
198});
199static SYSTEM_SCHEMAS: LazyLock<BTreeMap<&str, &Schema>> = LazyLock::new(|| {
200    [
201        &*MZ_CATALOG_SCHEMA,
202        &*PG_CATALOG_SCHEMA,
203        &*MZ_INTERNAL_SCHEMA,
204        &*INFORMATION_SCHEMA,
205        &*MZ_UNSAFE_SCHEMA,
206        &*MZ_CATALOG_UNSTABLE_SCHEMA,
207        &*MZ_INTROSPECTION_SCHEMA,
208    ]
209    .into_iter()
210    .map(|s| (&*s.name, s))
211    .collect()
212});
213
214/// Initializes the Catalog with some default objects.
215#[mz_ore::instrument]
216pub(crate) async fn initialize(
217    tx: &mut Transaction<'_>,
218    options: &BootstrapArgs,
219    initial_ts: EpochMillis,
220    catalog_content_version: String,
221) -> Result<(), CatalogError> {
222    // Collect audit events so we can commit them once at the very end.
223    let mut audit_events = vec![];
224
225    for (name, next_id) in [
226        (USER_ID_ALLOC_KEY.to_string(), DEFAULT_ALLOCATOR_ID),
227        (SYSTEM_ID_ALLOC_KEY.to_string(), DEFAULT_ALLOCATOR_ID),
228        (
229            DATABASE_ID_ALLOC_KEY.to_string(),
230            MATERIALIZE_DATABASE_ID_VAL + 1,
231        ),
232        (
233            SCHEMA_ID_ALLOC_KEY.to_string(),
234            max(&[
235                MZ_CATALOG_SCHEMA_ID,
236                PG_CATALOG_SCHEMA_ID,
237                PUBLIC_SCHEMA_ID,
238                MZ_INTERNAL_SCHEMA_ID,
239                INFORMATION_SCHEMA_ID,
240                MZ_UNSAFE_SCHEMA_ID,
241                MZ_CATALOG_UNSTABLE_SCHEMA_ID,
242                MZ_INTROSPECTION_SCHEMA_ID,
243            ])
244            .expect("known to be non-empty")
245                + 1,
246        ),
247        (USER_ROLE_ID_ALLOC_KEY.to_string(), DEFAULT_ALLOCATOR_ID),
248        (
249            USER_CLUSTER_ID_ALLOC_KEY.to_string(),
250            DEFAULT_USER_CLUSTER_ID.inner_id() + 1,
251        ),
252        (
253            SYSTEM_CLUSTER_ID_ALLOC_KEY.to_string(),
254            DEFAULT_ALLOCATOR_ID,
255        ),
256        (
257            USER_REPLICA_ID_ALLOC_KEY.to_string(),
258            DEFAULT_USER_REPLICA_ID.inner_id()
259                + u64::from(options.default_cluster_replication_factor),
260        ),
261        (
262            SYSTEM_REPLICA_ID_ALLOC_KEY.to_string(),
263            DEFAULT_ALLOCATOR_ID,
264        ),
265        (
266            USER_NETWORK_POLICY_ID_ALLOC_KEY.to_string(),
267            DEFAULT_ALLOCATOR_ID,
268        ),
269        (
270            USER_CLUSTER_REPLICA_SIZE_ID_ALLOC_KEY.to_string(),
271            DEFAULT_ALLOCATOR_ID,
272        ),
273        (
274            SYSTEM_CLUSTER_REPLICA_SIZE_ID_ALLOC_KEY.to_string(),
275            DEFAULT_ALLOCATOR_ID,
276        ),
277        (AUDIT_LOG_ID_ALLOC_KEY.to_string(), DEFAULT_ALLOCATOR_ID),
278        (STORAGE_USAGE_ID_ALLOC_KEY.to_string(), DEFAULT_ALLOCATOR_ID),
279        (OID_ALLOC_KEY.to_string(), FIRST_USER_OID.into()),
280    ] {
281        tx.insert_id_allocator(name, next_id)?;
282    }
283
284    for role in BUILTIN_ROLES {
285        tx.insert_builtin_role(
286            role.id,
287            role.name.to_string(),
288            role.attributes.clone(),
289            RoleMembership::new(),
290            RoleVars::default(),
291            role.oid,
292        )?;
293    }
294    tx.insert_builtin_role(
295        RoleId::Public,
296        PUBLIC_ROLE_NAME.as_str().to_lowercase(),
297        RoleAttributesRaw::new(),
298        RoleMembership::new(),
299        RoleVars::default(),
300        ROLE_PUBLIC_OID,
301    )?;
302
303    // If provided, generate a new Id for the bootstrap role.
304    let bootstrap_role = if let Some(role) = &options.bootstrap_role {
305        let attributes = RoleAttributesRaw::new();
306        let membership = RoleMembership::new();
307        let vars = RoleVars::default();
308
309        let (id, oid) = tx.insert_user_role(
310            role.to_string(),
311            attributes.clone(),
312            membership.clone(),
313            vars.clone(),
314            &HashSet::new(),
315        )?;
316
317        audit_events.push((
318            mz_audit_log::EventType::Create,
319            mz_audit_log::ObjectType::Role,
320            mz_audit_log::EventDetails::IdNameV1(mz_audit_log::IdNameV1 {
321                id: id.to_string(),
322                name: role.to_string(),
323            }),
324        ));
325
326        Some(Role {
327            id,
328            name: role.to_string(),
329            attributes: attributes.into(),
330            membership,
331            vars,
332            oid,
333        })
334    } else {
335        None
336    };
337
338    let default_privileges = [
339        // mz_support needs USAGE privileges on all clusters, databases, and schemas for
340        // debugging.
341        DefaultPrivilege {
342            object: DefaultPrivilegeObject {
343                role_id: RoleId::Public,
344                database_id: None,
345                schema_id: None,
346                object_type: mz_sql::catalog::ObjectType::Cluster,
347            },
348            acl_item: DefaultPrivilegeAclItem {
349                grantee: MZ_SUPPORT_ROLE_ID,
350                acl_mode: AclMode::USAGE,
351            },
352        },
353        DefaultPrivilege {
354            object: DefaultPrivilegeObject {
355                role_id: RoleId::Public,
356                database_id: None,
357                schema_id: None,
358                object_type: mz_sql::catalog::ObjectType::Database,
359            },
360            acl_item: DefaultPrivilegeAclItem {
361                grantee: MZ_SUPPORT_ROLE_ID,
362                acl_mode: AclMode::USAGE,
363            },
364        },
365        DefaultPrivilege {
366            object: DefaultPrivilegeObject {
367                role_id: RoleId::Public,
368                database_id: None,
369                schema_id: None,
370                object_type: mz_sql::catalog::ObjectType::Schema,
371            },
372            acl_item: DefaultPrivilegeAclItem {
373                grantee: MZ_SUPPORT_ROLE_ID,
374                acl_mode: AclMode::USAGE,
375            },
376        },
377        DefaultPrivilege {
378            object: DefaultPrivilegeObject {
379                role_id: RoleId::Public,
380                database_id: None,
381                schema_id: None,
382                object_type: mz_sql::catalog::ObjectType::Type,
383            },
384            acl_item: DefaultPrivilegeAclItem {
385                grantee: RoleId::Public,
386                acl_mode: AclMode::USAGE,
387            },
388        },
389    ];
390    tx.set_default_privileges(default_privileges.to_vec())?;
391    for DefaultPrivilege { object, acl_item } in default_privileges {
392        let object_type = match object.object_type {
393            ObjectType::Table => mz_audit_log::ObjectType::Table,
394            ObjectType::View => mz_audit_log::ObjectType::View,
395            ObjectType::MaterializedView => mz_audit_log::ObjectType::MaterializedView,
396            ObjectType::Source => mz_audit_log::ObjectType::Source,
397            ObjectType::Sink => mz_audit_log::ObjectType::Sink,
398            ObjectType::MetricSink => mz_audit_log::ObjectType::MetricSink,
399            ObjectType::Index => mz_audit_log::ObjectType::Index,
400            ObjectType::Type => mz_audit_log::ObjectType::Type,
401            ObjectType::Role => mz_audit_log::ObjectType::Role,
402            ObjectType::Cluster => mz_audit_log::ObjectType::Cluster,
403            ObjectType::ClusterReplica => mz_audit_log::ObjectType::ClusterReplica,
404            ObjectType::Secret => mz_audit_log::ObjectType::Secret,
405            ObjectType::Connection => mz_audit_log::ObjectType::Connection,
406            ObjectType::Database => mz_audit_log::ObjectType::Database,
407            ObjectType::Schema => mz_audit_log::ObjectType::Schema,
408            ObjectType::Func => mz_audit_log::ObjectType::Func,
409            ObjectType::NetworkPolicy => mz_audit_log::ObjectType::NetworkPolicy,
410        };
411        audit_events.push((
412            mz_audit_log::EventType::Grant,
413            object_type,
414            mz_audit_log::EventDetails::AlterDefaultPrivilegeV1(
415                mz_audit_log::AlterDefaultPrivilegeV1 {
416                    role_id: object.role_id.to_string(),
417                    database_id: object.database_id.map(|id| id.to_string()),
418                    schema_id: object.schema_id.map(|id| id.to_string()),
419                    grantee_id: acl_item.grantee.to_string(),
420                    privileges: acl_item.acl_mode.to_string(),
421                },
422            ),
423        ));
424    }
425
426    let mut db_privileges = vec![
427        MzAclItem {
428            grantee: RoleId::Public,
429            grantor: MZ_SYSTEM_ROLE_ID,
430            acl_mode: AclMode::USAGE,
431        },
432        MzAclItem {
433            grantee: MZ_SUPPORT_ROLE_ID,
434            grantor: MZ_SYSTEM_ROLE_ID,
435            acl_mode: AclMode::USAGE,
436        },
437        rbac::owner_privilege(mz_sql::catalog::ObjectType::Database, MZ_SYSTEM_ROLE_ID),
438    ];
439    // Optionally add a privilege for the bootstrap role.
440    if let Some(role) = &bootstrap_role {
441        db_privileges.push(MzAclItem {
442            grantee: role.id.clone(),
443            grantor: MZ_SYSTEM_ROLE_ID,
444            acl_mode: rbac::all_object_privileges(SystemObjectType::Object(
445                mz_sql::catalog::ObjectType::Database,
446            )),
447        })
448    };
449
450    let materialize_db_oid = tx.allocate_oid(&HashSet::new())?;
451    tx.insert_database(
452        MATERIALIZE_DATABASE_ID,
453        "materialize",
454        MZ_SYSTEM_ROLE_ID,
455        db_privileges,
456        materialize_db_oid,
457    )?;
458    audit_events.extend([
459        (
460            mz_audit_log::EventType::Create,
461            mz_audit_log::ObjectType::Database,
462            mz_audit_log::EventDetails::IdNameV1(mz_audit_log::IdNameV1 {
463                id: MATERIALIZE_DATABASE_ID.to_string(),
464                name: "materialize".to_string(),
465            }),
466        ),
467        (
468            mz_audit_log::EventType::Grant,
469            mz_audit_log::ObjectType::Database,
470            mz_audit_log::EventDetails::UpdatePrivilegeV1(mz_audit_log::UpdatePrivilegeV1 {
471                object_id: ObjectId::Database(MATERIALIZE_DATABASE_ID).to_string(),
472                grantee_id: RoleId::Public.to_string(),
473                grantor_id: MZ_SYSTEM_ROLE_ID.to_string(),
474                privileges: AclMode::USAGE.to_string(),
475            }),
476        ),
477    ]);
478    // Optionally add a privilege for the bootstrap role.
479    if let Some(role) = &bootstrap_role {
480        let role_id: RoleId = role.id.clone();
481        audit_events.push((
482            mz_audit_log::EventType::Grant,
483            mz_audit_log::ObjectType::Database,
484            mz_audit_log::EventDetails::UpdatePrivilegeV1(mz_audit_log::UpdatePrivilegeV1 {
485                object_id: ObjectId::Database(MATERIALIZE_DATABASE_ID).to_string(),
486                grantee_id: role_id.to_string(),
487                grantor_id: MZ_SYSTEM_ROLE_ID.to_string(),
488                privileges: rbac::all_object_privileges(SystemObjectType::Object(
489                    mz_sql::catalog::ObjectType::Database,
490                ))
491                .to_string(),
492            }),
493        ));
494    }
495
496    let public_schema_oid = tx.allocate_oid(&HashSet::new())?;
497    let public_schema = Schema {
498        id: SchemaId::User(PUBLIC_SCHEMA_ID),
499        oid: public_schema_oid,
500        database_id: Some(MATERIALIZE_DATABASE_ID),
501        name: "public".to_string(),
502        owner_id: MZ_SYSTEM_ROLE_ID,
503        privileges: vec![
504            MzAclItem {
505                grantee: RoleId::Public,
506                grantor: MZ_SYSTEM_ROLE_ID,
507                acl_mode: AclMode::USAGE,
508            },
509            MzAclItem {
510                grantee: MZ_SUPPORT_ROLE_ID,
511                grantor: MZ_SYSTEM_ROLE_ID,
512                acl_mode: AclMode::USAGE,
513            },
514            rbac::owner_privilege(mz_sql::catalog::ObjectType::Schema, MZ_SYSTEM_ROLE_ID),
515        ]
516        .into_iter()
517        // Optionally add the bootstrap role to the public schema.
518        .chain(bootstrap_role.as_ref().map(|role| MzAclItem {
519            grantee: role.id.clone(),
520            grantor: MZ_SYSTEM_ROLE_ID,
521            acl_mode: rbac::all_object_privileges(SystemObjectType::Object(
522                mz_sql::catalog::ObjectType::Schema,
523            )),
524        }))
525        .collect(),
526    };
527
528    for schema in SYSTEM_SCHEMAS.values().chain(iter::once(&&public_schema)) {
529        tx.insert_schema(
530            schema.id,
531            schema.database_id,
532            schema.name.clone(),
533            schema.owner_id,
534            schema.privileges.clone(),
535            schema.oid,
536        )?;
537    }
538    audit_events.push((
539        mz_audit_log::EventType::Create,
540        mz_audit_log::ObjectType::Schema,
541        mz_audit_log::EventDetails::SchemaV2(mz_audit_log::SchemaV2 {
542            id: PUBLIC_SCHEMA_ID.to_string(),
543            name: "public".to_string(),
544            database_name: Some("materialize".to_string()),
545        }),
546    ));
547    if let Some(role) = &bootstrap_role {
548        let role_id: RoleId = role.id.clone();
549        audit_events.push((
550            mz_audit_log::EventType::Grant,
551            mz_audit_log::ObjectType::Schema,
552            mz_audit_log::EventDetails::UpdatePrivilegeV1(mz_audit_log::UpdatePrivilegeV1 {
553                object_id: ObjectId::Schema((
554                    ResolvedDatabaseSpecifier::Id(MATERIALIZE_DATABASE_ID),
555                    SchemaSpecifier::Id(SchemaId::User(PUBLIC_SCHEMA_ID)),
556                ))
557                .to_string(),
558                grantee_id: role_id.to_string(),
559                grantor_id: MZ_SYSTEM_ROLE_ID.to_string(),
560                privileges: rbac::all_object_privileges(SystemObjectType::Object(
561                    mz_sql::catalog::ObjectType::Schema,
562                ))
563                .to_string(),
564            }),
565        ));
566    }
567
568    let mut cluster_privileges = vec![
569        MzAclItem {
570            grantee: RoleId::Public,
571            grantor: MZ_SYSTEM_ROLE_ID,
572            acl_mode: AclMode::USAGE,
573        },
574        MzAclItem {
575            grantee: MZ_SUPPORT_ROLE_ID,
576            grantor: MZ_SYSTEM_ROLE_ID,
577            acl_mode: AclMode::USAGE,
578        },
579        rbac::owner_privilege(mz_sql::catalog::ObjectType::Cluster, MZ_SYSTEM_ROLE_ID),
580    ];
581
582    // Optionally add a privilege for the bootstrap role.
583    if let Some(role) = &bootstrap_role {
584        cluster_privileges.push(MzAclItem {
585            grantee: role.id.clone(),
586            grantor: MZ_SYSTEM_ROLE_ID,
587            acl_mode: rbac::all_object_privileges(SystemObjectType::Object(
588                mz_sql::catalog::ObjectType::Cluster,
589            )),
590        });
591    };
592
593    tx.insert_network_policy(
594        DEFAULT_USER_NETWORK_POLICY_ID,
595        DEFAULT_USER_NETWORK_POLICY_NAME.to_string(),
596        DEFAULT_USER_NETWORK_POLICY_RULES
597            .into_iter()
598            .map(|(name, action, direction, ip_str)| NetworkPolicyRule {
599                name: name.to_string(),
600                action: action.clone(),
601                direction: direction.clone(),
602                address: PolicyAddress(
603                    IpNet::from_str(ip_str).expect("default policy must provide valid ip"),
604                ),
605            })
606            .collect::<Vec<NetworkPolicyRule>>(),
607        DEFAULT_USER_NETWORK_POLICY_PRIVILEGES.clone(),
608        MZ_SYSTEM_ROLE_ID,
609        NETWORK_POLICIES_DEFAULT_POLICY_OID,
610    )?;
611    // We created a network policy with a prefined ID user(1) and OID. We need
612    // to increment the id alloc key. It should be safe to assume that there's
613    // no user(1), as a sanity check, we'll assert this is the case.
614    let id = tx.get_and_increment_id(USER_NETWORK_POLICY_ID_ALLOC_KEY.to_string())?;
615    assert!(DEFAULT_USER_NETWORK_POLICY_ID == NetworkPolicyId::User(id));
616
617    audit_events.extend([(
618        mz_audit_log::EventType::Create,
619        mz_audit_log::ObjectType::NetworkPolicy,
620        mz_audit_log::EventDetails::IdNameV1(mz_audit_log::IdNameV1 {
621            id: DEFAULT_USER_NETWORK_POLICY_ID.to_string(),
622            name: DEFAULT_USER_NETWORK_POLICY_NAME.to_string(),
623        }),
624    )]);
625
626    tx.insert_user_cluster(
627        DEFAULT_USER_CLUSTER_ID,
628        DEFAULT_USER_CLUSTER_NAME,
629        Vec::new(),
630        MZ_SYSTEM_ROLE_ID,
631        cluster_privileges,
632        default_cluster_config(options)?,
633        &HashSet::new(),
634    )?;
635    audit_events.extend([
636        (
637            mz_audit_log::EventType::Create,
638            mz_audit_log::ObjectType::Cluster,
639            mz_audit_log::EventDetails::IdNameV1(mz_audit_log::IdNameV1 {
640                id: DEFAULT_USER_CLUSTER_ID.to_string(),
641                name: DEFAULT_USER_CLUSTER_NAME.to_string(),
642            }),
643        ),
644        (
645            mz_audit_log::EventType::Grant,
646            mz_audit_log::ObjectType::Cluster,
647            mz_audit_log::EventDetails::UpdatePrivilegeV1(mz_audit_log::UpdatePrivilegeV1 {
648                object_id: ObjectId::Cluster(DEFAULT_USER_CLUSTER_ID).to_string(),
649                grantee_id: RoleId::Public.to_string(),
650                grantor_id: MZ_SYSTEM_ROLE_ID.to_string(),
651                privileges: AclMode::USAGE.to_string(),
652            }),
653        ),
654    ]);
655
656    // Optionally add a privilege for the bootstrap role.
657    if let Some(role) = &bootstrap_role {
658        let role_id: RoleId = role.id.clone();
659        audit_events.push((
660            mz_audit_log::EventType::Grant,
661            mz_audit_log::ObjectType::Cluster,
662            mz_audit_log::EventDetails::UpdatePrivilegeV1(mz_audit_log::UpdatePrivilegeV1 {
663                object_id: ObjectId::Cluster(DEFAULT_USER_CLUSTER_ID).to_string(),
664                grantee_id: role_id.to_string(),
665                grantor_id: MZ_SYSTEM_ROLE_ID.to_string(),
666                privileges: rbac::all_object_privileges(SystemObjectType::Object(
667                    mz_sql::catalog::ObjectType::Cluster,
668                ))
669                .to_string(),
670            }),
671        ));
672    }
673
674    for i in 0..options.default_cluster_replication_factor {
675        let replica_id = ReplicaId::User(DEFAULT_USER_REPLICA_ID.inner_id() + u64::from(i));
676        let replica_name = format!("r{}", i + 1);
677        tx.insert_cluster_replica_with_id(
678            DEFAULT_USER_CLUSTER_ID,
679            replica_id,
680            &replica_name,
681            default_replica_config(options)?,
682            MZ_SYSTEM_ROLE_ID,
683        )?;
684        audit_events.push((
685            mz_audit_log::EventType::Create,
686            mz_audit_log::ObjectType::ClusterReplica,
687            mz_audit_log::EventDetails::CreateClusterReplicaV4(
688                mz_audit_log::CreateClusterReplicaV4 {
689                    cluster_id: DEFAULT_USER_CLUSTER_ID.to_string(),
690                    cluster_name: DEFAULT_USER_CLUSTER_NAME.to_string(),
691                    replica_name,
692                    replica_id: Some(replica_id.to_string()),
693                    logical_size: options.default_cluster_replica_size.to_string(),
694                    billed_as: None,
695                    internal: false,
696                    reason: CreateOrDropClusterReplicaReasonV1::System,
697                    scheduling_policies: None,
698                },
699            ),
700        ));
701    }
702
703    let system_privileges = [MzAclItem {
704        grantee: MZ_SYSTEM_ROLE_ID,
705        grantor: MZ_SYSTEM_ROLE_ID,
706        acl_mode: rbac::all_object_privileges(SystemObjectType::System),
707    }]
708    .into_iter()
709    // Optionally add system privileges for the bootstrap role.
710    .chain(bootstrap_role.as_ref().map(|role| MzAclItem {
711        grantee: role.id.clone(),
712        grantor: MZ_SYSTEM_ROLE_ID,
713        acl_mode: rbac::all_object_privileges(SystemObjectType::System),
714    }));
715    tx.set_system_privileges(system_privileges.clone().collect())?;
716    for system_privilege in system_privileges {
717        audit_events.push((
718            mz_audit_log::EventType::Grant,
719            mz_audit_log::ObjectType::System,
720            mz_audit_log::EventDetails::UpdatePrivilegeV1(mz_audit_log::UpdatePrivilegeV1 {
721                object_id: "SYSTEM".to_string(),
722                grantee_id: system_privilege.grantee.to_string(),
723                grantor_id: system_privilege.grantor.to_string(),
724                privileges: system_privilege.acl_mode.to_string(),
725            }),
726        ));
727    }
728
729    // Allocate an ID for each audit log event.
730    let mut audit_events_with_id = Vec::with_capacity(audit_events.len());
731    for (ty, obj, details) in audit_events {
732        let id = tx.get_and_increment_id(AUDIT_LOG_ID_ALLOC_KEY.to_string())?;
733        audit_events_with_id.push((id, ty, obj, details));
734    }
735
736    for (id, event_type, object_type, details) in audit_events_with_id {
737        tx.insert_audit_log_event(VersionedEvent::V1(EventV1 {
738            id,
739            event_type,
740            object_type,
741            details,
742            user: None,
743            occurred_at: initial_ts,
744        }));
745    }
746
747    for (key, value) in [
748        (USER_VERSION_KEY.to_string(), CATALOG_VERSION),
749        (SYSTEM_CONFIG_SYNCED_KEY.to_string(), 0),
750    ] {
751        tx.insert_config(key, value)?;
752    }
753
754    for (name, value) in [
755        (
756            CATALOG_CONTENT_VERSION_KEY.to_string(),
757            catalog_content_version,
758        ),
759        (
760            BUILTIN_MIGRATION_SHARD_KEY.to_string(),
761            ShardId::new().to_string(),
762        ),
763        (
764            EXPRESSION_CACHE_SHARD_KEY.to_string(),
765            ShardId::new().to_string(),
766        ),
767    ] {
768        tx.set_setting(name, Some(value))?;
769    }
770
771    if tx
772        .get_setting(MOCK_AUTHENTICATION_NONCE_KEY.to_string())
773        .is_none()
774    {
775        let mut nonce = [0u8; 24];
776        openssl::rand::rand_bytes(&mut nonce).expect("random number generation failed");
777        tx.set_setting(
778            MOCK_AUTHENTICATION_NONCE_KEY.to_string(),
779            Some(BASE64_STANDARD.encode(nonce)),
780        )?;
781    }
782
783    Ok(())
784}
785
786pub fn resolve_system_schema(name: &str) -> &Schema {
787    SYSTEM_SCHEMAS
788        .get(name)
789        .unwrap_or_else(|| panic!("unable to resolve system schema: {name}"))
790}
791
792/// Defines the default config for a Cluster.
793fn default_cluster_config(args: &BootstrapArgs) -> Result<ClusterConfig, CatalogError> {
794    Ok(ClusterConfig {
795        variant: ClusterVariant::Managed(ClusterVariantManaged {
796            size: args.default_cluster_replica_size.to_string(),
797            replication_factor: args.default_cluster_replication_factor,
798            availability_zones: vec![],
799            logging: ReplicaLogging {
800                log_logging: false,
801                interval: Some(Duration::from_secs(1)),
802            },
803            arrangement_compression: false,
804            optimizer_feature_overrides: Default::default(),
805            schedule: Default::default(),
806            auto_scaling_strategy: None,
807            reconfiguration: None,
808            burst: None,
809        }),
810        workload_class: None,
811    })
812}
813
814/// Defines the default config for a Cluster Replica.
815fn default_replica_config(args: &BootstrapArgs) -> Result<ReplicaConfig, CatalogError> {
816    Ok(ReplicaConfig {
817        location: ReplicaLocation::Managed {
818            size: args.default_cluster_replica_size.to_string(),
819            availability_zones: Vec::new(),
820            internal: false,
821            billed_as: None,
822            pending: false,
823        },
824        logging: ReplicaLogging {
825            log_logging: false,
826            interval: Some(Duration::from_secs(1)),
827        },
828        arrangement_compression: false,
829    })
830}
mz_catalog/durable/initialize.rs

mz_catalog/durable/
initialize.rs
