.*: Box all Futures that are larger than 8KiB

clippy.toml

@@ -83,3 +83,5 @@ disallowed-types = [
     { path = "std::collections::HashMap", reason = "use `std::collections::BTreeMap` or `mz_ore::collections::HashMap` instead" },
     { path = "std::collections::HashSet", reason = "use `std::collections::BTreeSet` or `mz_ore::collections::HashSet` instead" },
 ]
+
+future-size-threshold = 8192

misc/python/materialize/cli/gen-lints.py

@@ -176,6 +176,9 @@
     # Implementing `From` gives you `Into` for free, but the reverse is not
     # true.
     "clippy::from_over_into",
+    # Futures get compiled into state machines, which can end up being very large (10's of KB).
+    # It's inefficient to moves these around on the stack, so require that they get boxed.
+    "clippy::large_futures",
 ]
 
 

src/adapter-types/src/lib.rs

@@ -73,6 +73,7 @@
 #![warn(clippy::disallowed_macros)]
 #![warn(clippy::disallowed_types)]
 #![warn(clippy::from_over_into)]
+#![warn(clippy::large_futures)]
 // END LINT CONFIG
 
 //! Types for the adapter.

src/adapter/benches/catalog.rs

@@ -73,6 +73,7 @@
 #![warn(clippy::disallowed_macros)]
 #![warn(clippy::disallowed_types)]
 #![warn(clippy::from_over_into)]
+#![warn(clippy::large_futures)]
 // END LINT CONFIG
 
 use std::str::FromStr;

src/adapter/src/catalog.rs

@@ -4272,6 +4272,7 @@ mod tests {
     use std::time::{Duration, Instant};
     use std::{env, iter};
 
+    use futures::future::FutureExt;
     use itertools::Itertools;
     use tokio_postgres::types::Type;
     use tokio_postgres::NoTls;
@@ -4384,6 +4385,7 @@ mod tests {
             }
             catalog.expire().await;
         })
+        .boxed()
         .await
     }
 
@@ -4582,6 +4584,7 @@ mod tests {
             );
             catalog.expire().await;
         })
+        .boxed()
         .await
     }
 
@@ -4608,6 +4611,7 @@ mod tests {
             );
             catalog.expire().await;
         })
+        .boxed()
         .await;
     }
 
@@ -4787,6 +4791,7 @@ mod tests {
             );
             catalog.expire().await;
         })
+        .boxed()
         .await;
     }
 
@@ -4810,6 +4815,7 @@ mod tests {
             );
             catalog.expire().await;
         })
+        .boxed()
         .await;
     }
 
@@ -5208,7 +5214,7 @@ mod tests {
             catalog.expire().await;
         }
 
-        Catalog::with_debug(NOW_ZERO.clone(), inner).await
+        Catalog::with_debug(NOW_ZERO.clone(), inner).boxed().await
     }
 
     // Execute all builtin functions with all combinations of arguments from interesting datums.
@@ -5358,8 +5364,9 @@ mod tests {
             handles
         }
 
-        let handles =
-            Catalog::with_debug(NOW_ZERO.clone(), |catalog| async { inner(catalog) }).await;
+        let handles = Catalog::with_debug(NOW_ZERO.clone(), |catalog| async { inner(catalog) })
+            .boxed()
+            .await;
         for handle in handles {
             handle.await.expect("must succeed");
         }
@@ -5578,6 +5585,7 @@ mod tests {
             }
             catalog.expire().await;
         })
+        .boxed()
         .await
     }
 }

src/adapter/src/catalog/open.rs

@@ -1854,6 +1854,7 @@ fn default_logging_config() -> ReplicaLogging {
 mod builtin_migration_tests {
     use std::collections::{BTreeMap, BTreeSet};
 
+    use futures::FutureExt;
     use itertools::Itertools;
     use mz_catalog::memory::objects::Table;
 
@@ -2152,6 +2153,7 @@ mod builtin_migration_tests {
             );
             catalog.expire().await;
         })
+        .boxed()
         .await
     }
 

src/adapter/src/config/sync.rs

@@ -9,6 +9,7 @@
 
 use std::time::Duration;
 
