mz_catalog/

expr_cache.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.

//! A cache for optimized expressions.

use std::collections::{BTreeMap, BTreeSet};
use std::future::Future;
use std::sync::Arc;

use bytes::Bytes;
use mz_compute_types::dataflows::DataflowDescription;
use mz_durable_cache::{DurableCache, DurableCacheCodec};
use mz_dyncfg::ConfigSet;
use mz_expr::OptimizedMirRelationExpr;
use mz_ore::channel::trigger;
use mz_ore::soft_panic_or_log;
use mz_ore::task::spawn;
use mz_persist_client::PersistClient;
use mz_persist_client::cli::admin::{
    EXPRESSION_CACHE_FORCE_COMPACTION_FUEL, EXPRESSION_CACHE_FORCE_COMPACTION_WAIT,
};
use mz_persist_types::codec_impls::VecU8Schema;
use mz_persist_types::{Codec, ShardId};
use mz_repr::optimize::OptimizerFeatures;
use mz_repr::{GlobalId, RelationVersion};
use mz_transform::dataflow::DataflowMetainfo;
use mz_transform::notice::OptimizerNotice;
use semver::Version;
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;
use tracing::{debug, info, warn};

#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    Serialize,
    Deserialize
)]
enum ExpressionType {
    Local,
    Global,
}

/// The data that is cached per catalog object as a result of local optimizations.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LocalExpressions {
    pub local_mir: OptimizedMirRelationExpr,
    pub optimizer_features: OptimizerFeatures,
    /// The owning item's latest version, see [`ExpressionCache::open`].
    pub item_version: RelationVersion,
}

/// The data that is cached per catalog object as a result of global optimizations.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct GlobalExpressions {
    pub global_mir: DataflowDescription<OptimizedMirRelationExpr>,
    pub physical_plan: mz_compute_types::dataflows::LirDataflowDescription,
    pub dataflow_metainfos: DataflowMetainfo<Arc<OptimizerNotice>>,
    pub optimizer_features: OptimizerFeatures,
    /// The owning item's latest version, see [`ExpressionCache::open`].
    pub item_version: RelationVersion,
}

impl GlobalExpressions {
    fn index_imports(&self) -> impl Iterator<Item = &GlobalId> {
        self.global_mir
            .index_imports
            .keys()
            .chain(self.physical_plan.index_imports.keys())
    }
}

/// The item version to record with an item's cached expressions, the latest of its `versions`.
pub fn latest_item_version(versions: &BTreeMap<RelationVersion, GlobalId>) -> RelationVersion {
    versions
        .last_key_value()
        .map_or_else(RelationVersion::root, |(version, _)| *version)
}

#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    Serialize,
    Deserialize
)]
struct CacheKey {
    build_version: String,
    id: GlobalId,
    expr_type: ExpressionType,
}

#[derive(Debug, PartialEq, Eq)]
struct ExpressionCodec;

impl DurableCacheCodec for ExpressionCodec {
    type Key = CacheKey;
    // We use a raw bytes instead of `Expressions` so that there is no backwards compatibility
    // requirement on `Expressions` between build versions.
    type Val = Bytes;
    type KeyCodec = Bytes;
    type ValCodec = Bytes;

    fn schemas() -> (
        <Self::KeyCodec as Codec>::Schema,
        <Self::ValCodec as Codec>::Schema,
    ) {
        (VecU8Schema::default(), VecU8Schema::default())
    }

    fn encode(key: &Self::Key, val: &Self::Val) -> (Self::KeyCodec, Self::ValCodec) {
        let key = bincode::serialize(key).expect("must serialize");
        (Bytes::from(key), val.clone())
    }

    fn decode(key: &Self::KeyCodec, val: &Self::ValCodec) -> (Self::Key, Self::Val) {
        let key = bincode::deserialize(key).expect("must deserialize");
        (key, val.clone())
    }
}

/// Counts of the entries [`ExpressionCache::open`] removed, by reason.
#[derive(Debug, Default)]
struct RemovedEntries {
    unparsable_build_version: usize,
    dropped_item: usize,
    item_version_mismatch: usize,
    dropped_index_import: usize,
    prior_build_version: usize,
}

