dfir_lang/graph/ops/
sort.rs

Help
1use quote::quote_spanned;
2
3use super::{
4    DelayType, OperatorCategory, OperatorConstraints, OperatorWriteOutput, RANGE_0, RANGE_1,
5    WriteContextArgs,
6};
7
8/// Takes a stream as input and produces a sorted version of the stream as output.
9///
10/// ```dfir
11/// source_iter(vec![2, 3, 1])
12///     -> sort()
13///     -> assert_eq([1, 2, 3]);
14/// ```
15///
16/// `sort` is blocking. Only the values collected within a single tick will be sorted and
17/// emitted.
18pub const SORT: OperatorConstraints = OperatorConstraints {
19    name: "sort",
20    categories: &[OperatorCategory::Persistence],
21    hard_range_inn: RANGE_1,
22    soft_range_inn: RANGE_1,
23    hard_range_out: RANGE_1,
24    soft_range_out: RANGE_1,
25    num_args: 0,
26    persistence_args: RANGE_0,
27    type_args: RANGE_0,
28    is_external_input: false,
29    has_singleton_output: false,
30    flo_type: None,
31    ports_inn: None,
32    ports_out: None,
33    input_delaytype_fn: |_| Some(DelayType::Stratum),
34    write_fn: |&WriteContextArgs {
35                   op_span,
36                   ident,
37                   inputs,
38                   is_pull,
39                   work_fn,
40                   ..
41               },
42               _| {
43        assert!(is_pull);
44
45        let input = &inputs[0];
46        let write_iterator = quote_spanned! {op_span=>
47            // TODO(mingwei): unnecessary extra handoff into_iter() then collect().
48            // Fix requires handoff specialization.
49            let #ident = {
50                #work_fn(|| {
51                    let mut v = #input.collect::<::std::vec::Vec<_>>();
52                    v.sort_unstable();
53                    v
54                }).into_iter()
55            };
56        };
57        Ok(OperatorWriteOutput {
58            write_iterator,
59            ..Default::default()
60        })
61    },
62};
dfir_lang/graph/ops/sort.rs

dfir_lang/graph/ops/
sort.rs
