Method handle_startup

src/adapter/src/coord/command_handler.rs:789–899

(
        &mut self,
        tx: oneshot::Sender<Result<StartupResponse, AdapterError>>,
        user: User,
        conn_id: ConnectionId,
        secret_key: u32,
        uuid: uuid::Uuid,
        c

    #[mz_ore::instrument(level = "debug")]
    async fn handle_startup(
        &mut self,
        tx: oneshot::Sender<Result<StartupResponse, AdapterError>>,
        user: User,
        conn_id: ConnectionId,
        secret_key: u32,
        uuid: uuid::Uuid,
        client_ip: Option<IpAddr>,
        application_name: String,
        notice_tx: mpsc::UnboundedSender<AdapterNotice>,
    ) {
        // Early return if successful, otherwise cleanup any possible state.
        match self
            .handle_startup_inner(&user, &conn_id, &client_ip, &notice_tx)
            .await
        {
            Ok((role_id, superuser_attribute, session_defaults)) => {
                let session_type = metrics::session_type_label_value(&user);
                self.metrics
                    .active_sessions
                    .with_label_values(&[session_type])
                    .inc();
                let conn = ConnMeta {
                    secret_key,
                    notice_tx,
                    drop_sinks: BTreeSet::new(),
                    pending_cluster_alters: BTreeSet::new(),
                    connected_at: self.now(),
                    user,
                    application_name,
                    uuid,
                    client_ip,
                    conn_id: conn_id.clone(),
                    authenticated_role: role_id,
                    deferred_lock: None,
                };
                let update = self.catalog().state().pack_session_update(&conn, Diff::ONE);
                let update = self.catalog().state().resolve_builtin_table_update(update);
                self.begin_session_for_statement_logging(&conn);
                self.active_conns.insert(conn_id.clone(), conn);

                // Note: Do NOT await the notify here, we pass this back to
                // whatever requested the startup to prevent blocking startup
                // and the Coordinator on a builtin table update.
                let updates = vec![update];
                // It's not a hard error if our list is missing a builtin table, but we want to
                // make sure these two things stay in-sync.
                if mz_ore::assert::soft_assertions_enabled() {
                    let required_tables: BTreeSet<_> = super::appends::REQUIRED_BUILTIN_TABLES
                        .iter()
                        .map(|table| self.catalog().resolve_builtin_table(*table))
                        .collect();
                    let updates_tracked = updates
                        .iter()
                        .all(|update| required_tables.contains(&update.id));
                    let all_mz_internal = super::appends::REQUIRED_BUILTIN_TABLES
                        .iter()
                        .all(|table| table.schema == MZ_INTERNAL_SCHEMA);