/// Configuration needed to initialize an [`ExpressionCache`].
#[derive(Debug, Clone)]
pub struct ExpressionCacheConfig {
    pub build_version: Version,
    pub persist: PersistClient,
    pub shard_id: ShardId,
    /// Every `GlobalId` that may have entries, mapped to the latest version of the item it
    /// belongs to. All ids of an item map to that same version, which is the one its entries
    /// record, rather than the version the id itself denotes.
    pub current_items: BTreeMap<GlobalId, RelationVersion>,
    /// Whether to durably remove the entries of previous build versions.
    pub remove_prior_versions: bool,
    pub compact_shard: bool,
    pub dyncfgs: ConfigSet,
}

/// A durable cache of optimized expressions.
pub struct ExpressionCache {
    build_version: Version,
    durable_cache: DurableCache<ExpressionCodec>,
}

impl ExpressionCache {
    /// Creates a new [`ExpressionCache`] and reconciles all entries in current build version.
    /// Reconciliation removes entries whose id is not in `current_items` or whose recorded
    /// item version differs from the current one, and global entries that import an index
    /// that is not in `current_items`.
    ///
    /// An entry records the item version it was optimized for, because applying a materialized
    /// view replacement changes an item's definition while the item keeps its `GlobalId`s and
    /// bumps its [`RelationVersion`]. Dropping entries recorded at another version therefore
    /// catches stale entries whichever process wrote them. That includes the replacement's own
    /// entries: they are recorded at its root version, and the apply makes its id a later
    /// version of the target.
    ///
    /// The apply also invalidates the target's entries, but that alone would not do: it only
    /// reaches entries under the applying process's build version, while a 0dt deployment in
    /// its read-only phase caches under its own build version and reads those entries again
    /// when it reboots after promotion.
    ///
    /// If `remove_prior_versions` is `true`, then the entries of all previous build versions are
    /// durably removed from the cache.
    ///
    /// If `compact_shard` is `true`, then this function will block on fully compacting the backing
    /// persist shard.
    ///
    /// Returns all cached expressions in the current build version, after reconciliation.
    pub async fn open(
        ExpressionCacheConfig {
            build_version,
            persist,
            shard_id,
            current_items,
            remove_prior_versions,
            compact_shard,
            dyncfgs,
        }: ExpressionCacheConfig,
    ) -> (
        Self,
        BTreeMap<GlobalId, LocalExpressions>,
        BTreeMap<GlobalId, GlobalExpressions>,
    ) {
        let durable_cache = DurableCache::new(&persist, shard_id, "expressions").await;
        let mut cache = Self {
            build_version,
            durable_cache,
        };

        const RETRIES: usize = 100;
        for _ in 0..RETRIES {
            match cache
                .try_open(
                    &current_items,
                    remove_prior_versions,
                    compact_shard,
                    &dyncfgs,
                )
                .await
            {
                Ok((local_expressions, global_expressions)) => {
                    return (cache, local_expressions, global_expressions);
                }
                Err(err) => debug!("failed to open cache: {err} ... retrying"),
            }
        }

        panic!("Unable to open expression cache after {RETRIES} retries");
    }

