pub(crate) fn export_sink(
        &mut self,
        render_state: &mut RenderState,
        tokens: &mut RelevantTokens,
        import_ids: HashSet<GlobalId>,
        sink_id: GlobalId,
        sink: &SinkDesc,
        metrics: &SinkBaseMetrics,
    ) {
        let sink_render = get_sink_render_for(&sink.connector);

        // put together tokens that belong to the export
        let mut needed_source_tokens = Vec::new();
        let mut needed_additional_tokens = Vec::new();
        let mut needed_sink_tokens = Vec::new();
        for import_id in import_ids {
            if let Some(addls) = tokens.additional_tokens.get(&import_id) {
                needed_additional_tokens.extend_from_slice(addls);
            }
            if let Some(source_token) = tokens.source_tokens.get(&import_id) {
                needed_source_tokens.push(source_token.clone());
            }
        }

        let (collection, _err_collection) = self
            .lookup_id(expr::Id::Global(sink.from))
            .expect("Sink source collection not loaded")
            .as_collection();

        let collection = apply_sink_envelope(sink, &sink_render, collection);

        // TODO(benesch): errors should stream out through the sink,
        // if we figure out a protocol for that.

        let sink_token =
            sink_render.render_continuous_sink(render_state, sink, sink_id, collection, metrics);

        if let Some(sink_token) = sink_token {
            needed_sink_tokens.push(sink_token);
        }

        let tokens = Rc::new((
            needed_sink_tokens,
            needed_source_tokens,
            needed_additional_tokens,
        ));
        render_state
            .dataflow_tokens
            .insert(sink_id, Box::new(tokens));
    }
/// Export the sink described by `sink` from the rendering context.
Export the sink described by `sink` from the rendering context.
Export the sink described by ` sink ` from the rendering context .
pub(crate) fn export_sink(
        &mut self,
        render_state: &mut RenderState,
        tokens: &mut RelevantTokens,
        import_ids: HashSet<GlobalId>,
        sink_id: GlobalId,
        sink: &SinkDesc,
        metrics: &SinkBaseMetrics,
    ) {
        let sink_render = get_sink_render_for(&sink.connector);
        // put together tokens that belong to the export
        let mut needed_source_tokens = Vec::new();
        let mut needed_additional_tokens = Vec::new();
        let mut needed_sink_tokens = Vec::new();
        for import_id in import_ids {
            if let Some(addls) = tokens.additional_tokens.get(&import_id) {
                needed_additional_tokens.extend_from_slice(addls);
            }
            if let Some(source_token) = tokens.source_tokens.get(&import_id) {
                needed_source_tokens.push(source_token.clone());
            }
        }
        let (collection, _err_collection) = self
            .lookup_id(expr::Id::Global(sink.from))
            .expect("Sink source collection not loaded")
            .as_collection();
        let collection = apply_sink_envelope(sink, &sink_render, collection);
        // TODO(benesch): errors should stream out through the sink,
        // if we figure out a protocol for that.
        let sink_token =
            sink_render.render_continuous_sink(render_state, sink, sink_id, collection, metrics);
        if let Some(sink_token) = sink_token {
            needed_sink_tokens.push(sink_token);
        }
        let tokens = Rc::new((
            needed_sink_tokens,
            needed_source_tokens,
            needed_additional_tokens,
        ));
        render_state
            .dataflow_tokens
            .insert(sink_id, Box::new(tokens));
    }
pub ( crate ) fn export_sink ( & mut self , render_state : & mut RenderState , tokens : & mut RelevantTokens , import_ids : HashSet < GlobalId > , sink_id : GlobalId , sink : & SinkDesc , metrics : & SinkBaseMetrics , ) { let sink_render = get_sink_render_for ( & sink . connector ) ; // put together tokens that belong to the export let mut needed_source_tokens = Vec :: new ( ) ; let mut needed_additional_tokens = Vec :: new ( ) ; let mut needed_sink_tokens = Vec :: new ( ) ; for import_id in import_ids { if let Some ( addls ) = tokens . additional_tokens . get ( & import_id ) { needed_additional_tokens . extend_from_slice ( addls ) ; } if let Some ( source_token ) = tokens . source_tokens . get ( & import_id ) { needed_source_tokens . push ( source_token . clone ( ) ) ; } } let ( collection , _err_collection ) = self . lookup_id ( expr :: Id :: Global ( sink . from ) ) . expect ( " " ) . as_collection ( ) ; let collection = apply_sink_envelope ( sink , & sink_render , collection ) ; // TODO(benesch): errors should stream out through the sink, // if we figure out a protocol for that. let sink_token = sink_render . render_continuous_sink ( render_state , sink , sink_id , collection , metrics ) ; if let Some ( sink_token ) = sink_token { needed_sink_tokens . push ( sink_token ) ; } let tokens = Rc :: new ( ( needed_sink_tokens , needed_source_tokens , needed_additional_tokens , ) ) ; render_state . dataflow_tokens . insert ( sink_id , Box :: new ( tokens ) ) ; }
Export the sink described by `sink` from the rendering context.
Export the sink described by ` sink ` from the rendering context .
&mut self
render_state
&mut RenderState
tokens
&mut RelevantTokens
import_ids
HashSet<GlobalId>
sink_id
GlobalId
sink
&SinkDesc
metrics
&SinkBaseMetrics
&mut self
render_state
&mut RenderState
tokens
&mut RelevantTokens
import_ids
HashSet<GlobalId>
sink_id
GlobalId
sink
&SinkDesc
metrics
&SinkBaseMetrics