mz_adapter/coord/catalog_implications/

parsed_state_updates.rs

// Copyright Materialize, Inc. and contributors. All rights reserved.
//
// Use of this software is governed by the Business Source License
// included in the LICENSE file.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0.

//! Utilities for parsing and augmenting "raw" catalog changes
//! ([StateUpdateKind]), so that we can update in-memory adapter state, apply
//! implications, and apply derived commands to the controller(s).
//!
//! See [parse_state_update] for details.

use mz_catalog::builtin::BUILTIN_LOG_LOOKUP;
use mz_catalog::memory::objects::{
    CatalogItem, DataSourceDesc, StateDiff, StateUpdate, StateUpdateKind,
};
use mz_catalog::{durable, memory};
use mz_compute_client::logging::LogVariant;
use mz_controller_types::ClusterId;
use mz_ore::instrument;
use mz_repr::{CatalogItemId, GlobalId, Timestamp};
use mz_storage_types::connections::inline::IntoInlineConnection;
use mz_storage_types::sources::GenericSourceConnection;

// DO NOT add any more imports from `crate` outside of `crate::catalog`.
use crate::catalog::CatalogState;

/// An update that needs to be applied to a controller.
#[derive(Debug, Clone)]
pub struct ParsedStateUpdate {
    pub kind: ParsedStateUpdateKind,
    pub ts: Timestamp,
    pub diff: StateDiff,
}

/// An update that needs to be applied to a controller.
#[derive(Debug, Clone)]
pub enum ParsedStateUpdateKind {
    Item {
        durable_item: durable::objects::Item,
        parsed_item: memory::objects::CatalogItem,
        connection: Option<GenericSourceConnection>,
        parsed_full_name: String,
    },
    Cluster {
        durable_cluster: durable::objects::Cluster,
        parsed_cluster: memory::objects::Cluster,
    },
    ClusterReplica {
        durable_cluster_replica: durable::objects::ClusterReplica,
        parsed_cluster_replica: memory::objects::ClusterReplica,
    },
    IntrospectionSourceIndex {
        cluster_id: ClusterId,
        log: LogVariant,
        index_id: GlobalId,
    },
    /// A replica-scoped system-parameter override changed. The implication
    /// re-pushes the complete per-replica dyncfg layer from the catalog working
    /// copy, so it does not consume `durable`. We keep the row only so it shows
    /// up in the `tracing::trace!` of the parsed update.
    ReplicaSystemConfiguration {
        durable: durable::objects::ReplicaSystemConfiguration,
    },
    /// An environment-wide system-parameter changed. The implication re-runs the
    /// `SystemVars` callbacks against the committed values, so it does not
    /// consume `durable`. We keep the row only for the `tracing::trace!`.
    SystemConfiguration {
        durable: durable::objects::SystemConfiguration,
    },
}

/// Potentially generate a [ParsedStateUpdate] that corresponds to the given
/// change to the catalog.
///
/// This technically doesn't "parse" the given state update but uses the given
/// in-memory [CatalogState] as a shortcut. It already contains the parsed
/// representation of the item. In theory, we could re-construct the parsed
/// items by hand if we're given all the changes that lead to a given catalog
/// state.
///
/// For changes with a positive diff, the given [CatalogState] must reflect the
/// catalog state _after_ applying the catalog change to the catalog. For
/// negative changes, the given [CatalogState] must reflect the catalog state
/// _before_ applying the changes. This is so that we can easily extract the
/// state of an object before it is removed.
///
/// Will return `None` if the given catalog change is purely internal to the
/// catalog and does not have implications for anything else.
#[instrument(level = "debug")]
pub fn parse_state_update(
    catalog: &CatalogState,
    state_update: StateUpdate,
) -> Option<ParsedStateUpdate> {
    let kind = match state_update.kind {
        StateUpdateKind::Item(item) => Some(parse_item_update(catalog, item)),
        StateUpdateKind::Cluster(cluster) => Some(parse_cluster_update(catalog, cluster)),
        StateUpdateKind::ClusterReplica(replica) => {
            Some(parse_cluster_replica_update(catalog, replica))
        }
        StateUpdateKind::IntrospectionSourceIndex(isi) => {
            Some(parse_introspection_source_index_update(isi))
        }
        StateUpdateKind::ReplicaSystemConfiguration(durable) => {
            Some(ParsedStateUpdateKind::ReplicaSystemConfiguration { durable })
        }
        StateUpdateKind::SystemConfiguration(durable) => {
            Some(ParsedStateUpdateKind::SystemConfiguration { durable })
        }
        _ => {
            // The controllers are currently not interested in other kinds of
            // changes to the catalog. Cluster-scoped system-parameter overrides
            // are read at plan time and have no controller effect, so they stay
            // here too.
            None
        }
    };

    kind.map(|kind| ParsedStateUpdate {
        kind,
        ts: state_update.ts,
        diff: state_update.diff,
    })
}