+use futures::FutureExt;
 use tokio::time;
 
 use crate::config::{
@@ -32,7 +33,9 @@ pub async fn system_parameter_sync(
 
     // Ensure the frontend client is initialized.
     let mut frontend = Option::<SystemParameterFrontend>::None; // lazy initialize the frontend below
-    let mut backend = SystemParameterBackend::new(adapter_client).await?;
+
+    // Note: the Future is intentionally boxed because it is very large.
+    let mut backend = SystemParameterBackend::new(adapter_client).boxed().await?;
 
     // Tick every `tick_duration` ms, skipping missed ticks.
     let mut interval = time::interval(tick_interval);

src/adapter/src/coord/command_handler.rs

@@ -106,6 +106,8 @@ impl Coordinator {
 
                     self.handle_execute(portal_name, session, tx, outer_ctx_extra)
                         .instrument(span)
+                        // Note: the Future is intentionally boxed because it is very large.
+                        .boxed_local()
                         .await;
                 }
 
@@ -445,7 +447,10 @@ impl Coordinator {
             _ => {}
         }
 
-        self.handle_execute_inner(stmt, params, ctx).await
+        self.handle_execute_inner(stmt, params, ctx)
+            // Note: the Future is intentionally boxed because it is very large.
+            .boxed_local()
+            .await
     }
 
     #[tracing::instrument(level = "debug", skip(self, ctx))]

src/adapter/src/coord/message_handler.rs

@@ -42,9 +42,7 @@ use crate::util::{ComputeSinkId, ResultExt};
 use crate::{catalog, AdapterNotice, TimestampContext};
 
 impl Coordinator {
-    /// BOXED FUTURE: As of Nov 2023 the returned Future from this function was 74KB. This would
-    /// get stored on the stack which is bad for runtime performance, and blow up our stack usage.
-    /// Because of that we purposefully move this Future onto the heap (i.e. Box it).
+    /// Note: the returned Future is intentionally boxed because it is very large.
     pub(crate) fn handle_message<'a>(&'a mut self, msg: Message) -> LocalBoxFuture<'a, ()> {
         async move {
             match msg {
@@ -60,13 +58,22 @@ impl Coordinator {
                     }
                 }
                 Message::PurifiedStatementReady(ready) => {
-                    self.message_purified_statement_ready(ready).await
+                    self.message_purified_statement_ready(ready)
+                        // Note: the Future is intentionally boxed because it is very large.
+                        .boxed_local()
+                        .await
                 }
                 Message::CreateConnectionValidationReady(ready) => {
-                    self.message_create_connection_validation_ready(ready).await
+                    self.message_create_connection_validation_ready(ready)
+                        // Note: the Future is intentionally boxed because it is very large.
+                        .boxed_local()
+                        .await
                 }
                 Message::WriteLockGrant(write_lock_guard) => {
-                    self.message_write_lock_grant(write_lock_guard).await;
+                    self.message_write_lock_grant(write_lock_guard)
+                        // Note: the Future is intentionally boxed because it is very large.
+                        .boxed_local()
+                        .await;
                 }
                 Message::GroupCommitInitiate(span, permit) => {
                     // Add an OpenTelemetry link to our current span.
@@ -118,26 +125,35 @@ impl Coordinator {
                         real_time_recency_ts,
                         validity,
                     )
+                    // Note: the Future is intentionally boxed because it is very large.
+                    .boxed_local()
                     .await;
                 }
                 Message::RetireExecute { data, reason } => {
                     self.retire_execution(reason, data);
                 }
                 Message::ExecuteSingleStatementTransaction { ctx, stmt, params } => {
                     self.sequence_execute_single_statement_transaction(ctx, stmt, params)
+                        // Note: the Future is intentionally boxed because it is very large.
+                        .boxed_local()
                         .await;
                 }
                 Message::PeekStageReady { ctx, stage } => {
-                    self.sequence_peek_stage(ctx, stage).await;
+                    // Note: the Future is intentionally boxed because it is very large.
+                    self.sequence_peek_stage(ctx, stage).boxed_local().await;
                 }
                 Message::PeekStageDeprecatedReady { ctx, stage } => {
                     // Allow while the introduction of the new optimizer API in
                     // #20569 is in progress.
                     #[allow(deprecated)]
-                    self.sequence_peek_stage_deprecated(ctx, stage).await;
+                    self.sequence_peek_stage_deprecated(ctx, stage)
+                        // Note: the Future is intentionally boxed because it is very large.
+                        .boxed_local()
+                        .await;
                 }
                 Message::DrainStatementLog => {
-                    self.drain_statement_log().await;
+                    // Note: the Future is intentionally boxed because it is very large.
+                    self.drain_statement_log().boxed_local().await;
                 }
             }
         }
@@ -362,7 +378,10 @@ impl Coordinator {
             .iter()
             .all(|id| self.catalog().try_get_entry(id).is_some())
         {
-            self.handle_execute_inner(original_stmt, params, ctx).await;
+            self.handle_execute_inner(original_stmt, params, ctx)
+                // Note: the Future is intentionally boxed because it is very large.
+                .boxed_local()
+                .await;
             return;
         }
 
@@ -711,6 +730,8 @@ impl Coordinator {
                         global_mir_plan,
                     }),
                 )
