Method install

interactive_engine/executor/ir/runtime/src/assembly.rs:197–961

(
        &self, mut stream: Stream<Record>, plan: &[pb::PhysicalOpr],
    )

    }

    fn install(
        &self, mut stream: Stream<Record>, plan: &[pb::PhysicalOpr],
    ) -> Result<Stream<Record>, BuildJobError> {
        let mut prev_op_kind = pb::physical_opr::operator::OpKind::Root(pb::Root {});
        for op in &plan[..] {
            let op_kind = to_op_kind(op)?;
            match op_kind {
                OpKind::Repartition(repartition) => {
                    let repartition_strategy = repartition.strategy.as_ref().ok_or_else(|| {
                        FnGenError::from(ParsePbError::EmptyFieldError(
                            "Empty repartition strategy".to_string(),
                        ))
                    })?;
                    match repartition_strategy {
                        pb::repartition::Strategy::ToAnother(shuffle) => {
                            let router = self.udf_gen.gen_shuffle(shuffle)?;
                            stream = stream.repartition(move |t| router.route(t));
                        }
                        pb::repartition::Strategy::ToOthers(_) => stream = stream.broadcast(),
                    }
                }
                OpKind::Project(project) => {
                    let func = self.udf_gen.gen_project(project)?;
                    stream = stream.filter_map_with_name("Project", move |input| func.exec(input))?;
                }
                OpKind::Select(select) => {
                    let func = self.udf_gen.gen_filter(select)?;
                    stream = stream.filter(move |input| func.test(input))?;
                }
                OpKind::Unfold(unfold) => {
                    let func = self.udf_gen.gen_unfold(unfold)?;
                    stream = stream.flat_map_with_name("Unfold", move |input| func.exec(input))?;
                }
                OpKind::Limit(limit) => {
                    let range = limit.range.ok_or_else(|| {
                        FnGenError::from(ParsePbError::EmptyFieldError("pb::Limit::range".to_string()))
                    })?;
                    // e.g., `limit(10)` would be translate as `Range{lower=0, upper=10}`
                    if range.upper <= range.lower || range.lower != 0 {
                        Err(FnGenError::from(ParsePbError::ParseError(format!(
                            "range {:?} in Limit Operator",
                            range
                        ))))?;
                    }
                    stream = stream.limit(range.upper as u32)?;
                }
                OpKind::OrderBy(order) => {
                    let cmp = self.udf_gen.gen_cmp(order.clone())?;
                    if let Some(range) = order.limit {
                        if range.upper <= range.lower || range.lower != 0 {
                            Err(FnGenError::from(ParsePbError::ParseError(format!(
                                "range {:?} in Order Operator",
                                range
                            ))))?;
                        }
                        stream = stream.sort_limit_by(range.upper as u32, move |a, b| cmp.compare(a, b))?;
                    } else {
                        stream = stream.sort_by(move |a, b| cmp.compare(a, b))?;