fn parse_item_update(
    catalog: &CatalogState,
    durable_item: durable::objects::Item,
) -> ParsedStateUpdateKind {
    let (parsed_item, connection, parsed_full_name) =
        parse_item_update_common(catalog, &durable_item.id);

    ParsedStateUpdateKind::Item {
        durable_item,
        parsed_item,
        connection,
        parsed_full_name,
    }
}

fn parse_item_update_common(
    catalog: &CatalogState,
    item_id: &CatalogItemId,
) -> (CatalogItem, Option<GenericSourceConnection>, String) {
    let entry = catalog.get_entry(item_id);

    let parsed_item = entry.item().clone();
    let parsed_full_name = catalog
        .resolve_full_name(entry.name(), entry.conn_id())
        .to_string();

    let connection = match &parsed_item {
        memory::objects::CatalogItem::Source(source) => {
            if let DataSourceDesc::Ingestion { desc, .. }
            | DataSourceDesc::OldSyntaxIngestion { desc, .. } = &source.data_source
            {
                Some(desc.connection.clone().into_inline_connection(catalog))
            } else {
                None
            }
        }
        _ => None,
    };

    (parsed_item, connection, parsed_full_name)
}

fn parse_cluster_update(
    catalog: &CatalogState,
    durable_cluster: durable::objects::Cluster,
) -> ParsedStateUpdateKind {
    let parsed_cluster = catalog.get_cluster(durable_cluster.id);

    ParsedStateUpdateKind::Cluster {
        durable_cluster,
        parsed_cluster: parsed_cluster.clone(),
    }
}

fn parse_introspection_source_index_update(
    isi: durable::objects::IntrospectionSourceIndex,
) -> ParsedStateUpdateKind {
    let builtin_log = BUILTIN_LOG_LOOKUP
        .get(isi.name.as_str())
        .expect("introspection source index must reference a known log");

    ParsedStateUpdateKind::IntrospectionSourceIndex {
        cluster_id: isi.cluster_id,
        log: builtin_log.variant.clone(),
        index_id: isi.index_id,
    }
}

fn parse_cluster_replica_update(
    catalog: &CatalogState,
    durable_cluster_replica: durable::objects::ClusterReplica,
) -> ParsedStateUpdateKind {
    let parsed_cluster_replica = catalog.get_cluster_replica(
        durable_cluster_replica.cluster_id,
        durable_cluster_replica.replica_id,
    );

    ParsedStateUpdateKind::ClusterReplica {
        durable_cluster_replica,
        parsed_cluster_replica: parsed_cluster_replica.clone(),
    }
}

#[cfg(test)]
mod tests {
    use mz_catalog::durable::objects::{ReplicaSystemConfiguration, SystemConfiguration};
    use mz_catalog::memory::objects::{StateDiff, StateUpdate, StateUpdateKind};
    use mz_controller_types::ReplicaId;
    use mz_repr::Timestamp;

    use crate::catalog::Catalog;

    use super::{ParsedStateUpdateKind, parse_state_update};

    /// A replica-scoped system-parameter change must produce a parsed update so
    /// the controller push fires. It was previously dropped as a change the
    /// controllers were not interested in.
    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function on OS `linux`
    async fn replica_system_configuration_is_parsed() {
        Catalog::with_debug(|catalog| async move {
            let durable = ReplicaSystemConfiguration {
                replica_id: ReplicaId::User(1),
                name: "persist_pager".to_string(),
                value: "on".to_string(),
            };
            let update = StateUpdate {
                kind: StateUpdateKind::ReplicaSystemConfiguration(durable),
                ts: Timestamp::MIN,
                diff: StateDiff::Addition,
            };
            let parsed = parse_state_update(catalog.state(), update)
                .expect("replica system configuration must produce a parsed update");
            assert!(matches!(
                parsed.kind,
                ParsedStateUpdateKind::ReplicaSystemConfiguration { .. }
            ));
        })
        .await
    }

    /// An environment-wide system-parameter change must produce a parsed update
    /// so the `SystemVars` callback notification fires from
    /// `apply_catalog_implications`, including on a follower `environmentd` that
    /// only replays the committed diff. It was previously dropped as a change the
    /// controllers were not interested in.
    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function on OS `linux`
    async fn system_configuration_is_parsed() {
        Catalog::with_debug(|catalog| async move {
            let durable = SystemConfiguration {
                name: "max_connections".to_string(),
                value: "42".to_string(),
            };
            let update = StateUpdate {
                kind: StateUpdateKind::SystemConfiguration(durable),
                ts: Timestamp::MIN,
                diff: StateDiff::Addition,
            };
            let parsed = parse_state_update(catalog.state(), update)
                .expect("system configuration must produce a parsed update");
            assert!(matches!(
                parsed.kind,
                ParsedStateUpdateKind::SystemConfiguration { .. }
            ));
        })
        .await
    }
}