+                // Note: the Future is intentionally boxed because it is very large.
+                .boxed_local()
                 .await;
             }
             RealTimeRecencyContext::PeekDeprecated {
@@ -754,6 +775,8 @@ impl Coordinator {
                         dataflow_metainfo,
                     }),
                 )
+                // Note: the Future is intentionally boxed because it is very large.
+                .boxed_local()
                 .await;
             }
         }

src/adapter/src/coord/sequencer.rs

@@ -156,6 +156,8 @@ impl Coordinator {
                 }
                 Plan::CreateConnection(plan) => {
                     self.sequence_create_connection(ctx, plan, resolved_ids)
+                        // Note: the Future is intentionally boxed because it is very large.
+                        .boxed_local()
                         .await;
                 }
                 Plan::CreateDatabase(plan) => {
@@ -196,7 +198,10 @@ impl Coordinator {
                     ctx.retire(result);
                 }
                 Plan::CreateSink(plan) => {
-                    self.sequence_create_sink(ctx, plan, resolved_ids).await;
+                    self.sequence_create_sink(ctx, plan, resolved_ids)
+                        // Note: the Future is intentionally boxed because it is very large.
+                        .boxed_local()
+                        .await;
                 }
                 Plan::CreateView(plan) => {
                     let result = self
@@ -207,6 +212,8 @@ impl Coordinator {
                 Plan::CreateMaterializedView(plan) => {
                     let result = self
                         .sequence_create_materialized_view(ctx.session_mut(), plan, resolved_ids)
+                        // Note: the Future is intentionally boxed because it is very large.
+                        .boxed_local()
                         .await;
                     ctx.retire(result);
                 }
@@ -298,10 +305,16 @@ impl Coordinator {
                             AdapterNotice::ExplicitTransactionControlInImplicitTransaction,
                         );
                     }
-                    self.sequence_end_transaction(ctx, action).await;
+                    self.sequence_end_transaction(ctx, action)
+                        // Note: the Future is intentionally boxed because it is very large.
+                        .boxed_local()
+                        .await;
                 }
                 Plan::Select(plan) => {
-                    self.sequence_peek(ctx, plan, target_cluster).await;
+                    self.sequence_peek(ctx, plan, target_cluster)
+                        // Note: the Future is intentionally boxed because it is very large.
+                        .boxed_local()
+                        .await;
                 }
                 Plan::Subscribe(plan) => {
                     if enable_unified_optimizer_api {
@@ -327,6 +340,8 @@ impl Coordinator {
                 }
                 Plan::ShowColumns(show_columns_plan) => {
                     self.sequence_peek(ctx, show_columns_plan.select_plan, target_cluster)
+                        // Note: the Future is intentionally boxed because it is very large.
+                        .boxed_local()
                         .await;
                 }
                 Plan::CopyFrom(plan) => {
@@ -353,10 +368,12 @@ impl Coordinator {
                         .await;
                 }
                 Plan::Insert(plan) => {
-                    self.sequence_insert(ctx, plan).await;
+                    // Note: the Future is intentionally boxed because it is very large.
+                    self.sequence_insert(ctx, plan).boxed_local().await;
                 }
                 Plan::ReadThenWrite(plan) => {
-                    self.sequence_read_then_write(ctx, plan).await;
+                    // Note: the Future is intentionally boxed because it is very large.
+                    self.sequence_read_then_write(ctx, plan).boxed_local().await;
                 }
                 Plan::AlterNoop(plan) => {
                     ctx.retire(Ok(ExecuteResponse::AlteredObject(plan.object_type)));
@@ -439,6 +456,8 @@ impl Coordinator {
                 } => {
                     let result = self
                         .sequence_alter_source(ctx.session_mut(), alter_source, subsources)
+                        // Note: the Future is intentionally boxed because it is very large.
+                        .boxed_local()
                         .await;
                     ctx.retire(result);
                 }
@@ -627,7 +646,10 @@ impl Coordinator {
         let (sub_tx, sub_rx) = oneshot::channel();
         let sub_ct = ClientTransmitter::new(sub_tx, self.internal_cmd_tx.clone());
         let sub_ctx = ExecuteContext::from_parts(sub_ct, internal_cmd_tx, session, extra);
-        self.handle_execute_inner(stmt, params, sub_ctx).await;
+        self.handle_execute_inner(stmt, params, sub_ctx)
+            // Note: the Future is intentionally boxed because it is very large.
+            .boxed_local()
+            .await;
 
         // The response can need off-thread processing. Wait for it elsewhere so the coordinator can
         // continue processing.