| 963 | |
| 964 | impl<P: PartitionInfo, C: ClusterInfo> JobAssembly<Record> for IRJobAssembly<P, C> { |
| 965 | fn assemble(&self, plan: &JobDesc, worker: &mut Worker<Record, Vec<u8>>) -> Result<(), BuildJobError> { |
| 966 | worker.dataflow(move |input, output| { |
| 967 | let physical_plan = decode::<pb::PhysicalPlan>(&plan.plan)?; |
| 968 | if log_enabled!(log::Level::Debug) && pegasus::get_current_worker().index == 0 { |
| 969 | debug!("{:#?}", PhysicalPlanPrinter(&physical_plan)); |
| 970 | } |
| 971 | // input from a dummy record to trigger the computation |
| 972 | let source = input.input_from(vec![Record::default()])?; |
| 973 | let plan_len = physical_plan.plan.len(); |
| 974 | let stream = self.install(source, &physical_plan.plan[0..plan_len - 1])?; |
| 975 | let sink_opr = physical_plan.plan.last().ok_or_else(|| { |
| 976 | FnGenError::from(ParsePbError::EmptyFieldError("empty job plan".to_string())) |
| 977 | })?; |
| 978 | let ec = self.udf_gen.gen_sink(sink_opr.clone())?; |
| 979 | match ec { |
| 980 | Sinker::DefaultSinker(default_sinker) => stream |
| 981 | .map(move |record| default_sinker.exec(record))? |
| 982 | .sink_into(output), |
| 983 | #[cfg(feature = "with_v6d")] |
| 984 | Sinker::GraphSinker(graph_sinker) => { |
| 985 | return stream |
| 986 | .fold_partition(graph_sinker, || { |
| 987 | |mut accumulator, next| { |
| 988 | accumulator.accum(next)?; |
| 989 | Ok(accumulator) |
| 990 | } |
| 991 | })? |
| 992 | .map(|mut accumulator| Ok(accumulator.finalize()?))? |
| 993 | .into_stream()? |
| 994 | .map(|_r| Ok(vec![]))? |
| 995 | .sink_into(output) |
| 996 | } |
| 997 | } |
| 998 | }) |
| 999 | } |
| 1000 | } |
| 1001 | |
| 1002 | #[inline] |