Function main

experiments/src/bin/labelprop.rs:17–105

()

use Value::Eid;

fn main() {
    timely::execute(Configuration::Thread, move |worker| {
        let mut server = Server::<Duration, u64>::new(Default::default());
        let (send_results, results) = channel();

        let (x, y, z) = (0, 1, 2);
        let rules = vec![
            Rule {
                name: "label".to_string(),
                plan: Plan::Union(Union {
                    variables: vec![x, y],
                    plans: vec![
                        Plan::MatchA(x, ":node".to_string(), y),
                        Plan::Join(Join {
                            variables: vec![z],
                            left_plan: Box::new(Plan::MatchA(z, ":edge".to_string(), y)),
                            right_plan: Box::new(Plan::NameExpr(vec![x, z], "label".to_string())),
                        }),
                    ],
                }),
            },
            Rule {
                name: "labelprop".to_string(),
                plan: Plan::NameExpr(vec![x, y], "label".to_string()),
                // plan: Plan::Aggregate(Aggregate {
                //     variables: vec![x, y],
                //     key_variables: vec![],
                //     plan: Box::new(Plan::NameExpr(vec![x, y], "label".to_string())),
                //     aggregation_fn: AggregationFn::COUNT
                // })
            },
        ];

        let edge_source = Source::CsvFile(CsvFile {
            has_headers: false,
            delimiter: b' ',
            path: "/Users/niko/data/labelprop/edges.httpd_df".to_string(),
            eid_offset: 0,
            timestamp_offset: None,
            flexible: false,
            comment: None,
            schema: vec![(":edge".to_string(), (1, Eid(0)))],
        });
        let node_source = Source::CsvFile(CsvFile {
            has_headers: false,
            delimiter: b' ',
            path: "/Users/niko/data/labelprop/nodes.httpd_df".to_string(),
            eid_offset: 0,
            timestamp_offset: None,
            flexible: false,
            comment: None,
            schema: vec![(":node".to_string(), (1, Eid(0)))],
        });

        worker.dataflow::<Duration, _, _>(|scope| {
            server.register_source(edge_source, scope).unwrap();
            server.register_source(node_source, scope).unwrap();