WIP

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

@@ -40,6 +40,7 @@ rust_library(
         "//src/compute-types:mz_compute_types",
         "//src/controller:mz_controller",
         "//src/controller-types:mz_controller_types",
+        "//src/durable-cache:mz_durable_cache",
         "//src/dyncfg:mz_dyncfg",
         "//src/expr:mz_expr",
         "//src/kafka-util:mz_kafka_util",
@@ -104,6 +105,7 @@ rust_test(
         "//src/compute-types:mz_compute_types",
         "//src/controller:mz_controller",
         "//src/controller-types:mz_controller_types",
+        "//src/durable-cache:mz_durable_cache",
         "//src/dyncfg:mz_dyncfg",
         "//src/expr:mz_expr",
         "//src/kafka-util:mz_kafka_util",
@@ -153,6 +155,7 @@ rust_doc_test(
         "//src/compute-types:mz_compute_types",
         "//src/controller:mz_controller",
         "//src/controller-types:mz_controller_types",
+        "//src/durable-cache:mz_durable_cache",
         "//src/dyncfg:mz_dyncfg",
         "//src/expr:mz_expr",
         "//src/kafka-util:mz_kafka_util",
@@ -222,6 +225,7 @@ rust_test(
         "//src/compute-types:mz_compute_types",
         "//src/controller:mz_controller",
         "//src/controller-types:mz_controller_types",
+        "//src/durable-cache:mz_durable_cache",
         "//src/dyncfg:mz_dyncfg",
         "//src/expr:mz_expr",
         "//src/kafka-util:mz_kafka_util",
@@ -291,6 +295,7 @@ rust_test(
         "//src/compute-types:mz_compute_types",
         "//src/controller:mz_controller",
         "//src/controller-types:mz_controller_types",
+        "//src/durable-cache:mz_durable_cache",
         "//src/dyncfg:mz_dyncfg",
         "//src/expr:mz_expr",
         "//src/kafka-util:mz_kafka_util",
@@ -360,6 +365,7 @@ rust_test(
         "//src/compute-types:mz_compute_types",
         "//src/controller:mz_controller",
         "//src/controller-types:mz_controller_types",
+        "//src/durable-cache:mz_durable_cache",
         "//src/dyncfg:mz_dyncfg",
         "//src/expr:mz_expr",
         "//src/kafka-util:mz_kafka_util",

src/adapter/BUILD.bazel

@@ -43,6 +43,7 @@ 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-dyncfg = { path = "../dyncfg" }
 mz-expr = { path = "../expr" }
 mz-kafka-util = { path = "../kafka-util" }

src/adapter/Cargo.toml

@@ -96,6 +96,7 @@ mod migrate;
 
 mod apply;
 mod dataflow_expiration;
+mod expr_cache;
 mod open;
 mod state;
 mod transact;