Method optimize

src/adapter/src/optimize/subscribe.rs:173–290

(&mut self, plan: SubscribePlan)

    type To = GlobalMirPlan<Unresolved>;

    fn optimize(&mut self, plan: SubscribePlan) -> Result<Self::To, OptimizerError> {
        let output = plan.output;
        let plan = plan.from;
        let time = Instant::now();

        let mut df_builder = {
            let compute = self.compute_instance.clone();
            DataflowBuilder::new(&*self.catalog, compute).with_config(&self.config)
        };
        let mut df_desc = MirDataflowDescription::new(self.debug_name.clone());
        let mut df_meta = DataflowMetainfo::default();

        match plan {
            SubscribeFrom::Id(from_id) => {
                let from = self.catalog.get_entry(&from_id);
                let from_desc = from
                    .relation_desc()
                    .expect("subscribes can only be run on items with descs")
                    .into_owned();

                df_builder.import_into_dataflow(&from_id, &mut df_desc, &self.config.features)?;
                df_builder.maybe_reoptimize_imported_views(&mut df_desc, &self.config)?;

                // Make SinkDesc
                let sink_description = ComputeSinkDesc {
                    from: from_id,
                    from_desc,
                    connection: ComputeSinkConnection::Subscribe(SubscribeSinkConnection {
                        output: output.row_order().to_vec(),
                    }),
                    with_snapshot: self.with_snapshot,
                    up_to: self.up_to.map(Antichain::from_elem).unwrap_or_default(),
                    // No `FORCE NOT NULL` for subscribes
                    non_null_assertions: vec![],
                    // No `REFRESH` for subscribes
                    refresh_schedule: None,
                };
                df_desc.export_sink(self.sink_id, sink_description);
            }
            SubscribeFrom::Query { expr, desc } => {
                // TODO: Change the `expr` type to be `HirRelationExpr` and run
                // HIR ⇒ MIR lowering and decorrelation here. This would allow
                // us implement something like `EXPLAIN RAW PLAN FOR SUBSCRIBE.`
                //
                // let typ = expr.top_level_typ();
                // let expr = expr.lower(&self.config)?;

                // MIR ⇒ MIR optimization (local)
                let mut transform_ctx = TransformCtx::local(
                    &self.config.features,
                    &self.typecheck_ctx,
                    &mut df_meta,
                    Some(&mut self.metrics),
                    Some(self.view_id),
                );

                let expr = expr.lower(HirToMirConfig::from(&self.config), None)?;
                let expr = optimize_mir_local(expr, &mut transform_ctx)?;