Function decode_and_mfp 

fn decode_and_mfp<'scope, E, RT, CB, X>(
    cfg: PersistConfig,
    fetched: StreamVec<'scope, (Timestamp, Subtime), FetchedBlob<SourceData, (), Timestamp, StorageDiff>>,
    name: &str,
    until: Antichain<Timestamp>,
    map_filter_project: Option<&mut MfpPlan<X>>,
    record_time: fn((Timestamp, Subtime)) -> RT,
) -> (Stream<'scope, (Timestamp, Subtime), CB::Container>, StreamVec<'scope, (Timestamp, Subtime), (E, RT, Diff)>)
where
    E: ExchangeData + Ord + Clone + Debug + From<DataflowError> + From<EvalError>,
    RT: Ord + Clone + Debug + 'static,
    CB: ContainerBuilder + PushInto<(Row, RT, Diff)>,
    X: PlanExpr,

Decode fetched parts and apply map_filter_project, writing ok records into CB’s containers and err records into a separate output.

record_time picks what a record stores for its time. A reader that does not distinguish times within a millisecond passes |time| time.0, which drops the Subtime coordinate the enclosing scope refines with: that coordinate exists to pace flow control and stays on the capabilities, and keeping it in the records would force a re-encode to strip it later. A reader that builds a collection in the refined scope needs it, and passes |time| time.