Function purify_alter_source_refresh_references

src/sql/src/pure.rs:1665–1757

(
    desc: SourceDesc,
    resolved_source_name: ResolvedItemName,
    storage_configuration: &StorageConfiguration,
)

}

async fn purify_alter_source_refresh_references(
    desc: SourceDesc,
    resolved_source_name: ResolvedItemName,
    storage_configuration: &StorageConfiguration,
) -> Result<PurifiedStatement, PlanError> {
    let retrieved_source_references = match desc.connection {
        GenericSourceConnection::Postgres(pg_source_connection) => {
            // Get PostgresConnection for generating subsources.
            let pg_connection = &pg_source_connection.connection;

            let config = pg_connection
                .config(
                    &storage_configuration.connection_context.secrets_reader,
                    storage_configuration,
                    InTask::No,
                )
                .await?;

            let client = config
                .connect(
                    "postgres_purification",
                    &storage_configuration.connection_context.ssh_tunnel_manager,
                )
                .await?;
            let reference_client = SourceReferenceClient::Postgres {
                client: &client,
                publication: &pg_source_connection.publication,
                database: &pg_connection.database,
            };
            reference_client.get_source_references().await?
        }
        GenericSourceConnection::MySql(mysql_source_connection) => {
            let mysql_connection = &mysql_source_connection.connection;
            let config = mysql_connection
                .config(
                    &storage_configuration.connection_context.secrets_reader,
                    storage_configuration,
                    InTask::No,
                )
                .await?;

            let mut conn = config
                .connect(
                    "mysql purification",
                    &storage_configuration.connection_context.ssh_tunnel_manager,
                )
                .await?;

            let reference_client = SourceReferenceClient::MySql {
                conn: &mut conn,
                include_system_schemas: false,
            };
            reference_client.get_source_references().await?
        }
        GenericSourceConnection::SqlServer(sql_server_source) => {
            // Open a connection to the upstream SQL Server instance.
            let sql_server_connection = &sql_server_source.connection;
            let config = sql_server_connection