pub fn build_dataflow<A: Allocate>(
    timely_worker: &mut TimelyWorker<A>,
    render_state: &mut RenderState,
    dataflow: DataflowDescription<plan::Plan>,
    now: NowFn,
    source_metrics: &SourceBaseMetrics,
    sink_metrics: &SinkBaseMetrics,
) {
    let worker_logging = timely_worker.log_register().get("timely");
    let name = format!("Dataflow: {}", &dataflow.debug_name);
    let materialized_logging = timely_worker.log_register().get("materialized");

    timely_worker.dataflow_core(&name, worker_logging, Box::new(()), |_, scope| {
        // The scope.clone() occurs to allow import in the region.
        // We build a region here to establish a pattern of a scope inside the dataflow,
        // so that other similar uses (e.g. with iterative scopes) do not require weird
        // alternate type signatures.
        scope.clone().region_named(&name, |region| {
            let mut context = Context::for_dataflow(&dataflow, scope.addr().into_element());
            let mut tokens = RelevantTokens::default();

            assert!(
                !dataflow
                    .source_imports
                    .iter()
                    .map(|(id, _src)| id)
                    .duplicates()
                    .next()
                    .is_some(),
                "computation of unique IDs assumes a source appears no more than once per dataflow"
            );

            // Import declared sources into the rendering context.
            for (src_id, (src, orig_id)) in &dataflow.source_imports {
                context.import_source(
                    render_state,
                    &mut tokens,
                    region,
                    materialized_logging.clone(),
                    src_id.clone(),
                    src.clone(),
                    orig_id.clone(),
                    now.clone(),
                    source_metrics,
                );
            }

            // Import declared indexes into the rendering context.
            for (idx_id, idx) in &dataflow.index_imports {
                context.import_index(render_state, &mut tokens, scope, region, *idx_id, &idx.0);
            }

            // We first determine indexes and sinks to export, then build the declared object, and
            // finally export indexes and sinks. The reason for this is that we want to avoid
            // cloning the dataflow plan for `build_object`, which can be expensive.

            // Determine indexes to export
            let indexes = dataflow
                .index_exports
                .iter()
                .cloned()
                .map(|(idx_id, idx, _typ)| (idx_id, dataflow.get_imports(&idx.on_id), idx))
                .collect::<Vec<_>>();

            // Determine sinks to export
            let sinks = dataflow
                .sink_exports
                .iter()
                .cloned()
                .map(|(sink_id, sink)| (sink_id, dataflow.get_imports(&sink.from), sink))
                .collect::<Vec<_>>();

            // Build declared objects.
            for object in dataflow.objects_to_build {
                context.build_object(region, object);
            }

            // Export declared indexes.
            for (idx_id, imports, idx) in indexes {
                context.export_index(render_state, &mut tokens, imports, idx_id, &idx);
            }

            // Export declared sinks.
            for (sink_id, imports, sink) in sinks {
                context.export_sink(
                    render_state,
                    &mut tokens,
                    imports,
                    sink_id,
                    &sink,
                    sink_metrics,
                );
            }
        });
    })
}
/// Build a dataflow from a description.
Build a dataflow from a description.
Build a dataflow from a description .
pub fn build_dataflow<A: Allocate>(
    timely_worker: &mut TimelyWorker<A>,
    render_state: &mut RenderState,
    dataflow: DataflowDescription<plan::Plan>,
    now: NowFn,
    source_metrics: &SourceBaseMetrics,
    sink_metrics: &SinkBaseMetrics,
) {
    let worker_logging = timely_worker.log_register().get("timely");
    let name = format!("Dataflow: {}", &dataflow.debug_name);
    let materialized_logging = timely_worker.log_register().get("materialized");
    timely_worker.dataflow_core(&name, worker_logging, Box::new(()), |_, scope| {
        // The scope.clone() occurs to allow import in the region.
        // We build a region here to establish a pattern of a scope inside the dataflow,
        // so that other similar uses (e.g. with iterative scopes) do not require weird
        // alternate type signatures.
        scope.clone().region_named(&name, |region| {
            let mut context = Context::for_dataflow(&dataflow, scope.addr().into_element());
            let mut tokens = RelevantTokens::default();
            assert!(
                !dataflow
                    .source_imports
                    .iter()
                    .map(|(id, _src)| id)
                    .duplicates()
                    .next()
                    .is_some(),
                "computation of unique IDs assumes a source appears no more than once per dataflow"
            );
            // Import declared sources into the rendering context.
            for (src_id, (src, orig_id)) in &dataflow.source_imports {
                context.import_source(
                    render_state,
                    &mut tokens,
                    region,
                    materialized_logging.clone(),
                    src_id.clone(),
                    src.clone(),
                    orig_id.clone(),
                    now.clone(),
                    source_metrics,
                );
            }
            // Import declared indexes into the rendering context.
            for (idx_id, idx) in &dataflow.index_imports {
                context.import_index(render_state, &mut tokens, scope, region, *idx_id, &idx.0);
            }
            // We first determine indexes and sinks to export, then build the declared object, and
            // finally export indexes and sinks. The reason for this is that we want to avoid
            // cloning the dataflow plan for `build_object`, which can be expensive.
            // Determine indexes to export
            let indexes = dataflow
                .index_exports
                .iter()
                .cloned()
                .map(|(idx_id, idx, _typ)| (idx_id, dataflow.get_imports(&idx.on_id), idx))
                .collect::<Vec<_>>();
            // Determine sinks to export
            let sinks = dataflow
                .sink_exports
                .iter()
                .cloned()
                .map(|(sink_id, sink)| (sink_id, dataflow.get_imports(&sink.from), sink))
                .collect::<Vec<_>>();
            // Build declared objects.
            for object in dataflow.objects_to_build {
                context.build_object(region, object);
            }
            // Export declared indexes.
            for (idx_id, imports, idx) in indexes {
                context.export_index(render_state, &mut tokens, imports, idx_id, &idx);
            }
            // Export declared sinks.
            for (sink_id, imports, sink) in sinks {
                context.export_sink(
                    render_state,
                    &mut tokens,
                    imports,
                    sink_id,
                    &sink,
                    sink_metrics,
                );
            }
        });
    })
}
pub fn build_dataflow < A : Allocate > ( timely_worker : & mut TimelyWorker < A > , render_state : & mut RenderState , dataflow : DataflowDescription < plan :: Plan > , now : NowFn , source_metrics : & SourceBaseMetrics , sink_metrics : & SinkBaseMetrics , ) { let worker_logging = timely_worker . log_register ( ) . get ( " " ) ; let name = format ! ( " " dataflow debug_name ) ; let materialized_logging = timely_worker . log_register ( ) . get ( " " ) ; timely_worker . dataflow_core ( & name , worker_logging , Box :: new ( ( ) ) , | _ , scope | { // The scope.clone() occurs to allow import in the region. // We build a region here to establish a pattern of a scope inside the dataflow, // so that other similar uses (e.g. with iterative scopes) do not require weird // alternate type signatures. scope . clone ( ) . region_named ( & name , | region | { let mut context = Context :: for_dataflow ( & dataflow , scope . addr ( ) . into_element ( ) ) ; let mut tokens = RelevantTokens :: default ( ) ; assert ! ( dataflow source_imports iter ( ) map ( ( id _src ) id ) duplicates ( ) next ( ) is_some ( ) " " ) ; // Import declared sources into the rendering context. for ( src_id , ( src , orig_id ) ) in & dataflow . source_imports { context . import_source ( render_state , & mut tokens , region , materialized_logging . clone ( ) , src_id . clone ( ) , src . clone ( ) , orig_id . clone ( ) , now . clone ( ) , source_metrics , ) ; } // Import declared indexes into the rendering context. for ( idx_id , idx ) in & dataflow . index_imports { context . import_index ( render_state , & mut tokens , scope , region , * idx_id , & idx . 0 ) ; } // We first determine indexes and sinks to export, then build the declared object, and // finally export indexes and sinks. The reason for this is that we want to avoid // cloning the dataflow plan for `build_object`, which can be expensive. // Determine indexes to export let indexes = dataflow . index_exports . iter ( ) . cloned ( ) . map ( | ( idx_id , idx , _typ ) | ( idx_id , dataflow . get_imports ( & idx . on_id ) , idx ) ) . collect :: < Vec < _ > > ( ) ; // Determine sinks to export let sinks = dataflow . sink_exports . iter ( ) . cloned ( ) . map ( | ( sink_id , sink ) | ( sink_id , dataflow . get_imports ( & sink . from ) , sink ) ) . collect :: < Vec < _ > > ( ) ; // Build declared objects. for object in dataflow . objects_to_build { context . build_object ( region , object ) ; } // Export declared indexes. for ( idx_id , imports , idx ) in indexes { context . export_index ( render_state , & mut tokens , imports , idx_id , & idx ) ; } // Export declared sinks. for ( sink_id , imports , sink ) in sinks { context . export_sink ( render_state , & mut tokens , imports , sink_id , & sink , sink_metrics , ) ; } } ) ; } ) }
Build a dataflow from a description.
Build a dataflow from a description .
timely_worker
&mut TimelyWorker<A>
render_state
&mut RenderState
dataflow
DataflowDescription<plan::Plan>
now
NowFn
source_metrics
&SourceBaseMetrics
sink_metrics
&SinkBaseMetrics
timely_worker
&mut TimelyWorker<A>
render_state
&mut RenderState
dataflow
DataflowDescription<plan::Plan>
now
NowFn
source_metrics
&SourceBaseMetrics
sink_metrics
&SinkBaseMetrics