BOXED FUTURE: As of Nov 2023 the returned Future from this function was 34KB. 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).
(
&mut self,
mut ctx: ExecuteContext,
plan: Plan,
resolved_ids: ResolvedIds,
sql_impl_resolved_ids: ResolvedIds,
)
| 100 | /// get stored on the stack which is bad for runtime performance, and blow up our stack usage. |
| 101 | /// Because of that we purposefully move this Future onto the heap (i.e. Box it). |
| 102 | pub(crate) fn sequence_plan( |
| 103 | &mut self, |
| 104 | mut ctx: ExecuteContext, |
| 105 | plan: Plan, |
| 106 | resolved_ids: ResolvedIds, |
| 107 | sql_impl_resolved_ids: ResolvedIds, |
| 108 | ) -> LocalBoxFuture<'_, ()> { |
| 109 | async move { |
| 110 | let responses = ExecuteResponse::generated_from(&PlanKind::from(&plan)); |
| 111 | ctx.tx_mut().set_allowed(responses); |
| 112 | |
| 113 | if self.controller.read_only() && !plan.allowed_in_read_only() { |
| 114 | ctx.retire(Err(AdapterError::ReadOnly)); |
| 115 | return; |
| 116 | } |
| 117 | |
| 118 | // Check if we're still waiting for any of the builtin table appends from when we |
| 119 | // started the Session to complete. |
| 120 | if let Some((dependencies, wait_future)) = |
| 121 | super::appends::waiting_on_startup_appends(self.catalog(), ctx.session_mut(), &plan) |
| 122 | { |
| 123 | let conn_id = ctx.session().conn_id(); |
| 124 | tracing::debug!(%conn_id, "deferring plan for startup appends"); |
| 125 | |
| 126 | let role_metadata = ctx.session().role_metadata().clone(); |
| 127 | let validity = |
| 128 | PlanValidity::new(&self.catalog, dependencies, None, None, role_metadata); |
| 129 | let deferred_plan = DeferredPlan { |
| 130 | ctx, |
| 131 | plan, |
| 132 | validity, |
| 133 | requires_locks: BTreeSet::default(), |
| 134 | resolved_ids, |
| 135 | sql_impl_resolved_ids, |
| 136 | }; |
| 137 | // Defer op accepts an optional write lock, but there aren't any writes occurring |
| 138 | // here, since the map to `None`. |
| 139 | let acquire_future = wait_future.map(|()| None); |
| 140 | |
| 141 | self.defer_op(acquire_future, DeferredOp::Plan(deferred_plan)); |
| 142 | |
| 143 | // Return early because our op is deferred on waiting for the builtin writes to |
| 144 | // complete. |
| 145 | return; |
| 146 | }; |
| 147 | |
| 148 | // Scope the borrow of the Catalog because we need to mutate the Coordinator state below. |
| 149 | let target_cluster = match ctx.session().transaction().cluster() { |
| 150 | // Use the current transaction's cluster. |
| 151 | Some(cluster_id) => TargetCluster::Transaction(cluster_id), |
| 152 | // If there isn't a current cluster set for a transaction, then try to auto route. |
| 153 | None => { |
| 154 | let session_catalog = self.catalog.for_session(ctx.session()); |
| 155 | catalog_serving::auto_run_on_catalog_server( |
| 156 | &session_catalog, |
| 157 | ctx.session(), |
| 158 | &plan, |
| 159 | ) |
no test coverage detected