    async fn try_open(
        &mut self,
        current_items: &BTreeMap<GlobalId, RelationVersion>,
        remove_prior_versions: bool,
        compact_shard: bool,
        dyncfgs: &ConfigSet,
    ) -> Result<
        (
            BTreeMap<GlobalId, LocalExpressions>,
            BTreeMap<GlobalId, GlobalExpressions>,
        ),
        mz_durable_cache::Error,
    > {
        let mut keys_to_remove = Vec::new();
        let mut local_expressions = BTreeMap::new();
        let mut global_expressions = BTreeMap::new();
        let mut removed = RemovedEntries::default();
        let mut unreadable = 0;

        for (key, expressions) in self.durable_cache.entries_local() {
            let build_version = match key.build_version.parse::<Version>() {
                Ok(build_version) => build_version,
                Err(err) => {
                    warn!("unable to parse build version: {key:?}: {err:?}");
                    removed.unparsable_build_version += 1;
                    keys_to_remove.push((key.clone(), None));
                    continue;
                }
            };
            if build_version == self.build_version {
                // Only deserialize the current build version.
                match key.expr_type {
                    ExpressionType::Local => {
                        let expressions: LocalExpressions = match bincode::deserialize(expressions)
                        {
                            Ok(expressions) => expressions,
                            Err(err) => {
                                soft_panic_or_log!(
                                    "unable to deserialize local expressions: ({key:?}, {expressions:?}): {err:?}"
                                );
                                unreadable += 1;
                                continue;
                            }
                        };
                        match current_items.get(&key.id) {
                            None => {
                                removed.dropped_item += 1;
                                keys_to_remove.push((key.clone(), None));
                            }
                            // The entry was optimized for another definition of the item.
                            Some(version) if *version != expressions.item_version => {
                                removed.item_version_mismatch += 1;
                                keys_to_remove.push((key.clone(), None));
                            }
                            Some(_) => {
                                local_expressions.insert(key.id, expressions);
                            }
                        }
                    }
                    ExpressionType::Global => {
                        let expressions: GlobalExpressions = match bincode::deserialize(expressions)
                        {
                            Ok(expressions) => expressions,
                            Err(err) => {
                                soft_panic_or_log!(
                                    "unable to deserialize global expressions: ({key:?}, {expressions:?}): {err:?}"
                                );
                                unreadable += 1;
                                continue;
                            }
                        };
                        match current_items.get(&key.id) {
                            None => {
                                removed.dropped_item += 1;
                                keys_to_remove.push((key.clone(), None));
                            }
                            Some(version) if *version != expressions.item_version => {
                                removed.item_version_mismatch += 1;
                                keys_to_remove.push((key.clone(), None));
                            }
                            // The plan reads an index that was dropped while the dataflow kept
                            // running, so the consumer has to be re-optimized at the next boot.
                            Some(_)
                                if !expressions
                                    .index_imports()
                                    .all(|id| current_items.contains_key(id)) =>
                            {
                                removed.dropped_index_import += 1;
                                keys_to_remove.push((key.clone(), None));
                            }
                            Some(_) => {
                                global_expressions.insert(key.id, expressions);
                            }
                        }
                    }
                }
            } else if remove_prior_versions {
                removed.prior_build_version += 1;
                keys_to_remove.push((key.clone(), None));
            }
        }
        let keys_to_remove: Vec<_> = keys_to_remove
            .iter()
            .map(|(key, expressions)| (key, expressions.as_ref()))
            .collect();
        self.durable_cache.try_set_many(&keys_to_remove).await?;
        let RemovedEntries {
            unparsable_build_version,
            dropped_item,
            item_version_mismatch,
            dropped_index_import,
            prior_build_version,
        } = removed;
        info!(
            build_version = %self.build_version,
            remove_prior_versions,
            kept_local = local_expressions.len(),
            kept_global = global_expressions.len(),
            skipped_unreadable = unreadable,
            removed_unparsable_build_version = unparsable_build_version,
            removed_dropped_item = dropped_item,
            removed_item_version_mismatch = item_version_mismatch,
            removed_dropped_index_import = dropped_index_import,
            removed_prior_build_version = prior_build_version,
            "reconciled the expression cache"
        );

        if remove_prior_versions {
            // We've purged old build versions from the cache. Upgrade the backing Persist version
            // as well.
            self.durable_cache.upgrade_version().await;
        }

        if compact_shard {
            let fuel = EXPRESSION_CACHE_FORCE_COMPACTION_FUEL.handle(dyncfgs);
            let wait = EXPRESSION_CACHE_FORCE_COMPACTION_WAIT.handle(dyncfgs);
            self.durable_cache
                .dangerous_compact_shard(move || fuel.get(), move || wait.get())
                .await;
        }

        Ok((local_expressions, global_expressions))
    }

