MCPcopy Create free account
hub / github.com/MaterializeInc/materialize / sequence_plan

Method sequence_plan

src/adapter/src/coord/sequencer.rs:102–694  ·  view source on GitHub ↗

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,
    )

Source from the content-addressed store, hash-verified

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 )

Callers 5

handle_commandMethod · 0.80
handle_execute_innerMethod · 0.80
try_deferredMethod · 0.80

Calls 15

check_planFunction · 0.85
takeFunction · 0.85
spawnFunction · 0.85
set_allowedMethod · 0.80
tx_mutMethod · 0.80
allowed_in_read_onlyMethod · 0.80
role_metadataMethod · 0.80
defer_opMethod · 0.80

Tested by

no test coverage detected