adapter: Implement expression cache (#30122)

This commit adds a module for an expression cache. It adds the
functionality to durably cache optimized expressions. It uses the
durable_cache module in its implementation.

The expression cache is run on its own task, so that callers can insert
new entries without blocking on the insert completing.

The adapter doesn't use the expression cache yet, that is saved for a
later commit.

Works towards resolving #MaterializeInc/database-issues/issues/8384

Author: jkosh44

Cargo.lock

@@ -70,16 +70,7 @@ struct Expressions {
     physical_plan: DataflowDescription<mz_compute_types::plan::Plan>,
     dataflow_metainfos: DataflowMetainfo<Arc<OptimizerNotice>>,
     notices: SmallVec<[Arc<OptimizerNotice>; 4]>,
-    optimizer_feature_overrides: OptimizerFeatures,
-}
-
-struct NewEntry {
-    /// `GlobalId` of the new expression.
-    id: GlobalId,
-    /// New `Expressions` to cache.
-    expressions: Expressions,
-    /// `GlobalId`s to invalidate as a result of the new entry.
-    invalidate_ids: BTreeSet<GlobalId>,
+    optimizer_feature: OptimizerFeatures,
 }
 
 struct ExpressionCache {
@@ -100,13 +91,12 @@ impl ExpressionCache {
     /// Returns all cached expressions in the current deploy generation, after reconciliation.
     fn open(&mut self, current_ids: &BTreeSet<GlobalId>, optimizer_features: &OptimizerFeatures, remove_prior_gens: bool) -> Vec<(GlobalId, Expressions)>;
 
-    /// Durably inserts `expressions` into current deploy generation. This may also invalidate
-    /// entries giving by `expressions`.
-    ///
-    /// Returns a [`Future`] that completes once the changes have been made durable.
+    /// Durably removes all entries given by `invalidate_ids` and inserts `new_entries` into
+    /// current deploy generation.
     ///
-    /// Panics if any `GlobalId` already exists in the cache.
-    fn insert_expressions(&mut self, expressions: Vec<NewEntry>) -> impl Future<Output=()>;
+    /// If there is a duplicate ID in both `invalidate_ids` and `new_entries`, then the final value
+    /// will be taken from `new_entries`.
+    fn insert_expressions(&mut self, new_entries: Vec<(GlobalId, Expressions)>, invalidate_ids: BTreeSet<GlobalId>);
 
     /// Durably remove and return all entries in current deploy generation that depend on an ID in
     /// `dropped_ids` .

doc/developer/design/20241008_expression_cache.md

@@ -36,8 +36,10 @@ rust_library(
         "//src/build-info:mz_build_info",
         "//src/cloud-resources:mz_cloud_resources",
         "//src/compute-client:mz_compute_client",
+        "//src/compute-types:mz_compute_types",
         "//src/controller:mz_controller",
         "//src/controller-types:mz_controller_types",
+        "//src/durable-cache:mz_durable_cache",
         "//src/expr:mz_expr",
         "//src/orchestrator:mz_orchestrator",
         "//src/ore:mz_ore",
@@ -53,6 +55,7 @@ rust_library(
         "//src/storage-client:mz_storage_client",
         "//src/storage-controller:mz_storage_controller",
         "//src/storage-types:mz_storage_types",
+        "//src/transform:mz_transform",
     ] + all_crate_deps(normal = True),
 )
 
@@ -87,8 +90,10 @@ rust_test(
         "//src/build-tools:mz_build_tools",
         "//src/cloud-resources:mz_cloud_resources",
         "//src/compute-client:mz_compute_client",
+        "//src/compute-types:mz_compute_types",
         "//src/controller:mz_controller",
         "//src/controller-types:mz_controller_types",
+        "//src/durable-cache:mz_durable_cache",
         "//src/expr:mz_expr",
         "//src/orchestrator:mz_orchestrator",
         "//src/ore:mz_ore",
@@ -105,6 +110,7 @@ rust_test(
         "//src/storage-client:mz_storage_client",
         "//src/storage-controller:mz_storage_controller",
         "//src/storage-types:mz_storage_types",
+        "//src/transform:mz_transform",
     ] + all_crate_deps(
         normal = True,
         normal_dev = True,
@@ -121,8 +127,10 @@ rust_doc_test(
         "//src/build-tools:mz_build_tools",
         "//src/cloud-resources:mz_cloud_resources",
         "//src/compute-client:mz_compute_client",
+        "//src/compute-types:mz_compute_types",
         "//src/controller:mz_controller",
         "//src/controller-types:mz_controller_types",
+        "//src/durable-cache:mz_durable_cache",
         "//src/expr:mz_expr",
         "//src/orchestrator:mz_orchestrator",
         "//src/ore:mz_ore",
@@ -139,6 +147,7 @@ rust_doc_test(
         "//src/storage-client:mz_storage_client",
         "//src/storage-controller:mz_storage_controller",
         "//src/storage-types:mz_storage_types",
+        "//src/transform:mz_transform",
     ] + all_crate_deps(
         normal = True,
         normal_dev = True,
@@ -193,8 +202,10 @@ rust_test(
         "//src/build-tools:mz_build_tools",
         "//src/cloud-resources:mz_cloud_resources",
         "//src/compute-client:mz_compute_client",
+        "//src/compute-types:mz_compute_types",
         "//src/controller:mz_controller",
         "//src/controller-types:mz_controller_types",
+        "//src/durable-cache:mz_durable_cache",
         "//src/expr:mz_expr",
         "//src/orchestrator:mz_orchestrator",
         "//src/ore:mz_ore",
@@ -211,6 +222,7 @@ rust_test(
         "//src/storage-client:mz_storage_client",
         "//src/storage-controller:mz_storage_controller",
         "//src/storage-types:mz_storage_types",
+        "//src/transform:mz_transform",
     ] + all_crate_deps(
         normal = True,
         normal_dev = True,
@@ -247,8 +259,10 @@ rust_test(
         "//src/build-tools:mz_build_tools",
         "//src/cloud-resources:mz_cloud_resources",
         "//src/compute-client:mz_compute_client",
+        "//src/compute-types:mz_compute_types",
         "//src/controller:mz_controller",
         "//src/controller-types:mz_controller_types",
+        "//src/durable-cache:mz_durable_cache",
         "//src/expr:mz_expr",
         "//src/orchestrator:mz_orchestrator",
         "//src/ore:mz_ore",
@@ -265,6 +279,7 @@ rust_test(
         "//src/storage-client:mz_storage_client",
         "//src/storage-controller:mz_storage_controller",
         "//src/storage-types:mz_storage_types",
+        "//src/transform:mz_transform",
     ] + all_crate_deps(
         normal = True,
         normal_dev = True,
@@ -301,8 +316,10 @@ rust_test(
         "//src/build-tools:mz_build_tools",
         "//src/cloud-resources:mz_cloud_resources",
         "//src/compute-client:mz_compute_client",
+        "//src/compute-types:mz_compute_types",
         "//src/controller:mz_controller",
         "//src/controller-types:mz_controller_types",
+        "//src/durable-cache:mz_durable_cache",
         "//src/expr:mz_expr",
         "//src/orchestrator:mz_orchestrator",
         "//src/ore:mz_ore",
@@ -319,6 +336,7 @@ rust_test(
         "//src/storage-client:mz_storage_client",
         "//src/storage-controller:mz_storage_controller",
         "//src/storage-types:mz_storage_types",
+        "//src/transform:mz_transform",
     ] + all_crate_deps(
         normal = True,
         normal_dev = True,

src/catalog/BUILD.bazel

@@ -27,8 +27,10 @@ mz-audit-log = { path = "../audit-log" }
 mz-build-info = { path = "../build-info" }
 mz-cloud-resources = { path = "../cloud-resources" }
 mz-compute-client = { path = "../compute-client" }
+mz-compute-types = { path = "../compute-types" }
 mz-controller = { path = "../controller" }
 mz-controller-types = { path = "../controller-types" }
+mz-durable-cache = { path = "../durable-cache" }
 mz-expr = { path = "../expr" }
 mz-orchestrator = { path = "../orchestrator" }
 mz-ore = { path = "../ore", features = ["chrono", "async", "tracing_"] }
@@ -44,6 +46,7 @@ mz-sql-parser = { path = "../sql-parser" }
 mz-storage-client = { path = "../storage-client" }
 mz-storage-controller = { path = "../storage-controller" }
 mz-storage-types = { path = "../storage-types" }
+mz-transform = { path = "../transform" }
 paste = "1.0.11"
 prometheus = { version = "0.13.3", default-features = false }
 proptest = { version = "1.0.0", default-features = false, features = ["std"] }
@@ -55,11 +58,13 @@ semver = { version = "1.0.16" }
 serde = "1.0.152"
 serde_json = "1.0.125"
 serde_plain = "1.0.1"
+smallvec = { version = "1.10.0", features = ["union"] }
 static_assertions = "1.1"
 sha2 = "0.10.6"
+thiserror = "1.0.37"
 timely = "0.12.0"
+tokio = { version = "1.38.0" }
 tracing = "0.1.37"
-thiserror = "1.0.37"
 uuid = "1.2.2"
 workspace-hack = { version = "0.0.0", path = "../workspace-hack" }
 
@@ -69,7 +74,6 @@ insta = "1.32"
 mz-build-tools = { path = "../build-tools", default-features = false }
 mz-postgres-util = { path = "../postgres-util" }
 similar-asserts = "1.4"
-tokio = { version = "1.38.0" }
 tokio-postgres = { version = "0.7.8" }
 
 [build-dependencies]

src/catalog/Cargo.toml

@@ -36,7 +36,7 @@ pub use crate::durable::objects::{
     Role, Schema, SourceReference, SourceReferences, StorageCollectionMetadata,
     SystemConfiguration, SystemObjectDescription, SystemObjectMapping, UnfinalizedShard,
 };
-pub use crate::durable::persist::builtin_migration_shard_id;
+pub use crate::durable::persist::{builtin_migration_shard_id, expression_cache_shard_id};
 use crate::durable::persist::{Timestamp, UnopenedPersistCatalogState};
 pub use crate::durable::transaction::Transaction;
 use crate::durable::transaction::TransactionBatch;

src/catalog/src/durable.rs

@@ -117,6 +117,8 @@ const CATALOG_SEED: usize = 1;
 const UPGRADE_SEED: usize = 2;
 /// Seed used to generate the persist shard ID for builtin table migrations.
 const BUILTIN_MIGRATION_SEED: usize = 3;
+/// Seed used to generate the persist shard ID for the expression cache.
+const EXPRESSION_CACHE_SEED: usize = 4;
 
 /// Durable catalog mode that dictates the effect of mutable operations.
 #[derive(Debug, Copy, Clone, Eq, PartialEq)]
@@ -1701,6 +1703,12 @@ pub fn builtin_migration_shard_id(organization_id: Uuid) -> ShardId {
     shard_id(organization_id, BUILTIN_MIGRATION_SEED)
 }
 
+/// Deterministically generate an expression cache shard ID for the given
+/// `organization_id`.
+pub fn expression_cache_shard_id(organization_id: Uuid) -> ShardId {
+    shard_id(organization_id, EXPRESSION_CACHE_SEED)
+}
+
 /// Deterministically generate a shard ID for the given `organization_id` and `seed`.
 fn shard_id(organization_id: Uuid, seed: usize) -> ShardId {
     let hash = sha2::Sha256::digest(format!("{organization_id}{seed}")).to_vec();