Core planning via nodedb-sql: parse → plan → optimize → convert. Returns the compiled physical tasks and the [`super::descriptor_set::DescriptorVersionSet`] recording every descriptor the planner touched. The version set is used as the plan-cache key AND as the input to `SharedState::acquire_plan_lease_scope` so cache hits and fresh plans share the same lease-acquisition path.
(
&self,
sql: &str,
tenant_id: crate::types::TenantId,
database_id: crate::types::DatabaseId,
)
| 251 | /// `SharedState::acquire_plan_lease_scope` so cache hits |
| 252 | /// and fresh plans share the same lease-acquisition path. |
| 253 | fn plan_with_nodedb_sql( |
| 254 | &self, |
| 255 | sql: &str, |
| 256 | tenant_id: crate::types::TenantId, |
| 257 | database_id: crate::types::DatabaseId, |
| 258 | ) -> crate::Result<( |
| 259 | Vec<nodedb_physical::physical_task::PhysicalTask>, |
| 260 | super::descriptor_set::DescriptorVersionSet, |
| 261 | )> { |
| 262 | let inputs = match &self.catalog_inputs { |
| 263 | Some(i) => i, |
| 264 | None => { |
| 265 | return Err(crate::Error::PlanError { |
| 266 | detail: "no catalog available for SQL planning".into(), |
| 267 | }); |
| 268 | } |
| 269 | }; |
| 270 | // Fresh adapter per plan call: the adapter's |
| 271 | // `recorded_versions` field is per-plan state, and |
| 272 | // two concurrent plans through a shared QueryContext |
| 273 | // would otherwise interleave their recorded sets. |
| 274 | let catalog = inputs.build_adapter(tenant_id.as_u64(), database_id); |
| 275 | let plans = nodedb_sql::plan_sql(sql, &catalog).map_err(|e| match e { |
| 276 | nodedb_sql::SqlError::RetryableSchemaChanged { descriptor } => { |
| 277 | crate::Error::RetryableSchemaChanged { descriptor } |
| 278 | } |
| 279 | nodedb_sql::SqlError::CollectionDeactivated { |
| 280 | name, |
| 281 | retention_expires_at_ns, |
| 282 | .. |
| 283 | } => crate::Error::CollectionDeactivated { |
| 284 | tenant_id, |
| 285 | collection: name, |
| 286 | retention_expires_at_ns, |
| 287 | }, |
| 288 | other => crate::Error::PlanError { |
| 289 | detail: format!("{other}"), |
| 290 | }, |
| 291 | })?; |
| 292 | let version_set = catalog.take_recorded_versions(); |
| 293 | let ctx = super::sql_plan_convert::ConvertContext { |
| 294 | retention_registry: self.retention_registry.clone(), |
| 295 | array_catalog: self.array_catalog.clone(), |
| 296 | credentials: self |
| 297 | .catalog_inputs |
| 298 | .as_ref() |
| 299 | .map(|i| Arc::clone(&i.credentials)), |
| 300 | wal: self.wal.clone(), |
| 301 | surrogate_assigner: self.surrogate_assigner.clone(), |
| 302 | cluster_enabled: self.cluster_enabled, |
| 303 | bitemporal_retention_registry: self.bitemporal_retention_registry.clone(), |
| 304 | max_vector_dim: self |
| 305 | .max_vector_dim |
| 306 | .load(std::sync::atomic::Ordering::Relaxed), |
| 307 | database_id, |
| 308 | }; |
| 309 | let tasks = super::sql_plan_convert::convert(&plans, tenant_id, &ctx)?; |
| 310 | Ok((tasks, version_set)) |
no test coverage detected