(
&mut self,
ordered_catalog_entries: &[CatalogEntry],
mut cached_global_exprs: BTreeMap<GlobalId, GlobalExpressions>,
)
| 3441 | /// Returns a map of expressions that were not cached. |
| 3442 | #[instrument] |
| 3443 | fn bootstrap_dataflow_plans( |
| 3444 | &mut self, |
| 3445 | ordered_catalog_entries: &[CatalogEntry], |
| 3446 | mut cached_global_exprs: BTreeMap<GlobalId, GlobalExpressions>, |
| 3447 | ) -> Result<BTreeMap<GlobalId, GlobalExpressions>, AdapterError> { |
| 3448 | // The optimizer expects to be able to query its `ComputeInstanceSnapshot` for |
| 3449 | // collections the current dataflow can depend on. But since we don't yet install anything |
| 3450 | // on compute instances, the snapshot information is incomplete. We fix that by manually |
| 3451 | // updating `ComputeInstanceSnapshot` objects to ensure they contain collections previously |
| 3452 | // optimized. |
| 3453 | let mut instance_snapshots = BTreeMap::new(); |
| 3454 | let mut uncached_expressions = BTreeMap::new(); |
| 3455 | |
| 3456 | let optimizer_config = |catalog: &Catalog, cluster_id| { |
| 3457 | let system_config = catalog.system_config(); |
| 3458 | let overrides = catalog.get_cluster(cluster_id).config.features(); |
| 3459 | OptimizerConfig::from(system_config) |
| 3460 | .override_from(&overrides) |
| 3461 | // A cluster-scoped LaunchDarkly rule beats a manual `FEATURES` |
| 3462 | // pin. |
| 3463 | .override_from( |
| 3464 | &catalog |
| 3465 | .state() |
| 3466 | .cluster_scoped_optimizer_overrides(cluster_id), |
| 3467 | ) |
| 3468 | }; |
| 3469 | |
| 3470 | for entry in ordered_catalog_entries { |
| 3471 | match entry.item() { |
| 3472 | CatalogItem::Index(idx) => { |
| 3473 | // Collect optimizer parameters. |
| 3474 | let compute_instance = |
| 3475 | instance_snapshots.entry(idx.cluster_id).or_insert_with(|| { |
| 3476 | self.instance_snapshot(idx.cluster_id) |
| 3477 | .expect("compute instance exists") |
| 3478 | }); |
| 3479 | let global_id = idx.global_id(); |
| 3480 | |
| 3481 | // The index may already be installed on the compute instance. For example, |
| 3482 | // this is the case for introspection indexes. |
| 3483 | if compute_instance.contains_collection(&global_id) { |
| 3484 | continue; |
| 3485 | } |
| 3486 | |
| 3487 | let optimizer_config = optimizer_config(&self.catalog, idx.cluster_id); |
| 3488 | |
| 3489 | let (optimized_plan, physical_plan, metainfo) = |
| 3490 | match cached_global_exprs.remove(&global_id) { |
| 3491 | Some(global_expressions) |
| 3492 | if global_expressions.optimizer_features |
| 3493 | == optimizer_config.features => |
| 3494 | { |
| 3495 | debug!("global expression cache hit for {global_id:?}"); |
| 3496 | ( |
| 3497 | global_expressions.global_mir, |
| 3498 | global_expressions.physical_plan, |
| 3499 | global_expressions.dataflow_metainfos, |
| 3500 | ) |
no test coverage detected