adapter: define OptimizePeek

Define an `OptimizePeek` struct which implements an end-to-end
optimizer for `Peek` plans statements using the `Optimize` API.

src/adapter/src/coord.rs

@@ -1412,7 +1412,8 @@ impl Coordinator {
                         );
 
                         // MIR ⇒ MIR optimization (global)
-                        let index_plan = optimize::Index::new(entry.name(), &idx.on, &idx.keys);
+                        let index_plan =
+                            optimize::index::Index::new(entry.name(), &idx.on, &idx.keys);
                         let global_mir_plan = optimizer.optimize(index_plan)?;
                         // Timestamp selection
                         let as_of = self.bootstrap_index_as_of(

src/adapter/src/coord/sequencer/inner.rs

@@ -1160,7 +1160,7 @@ impl Coordinator {
         );
 
         // MIR ⇒ MIR optimization (global)
-        let index_plan = optimize::Index::new(&name, &on, &keys);
+        let index_plan = optimize::index::Index::new(&name, &on, &keys);
         let global_mir_plan = optimizer.optimize(index_plan)?;
         // Timestamp selection
         let since = self.least_valid_read(&global_mir_plan.id_bundle());
@@ -3705,7 +3705,7 @@ impl Coordinator {
 
         let (df_desc, df_meta, used_indexes) = catch_unwind(broken, "optimize", || {
             // MIR ⇒ MIR optimization (global)
-            let index_plan = optimize::Index::new(&name, &index.on, &index.keys);
+            let index_plan = optimize::index::Index::new(&name, &index.on, &index.keys);
             let global_mir_plan = optimizer.optimize(index_plan)?;
 
             // Collect the list of indexes used by the dataflow at this point

src/adapter/src/optimize/mod.rs

@@ -50,14 +50,16 @@
 //! For details, see the `20230714_optimizer_interface.md` design doc in this
 //! repository.
 
-mod index;
-mod materialized_view;
-mod subscribe;
-mod view;
+pub mod index;
+pub mod materialized_view;
+pub mod peek;
+pub mod subscribe;
+pub mod view;
 
 // Re-export optimzier structs
-pub use index::{Index, OptimizeIndex};
+pub use index::OptimizeIndex;
 pub use materialized_view::OptimizeMaterializedView;
+pub use peek::OptimizePeek;
 pub use subscribe::OptimizeSubscribe;
 pub use view::OptimizeView;
 
@@ -90,11 +92,11 @@ use crate::AdapterError;
 /// The `'s: 'ctx` bound in the `optimize` method call ensures that an optimizer
 /// instance can run an optimization stage that produces a `Self::To` with
 /// `&'ctx` references.
-pub trait Optimize<From>: Send + Sync
+pub trait Optimize<From>: Send
 where
-    From: Send + Sync,
+    From: Send,
 {
-    type To: Send + Sync;
+    type To: Send;
 
     /// Execute the optimization stage, transforming the input plan of type
     /// `From` to an output plan of type `To`.
@@ -112,7 +114,7 @@ where
 }
 
 // Feature flags for the optimizer.
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct OptimizerConfig {
     /// The mode in which the optimizer runs.
     pub mode: OptimizeMode,
@@ -126,13 +128,14 @@ pub struct OptimizerConfig {
     /// The collection of type information happens in MIR ⇒ LIR lowering.
     pub enable_specialized_arrangements: bool,
     /// An exclusive upper bound on the number of results we may return from a
-    /// Persist fast-path peek.
+    /// Persist fast-path peek. Required by the `create_fast_path_plan` call in
+    /// [`OptimizePeek`].
     pub persist_fast_path_limit: usize,
     /// Enable outer join lowering implemented in #22343.
     pub enable_new_outer_join_lowering: bool,
 }
 
-#[derive(Clone, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub enum OptimizeMode {
     /// A mode where the optimized statement is executed.
     Execute,