fn optimize_dataflow_demand(dataflow: &mut DataflowDesc) -> Result<(), TransformError> {
    // Maps id -> union of known columns demanded from the source/view with the
    // corresponding id.
    let mut demand = HashMap::new();

    // Demand all columns of inputs to sinks.
    for (_id, sink) in dataflow.sink_exports.iter() {
        let input_id = sink.from;
        demand
            .entry(Id::Global(input_id))
            .or_insert_with(BTreeSet::new)
            .extend(0..dataflow.arity_of(&input_id));
    }

    // Demand all columns of inputs to exported indexes.
    for (_id, desc, _typ) in dataflow.index_exports.iter() {
        let input_id = desc.on_id;
        demand
            .entry(Id::Global(input_id))
            .or_insert_with(BTreeSet::new)
            .extend(0..dataflow.arity_of(&input_id));
    }

    optimize_dataflow_demand_inner(
        dataflow
            .objects_to_build
            .iter_mut()
            .rev()
            .map(|build_desc| (Id::Global(build_desc.id), build_desc.view.as_inner_mut())),
        &mut demand,
    )?;

    // Push demand information into the SourceDesc.
    for (source_id, (source_desc, _)) in dataflow.source_imports.iter_mut() {
        if let Some(columns) = demand.get(&Id::Global(*source_id)).clone() {
            // Install no-op demand information if none exists.
            if source_desc.operators.is_none() {
                source_desc.operators = Some(LinearOperator {
                    predicates: Vec::new(),
                    projection: (0..source_desc.bare_desc.arity()).collect(),
                })
            }
            // Restrict required columns by those identified as demanded.
            if let Some(operator) = &mut source_desc.operators {
                operator.projection.retain(|col| columns.contains(col));
            }
        }
    }

    Ok(())
}
/// Pushes demand information from published outputs to dataflow inputs,
/// projecting away unnecessary columns.
///
/// Dataflows that exist for the sake of generating plan explanations do not
/// have published outputs. In this case, we push demand information from views
/// not depended on by other views to dataflow inputs.
Pushes demand information from published outputs to dataflow inputs,
projecting away unnecessary columns.

Dataflows that exist for the sake of generating plan explanations do not
have published outputs. In this case, we push demand information from views
not depended on by other views to dataflow inputs.
Pushes demand information from published outputs to dataflow inputs projecting away unnecessary columns . Dataflows that exist for the sake of generating plan explanations do not have published outputs . In this case we push demand information from views not depended on by other views to dataflow inputs .
fn optimize_dataflow_demand(dataflow: &mut DataflowDesc) -> Result<(), TransformError> {
    // Maps id -> union of known columns demanded from the source/view with the
    // corresponding id.
    let mut demand = HashMap::new();
    // Demand all columns of inputs to sinks.
    for (_id, sink) in dataflow.sink_exports.iter() {
        let input_id = sink.from;
        demand
            .entry(Id::Global(input_id))
            .or_insert_with(BTreeSet::new)
            .extend(0..dataflow.arity_of(&input_id));
    }
    // Demand all columns of inputs to exported indexes.
    for (_id, desc, _typ) in dataflow.index_exports.iter() {
        let input_id = desc.on_id;
        demand
            .entry(Id::Global(input_id))
            .or_insert_with(BTreeSet::new)
            .extend(0..dataflow.arity_of(&input_id));
    }
    optimize_dataflow_demand_inner(
        dataflow
            .objects_to_build
            .iter_mut()
            .rev()
            .map(|build_desc| (Id::Global(build_desc.id), build_desc.view.as_inner_mut())),
        &mut demand,
    )?;
    // Push demand information into the SourceDesc.
    for (source_id, (source_desc, _)) in dataflow.source_imports.iter_mut() {
        if let Some(columns) = demand.get(&Id::Global(*source_id)).clone() {
            // Install no-op demand information if none exists.
            if source_desc.operators.is_none() {
                source_desc.operators = Some(LinearOperator {
                    predicates: Vec::new(),
                    projection: (0..source_desc.bare_desc.arity()).collect(),
                })
            }
            // Restrict required columns by those identified as demanded.
            if let Some(operator) = &mut source_desc.operators {
                operator.projection.retain(|col| columns.contains(col));
            }
        }
    }
    Ok(())
}
fn optimize_dataflow_demand ( dataflow : & mut DataflowDesc ) -> Result < ( ) , TransformError > { // Maps id -> union of known columns demanded from the source/view with the // corresponding id. let mut demand = HashMap :: new ( ) ; // Demand all columns of inputs to sinks. for ( _id , sink ) in dataflow . sink_exports . iter ( ) { let input_id = sink . from ; demand . entry ( Id :: Global ( input_id ) ) . or_insert_with ( BTreeSet :: new ) . extend ( 0 .. dataflow . arity_of ( & input_id ) ) ; } // Demand all columns of inputs to exported indexes. for ( _id , desc , _typ ) in dataflow . index_exports . iter ( ) { let input_id = desc . on_id ; demand . entry ( Id :: Global ( input_id ) ) . or_insert_with ( BTreeSet :: new ) . extend ( 0 .. dataflow . arity_of ( & input_id ) ) ; } optimize_dataflow_demand_inner ( dataflow . objects_to_build . iter_mut ( ) . rev ( ) . map ( | build_desc | ( Id :: Global ( build_desc . id ) , build_desc . view . as_inner_mut ( ) ) ) , & mut demand , ) ? ; // Push demand information into the SourceDesc. for ( source_id , ( source_desc , _ ) ) in dataflow . source_imports . iter_mut ( ) { if let Some ( columns ) = demand . get ( & Id :: Global ( * source_id ) ) . clone ( ) { // Install no-op demand information if none exists. if source_desc . operators . is_none ( ) { source_desc . operators = Some ( LinearOperator { predicates : Vec :: new ( ) , projection : ( 0 .. source_desc . bare_desc . arity ( ) ) . collect ( ) , } ) } // Restrict required columns by those identified as demanded. if let Some ( operator ) = & mut source_desc . operators { operator . projection . retain ( | col | columns . contains ( col ) ) ; } } } Ok ( ( ) ) }
Pushes demand information from published outputs to dataflow inputs,
projecting away unnecessary columns.
Pushes demand information from published outputs to dataflow inputs projecting away unnecessary columns .
dataflow
&mut DataflowDesc
dataflow
&mut DataflowDesc