    /// Durably removes all entries given by `invalidate_ids` and inserts `new_local_expressions`
    /// and `new_global_expressions` into current build version.
    ///
    /// If there is a duplicate ID in both `invalidate_ids` and one of the new expressions vector,
    /// then the final value will be taken from the new expressions vector.
    async fn update(
        &mut self,
        new_local_expressions: Vec<(GlobalId, LocalExpressions)>,
        new_global_expressions: Vec<(GlobalId, GlobalExpressions)>,
        invalidate_ids: BTreeSet<GlobalId>,
    ) {
        let mut entries = BTreeMap::new();
        let build_version = self.build_version.to_string();
        // Important to do `invalidate_ids` first, so that `new_X_expressions` overwrites duplicate
        // keys.
        for id in invalidate_ids {
            entries.insert(
                CacheKey {
                    id,
                    build_version: build_version.clone(),
                    expr_type: ExpressionType::Local,
                },
                None,
            );
            entries.insert(
                CacheKey {
                    id,
                    build_version: build_version.clone(),
                    expr_type: ExpressionType::Global,
                },
                None,
            );
        }
        for (id, expressions) in new_local_expressions {
            let expressions = match bincode::serialize(&expressions) {
                Ok(expressions) => Bytes::from(expressions),
                Err(err) => {
                    soft_panic_or_log!(
                        "unable to serialize local expressions: {expressions:?}: {err:?}"
                    );
                    continue;
                }
            };
            entries.insert(
                CacheKey {
                    id,
                    build_version: build_version.clone(),
                    expr_type: ExpressionType::Local,
                },
                Some(expressions),
            );
        }
        for (id, expressions) in new_global_expressions {
            let expressions = match bincode::serialize(&expressions) {
                Ok(expressions) => Bytes::from(expressions),
                Err(err) => {
                    soft_panic_or_log!(
                        "unable to serialize global expressions: {expressions:?}: {err:?}"
                    );
                    continue;
                }
            };
            entries.insert(
                CacheKey {
                    id,
                    build_version: build_version.clone(),
                    expr_type: ExpressionType::Global,
                },
                Some(expressions),
            );
        }
        let entries: Vec<_> = entries
            .iter()
            .map(|(key, expressions)| (key, expressions.as_ref()))
            .collect();
        self.durable_cache.set_many(&entries).await
    }
}

/// Operations to perform on the cache.
enum CacheOperation {
    /// See [`ExpressionCache::update`].
    Update {
        new_local_expressions: Vec<(GlobalId, LocalExpressions)>,
        new_global_expressions: Vec<(GlobalId, GlobalExpressions)>,
        invalidate_ids: BTreeSet<GlobalId>,
        trigger: trigger::Trigger,
    },
}

#[derive(Debug, Clone)]
pub struct ExpressionCacheHandle {
    tx: mpsc::UnboundedSender<CacheOperation>,
}

impl ExpressionCacheHandle {
    /// Spawns a task responsible for managing the expression cache. See [`ExpressionCache::open`].
    ///
    /// Returns a handle to interact with the cache and the initial contents of the cache.
    pub async fn spawn_expression_cache(
        config: ExpressionCacheConfig,
    ) -> (
        Self,
        BTreeMap<GlobalId, LocalExpressions>,
        BTreeMap<GlobalId, GlobalExpressions>,
    ) {
        let (mut cache, local_expressions, global_expressions) =
            ExpressionCache::open(config).await;
        let (tx, mut rx) = mpsc::unbounded_channel();
        spawn(|| "expression-cache-task", async move {
            while let Some(op) = rx.recv().await {
                match op {
                    CacheOperation::Update {
                        new_local_expressions,
                        new_global_expressions,
                        invalidate_ids,
                        trigger: _trigger,
                    } => {
                        cache
                            .update(
                                new_local_expressions,
                                new_global_expressions,
                                invalidate_ids,
                            )
                            .await
                    }
                }
            }
        });

        (Self { tx }, local_expressions, global_expressions)
    }

    pub fn update(
        &self,
        new_local_expressions: Vec<(GlobalId, LocalExpressions)>,
        new_global_expressions: Vec<(GlobalId, GlobalExpressions)>,
        invalidate_ids: BTreeSet<GlobalId>,
    ) -> impl Future<Output = ()> + use<> {
        let (trigger, trigger_rx) = trigger::channel();
        let op = CacheOperation::Update {
            new_local_expressions,
            new_global_expressions,
            invalidate_ids,
            trigger,
        };
        // If the send fails, then we must be shutting down.
        let _ = self.tx.send(op);
        trigger_rx
    }
}

#[cfg(test)]
mod tests {
    use std::collections::{BTreeMap, BTreeSet};

    use bytes::Bytes;
    use mz_compute_types::dataflows::{DataflowDescription, IndexDesc, IndexImport};
    use mz_durable_cache::DurableCacheCodec;
    use mz_dyncfg::ConfigSet;
    use mz_expr::{MirRelationExpr, OptimizedMirRelationExpr};
    use mz_persist_client::PersistClient;
    use mz_persist_types::ShardId;
    use mz_repr::{Datum, GlobalId, ReprRelationType, ReprScalarType};
    use semver::Version;

    use super::*;

    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] // unsupported operation: returning ready events from epoll_wait is not yet implemented
    async fn expression_cache() {
        let first_build_version = Version::new(0, 1, 0);
        let second_build_version = Version::new(0, 2, 0);
        let persist = PersistClient::new_for_tests().await;
        let shard_id = ShardId::new();

        let mut current_items = BTreeMap::new();
        let mut remove_prior_versions = false;
        // Compacting the shard takes too long, so we leave it to integration tests.
        let compact_shard = false;
        let dyncfgs = &mz_persist_client::cfg::all_dyncfgs(ConfigSet::default());
        let spawn = |build_version: &Version,
                     current_items: &BTreeMap<GlobalId, RelationVersion>,
                     remove_prior_versions: bool| {
            ExpressionCacheHandle::spawn_expression_cache(ExpressionCacheConfig {
                build_version: build_version.clone(),
                persist: persist.clone(),
                shard_id,
                current_items: current_items.clone(),
                remove_prior_versions,
                compact_shard,
                dyncfgs: dyncfgs.clone(),
            })
        };

        let mut next_id = 0;

        let (mut local_exps, mut global_exps) = {
            // Open a new empty cache.
            let (cache, local_exprs, global_exprs) =
                spawn(&first_build_version, &current_items, remove_prior_versions).await;
            assert_eq!(local_exprs, BTreeMap::new(), "new cache should be empty");
            assert_eq!(global_exprs, BTreeMap::new(), "new cache should be empty");

            // Insert some expressions into the cache.
            let mut local_exps = BTreeMap::new();
            let mut global_exps = BTreeMap::new();
            for _ in 0..4 {
                let id = GlobalId::User(next_id);
                let local_exp = gen_local_expressions();
                let global_exp = gen_global_expressions();

                cache
                    .update(
                        vec![(id, local_exp.clone())],
                        vec![(id, global_exp.clone())],
                        BTreeSet::new(),
                    )
                    .await;

                current_items.insert(id, RelationVersion::root());
                current_items.extend(
                    global_exp
                        .index_imports()
                        .map(|id| (*id, RelationVersion::root())),
                );
                local_exps.insert(id, local_exp);
                global_exps.insert(id, global_exp);

                next_id += 1;
            }
            (local_exps, global_exps)
        };

        {
            // Re-open the cache.
            let (_cache, local_entries, global_entries) =
                spawn(&first_build_version, &current_items, remove_prior_versions).await;
            assert_eq!(
                local_entries, local_exps,
                "local expression with non-matching optimizer features should be removed during reconciliation"
            );
            assert_eq!(
                global_entries, global_exps,
                "global expression with non-matching optimizer features should be removed during reconciliation"
            );
        }

        {
            // Simulate dropping an object.
            let id_to_remove = local_exps.keys().next().expect("not empty").clone();
            current_items.remove(&id_to_remove);
            let _removed_local_exp = local_exps.remove(&id_to_remove);
            let _removed_global_exp = global_exps.remove(&id_to_remove);

            // Re-open the cache.
            let (_cache, local_entries, global_entries) =
                spawn(&first_build_version, &current_items, remove_prior_versions).await;
            assert_eq!(
                local_entries, local_exps,
                "dropped local objects should be removed during reconciliation"
            );
            assert_eq!(
                global_entries, global_exps,
                "dropped global objects should be removed during reconciliation"
            );
        }

        {
            // Simulate applying a replacement: the item keeps its id at a bumped version.
            let id_to_bump = global_exps.keys().next_back().expect("not empty").clone();
            current_items.insert(id_to_bump, RelationVersion::root().bump());
            let _removed_local_exp = local_exps.remove(&id_to_bump);
            let _removed_global_exp = global_exps.remove(&id_to_bump);

            // Re-open the cache.
            let (_cache, local_entries, global_entries) =
                spawn(&first_build_version, &current_items, remove_prior_versions).await;
            assert_eq!(
                local_entries, local_exps,
                "local expressions of an earlier item version should be removed during reconciliation"
            );
            assert_eq!(
                global_entries, global_exps,
                "global expressions of an earlier item version should be removed during reconciliation"
            );

            // The removal is durable: restoring the item version does not bring the entries back.
            current_items.insert(id_to_bump, RelationVersion::root());
            let (_cache, local_entries, global_entries) =
                spawn(&first_build_version, &current_items, remove_prior_versions).await;
            assert_eq!(
                local_entries, local_exps,
                "local expressions of an earlier item version should be durably removed"
            );
            assert_eq!(
                global_entries, global_exps,
                "global expressions of an earlier item version should be durably removed"
            );
        }

        {
            // Simulate dropping an object dependency.
            let global_exp_to_remove = global_exps.keys().next().expect("not empty").clone();
            let removed_global_exp = global_exps
                .remove(&global_exp_to_remove)
                .expect("known to exist");
            let dependency_to_remove = removed_global_exp
                .index_imports()
                .next()
                .expect("generator always makes non-empty index imports");
            current_items.remove(dependency_to_remove);

            // If the dependency is also tracked in the cache remove it.
            let _removed_local_exp = local_exps.remove(dependency_to_remove);
            let _removed_global_exp = global_exps.remove(dependency_to_remove);
            // Remove any other exps that depend on dependency.
            global_exps.retain(|_, exp| {
                let index_imports: BTreeSet<_> = exp.index_imports().collect();
                !index_imports.contains(&dependency_to_remove)
            });

            // Re-open the cache.
            let (_cache, local_entries, global_entries) =
                spawn(&first_build_version, &current_items, remove_prior_versions).await;
            assert_eq!(
                local_entries, local_exps,
                "dropped object dependencies should NOT remove local expressions"
            );
            assert_eq!(
                global_entries, global_exps,
                "dropped object dependencies should remove global expressions"
            );
        }

        let (new_gen_local_exps, new_gen_global_exps) = {
            // Open the cache at a new build version.
            let (cache, local_entries, global_entries) =
                spawn(&second_build_version, &current_items, remove_prior_versions).await;
            assert_eq!(
                local_entries,
                BTreeMap::new(),
                "new build version should be empty"
            );
            assert_eq!(
                global_entries,
                BTreeMap::new(),
                "new build version should be empty"
            );

            // Insert some expressions at the new build version.
            let mut local_exps = BTreeMap::new();
            let mut global_exps = BTreeMap::new();
            for _ in 0..2 {
                let id = GlobalId::User(next_id);
                let local_exp = gen_local_expressions();
                let global_exp = gen_global_expressions();

                cache
                    .update(
                        vec![(id, local_exp.clone())],
                        vec![(id, global_exp.clone())],
                        BTreeSet::new(),
                    )
                    .await;

                current_items.insert(id, RelationVersion::root());
                current_items.extend(
                    global_exp
                        .index_imports()
                        .map(|id| (*id, RelationVersion::root())),
                );
                local_exps.insert(id, local_exp);
                global_exps.insert(id, global_exp);

                next_id += 1;
            }
            (local_exps, global_exps)
        };

        {
            // Re-open the cache at the first build version.
            let (_cache, local_entries, global_entries) =
                spawn(&first_build_version, &current_items, remove_prior_versions).await;
            assert_eq!(
                local_entries, local_exps,
                "Previous build version local expressions should still exist"
            );
            assert_eq!(
                global_entries, global_exps,
                "Previous build version global expressions should still exist"
            );
        }

        {
            // Open the cache at a new build version and clear previous build versions.
            remove_prior_versions = true;
            let (_cache, local_entries, global_entries) =
                spawn(&second_build_version, &current_items, remove_prior_versions).await;
            assert_eq!(
                local_entries, new_gen_local_exps,
                "new build version local expressions should be persisted"
            );
            assert_eq!(
                global_entries, new_gen_global_exps,
                "new build version global expressions should be persisted"
            );
        }

        {
            // Re-open the cache at the first build version.
            let (_cache, local_entries, global_entries) =
                spawn(&first_build_version, &current_items, remove_prior_versions).await;
            assert_eq!(
                local_entries,
                BTreeMap::new(),
                "Previous build version local expressions should be cleared"
            );
            assert_eq!(
                global_entries,
                BTreeMap::new(),
                "Previous build version global expressions should be cleared"
            );
        }
    }

    #[mz_ore::test]
    fn local_expr_cache_roundtrip() {
        let key = CacheKey {
            id: GlobalId::User(1),
            build_version: "1.2.3".into(),
            expr_type: ExpressionType::Local,
        };
        let val = gen_local_expressions();

        let bincode_val = Bytes::from(bincode::serialize(&val).expect("must serialize"));
        let (encoded_key, encoded_val) = ExpressionCodec::encode(&key, &bincode_val);
        let (decoded_key, decoded_val) = ExpressionCodec::decode(&encoded_key, &encoded_val);
        let decoded_val: LocalExpressions =
            bincode::deserialize(&decoded_val).expect("local expressions should roundtrip");

        assert_eq!(key, decoded_key);
        assert_eq!(val, decoded_val);
    }

    #[mz_ore::test]
    fn global_expr_cache_roundtrip() {
        let key = CacheKey {
            id: GlobalId::User(1),
            build_version: "1.2.3".into(),
            expr_type: ExpressionType::Global,
        };
        let val = gen_global_expressions();

        let bincode_val = Bytes::from(bincode::serialize(&val).expect("must serialize"));
        let (encoded_key, encoded_val) = ExpressionCodec::encode(&key, &bincode_val);
        let (decoded_key, decoded_val) = ExpressionCodec::decode(&encoded_key, &encoded_val);
        let decoded_val: GlobalExpressions =
            bincode::deserialize(&decoded_val).expect("global expressions should roundtrip");

        assert_eq!(key, decoded_key);
        assert_eq!(val, decoded_val);
    }

    /// Generate a random [`LocalExpressions`] value.
    ///
    /// The returned values are mostly hardcoded and only differ in a single randomized number.
    /// That's sufficient for the expr cache tests, since the cache mostly treats expressions as
    /// opaque objects.
    fn gen_local_expressions() -> LocalExpressions {
        let datum = Datum::UInt64(rand::random());

        LocalExpressions {
            local_mir: OptimizedMirRelationExpr(MirRelationExpr::constant(
                vec![vec![datum]],
                ReprRelationType::new(vec![ReprScalarType::UInt64.nullable(false)]),
            )),
            optimizer_features: Default::default(),
            item_version: RelationVersion::root(),
        }
    }

    /// Generate a random [`GlobalExpressions`] value.
    ///
    /// The returned values are mostly hardcoded and only differ in a single randomized string.
    /// That's sufficient for the expr cache tests, since the cache mostly treats expressions as
    /// opaque objects.
    fn gen_global_expressions() -> GlobalExpressions {
        let name = format!("test-{}", rand::random::<u64>());

        let mut global_mir = DataflowDescription::new(name.clone());
        let mut physical_plan = DataflowDescription::new(name);

        // Add pieces expected by tests.
        let index_imports = BTreeMap::from_iter([(
            GlobalId::User(2),
            IndexImport {
                desc: IndexDesc {
                    on_id: GlobalId::User(1),
                    key: Default::default(),
                },
                typ: ReprRelationType::empty(),
                monotonic: false,
                with_snapshot: true,
            },
        )]);
        global_mir.index_imports = index_imports.clone();
        physical_plan.index_imports = index_imports;

        GlobalExpressions {
            global_mir,
            physical_plan,
            dataflow_metainfos: Default::default(),
            optimizer_features: Default::default(),
            item_version: RelationVersion::root(),
        }
    }
}